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