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.smartmeter.internal;
  14
  15 import java.time.Duration;
  16 import java.util.ArrayList;
  17 import java.util.Collection;
  18 import java.util.HashMap;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Map.Entry;
  22 import java.util.concurrent.CopyOnWriteArrayList;
  23 import java.util.concurrent.ScheduledExecutorService;
  24 import java.util.concurrent.TimeUnit;
  25 import java.util.concurrent.TimeoutException;
  26 import java.util.function.Supplier;
  27
  28 import javax.measure.Quantity;
  29
  30 import org.eclipse.jdt.annotation.NonNullByDefault;
  31 import org.eclipse.jdt.annotation.Nullable;
  32 import org.openhab.binding.smartmeter.connectors.IMeterReaderConnector;
  33 import org.openhab.binding.smartmeter.internal.helper.ProtocolMode;
  34 import org.openhab.core.io.transport.serial.SerialPortManager;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 import io.reactivex.Flowable;
  39 import io.reactivex.disposables.Disposable;
  40 import io.reactivex.plugins.RxJavaPlugins;
  41 import io.reactivex.schedulers.Schedulers;
  42
  43 /**
  44  * This represents a meter device.
  45  * All read values of the device are cached here and can be obtained. The reading can be started with
  46  * {@link #readValues(ScheduledExecutorService, Duration)}
  47  *
  48  * @author Matthias Steigenberger - Initial contribution
  49  *
  50  * @param <T> The type of Payload which is read from the device.
  51  */
  52 @NonNullByDefault
  53 public abstract class MeterDevice<T> {
  54
  55     private static final int RETRY_DELAY = 2;
  56     private final Logger logger = LoggerFactory.getLogger(MeterDevice.class);
  57     /**
  58      * Controls wether the device info is logged to the OSGi console.
  59      */
  60     private boolean printMeterInfo;
  61     /**
  62      * Map of all values captured from the device during the read request.
  63      */
  64     private Map<String, MeterValue<?>> valueCache;
  65     private byte @Nullable [] initMessage;
  66     /**
  67      * The id of the SML device from openHAB configuration.
  68      */
  69     private String deviceId;
  70     /**
  71      * Used to establish the device connection
  72      */
  73     IMeterReaderConnector<T> connector;
  74     private List<MeterValueListener> valueChangeListeners;
  75
  76     public MeterDevice(Supplier<SerialPortManager> serialPortManagerSupplier, String deviceId, String serialPort,
  77             byte @Nullable [] initMessage, int baudrate, int baudrateChangeDelay, ProtocolMode protocolMode) {
  78         super();
  79         this.deviceId = deviceId;
  80         this.valueCache = new HashMap<>();
  81         this.valueChangeListeners = new CopyOnWriteArrayList<>();
  82         this.printMeterInfo = true;
  83         this.connector = createConnector(serialPortManagerSupplier, serialPort, baudrate, baudrateChangeDelay,
  84                 protocolMode);
  85         RxJavaPlugins.setErrorHandler(error -> {
  86             logger.error("Fatal error occured", error);
  87         });
  88     }
  89
  90     /**
  91      * Creates the actual connector that handles the serial port communication and protocol.
  92      *
  93      * @param serialPortManagerSupplier Supplies the {@link SerialPortManager} which is used to obtain the serial port
  94      *            implementation
  95      * @param serialPort The name of the port to communicate with.
  96      * @param baudrate The Baudrate to set for communication.
  97      * @param baudrateChangeDelay The delay which is used before changing the baudrate (used only for specific
  98      *            protocols).
  99      * @param protocolMode The {@link ProtocolMode} to use.
 100      * @return The connector which handles the serial port communication.
 101      */
 102     protected abstract IMeterReaderConnector<T> createConnector(Supplier<SerialPortManager> serialPortManagerSupplier,
 103             String serialPort, int baudrate, int baudrateChangeDelay, ProtocolMode protocolMode);
 104
 105     /**
 106      * Gets the configured deviceId.
 107      *
 108      * @return the id of the SmlDevice from openHAB configuration.
 109      */
 110     public String getDeviceId() {
 111         return deviceId;
 112     }
 113
 114     /**
 115      * Returns the specified OBIS value if available.
 116      *
 117      * @param obis the OBIS code which value should be retrieved.
 118      * @return the OBIS value as String if available - otherwise null.
 119      */
 120     @Nullable
 121     public String getValue(String obisId) {
 122         MeterValue<?> smlValue = getMeterValue(obisId);
 123         if (smlValue != null) {
 124             return smlValue.getValue();
 125         }
 126         return null;
 127     }
 128
 129     /**
 130      * Returns the specified OBIS value if available.
 131      *
 132      * @param obis the OBIS code which value should be retrieved.
 133      * @return the OBIS value if available - otherwise null.
 134      */
 135     @SuppressWarnings("unchecked")
 136     @Nullable
 137     public <Q extends Quantity<Q>> MeterValue<Q> getMeterValue(String obisId) {
 138         if (valueCache.containsKey(obisId)) {
 139             return (MeterValue<Q>) valueCache.get(obisId);
 140         }
 141         return null;
 142     }
 143
 144     /**
 145      * Gets all currently available OBIS codes.
 146      *
 147      * @return All cached OBIS codes.
 148      */
 149     public Collection<String> getObisCodes() {
 150         return new ArrayList<>(this.valueCache.keySet());
 151     }
 152
 153     /**
 154      * Read values from this device a store them locally against their OBIS code.
 155      *
 156      * If there is an error in reading, it will be retried {@value #NUMBER_OF_RETRIES} times. The retry will be delayed
 157      * by {@code period} seconds.
 158      * If its still failing, the connection will be closed and opened again.
 159      *
 160      * @return The {@link Disposable} which needs to be disposed whenever not used anymore.
 161      *
 162      */
 163     public Disposable readValues(long timeout, ScheduledExecutorService executorService, Duration period) {
 164         return Flowable.fromPublisher(connector.getMeterValues(initMessage, period, executorService))
 165                 .timeout(timeout + period.toMillis(), TimeUnit.MILLISECONDS, Schedulers.from(executorService))
 166                 .doOnSubscribe(sub -> {
 167                     logger.debug("Opening connection to {}", getDeviceId());
 168                     connector.openConnection();
 169                 }).doOnError(ex -> {
 170                     if (ex instanceof TimeoutException) {
 171                         logger.debug("Timeout occured for {}; {}", getDeviceId(), ex.getMessage());
 172                     } else {
 173                         logger.debug("Failed to read: {}. Closing connection and trying again in {} seconds...; {}",
 174                                 ex.getMessage(), RETRY_DELAY, getDeviceId(), ex);
 175                     }
 176                     connector.closeConnection();
 177                     notifyReadingError(ex);
 178                 }).doOnCancel(connector::closeConnection).doOnComplete(connector::closeConnection).share()
 179                 .retryWhen(
 180                         publisher -> publisher.delay(RETRY_DELAY, TimeUnit.SECONDS, Schedulers.from(executorService)))
 181                 .subscribeOn(Schedulers.from(executorService), true).subscribe((value) -> {
 182                     Map<String, MeterValue<?>> obisCodes = new HashMap<>(valueCache);
 183                     clearValueCache();
 184                     populateValueCache(value);
 185                     printInfo();
 186                     Collection<String> newObisCodes = getObisCodes();
 187                     // notify every removed obis code.
 188                     obisCodes.values().stream().filter((val) -> !newObisCodes.contains(val.getObisCode()))
 189                             .forEach((val) -> notifyValuesRemoved(val));
 190                 });
 191     }
 192
 193     /**
 194      * Deletes all cached values.
 195      *
 196      * The method will always be called before new values are populated.
 197      */
 198     protected void clearValueCache() {
 199         valueCache.clear();
 200     }
 201
 202     /**
 203      * Called whenever a new value was made available. The value cache needs to be filled here with
 204      * {@link #addObisCache(MeterValue)}.
 205      *
 206      * @param payload The actual payload value.
 207      */
 208     protected abstract <Q extends Quantity<Q>> void populateValueCache(T payload);
 209
 210     /**
 211      * Adds a {@link MeterValue} to the current cache.
 212      *
 213      * @param value The value to add.
 214      */
 215     protected <Q extends Quantity<Q>> void addObisCache(MeterValue<Q> value) {
 216         logger.debug("Value changed: {}", value);
 217         this.valueCache.put(value.getObisCode(), value);
 218         this.valueChangeListeners.forEach((listener) -> {
 219             try {
 220                 listener.valueChanged(value);
 221             } catch (Exception e) {
 222                 logger.error("Meter listener failed", e);
 223             }
 224         });
 225     }
 226
 227     @Override
 228     public String toString() {
 229         StringBuilder stringBuilder = new StringBuilder();
 230         stringBuilder.append("Device: ");
 231         stringBuilder.append(getDeviceId());
 232         stringBuilder.append(System.lineSeparator());
 233
 234         for (Entry<String, MeterValue<?>> entry : valueCache.entrySet()) {
 235             stringBuilder.append("Obis: " + entry.getKey() + " " + entry.getValue().toString());
 236             stringBuilder.append(System.lineSeparator());
 237         }
 238         return stringBuilder.toString();
 239     }
 240
 241     /**
 242      * Adds a {@link MeterValueListener} to the list of listeners which gets notified on new values being read.
 243      *
 244      * @param valueChangeListener The new {@link MeterValueListener}
 245      */
 246     public void addValueChangeListener(MeterValueListener valueChangeListener) {
 247         this.valueChangeListeners.add(valueChangeListener);
 248     }
 249
 250     /**
 251      * Removes a {@link MeterValueListener} from the list of listeners.
 252      *
 253      * @param valueChangeListener The listener to remove.
 254      */
 255     public void removeValueChangeListener(MeterValueListener valueChangeListener) {
 256         this.valueChangeListeners.remove(valueChangeListener);
 257     }
 258
 259     private <Q extends Quantity<Q>> void notifyValuesRemoved(MeterValue<Q> value) {
 260         this.valueChangeListeners.forEach((listener) -> listener.valueRemoved(value));
 261     }
 262
 263     private void notifyReadingError(Throwable e) {
 264         this.valueChangeListeners.forEach((listener) -> listener.errorOccurred(e));
 265     }
 266
 267     /**
 268      * Logs the object information with all given SML values to OSGi console.
 269      *
 270      * It's only called once - except the config was updated.
 271      */
 272     protected void printInfo() {
 273         if (this.getPrintMeterInfo()) {
 274             logger.info("Read out following values: {}", toString());
 275             setPrintMeterInfo(false);
 276         }
 277     }
 278
 279     /**
 280      * Gets if the object information has to be logged to OSGi console.
 281      *
 282      * @return true if the object information should be logged, otherwise false.
 283      */
 284     private Boolean getPrintMeterInfo() {
 285         return this.printMeterInfo;
 286     }
 287
 288     /**
 289      * Sets if the object information has to be logged to OSGi console.
 290      */
 291     private void setPrintMeterInfo(Boolean printMeterInfo) {
 292         this.printMeterInfo = printMeterInfo;
 293     }
 294 }