git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2023 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.binding.nikobus.internal.handler;
  14
  15 import static org.openhab.binding.nikobus.internal.NikobusBindingConstants.CONFIG_REFRESH_INTERVAL;
  16
  17 import java.io.IOException;
  18 import java.io.OutputStream;
  19 import java.util.Collection;
  20 import java.util.Collections;
  21 import java.util.HashMap;
  22 import java.util.LinkedList;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.concurrent.ScheduledFuture;
  26 import java.util.concurrent.TimeUnit;
  27 import java.util.concurrent.TimeoutException;
  28 import java.util.function.Consumer;
  29 import java.util.stream.Collectors;
  30 import java.util.stream.Stream;
  31
  32 import org.eclipse.jdt.annotation.NonNullByDefault;
  33 import org.eclipse.jdt.annotation.Nullable;
  34 import org.openhab.binding.nikobus.internal.NikobusBindingConstants;
  35 import org.openhab.binding.nikobus.internal.discovery.NikobusDiscoveryService;
  36 import org.openhab.binding.nikobus.internal.protocol.NikobusCommand;
  37 import org.openhab.binding.nikobus.internal.protocol.NikobusConnection;
  38 import org.openhab.binding.nikobus.internal.utils.Utils;
  39 import org.openhab.core.io.transport.serial.SerialPortManager;
  40 import org.openhab.core.thing.Bridge;
  41 import org.openhab.core.thing.ChannelUID;
  42 import org.openhab.core.thing.Thing;
  43 import org.openhab.core.thing.ThingStatus;
  44 import org.openhab.core.thing.ThingStatusDetail;
  45 import org.openhab.core.thing.binding.BaseBridgeHandler;
  46 import org.openhab.core.thing.binding.ThingHandler;
  47 import org.openhab.core.thing.binding.ThingHandlerService;
  48 import org.openhab.core.types.Command;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51
  52 /**
  53  * The {@link NikobusPcLinkHandler} is responsible for handling commands, which
  54  * are sent or received from the PC-Link Nikobus component.
  55  *
  56  * @author Boris Krivonog - Initial contribution
  57  */
  58 @NonNullByDefault
  59 public class NikobusPcLinkHandler extends BaseBridgeHandler {
  60     private final Logger logger = LoggerFactory.getLogger(NikobusPcLinkHandler.class);
  61     private final Map<String, Runnable> commandListeners = Collections.synchronizedMap(new HashMap<>());
  62     private final LinkedList<NikobusCommand> pendingCommands = new LinkedList<>();
  63     private final StringBuilder stringBuilder = new StringBuilder();
  64     private final SerialPortManager serialPortManager;
  65     private @Nullable NikobusConnection connection;
  66     private @Nullable NikobusCommand currentCommand;
  67     private @Nullable ScheduledFuture<?> scheduledRefreshFuture;
  68     private @Nullable ScheduledFuture<?> scheduledSendCommandWatchdogFuture;
  69     private @Nullable String ack;
  70     private @Nullable Consumer<String> unhandledCommandsProcessor;
  71     private int refreshThingIndex = 0;
  72
  73     public NikobusPcLinkHandler(Bridge bridge, SerialPortManager serialPortManager) {
  74         super(bridge);
  75         this.serialPortManager = serialPortManager;
  76     }
  77
  78     @Override
  79     public void initialize() {
  80         ack = null;
  81         stringBuilder.setLength(0);
  82
  83         updateStatus(ThingStatus.UNKNOWN);
  84
  85         String portName = (String) getConfig().get(NikobusBindingConstants.CONFIG_PORT_NAME);
  86         if (portName == null) {
  87             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.CONFIGURATION_ERROR, "Port must be set!");
  88             return;
  89         }
  90
  91         connection = new NikobusConnection(serialPortManager, portName, this::processReceivedValue);
  92
  93         int refreshInterval = ((Number) getConfig().get(CONFIG_REFRESH_INTERVAL)).intValue();
  94         scheduledRefreshFuture = scheduler.scheduleWithFixedDelay(this::refresh, refreshInterval, refreshInterval,
  95                 TimeUnit.SECONDS);
  96     }
  97
  98     @Override
  99     public void dispose() {
 100         super.dispose();
 101
 102         Utils.cancel(scheduledSendCommandWatchdogFuture);
 103         scheduledSendCommandWatchdogFuture = null;
 104
 105         Utils.cancel(scheduledRefreshFuture);
 106         scheduledRefreshFuture = null;
 107
 108         NikobusConnection connection = this.connection;
 109         this.connection = null;
 110
 111         if (connection != null) {
 112             connection.close();
 113         }
 114     }
 115
 116     @Override
 117     public void handleCommand(ChannelUID channelUID, Command command) {
 118         // Noop.
 119     }
 120
 121     @Override
 122     public Collection<Class<? extends ThingHandlerService>> getServices() {
 123         return Collections.singleton(NikobusDiscoveryService.class);
 124     }
 125
 126     private void processReceivedValue(byte value) {
 127         logger.trace("Received {}", value);
 128
 129         if (value == 13) {
 130             String command = stringBuilder.toString();
 131             stringBuilder.setLength(0);
 132
 133             logger.debug("Received command '{}', ack = '{}'", command, ack);
 134
 135             try {
 136                 if (command.startsWith("$")) {
 137                     String ack = this.ack;
 138                     this.ack = null;
 139
 140                     processResponse(command, ack);
 141                 } else {
 142                     Runnable listener = commandListeners.get(command);
 143                     if (listener != null) {
 144                         listener.run();
 145                     } else {
 146                         Consumer<String> processor = unhandledCommandsProcessor;
 147                         if (processor != null) {
 148                             processor.accept(command);
 149                         }
 150                     }
 151                 }
 152             } catch (RuntimeException e) {
 153                 logger.debug("Processing command '{}' failed due {}", command, e.getMessage(), e);
 154             }
 155         } else {
 156             stringBuilder.append((char) value);
 157
 158             // Take ACK part, i.e. "$0512"
 159             if (stringBuilder.length() == 5) {
 160                 String payload = stringBuilder.toString();
 161                 if (payload.startsWith("$05")) {
 162                     ack = payload;
 163                     logger.debug("Received ack '{}'", ack);
 164                     stringBuilder.setLength(0);
 165                 }
 166             } else if (stringBuilder.length() > 128) {
 167                 // Fuse, if for some reason we don't receive \r don't fill buffer.
 168                 stringBuilder.setLength(0);
 169                 logger.warn("Resetting read buffer, should not happen, am I connected to Nikobus?");
 170             }
 171         }
 172     }
 173
 174     public void addListener(String command, Runnable listener) {
 175         if (commandListeners.put(command, listener) != null) {
 176             logger.warn("Multiple registrations for '{}'", command);
 177         }
 178     }
 179
 180     public void removeListener(String command) {
 181         commandListeners.remove(command);
 182     }
 183
 184     public void setUnhandledCommandProcessor(Consumer<String> processor) {
 185         if (unhandledCommandsProcessor != null) {
 186             logger.debug("Unexpected override of unhandledCommandsProcessor");
 187         }
 188         unhandledCommandsProcessor = processor;
 189     }
 190
 191     public void resetUnhandledCommandProcessor() {
 192         unhandledCommandsProcessor = null;
 193     }
 194
 195     private void processResponse(String commandPayload, @Nullable String ack) {
 196         NikobusCommand command;
 197         synchronized (pendingCommands) {
 198             command = currentCommand;
 199         }
 200
 201         if (command == null) {
 202             logger.debug("Processing response but no command pending");
 203             return;
 204         }
 205
 206         NikobusCommand.ResponseHandler responseHandler = command.getResponseHandler();
 207         if (responseHandler == null) {
 208             logger.debug("No response expected for current command");
 209             return;
 210         }
 211
 212         if (ack == null) {
 213             logger.debug("No ack received");
 214             return;
 215         }
 216
 217         String requestCommandId = command.getPayload().substring(3, 5);
 218         String ackCommandId = ack.substring(3, 5);
 219         if (!ackCommandId.equals(requestCommandId)) {
 220             logger.debug("Unexpected command's ack '{}' != '{}'", requestCommandId, ackCommandId);
 221             return;
 222         }
 223
 224         // Check if response has expected length.
 225         if (commandPayload.length() != responseHandler.getResponseLength()) {
 226             logger.debug("Unexpected response length");
 227             return;
 228         }
 229
 230         if (!commandPayload.startsWith(responseHandler.getResponseCode())) {
 231             logger.debug("Unexpected response command code");
 232             return;
 233         }
 234
 235         String requestCommandAddress = command.getPayload().substring(5, 9);
 236         String ackCommandAddress = commandPayload.substring(responseHandler.getAddressStart(),
 237                 responseHandler.getAddressStart() + 4);
 238         if (!requestCommandAddress.equals(ackCommandAddress)) {
 239             logger.debug("Unexpected response address");
 240             return;
 241         }
 242
 243         if (responseHandler.complete(commandPayload)) {
 244             resetProcessingAndProcessNext();
 245         }
 246     }
 247
 248     public void sendCommand(NikobusCommand command) {
 249         synchronized (pendingCommands) {
 250             pendingCommands.addLast(command);
 251         }
 252
 253         scheduler.submit(this::processCommand);
 254     }
 255
 256     private void processCommand() {
 257         NikobusCommand command;
 258         synchronized (pendingCommands) {
 259             if (currentCommand != null) {
 260                 return;
 261             }
 262
 263             command = pendingCommands.pollFirst();
 264             if (command == null) {
 265                 return;
 266             }
 267
 268             currentCommand = command;
 269         }
 270         sendCommand(command, 3);
 271     }
 272
 273     private void sendCommand(NikobusCommand command, int retry) {
 274         logger.debug("Sending retry = {}, command '{}'", retry, command.getPayload());
 275
 276         NikobusConnection connection = this.connection;
 277         if (connection == null) {
 278             return;
 279         }
 280
 281         try {
 282             connectIfNeeded(connection);
 283
 284             OutputStream outputStream = connection.getOutputStream();
 285             if (outputStream == null) {
 286                 return;
 287             }
 288             outputStream.write(command.getPayload().getBytes());
 289             outputStream.flush();
 290         } catch (IOException e) {
 291             logger.debug("Sending command failed due {}", e.getMessage(), e);
 292             connection.close();
 293             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR, e.getMessage());
 294         } finally {
 295             NikobusCommand.ResponseHandler responseHandler = command.getResponseHandler();
 296             if (responseHandler == null) {
 297                 resetProcessingAndProcessNext();
 298             } else if (retry > 0) {
 299                 scheduleSendCommandTimeout(() -> {
 300                     if (!responseHandler.isCompleted()) {
 301                         sendCommand(command, retry - 1);
 302                     }
 303                 });
 304             } else {
 305                 scheduleSendCommandTimeout(() -> processTimeout(responseHandler));
 306             }
 307         }
 308     }
 309
 310     private void scheduleSendCommandTimeout(Runnable command) {
 311         scheduledSendCommandWatchdogFuture = scheduler.schedule(command, 2, TimeUnit.SECONDS);
 312     }
 313
 314     private void processTimeout(NikobusCommand.ResponseHandler responseHandler) {
 315         if (responseHandler.completeExceptionally(new TimeoutException("Waiting for response timed-out."))) {
 316             resetProcessingAndProcessNext();
 317         }
 318     }
 319
 320     private void resetProcessingAndProcessNext() {
 321         Utils.cancel(scheduledSendCommandWatchdogFuture);
 322         synchronized (pendingCommands) {
 323             currentCommand = null;
 324         }
 325         scheduler.submit(this::processCommand);
 326     }
 327
 328     private void refresh() {
 329         List<Thing> things = getThing().getThings().stream()
 330                 .filter(thing -> thing.getHandler() instanceof NikobusModuleHandler).collect(Collectors.toList());
 331
 332         // if there are command listeners (buttons) then we need an open connection even if no modules exist
 333         if (!commandListeners.isEmpty()) {
 334             NikobusConnection connection = this.connection;
 335             if (connection == null) {
 336                 return;
 337             }
 338             try {
 339                 connectIfNeeded(connection);
 340             } catch (IOException e) {
 341                 connection.close();
 342                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR, e.getMessage());
 343                 return;
 344             }
 345         }
 346
 347         if (things.isEmpty()) {
 348             logger.debug("Nothing to refresh");
 349             return;
 350         }
 351
 352         refreshThingIndex = (refreshThingIndex + 1) % things.size();
 353
 354         ThingHandler thingHandler = things.get(refreshThingIndex).getHandler();
 355         if (thingHandler == null) {
 356             return;
 357         }
 358
 359         NikobusModuleHandler handler = (NikobusModuleHandler) thingHandler;
 360         handler.refreshModule();
 361     }
 362
 363     private synchronized void connectIfNeeded(NikobusConnection connection) throws IOException {
 364         if (!connection.isConnected()) {
 365             connection.connect();
 366
 367             // Send connection sequence, mimicking the Nikobus software. If this is not
 368             // sent, PC-Link sometimes does not forward button presses via serial interface.
 369             Stream.of(new String[] { "++++", "ATH0", "ATZ", "$10110000B8CF9D", "#L0", "#E0", "#L0", "#E1" })
 370                     .map(NikobusCommand::new).forEach(this::sendCommand);
 371
 372             updateStatus(ThingStatus.ONLINE);
 373         }
 374     }
 375 }