registers changed

ENV

@@ -0,0 +1,2 @@
+export LD_LIBRARY_PATH=/home/pi/modbusmaster/pyserialext/
+export PYTHONPATH=/home/pi/modbusmaster/pyserialext/

pyserialext/writec.c

@@ -9,17 +9,17 @@ const uint8_t DE_PIN = 0;
 int init() {
   wiringPiSetup();
   pinMode(DE_PIN, OUTPUT);
-  digitalWrite(DE_PIN, HIGH);
+  digitalWrite(DE_PIN, LOW);
 }
 
 ssize_t writec(int fd, char *buf, size_t count) {
-  digitalWrite(DE_PIN, LOW);
+  digitalWrite(DE_PIN, HIGH);
   ssize_t r = write(fd, buf, count);
   uint8_t lsr;
   do {
     int r = ioctl(fd, TIOCSERGETLSR, &lsr);
   } while (!(lsr & TIOCSER_TEMT));
-  digitalWrite(DE_PIN, HIGH);
+  digitalWrite(DE_PIN, LOW);
 
   return r;
 }

readme.md

@@ -1,5 +1,8 @@
 # modbusmaster
 
+
+![The whole setup](./docs/2019-06-14_22-06-52-1.jpg)
+
 ## Disable Bluetooth on RPi3
 
 Add at the end of `/boot/config`:
@@ -11,6 +14,7 @@ Remove mentions of `serial0` from `/boot/cmdline`.
 
 ## Enable rs485 mode
 
+<del>
 Use the submodule rpirtscts to enable to alternate functions of the related
 pins at the RPi MCU. It is submoduled here, can be found directly at https://github.com/mholling/rpirtscts
 
@@ -23,6 +27,10 @@ This needs to be done at every boot.
 Kudos to danjperron, cmp. https://www.raspberrypi.org/forums/viewtopic.php?f=98&t=224533&hilit=rs+485#p1383709
 
 (Note, please: This software is under the GPL 3.0 license. However, I do not derive from this software, I use it in an unchanged way. It is not integrated into my sources, it just needs to be called once the RPi has booted.)
+</del>
+
+The approach to enable the transmitter has been changed due to timing issues with disabling the transmitter after the TX phase when talking to Modbus devices (see below), so this step is not longer required.
+
 
 ## Pinout
 
@@ -38,6 +46,10 @@ TX is at GPIO14, RX is at GPIO15 and RTS (control line for transmitter enable) i
 
 ## Python snippet to test
 
+### Pure RS485 test 
+
+#### Test 1: Transmit only
+
 (See also `./snippets/test1.py`.)
 
     import serial.rs485
@@ -52,4 +64,115 @@ Find an signal screenshot here:
 Channel 1 in yellow has the TX line (GPIO14), channel 3 in purple has the RTS (GPIO17, transmitter enable line).
 
 
+#### Test 2: Transmit and receive
+
+(See also `./snippets/test2.py`.)
+
+Here in contrast to the last script first an octet is received and then echoed.
+
+    import serial.rs485
+    ser=serial.rs485.RS485(port='/dev/ttyAMA0',baudrate=2400)
+    ser.rs485_mode = serial.rs485.RS485Settings(False,True)
+    ser.write('a test'.encode('utf-8'))
+
+    while True:
+        c = ser.read(1)
+        ser.write(c)
+        print(c, end='')
+
+The signal graph is here:
+
+![Test2 Signals](./docs/osci2.png)
+
+Channel 1 in yellow has the TX line (GPIO14), channel 2 in blue has the RX line and channel 3 in purple has the RTS (GPIO17, transmitter enable line).
+
+It is obvious that the DE (transmitter enable line, RTS controlled by `pyserial` in RS485 mode) is hold active a good while after all data already have been transmitted.
+
+
+#### Test 3: First Modbus communication
+
+(See also `./snippets/test3.py`.)
+
+    from pymodbus.client.sync import ModbusSerialClient
+    import serial.rs485
+
+    ser=serial.rs485.RS485(port='/dev/ttyAMA0',baudrate=1200)
+    ser.rs485_mode = serial.rs485.RS485Settings(rts_level_for_tx=False, 
+                                                rts_level_for_rx=True,
+                                                delay_before_tx=0.005,
+                                                delay_before_rx=-0.0)
+
+    client = ModbusSerialClient(method='rtu')
+    client.socket = ser
+    client.connect()
+    result = client.read_holding_registers(address=0x2000, count=2, unit=1)
+    print(result)
+    print(result.registers)
+    client.close()
+
+Here, the port is initialized in RS485 mode and a device is queried.
+
+Sometimes this works:
+
+![Modbus comm ok](./docs/modbus-ok.png)
+
+But sometimes is does not work:
+
+![Modbus comm not ok](./docs/modbus-not-ok.png)
+
+The long hold time of about 18ms of the DE (transmitter enable, RTS line) becomes a problem, the response of the device already starts when the transmitter of the master is still enabled and thus the receiver of the master is still disabled.
+
+A couple of experiments with deriving from the `RS485` class of `pyserial` and moving the time critical code (disabling the transmitter after the transmit) into C code failed. It wasn't faster at all. It became obvious that the system call `tcdrain`, which waits for all octets in the buffer to be transmitted returns very late.
+
+Finally, the solution was to get away from the RS485 mode in `pyserial` and instead use the line status register of the UART via a system call to see whether the transmit register is empty and switch the DE line of the transmitter not longer with the RTS functionality but directly using `wiringPi`.
+
+Derived class from `RS485` in `pyserial` (maybe this can be switched to the regular `Serial` class ...)
+
+    class RS485Ext(serial.rs485.RS485):
+        def __init__(self, *args, **kwargs):
+            super(RS485Ext, self).__init__(*args, **kwargs)
+            self.writec = ctypes.cdll.LoadLibrary('writec.so')
+            r = self.writec.init()
+
+        def write(self, b):
+            d = serial.serialutil.to_bytes(b)
+            r = self.writec.writec(self.fileno(), d, len(d))
+            return r
+
+
+C code of the function `writec` in the library `writec`:
+
+    ssize_t writec(int fd, char *buf, size_t count) {
+      digitalWrite(DE_PIN, HIGH);
+      ssize_t r = write(fd, buf, count);
+      uint8_t lsr;
+      do {
+        int r = ioctl(fd, TIOCSERGETLSR, &lsr);
+      } while (!(lsr & TIOCSER_TEMT));
+      digitalWrite(DE_PIN, LOW);
+
+      return r;
+    }
+
+This change brought the break through:
+
+![Delta between TX end and DE disable](./docs/rs485_should_be_ok_now.JPG)
+
+The delta between TX end and DE disable is now only 2.2ms (instead of nearly 20ms before).
+
+And finally Modbus communication works a lot more reliable:
+
+![Modbus comm nearly really ok](./docs/modbus-really-ok.png)
+
+However, still a lot of errors in the Modbus communication, certainly because of the dirty RX signal (blue). Furthermore it appears that the communication fails completely as soon as the termination resistor of 120 Ohms was placed.
+
+A hint from the application AN-960 from Analog Devices (https://www.analog.com/media/en/technical-documentation/application-notes/AN-960.pdf) helped here. In chapter "Fail-Safe Biasing" it is explained that, when no transmitter is active at all the lines are completely floating and thus the RX signal behind the transceiver becomes dirty, as in the image above.
+
+Pulling the A line to `Vcc` and the B line to `Gnd` using 1.5kOhm resistors, as proposed, solved this problem. And now also the termination resistor worked as expected.
+
+![Now it really works](./docs/nice-signals-with-fail-safe-resistors.png)
+
+
+
+
 

schema/conf-prov.sql

@@ -0,0 +1,212 @@
+-- Configuration and Provisioning Schema
+
+
+DROP TABLE tReadDatapoint;
+
+CREATE TABLE tReadDatapoint (
+    id INTEGER PRIMARY KEY AUTO_INCREMENT,
+    unit INTEGER NOT NULL,
+    address INTEGER NOT NULL,
+    count INTEGER NOT NULL,
+    converter VARCHAR(10) NOT NULL,
+    label VARCHAR(128) NOT NULL,
+    scanRate TIME(3) DEFAULT '00:00:01.000',
+    topic VARCHAR(256) NOT NULL,
+    lastContact TIMESTAMP(3) NOT NULL DEFAULT '2000-01-01 00:00:01.000',
+    lastError VARCHAR(512),
+    lastValue VARCHAR(512),
+    backoff TIME(3) DEFAULT '00:00:00.000',
+    available BOOLEAN DEFAULT TRUE,
+    retries INTEGER NOT NULL DEFAULT 0,
+    giveUpCount INTEGER NOT NULL DEFAULT 0,
+    active BOOLEAN NOT NULL DEFAULT TRUE,
+    CONSTRAINT uniqueReadDatapoint UNIQUE (unit, address, count, label)
+);
+
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(4, 0x2000, 2, 'F', '(ERR) Unavailable device', 'IoT/ModbusMaster1/UnavailableDevice', '00:00:01.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(1, 0x2000, 4, 'F', '(ERR) Wrong register size', 'IoT/ModbusMaster1/WrongRegisterSize', '00:05:00.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(1, 0x2000, 2, 'F', 'Voltage', 'IoT/ModbusMaster1/Voltage', '00:05:00.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(1, 0x2020, 2, 'F', 'Frequency', 'IoT/ModbusMaster1/Frequency', '00:05:00.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(1, 0x2060, 2, 'F', 'Current', 'IoT/ModbusMaster1/Current', '00:05:00.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(3, 0x0004, 2, 'RF', 'Resistance Channel 1', 'IoT/ModbusMaster1/Channel1/Resistance', '00:00:01.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(3, 0x000C, 2, 'RF', 'Temperature Channel 1', 'IoT/ModbusMaster1/Channel1/Temperature', '00:00:01.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(3, 0x0014, 2, 'RF', 'Resistance Channel 2', 'IoT/ModbusMaster1/Channel2/Resistance', '00:00:01.000');
+INSERT INTO tReadDatapoint (unit, address, count, converter, label, topic, scanRate)
+    VALUES(3, 0x001C, 2, 'RF', 'Temperature Channel 2', 'IoT/ModbusMaster1/Channel2/Temperature', '00:00:01.000');
+
+
+DROP TABLE tWriteDatapoint;
+
+CREATE TABLE tWriteDatapoint (
+    id INTEGER PRIMARY KEY AUTO_INCREMENT,
+    unit INTEGER NOT NULL,
+    address INTEGER NOT NULL,
+    count INTEGER NOT NULL,
+    converter VARCHAR(10) NOT NULL,
+    label VARCHAR(128) NOT NULL,
+    topic VARCHAR(256) NOT NULL,
+    lastContact TIMESTAMP(3) NOT NULL DEFAULT '2000-01-01 00:00:01.000',
+    lastError VARCHAR(512),
+    value VARCHAR(512),
+    retries INTEGER NOT NULL DEFAULT 0,
+    active BOOLEAN NOT NULL DEFAULT TRUE,
+    CONSTRAINT uniqueWriteDatapoint UNIQUE (unit, address, count, label)
+);
+
+
+
+INSERT INTO tWriteDatapoint (unit, address, count, converter, label, topic, active)
+    VALUES(5, 0x0000, 1, 'B', 'Relay 1', 'IoT/ModbusMaster1/Relay1', FALSE);
+
+
+CREATE OR REPLACE VIEW vReadDatapointsToBeHandled AS 
+    SELECT id, unit, address, count, converter
+        FROM tReadDatapoint 
+        WHERE available AND
+              active AND
+              ADDTIME(lastContact, ADDTIME(scanRate, backoff)) < NOW(3)
+        ORDER BY scanRate;
+
+CREATE OR REPLACE VIEW vWriteDatapintsToBeHandled AS
+    SELECT id, unit, address, count, converter, value
+        FROM tWriteDatapoint
+        WHERE active;
+
+
+
+DROP TABLE tReadNotification;
+
+CREATE TABLE tReadNotification (
+    id INTEGER PRIMARY KEY AUTO_INCREMENT,
+    readDatapointId INTEGER NOT NULL REFERENCES tReadDatapoint(id),
+    notificationType VARCHAR(1),
+    CONSTRAINT checkNotificationType CHECK (notificationtype IN ('V', 'F', 'R')) -- value, failure, return
+);
+
+
+DROP TABLE tWrittenNotification;
+
+CREATE TABLE tWrittenNotification (
+    id INTEGER PRIMARY KEY AUTO_INCREMENT,
+    writeDatapointId INTEGER NOT NULL REFERENCES tWriteDatapoint(id),
+    notificationType VARCHAR(1),
+    CONSTRAINT checkNotificationType CHECK (notificationtype IN ('S', 'F')) -- success, failure
+);
+
+
+
+
+DELIMITER $$
+CREATE OR REPLACE PROCEDURE prWriteFeedback (IN p_id INTEGER, IN p_lastError VARCHAR(512))
+    MODIFIES SQL DATA
+    BEGIN
+        DECLARE v_retries INTEGER;
+        DECLARE v_active BOOLEAN;
+
+        IF p_lastError = '' OR p_lastError IS NULL THEN
+            UPDATE tWriteDatapoint
+                SET lastError = NULL,
+                    lastContact = NOW(3),
+                    retries = 0,
+                    active = FALSE
+                WHERE id = p_id;
+            INSERT INTO tWrittenNotification (writeDatapointId, notificationType) VALUES (p_id, 'S');
+        ELSE
+            SELECT retries
+                INTO v_retries
+                FROM tWriteDatapoint
+                WHERE id = p_id;
+
+            SET v_retries := v_retries + 1;
+
+            IF v_retries >= 5 THEN
+                SET v_retries := 0;
+                SET v_active := FALSE;
+            ELSE
+                SET v_active := TRUE;
+            END IF;
+
+            UPDATE tWriteDatapoint
+                SET lastError = p_lastError,
+                    retries = v_retries,
+                    active = v_active
+                WHERE id = p_id;
+
+            IF NOT v_active THEN
+                INSERT INTO tWrittenNotification (writeDatapointId, notificationType) VALUES(p_id, 'F');
+            END IF;
+        END IF;
+    END; $$
+DELIMITER ;
+
+
+
+
+DELIMITER $$
+CREATE OR REPLACE PROCEDURE prReadFeedback (IN p_id INTEGER, IN p_lastValue VARCHAR(512), IN p_lastError VARCHAR(512))
+    MODIFIES SQL DATA
+    BEGIN
+        DECLARE v_retries INTEGER;
+        DECLARE v_backoff TIME(3);
+        DECLARE v_scanRate TIME(3);
+        DECLARE v_giveUpCount INTEGER;
+        DECLARE v_available BOOLEAN;
+
+        IF p_lastError = '' OR p_lastError IS NULL THEN
+            UPDATE tReadDatapoint
+                SET lastError = NULL,
+                    lastContact = NOW(3),
+                    lastValue = p_lastValue,
+                    retries = 0,
+                    backoff = '00:00:00.000',
+                    giveUpCount = 0
+                WHERE id = p_id;
+            INSERT INTO tReadNotification (readDatapointId, notificationType) VALUES(p_id, 'V');
+        ELSE
+            SELECT retries, backoff, scanRate, giveUpCount
+                INTO v_retries, v_backoff, v_scanRate, v_giveUpCount
+                FROM tReadDatapoint
+                WHERE id = p_id;
+
+            SET v_retries := v_retries + 1;
+
+            IF v_retries >= 5 THEN
+                IF v_backoff = '00:00:00.000' THEN
+                    SET v_backoff = v_scanRate;
+                ELSE
+                    SET v_backoff = ADDTIME(v_backoff, v_backoff);
+                END IF;
+                SET v_retries := 0;
+                SET v_giveUpCount := v_giveUpCount + 1;
+                SET v_available := TRUE;
+            END IF;
+            IF v_giveUpCount = 10 THEN
+                SET v_available := FALSE;
+                SET v_giveUpCount := 0;
+                SET v_backoff := '00:00:00.000';
+            END IF;
+
+            UPDATE tReadDatapoint
+                SET lastError = p_lastError,
+                    retries = v_retries,
+                    backoff = v_backoff,
+                    giveUpCount = v_giveUpCount,
+                    available = v_available
+                WHERE id = p_id;
+
+            IF NOT v_available THEN
+                INSERT INTO tReadNotification (readDatapointId, notificationType) VALUES(p_id, 'F');
+            END IF;
+        END IF;
+    END; $$
+DELIMITER ;
+
+

snippets/test5.py

@@ -1,12 +1,99 @@
 from pymodbus.client.sync import ModbusSerialClient
+from pymodbus.pdu import ExceptionResponse
 import RS485Ext
+import struct
+import time
 
-ser=RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=1200)
+def registersToIeeeFloat(i):
+  return struct.unpack('f', bytes(
+                                   [((x & 0xff00) >> 8) if y else (x & 0x00ff) 
+                                     for x in i[::-1] 
+                                     for y in [False, True]
+                                   ]
+                                 )
+                      )[0]
+
+def registersToIeeeFloatReverse(i):
+  return struct.unpack('f', bytes(
+                                   [((x & 0xff00) >> 8) if y else (x & 0x00ff) 
+                                     for x in i 
+                                     for y in [False, True]
+                                   ]
+                                 )
+                      )[0]
+
+class ModbusException(Exception):
+  def __init__(self, resp):
+    self.msg = str(result)
+
+  def __str__(self):
+    return self.msg
+
+
+
+ser=RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=1200, stopbits=1)
 
 client = ModbusSerialClient(method='rtu')
 client.socket = ser
 client.connect()
-result = client.read_holding_registers(address=0x2000, count=2, unit=1)
-print(result)
-print(result.registers)
-client.close()
+
+delay = 0.05
+
+try:
+  # BG-Tech, Voltage
+  result = client.read_holding_registers(address=0x2000, count=2, unit=1)
+  if type(result) == ExceptionResponse:
+    raise ModbusException(result)
+  print("Voltage: {:.2f}".format(registersToIeeeFloat(result.registers)))
+
+  time.sleep(delay)
+
+  # BG-Tech, Frequency
+  result = client.read_holding_registers(address=0x2020, count=2, unit=1)
+  if type(result) == ExceptionResponse:
+    raise ModbusException(result)
+  print("Frequency: {:.2f}".format(registersToIeeeFloat(result.registers)))
+
+  time.sleep(delay)
+
+  # BG-Tech, Current
+  result = client.read_holding_registers(address=0x2060, count=2, unit=1)
+  if type(result) == ExceptionResponse:
+    raise ModbusException(result)
+  print("Current: {:.2f}".format(registersToIeeeFloat(result.registers)))
+
+  time.sleep(delay)
+
+  # Hottis Thermometer, Resistance Channel 1
+  result = client.read_holding_registers(address=0x0004, count=2, unit=3)
+  if type(result) == ExceptionResponse:
+    raise ModbusException(result)
+  print("Resistance Channel 1: {:.2f}".format(registersToIeeeFloatReverse(result.registers)))
+
+  time.sleep(delay)
+
+  # Hottis Thermometer, Temperature Channel 1
+  result = client.read_holding_registers(address=0x000c, count=2, unit=3)
+  if type(result) == ExceptionResponse:
+    raise ModbusException(result)
+  print("Temperature Channel 1: {:.2f}".format(registersToIeeeFloatReverse(result.registers)))
+
+  time.sleep(delay)
+
+  # Hottis Thermometer, Resistance Channel 2
+  result = client.read_holding_registers(address=0x0014, count=2, unit=3)
+  if type(result) == ExceptionResponse:
+    raise ModbusException(result)
+  print("Resistance Channel 2: {:.2f}".format(registersToIeeeFloatReverse(result.registers)))
+
+  time.sleep(delay)
+
+  # Hottis Thermometer, Temperature Channel 2
+  result = client.read_holding_registers(address=0x001c, count=2, unit=3)
+  if type(result) == ExceptionResponse:
+    raise ModbusException(result)
+  print("Temperature Channel 2: {:.2f}".format(registersToIeeeFloatReverse(result.registers)))
+except ModbusException as e:
+  print("ERROR: %s" % e)
+finally:
+  client.close()

snippets/test6.py

@@ -0,0 +1,21 @@
+from pymodbus.client.sync import ModbusSerialClient
+import RS485Ext
+import time
+
+ser=RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=9600, stopbits=1)
+
+client = ModbusSerialClient(method='rtu')
+client.socket = ser
+client.connect()
+
+while True:
+    try:
+        result = client.read_holding_registers(address=0x9c42, count=1, unit=1)
+        print(result)
+        print(result.registers)
+    except Exception as e:
+        print("ERROR: %s" % str(e))
+    time.sleep(2)
+
+client.close()
+

snippets/test6b.py

@@ -0,0 +1,23 @@
+from pymodbus.client.sync import ModbusSerialClient
+import RS485Ext
+import time
+
+ser=RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=9600, stopbits=1)
+
+client = ModbusSerialClient(method='rtu')
+client.socket = ser
+client.connect()
+
+try:
+   result = client.read_holding_registers(address=0x9c43, count=1, unit=4)
+   # result = client.read_holding_registers(address=0x0102, count=1, unit=5)
+   # result = client.read_input_registers(address=0x0002, count=1, unit=5)
+   # result = client.read_discrete_inputs(address=0x0000, count=1, unit=4)
+   print(result)
+   print(result.registers)
+   # print(result.bits)
+except Exception as e:
+   print("ERROR: %s" % str(e))
+
+client.close()
+

snippets/test6c.py

@@ -0,0 +1,19 @@
+from pymodbus.client.sync import ModbusSerialClient
+import RS485Ext
+import time
+
+ser=RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=9600, stopbits=1)
+
+client = ModbusSerialClient(method='rtu')
+client.socket = ser
+client.connect()
+
+try:
+   result = client.write_register(address=0x9c48, unit=4, value=0x000F)
+   # result = client.write_coil(address=0x0000, unit=4, value=1)
+   print(result)
+except Exception as e:
+   print("ERROR: %s" % str(e))
+
+client.close()
+

snippets/test6d.py

@@ -0,0 +1,25 @@
+from pymodbus.client.sync import ModbusSerialClient
+import RS485Ext
+import time
+
+ser=RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=1200, stopbits=1)
+
+client = ModbusSerialClient(method='rtu')
+client.socket = ser
+client.connect()
+
+v = 0
+while True:
+    if v == 0:
+        v = 1
+    else:
+        v = 0
+    try:
+        result = client.write_coil(address=0x0000, unit=4, value=v)
+        print(result)
+    except Exception as e:
+        print("ERROR: %s" % str(e))
+    time.sleep(0.5)
+
+client.close()
+

snippets/test6e.py

@@ -0,0 +1,21 @@
+from pymodbus.client.sync import ModbusSerialClient
+import RS485Ext
+import time
+
+ser=RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=1200, stopbits=1)
+
+client = ModbusSerialClient(method='rtu')
+client.socket = ser
+client.connect()
+
+while True:
+    try:
+        result = client.read_discrete_inputs(address=0x0000, count=1, unit=4)
+        print(result)
+        print(result.bits)
+    except Exception as e:
+        print("ERROR: %s" % str(e))
+    time.sleep(0.5)
+
+client.close()
+

snippets/test7.py

@@ -0,0 +1,117 @@
+from pymodbus.client.sync import ModbusSerialClient
+from pymodbus.pdu import ExceptionResponse
+from pymodbus.exceptions import ModbusIOException
+import RS485Ext
+import struct
+import time
+
+def registersToIeeeFloat(i):
+  return struct.unpack('f', bytes(
+                                   [((x & 0xff00) >> 8) if y else (x & 0x00ff) 
+                                     for x in i[::-1] 
+                                     for y in [False, True]
+                                   ]
+                                 )
+                      )[0]
+
+def registersToIeeeFloatReverse(i):
+  return struct.unpack('f', bytes(
+                                   [((x & 0xff00) >> 8) if y else (x & 0x00ff) 
+                                     for x in i 
+                                     for y in [False, True]
+                                   ]
+                                 )
+                      )[0]
+
+def dataConverter(t, d):
+  if t == 'F':
+    return registersToIeeeFloat(d)
+  elif t == 'RF':
+    return registersToIeeeFloatReverse(d)
+  else:
+    raise Exception("Converter '{0}' is not supported".format(t))
+
+
+class ModbusException(Exception):
+  def __init__(self, resp):
+    self.msg = str(result)
+
+  def __str__(self):
+    return self.msg
+
+class ModbusRequestDefinition(object):
+  def __init__(self, kind, unit, address, count, converter, label):
+    self.kind = kind
+    self.unit = unit
+    self.address = address
+    self.count = count
+    self.converter = converter
+    self.label = label
+  
+reqs = [
+#  ModbusRequestDefinition(4, 0x2000, 2, 'F', '(ERR) Unavailable device'),
+#  ModbusRequestDefinition(1, 0x2000, 4, 'F', '(ERR) Wrong register size'),
+#  ModbusRequestDefinition(1, 0x2000, 2, 'F', 'Voltage'),
+#  ModbusRequestDefinition(1, 0x2020, 2, 'F', 'Frequency'),
+#  ModbusRequestDefinition(1, 0x2060, 2, 'F', 'Current'),
+#  ModbusRequestDefinition('H', 3, 0x0004, 2, 'RF', 'Resistance Channel 1'),
+#  ModbusRequestDefinition('H', 3, 0x000C, 2, 'RF', 'Temperature Channel 1'),
+#  ModbusRequestDefinition('H', 3, 0x0014, 2, 'RF', 'Resistance Channel 2'),
+#  ModbusRequestDefinition('H', 3, 0x001C, 2, 'RF', 'Temperature Channel 2'),
+  ModbusRequestDefinition('D', 4, 0x0000, 1, '', 'Discrete Input'),
+  ModbusRequestDefinition('I', 5, 0x0001, 1, '', 'Temperature'),
+  ModbusRequestDefinition('I', 5, 0x0002, 1, '', 'Humidity'),
+]
+
+
+def getSerial():
+  return RS485Ext.RS485Ext(port='/dev/ttyAMA0', baudrate=9600, stopbits=1,
+                           timeout=1)
+
+client = ModbusSerialClient(method='rtu')
+client.socket = getSerial()
+client.connect()
+
+delay = 0.05
+period = 0.5
+
+while True:
+  for req in reqs:
+    try:
+      time.sleep(delay)
+      if req.kind == 'H':
+        # print("Trying to read: {0} {1} {2}".format(req.address, req.count, req.unit))
+        result = client.read_holding_registers(address=req.address, 
+                                               count=req.count, 
+                                               unit=req.unit)
+        if type(result) in [ExceptionResponse, ModbusIOException]:
+          raise ModbusException(result)
+        print("{0}: {1:.2f}".format(req.label, 
+                                    dataConverter(req.converter, 
+                                                  result.registers)))
+      elif req.kind == 'D':
+        result = client.read_discrete_inputs(address=req.address,
+                                             count=req.count,
+                                             unit=req.unit)
+        if type(result) in [ExceptionResponse, ModbusIOException]:
+          raise ModbusException(result)
+        print("{0}: {1!s}".format(req.label, result.bits))        
+      elif req.kind == 'I':
+        result = client.read_input_registers(address=req.address,
+                                             count=req.count,
+                                             unit=req.unit)
+        if type(result) in [ExceptionResponse, ModbusIOException]:
+          raise ModbusException(result)
+        print("{0}: {1}".format(req.label, result.registers))        
+    except ModbusException as e:
+      print("ERROR when querying '{0}': {1!s}".format(req.label, e))
+      if client.socket is None:
+        print("renew socket")
+        client.socket = getSerial()
+
+  print("-------------")
+  time.sleep(period)
+
+
+client.close()
+

snippets/test8.py

@@ -0,0 +1,105 @@
+import unittest
+import json
+
+class A(object):
+    def __init__(self, arg1=None, arg2=None):
+        self.arg1 = arg1
+        self.arg2 = arg2
+        self.arg3 = self.arg1 + self.arg2
+
+    def __str__(self):
+        return "A: {0!s} {1!s} {2!s}".format(self.arg1, self.arg2, self.arg3)
+
+    def jsonify(self):
+        return {'type':self.__class__.__name__, 'args': {'arg1':self.arg1, 'arg2':self.arg2}}
+
+class MyEncoder(json.JSONEncoder):
+    def default(self, o):
+        try:
+            return o.jsonify()
+        except TypeError or AttributeError:
+            return super().default(o)
+
+
+def objectFactory(j):
+    klass = eval(j['type'])
+    o = klass(**j['args'])
+    return o
+
+def MyDecoder(j):
+    if type(j) == dict and 'type' in j:
+        return objectFactory(j)
+    else:
+        return j
+
+class Tests(unittest.TestCase):
+    def test_a1(self):
+        a1 = A(1, 2)
+        self.assertEqual(a1.arg1, 1)
+        self.assertEqual(a1.arg2, 2)
+        self.assertEqual(a1.arg3, 3)
+
+    def test_a2(self):
+        a2 = A(**{'arg1':2, 'arg2':4})
+        self.assertEqual(a2.arg1, 2)
+        self.assertEqual(a2.arg2, 4)
+        self.assertEqual(a2.arg3, 6)
+
+    def test_a3(self):
+        j = '{ "type": "A", "args": { "arg1": 3, "arg2": 5 } }'
+        jj = json.loads(j)
+        klass = eval(jj['type'])
+        self.assertEqual(A, klass)
+        a3 = klass(**jj['args'])
+        self.assertEqual(a3.arg1, 3)
+        self.assertEqual(a3.arg2, 5)
+        self.assertEqual(a3.arg3, 8)
+
+    def test_a4(self):
+        j = '{ "type": "A", "args": { "arg1": 3, "arg2": 5 } }'
+        jj = json.loads(j)
+        klass = eval(jj['type'])
+        self.assertEqual(A, klass)
+        a3 = klass(**jj['args'])
+        self.assertEqual(a3.arg1, 3)
+        self.assertEqual(a3.arg2, 5)
+        self.assertEqual(a3.arg3, 8)
+
+        jjjj = json.dumps(a3, cls=MyEncoder)
+
+        jjj = json.loads(jjjj)
+        klass = eval(jjj['type'])
+        self.assertEqual(A, klass)
+        a3 = klass(**jjj['args'])
+        self.assertEqual(a3.arg1, 3)
+        self.assertEqual(a3.arg2, 5)
+        self.assertEqual(a3.arg3, 8)
+
+    def test_a5(self):
+        jList = []
+        jList.append(objectFactory({'type':'A', 'args': {'arg1':1, 'arg2':2}}))
+        jList.append(objectFactory({'type':'A', 'args': {'arg1':2, 'arg2':3}}))
+        jList.append(objectFactory({'type':'A', 'args': {'arg1':3, 'arg2':4}}))
+
+        js = json.dumps(jList, cls=MyEncoder, sort_keys=True, indent=4)
+
+        print(js)
+
+        jResultList = json.loads(js, object_hook=MyDecoder)
+
+        self.assertEqual(jResultList[0].arg1, 1)
+        self.assertEqual(jResultList[0].arg2, 2)
+        self.assertEqual(jResultList[0].arg3, 3)
+
+        self.assertEqual(jResultList[1].arg1, 2)
+        self.assertEqual(jResultList[1].arg2, 3)
+        self.assertEqual(jResultList[1].arg3, 5)
+
+        self.assertEqual(jResultList[2].arg1, 3)
+        self.assertEqual(jResultList[2].arg2, 4)
+        self.assertEqual(jResultList[2].arg3, 7)
+
+
+
+if __name__ == '__main__':
+    unittest.main()

snippets/test9.py

@@ -0,0 +1,14 @@
+class A(object):
+    def __init__(self):
+        self.a = 1
+    def x(self):
+        return self.a
+
+class B(A):
+    def __init__(self):
+        self.a = 2
+    def x(self):
+        return self.a
+    def y(self):
+        return super().x()
+

src/CmdServer.py

@@ -0,0 +1,392 @@
+import threading
+import socketserver
+import cmd
+import re
+import io
+import datetime
+import RegisterDatapoint
+import logging
+
+class CmdInterpreterException(ValueError): pass
+
+def parseIntArbitraryBase(s):
+    i = 0
+    if s.startswith('0x'):
+        i = int(s, 16)
+    elif s.startswith('0b'):
+        i = int(s, 2)
+    else:
+        i = int(s, 10)
+    return i
+
+class CmdInterpreter(cmd.Cmd):
+    def __init__(self, infile, outfile, config, notifier, registers):
+        super().__init__(stdin=infile, stdout=outfile)
+        self.use_rawinput = False
+        self.config = config
+        self.notifier = notifier
+        self.registers = registers
+        self.prompt = "test8> "
+        self.intro = "test8 admin interface"
+        self.splitterRe = re.compile('\s+')
+        self.logger = logging.getLogger('CmdInterpreter')
+
+    def __print(self, text):
+        self.stdout.write(text)
+
+    def __println(self, text):
+        self.stdout.write(text)
+        self.stdout.write("\n\r")
+
+    def do_notify(self, arg):
+        self.notifier.notify()
+
+    def help_notify(self):
+        self.__println("Notifies threads using the list of datapoints about changes in this list.")
+        self.__println("Call after modifications on the list.")
+
+    def do_quit(self, arg):
+        self.__println("Bye!")
+        return True
+    
+    def do_add_hr(self, arg):
+        try:
+            (label, unit, address, count, scanrate, readTopic, writeTopic, feedbackTopic, converter) = self.splitterRe.split(arg)
+            self.__println("Label:         {0}".format(label))
+            self.__println("Unit:          {0}".format(unit))
+            self.__println("Address:       {0}".format(address))
+            self.__println("Count:         {0}".format(count))
+            self.__println("ScanRate:      {0}".format(scanrate))
+            self.__println("ReadTopic:     {0}".format(readTopic))
+            self.__println("WriteTopic:    {0}".format(writeTopic))
+            self.__println("FeedbackTopic: {0}".format(feedbackTopic))
+            self.__println("Converter:     {0}".format(converter))
+
+            if readTopic == 'None':
+                readTopic = None
+            if writeTopic == 'None':
+                writeTopic = None
+            if feedbackTopic == 'None':
+                feedbackTopic = None
+            if converter == 'None':
+                converter = None
+            unit = parseIntArbitraryBase(unit)
+            address = parseIntArbitraryBase(address)
+            count = parseIntArbitraryBase(count)
+            scanrate = float(scanrate)
+            r = RegisterDatapoint.HoldingRegisterDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), readTopic, writeTopic, feedbackTopic, converter)
+            self.registers.append(r)
+        except ValueError as e:
+            self.__println("ERROR: {0!s}, {1!s}".format(e.__class__.__name__, e))
+
+    def help_add_hr(self):
+        # HoldingRegisterDatapoint('Voltage', 1, 0x2000, 2, datetime.timedelta(seconds=10), 'Pub/Voltage', None, None),
+        self.__println("Usage: add_hr <Label> <Unit> <Address> <Count> <ScanRate>")
+        self.__println("              <ReadTopic> <WriteTopic> <FeedbackTopic>")
+        self.__println("              <Converter>")
+        self.__println("Adds a holding register")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+        self.__println("---------------------------------------------------------------------")
+        self.__println("<Label>                      Descriptive label")
+        self.__println("<Unit>                       Modbus address of the device")
+        self.__println("<Address>                    Register address within the device")
+        self.__println("<Count>                      Count of registers to be read or write in words")
+        self.__println("<ScanRate>                   Scanrate in seconds (float), for write datapoints")
+        self.__println("                             set to zero (0)")
+        self.__println("<ReadTopic>                  Topic to publish read data")
+        self.__println("<WriteTopic>                 Topic to be subscribe to receive data to be")
+        self.__println("                             written")
+        self.__println("<FeedbackTopic>              Topic to publish feedback after a write process,")
+        self.__println("<Converter>                  Converter for data")
+
+    
+    def do_add_coil(self, arg):
+        try:
+            (label, unit, address, scanrate, readTopic, writeTopic, feedbackTopic) = self.splitterRe.split(arg)
+            self.__println("Label:         {0}".format(label))
+            self.__println("Unit:          {0}".format(unit))
+            self.__println("Address:       {0}".format(address))
+            self.__println("ScanRate:      {0}".format(scanrate))
+            self.__println("ReadTopic:     {0}".format(readTopic))
+            self.__println("WriteTopic:    {0}".format(writeTopic))
+            self.__println("FeedbackTopic: {0}".format(feedbackTopic))
+
+            if readTopic == 'None':
+                readTopic = None
+            if writeTopic == 'None':
+                writeTopic = None
+            if feedbackTopic == 'None':
+                feedbackTopic = None
+            unit = parseIntArbitraryBase(unit)
+            address = parseIntArbitraryBase(address)
+            scanrate = float(scanrate)
+            r = RegisterDatapoint.CoilDatapoint(label, unit, address, datetime.timedelta(seconds=scanrate), readTopic, writeTopic, feedbackTopic)
+            self.registers.append(r)
+        except ValueError as e:
+            self.__println("ERROR: {0!s}, {1!s}".format(e.__class__.__name__, e))
+
+    def help_add_coil(self):
+        self.__println("Usage: add_coil <Label> <Unit> <Address> <ScanRate>")
+        self.__println("                <ReadTopic> <WriteTopic> <FeedbackTopic>")
+        self.__println("Adds a coil")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+        self.__println("---------------------------------------------------------------------")
+        self.__println("<Label>                      Descriptive label")
+        self.__println("<Unit>                       Modbus address of the device")
+        self.__println("<Address>                    Register address within the device")
+        self.__println("<ScanRate>                   Scanrate in seconds (float), for write datapoints")
+        self.__println("                             set to zero (0)")
+        self.__println("<ReadTopic>                  Topic to publish read data")
+        self.__println("<WriteTopic>                 Topic to be subscribe to receive data to be")
+        self.__println("                             written")
+        self.__println("<FeedbackTopic>              Topic to publish feedback after a write process,")
+    
+    def do_add_ir(self, arg):
+        try:
+            (label, unit, address, count, scanrate, updateOnly, readTopic, converter) = self.splitterRe.split(arg)
+            self.__println("Label:         {0}".format(label))
+            self.__println("Unit:          {0}".format(unit))
+            self.__println("Address:       {0}".format(address))
+            self.__println("Count:         {0}".format(count))
+            self.__println("ScanRate:      {0}".format(scanrate))
+            self.__println("UpdateOnly:    {0}".format(updateOnly))
+            self.__println("ReadTopic:     {0}".format(readTopic))
+            self.__println("Converter:     {0}".format(converter))
+
+            if readTopic == 'None':
+                readTopic = None
+            if converter == 'None':
+                converter = None
+            if updateOnly in ['true', 'True', 'yes', 'Yes']:
+                updateOnly = True
+            elif updateOnly in ['false', 'False', 'no', 'No']:
+                updateOnly = False
+            else:
+                raise CmdInterpreterException('updateOnly must be true or false, yes or no')
+            unit = parseIntArbitraryBase(unit)
+            address = parseIntArbitraryBase(address)
+            count = parseIntArbitraryBase(count)
+            scanrate = float(scanrate)
+            r = RegisterDatapoint.InputRegisterDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), updateOnly, readTopic, converter)
+            self.registers.append(r)
+        except ValueError as e:
+            self.__println("ERROR: {0!s}, {1!s}".format(e.__class__.__name__, e))
+
+    def help_add_ir(self):
+        self.__println("Usage: add_ir <Label> <Unit> <Address> <Count> <ScanRate>")
+        self.__println("              <UpdateOnly> <ReadTopic> <Converter>")
+        self.__println("Adds an input register")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+        self.__println("---------------------------------------------------------------------")
+        self.__println("<Label>                      Descriptive label")
+        self.__println("<Unit>                       Modbus address of the device")
+        self.__println("<Address>                    Register address within the device")
+        self.__println("<Count>                      Count of registers to be read in words")
+        self.__println("<ScanRate>                   Scanrate in seconds (float)")
+        self.__println("<UpdateOnly>                 Publish only when value has changed")
+        self.__println("<ReadTopic>                  Topic to publish read data")
+        self.__println("<Converter>                  Converter for data")
+
+    def do_add_di(self, arg):
+        try:
+            (label, unit, address, count, scanrate, updateOnly, readTopic, bitCount) = self.splitterRe.split(arg)
+            self.__println("Label:         {0}".format(label))
+            self.__println("Unit:          {0}".format(unit))
+            self.__println("Address:       {0}".format(address))
+            self.__println("Count:         {0}".format(count))
+            self.__println("ScanRate:      {0}".format(scanrate))
+            self.__println("UpdateOnly:    {0}".format(updateOnly))
+            self.__println("ReadTopic:     {0}".format(readTopic))
+            self.__println("BitCount:      {0}".format(bitCount))
+
+            if readTopic == 'None':
+                readTopic = None
+            if updateOnly in ['true', 'True', 'yes', 'Yes']:
+                updateOnly = True
+            elif updateOnly in ['false', 'False', 'no', 'No']:
+                updateOnly = False
+            else:
+                raise CmdInterpreterException('updateOnly must be true or false, yes or no')
+            unit = parseIntArbitraryBase(unit)
+            address = parseIntArbitraryBase(address)
+            count = parseIntArbitraryBase(count)
+            scanrate = float(scanrate)
+            bitCount = int(bitCount)
+            r = RegisterDatapoint.DiscreteInputDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), updateOnly, readTopic, None, bitCount)
+            self.registers.append(r)
+        except ValueError as e:
+            self.__println("ERROR: {0!s}, {1!s}".format(e.__class__.__name__, e))
+
+    def help_add_di(self):
+        self.__println("Usage: add_di <Label> <Unit> <Address> <Count> <ScanRate>")
+        self.__println("              <UpdateOnly> <ReadTopic> <bitCount>")
+        self.__println("Adds a discrete input")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+        self.__println("---------------------------------------------------------------------")
+        self.__println("<Label>                      Descriptive label")
+        self.__println("<Unit>                       Modbus address of the device")
+        self.__println("<Address>                    Register address within the device")
+        self.__println("<Count>                      Count of registers to be read in words")
+        self.__println("<ScanRate>                   Scanrate in seconds (float)")
+        self.__println("<UpdateOnly>                 Publish only when value has changed")
+        self.__println("<ReadTopic>                  Topic to publish read data")
+        self.__println("<BitCount>                   Number of bit to be considered")
+
+    def do_list(self, arg):
+        for i, r in enumerate(self.registers):
+            self.__println("#{0}: {1!s}".format(i, r))
+    
+    def help_list(self):
+        self.__println("Usage: list")
+        self.__println("-----------")
+        self.__println("List the configured datapoints")
+
+    def do_reset(self, arg):
+        for r in self.registers:
+            r.errorCount = 0
+            r.processCount = 0
+    
+    def help_reset(self):
+        self.__println("Usage: reset")
+        self.__println("-----------")
+        self.__println("Resets the statistics of configured datapoints")
+
+    def do_stats(self, arg):
+        for i, r in enumerate(self.registers):
+            if r.processCount == 0:
+                ratio = -1
+            else:
+                ratio = float(r.errorCount) / float(r.processCount)
+            self.__println("#{0:2d}: {1:15s} ({2:2d}, {3:5d}), pc: {4:7d}, ec: {5:7d}, q: {6:1.4f}"
+                           .format(i, r.label, r.unit, r.address,
+                                   r.processCount, r.errorCount, ratio))
+
+    def help_stats(self):
+        self.__println("Usage: stats")
+        self.__println("-----------")
+        self.__println("List the statistics of configured datapoints")
+
+
+
+    def do_change(self, arg):
+        (idx, key, typ, value) = self.splitterRe.split(arg)
+        try:
+            i = int(idx)
+            r = self.registers[i]
+
+            if typ == 'I':
+                value = parseIntArbitraryBase(value)
+            elif typ == 'F':
+                value = float(value)
+            elif typ == 'B':
+                if value in ['true', 'True', 'yes', 'Yes']:
+                    value = True
+                elif value in ['false', 'False', 'no', 'No']:
+                    value = False
+                else:
+                    raise CmdInterpreterException('boolean value must be true or false, yes or no')
+            elif typ == 'S':
+                # string
+                pass
+            elif typ == 'T':
+                value = datetime.timedelta(seconds=float(value))
+            elif typ == 'N':
+                value = None
+            else:
+                raise CmdInterpreterException('unknown type specifier, must be I, F, B, S or T')
+            
+            if key not in r.__dict__:
+                raise CmdInterpreterException('selected datapoint does not support key')
+            
+            r.__dict__[key] = value
+        except ValueError as e:
+            self.__println("ERROR: {0!s}, {1!s}".format(e.__class__.__name__, e))
+
+    def help_change(self):
+        self.__println("Usage: change <idx> <key> <type> <value>")
+        self.__println("Changes on attribute of a datapoint")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+        self.__println("---------------------------------------------------------------------")
+        self.__println("<idx>                      Index, use list command to find")
+        self.__println("<key>                      Name of attribute")
+        self.__println("<type>                     Type of attribute")
+        self.__println("                           I .. Integer")
+        self.__println("                           F .. Float")
+        self.__println("                           B .. Boolean")
+        self.__println("                           T .. Timedelta, give in seconds")
+        self.__println("                           S .. String")
+        self.__println("                           N .. None (Value must be given but is not")
+        self.__println("                                      considered)")
+        self.__println("<value>                    New value")
+
+
+
+
+    def do_del(self, arg):
+        try:
+            i = int(arg)
+            r = self.registers[i]
+            self.registers.remove(r)
+            self.__println("{0!s} removed".format(r))
+        except ValueError as e:
+            self.__println("ERROR: {0!s}".format(e))
+
+    def help_del(self):
+        self.__println("Usage: del <idx>")
+        self.__println("Removes an item from the list of datapoints by its index, see list command.")
+        self.__println("Be aware: indexes have been changed, rerun list before removing the next item.")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+
+    def do_save(self, arg):
+        RegisterDatapoint.saveRegisterList(self.registers, self.config.registerFile)
+
+    def help_save(self):
+        self.__println("Usage: save")
+        self.__println("Saves a modified register list into the register file.")
+
+    def do_load(self, arg):
+        try:
+            registers = RegisterDatapoint.loadRegisterList(self.config.registerFile)
+            self.registers = registers
+        except Exception as e:
+            self.__println("Unable to load register list: {0!s}".format(e))
+
+    def help_load(self):
+        self.__println("Usage: load")
+        self.__println("Reload the register file, overwrite all unsaved changes.")
+
+
+
+class CmdHandle(socketserver.StreamRequestHandler):
+    def handle(self):
+        logger = logging.getLogger('CmdHandle')
+        cmd = CmdInterpreter(io.TextIOWrapper(self.rfile), io.TextIOWrapper(self.wfile), self.server.userData.config, 
+                             self.server.userData.notifier, self.server.userData.registers)
+        try:
+            cmd.cmdloop()
+            logger.info("Cmd handle terminated")
+        except ConnectionAbortedError as e:
+            logger.info("Cmd handle externally interrupted")
+
+class MyThreadingTCPServer(socketserver.ThreadingTCPServer):
+    def __init__(self, host, handler, userData):
+        super().__init__(host, handler)
+        self.userData = userData
+
+class MyCmdUserData(object):
+    def __init__(self, config, notifier, registers):
+        self.config = config
+        self.notifier = notifier
+        self.registers = registers
+
+class CmdServer(threading.Thread):
+    def __init__(self, config, notifier, registers):
+        super().__init__()
+        self.config = config
+        self.server = MyThreadingTCPServer((config.cmdAddress, config.cmdPort), CmdHandle, MyCmdUserData(config, notifier, registers))
+        self.daemon = True
+
+    def start(self):
+        self.server.serve_forever()
+
+

src/CommunicationProcessor.py

@@ -0,0 +1,59 @@
+import threading
+import datetime
+# import RS485Ext
+import RegisterDatapoint
+from pymodbus.client.sync import ModbusSerialClient
+import wiringpi
+import MyRS485
+import time
+import logging
+
+
+ERROR_PIN = 29
+
+class CommunicationProcessor(threading.Thread):
+    def __init__(self, config, queue, pubQueue):
+        super().__init__()
+        self.config = config
+        self.queue = queue
+        self.pubQueue = pubQueue
+        wiringpi.wiringPiSetup()
+        wiringpi.pinMode(ERROR_PIN, wiringpi.OUTPUT)
+        self.daemon = True
+        if self.config.modbusDebug:
+            logging.getLogger('pymodbus').setLevel(logging.DEBUG)
+        else:
+            logging.getLogger('pymodbus').setLevel(logging.ERROR)
+        self.logger = logging.getLogger('CommunicationProcessor')
+
+    def __getSerial(self):
+        # return RS485Ext.RS485Ext(port=self.config.serialPort, baudrate=self.config.serialBaudRate, stopbits=1,
+        #                         timeout=1)
+        return MyRS485.MyRS485(port=self.config.serialPort, baudrate=self.config.serialBaudRate, stopbits=1,
+                               timeout=1)
+
+
+    def run(self):
+        client = ModbusSerialClient(method='rtu')
+        client.socket = self.__getSerial()
+        client.connect()
+
+        while True:
+            r = self.queue.get()
+            try:
+                wiringpi.digitalWrite(ERROR_PIN, wiringpi.LOW)
+                self.logger.debug("Dequeued: {0!s}".format(r))
+                r.enqueued = False
+                r.process(client, self.pubQueue)
+            except RegisterDatapoint.DatapointException as e:
+                wiringpi.digitalWrite(ERROR_PIN, wiringpi.HIGH)
+                self.logger.error("ERROR when processing '{0}': {1!s}".format(r.label, e))
+                if client.socket is None:
+                    self.logger.error("renew socket")
+                    client.socket = self.__getSerial()
+            finally:
+                time.sleep(self.config.interCommDelay)
+
+
+
+

src/Config.py

@@ -0,0 +1,15 @@
+
+class Config(object):
+    def __init__(self):
+        self.logFile = '/tmp/mbm.log'
+        self.modbusDebug = False
+        self.mqttBrokerHost = '172.16.2.16'
+        self.mqttBrokerPort = 1883
+        self.mqttLogin = ''
+        self.mqttPassword = ''
+        self.cmdAddress = '127.0.0.1'
+        self.cmdPort = 9999
+        self.registerFile = 'registers.json'
+        self.serialPort = '/dev/ttyAMA0'
+        self.serialBaudRate = 9600
+        self.interCommDelay = 0.025

src/Converters.py

@@ -0,0 +1,18 @@
+# in:  from Modbus to MQTT, input is a list of 16bit integers, output shall be the desired format 
+#      to be sent in the MQTT message
+# out: from MQTT to Modbus, input is the format received from MQTT, output shall be a list of 
+#      16bit integers to be written to the Modbus slave
+
+from struct import pack, unpack
+
+
+Converters = {
+    "dht20TOFloat": {
+        "in": lambda x : float(x[0]) / 10.0,
+        "out": None
+    },
+    "uint32": {
+        "in": lambda x : unpack('L', pack('HH', *x))[0],
+        "out": lambda x : unpack('HH', pack('L', int(x)))
+    }
+}

src/MqttProcessor.py

@@ -0,0 +1,88 @@
+import threading
+import paho.mqtt.client as mqtt
+from NotificationForwarder import AbstractNotificationReceiver
+import logging
+
+
+class PublishItem(object):
+    def __init__(self, topic, payload):
+        self.topic = topic
+        self.payload = payload
+
+def mqttOnConnectCallback(client, userdata, flags, rc):
+    userdata.onConnect()
+
+def mqttOnMessageCallback(client, userdata, message):
+    userdata.onMessage(message.topic, message.payload)
+
+def mqttOnDisconnectCallback(client, userdata, rc):
+    userdata.onDisconnect(rc)
+
+class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
+    def __init__(self, config, registers, queue, pubQueue):
+        super().__init__()
+        self.config = config
+        self.registers = registers
+        self.queue = queue
+        self.pubQueue = pubQueue
+        self.client = mqtt.Client(userdata=self)
+        self.subscriptions = []
+        self.topicRegisterMap ={}
+        self.daemon = True
+        self.logger = logging.getLogger('MqttProcessor')
+
+    def __processUpdatedRegisters(self, force=False):
+        self.logger.debug("MqttProcessor.__updateSubscriptions")
+
+        subscribeTopics = [ r.subscribeTopic for r in self.registers if hasattr(r,'subscribeTopic') and r.subscribeTopic]
+        self.logger.debug("Topics: {0!s}".format(subscribeTopics))
+
+        for subscribeTopic in subscribeTopics:
+            if (subscribeTopic not in self.subscriptions) or force:
+                self.logger.debug("Subscribe to {0}".format(subscribeTopic))
+                self.client.subscribe(subscribeTopic)
+                self.subscriptions.append(subscribeTopic)
+
+        for subscription in self.subscriptions:
+            if (subscription not in subscribeTopics) and not force:
+                self.logger.debug("Unsubscribe from {0}".format(subscription))
+                self.client.unsubscribe(subscription)
+                self.subscriptions.remove(subscription)
+
+        self.topicRegisterMap = { r.subscribeTopic: r for r in self.registers if hasattr(r,'subscribeTopic') and r.subscribeTopic }
+
+    def receiveNotification(self, arg):
+        self.logger.info("MqttProcessor:registersChanged")
+        self.__processUpdatedRegisters()
+
+    def run(self):
+        self.client.on_message = mqttOnMessageCallback
+        self.client.on_connect = mqttOnConnectCallback
+        self.client.on_disconnect = mqttOnDisconnectCallback
+        if self.config.mqttLogin and self.config.mqttPassword:
+            self.client.username_pw_set(self.config.mqttLogin, self.config.mqttPassword)
+        self.client.connect(self.config.mqttBrokerHost, self.config.mqttBrokerPort)
+        self.client.loop_start()
+
+        while True:
+            pubItem = self.pubQueue.get()
+            if isinstance(pubItem, PublishItem):
+                self.client.publish(pubItem.topic, pubItem.payload)
+            else:
+                self.logger.error("Invalid object in publish queue")
+
+
+    def onConnect(self):
+        # print("MqttProcessor.onConnect")
+        self.__processUpdatedRegisters(force=True)
+
+    def onDisconnect(self, rc):
+        self.logger.error("Disconnected from MQTT broker: {0}".format(rc))
+
+    def onMessage(self, topic, payload):
+        # print("MqttProcessor.onMessage")
+        r = self.topicRegisterMap[topic]
+        self.logger.debug("{0}: {1!s} -> {2!s}".format(topic, payload, r))
+        r.onMessage(payload)
+        self.queue.put(r)
+

src/MyPriorityQueue.py

@@ -0,0 +1,22 @@
+import queue
+
+
+class MyPriorityQueueItem(object):
+    def __init__(self, itemWithPriority):
+        self.itemWithPriority = itemWithPriority
+
+    def __lt__(self, other): return self.itemWithPriority.priority < other.itemWithPriority.priority
+    def __le__(self, other): return self.itemWithPriority.priority <= other.itemWithPriority.priority
+    def __eq__(self, other): return self.itemWithPriority.priority == other.itemWithPriority.priority
+    def __ne__(self, other): return self.itemWithPriority.priority != other.itemWithPriority.priority
+    def __gt__(self, other): return self.itemWithPriority.priority > other.itemWithPriority.priority
+    def __ge__(self, other): return self.itemWithPriority.priority >= other.itemWithPriority.priority
+
+class MyPriorityQueue(queue.PriorityQueue):
+    def _put(self, itemWithPriority):
+        i = MyPriorityQueueItem(itemWithPriority)
+        super()._put(i)
+
+    def _get(self):
+        i = super()._get()
+        return i.itemWithPriority

src/MyRS485.py

@@ -0,0 +1,24 @@
+import serial
+import wiringpi
+import array
+import fcntl
+import termios
+
+DE_PIN = 0
+
+class MyRS485(serial.Serial):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        wiringpi.wiringPiSetup()
+        wiringpi.pinMode(DE_PIN, wiringpi.OUTPUT)
+        self.buf = array.array('h', [0])
+
+    def write(self, b):
+        wiringpi.digitalWrite(DE_PIN, wiringpi.HIGH)
+        super().write(b)
+        while True:
+            fcntl.ioctl(self.fileno(), termios.TIOCSERGETLSR, self.buf, 1)
+            if self.buf[0] & termios.TIOCSER_TEMT:
+                break
+        wiringpi.digitalWrite(DE_PIN, wiringpi.LOW)
+

src/NotificationForwarder.py

@@ -0,0 +1,15 @@
+
+class AbstractNotificationReceiver(object):
+    def receiveNotification(self, arg):
+        raise NotImplementedError
+
+class NotificationForwarder(object):
+    def __init__(self):
+        self.receivers = []
+
+    def register(self, receiver):
+        self.receivers.append(receiver)
+
+    def notify(self, arg=None):
+        for r in self.receivers:
+            r.receiveNotification(arg)

src/RegisterDatapoint.py

@@ -0,0 +1,291 @@
+import datetime
+from pymodbus.pdu import ExceptionResponse
+from pymodbus.exceptions import ModbusIOException
+import MqttProcessor
+import logging
+import json
+import Converters
+
+class DatapointException(Exception): pass
+
+class AbstractModbusDatapoint(object):
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, converter=None):
+        self.argList = ['label', 'unit', 'address', 'count', 'scanRate', 'converter']
+        self.label = label
+        self.unit = unit
+        self.address = address
+        self.count = count
+        self.converter = converter
+        if type(scanRate) == float:
+            self.scanRate = datetime.timedelta(seconds=scanRate)
+        else:
+            self.scanRate = scanRate
+        self.type = 'abstract data point'
+        self.enqueued = False
+        self.lastContact = None
+        self.errorCount = 0
+        self.processCount = 0
+        if self.scanRate:
+            self.priority = 1
+        else:
+            self.priority = 0
+
+    def __str__(self):
+        return ("{0}, {1}: unit: {2},  address: {3}, count: {4}, scanRate: {5}, "
+                "enqueued: {6}, lastContact: {7}, errorCount: {8}, processCount: {9}, "
+                "converter: {10}"
+                .format(self.type, self.label, self.unit, self.address, self.count,
+                        self.scanRate, self.enqueued, self.lastContact,
+                        self.errorCount, self.processCount, self.converter))
+
+    def jsonify(self):
+        return {'type':self.__class__.__name__, 
+                'args': { k: getattr(self, k) for k in self.argList }
+               }
+
+    def process(self, client):
+        raise NotImplementedError
+
+
+
+class HoldingRegisterDatapoint(AbstractModbusDatapoint):
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, 
+                 publishTopic=None, subscribeTopic=None, feedbackTopic=None, converter=None):
+        super().__init__(label, unit, address, count, scanRate, converter)
+        self.argList = self.argList + ['publishTopic', 'subscribeTopic', 'feedbackTopic']
+        self.publishTopic = publishTopic
+        self.subscribeTopic = subscribeTopic
+        self.feedbackTopic = feedbackTopic
+        self.writeRequestValue = None
+        self.type = 'holding register'
+
+    def __str__(self):
+        return ("[{0!s}, publishTopic: {1}, subscribeTopic: {2}, feedbackTopic: {3}, "
+                "writeRequestValue: {4!s}"
+                .format(super().__str__(), self.publishTopic, self.subscribeTopic, self.feedbackTopic,
+                        self.writeRequestValue))
+
+    def process(self, client, pubQueue):
+        logger = logging.getLogger('HoldingRegisterDatapoint')
+        if self.writeRequestValue:
+            # perform write operation
+            logger.debug("Holding register, perform write operation")
+            self.processCount += 1
+            values = None
+            logger.debug("{0}: raw: {1!s}".format(self.label, self.writeRequestValue))
+            if self.converter and Converters.Converters[self.converter]['out']:
+                try:
+                    values = Converters.Converters[self.converter]['out'](self.writeRequestValue)
+                    logger.debug("{0}: converted: {1!s}".format(self.label, values))
+                except Exception as e:
+                    raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
+            else:
+                values = [int(self.writeRequestValue)]
+            result = client.write_registers(address=self.address,
+                                           unit=self.unit,
+                                           values=values)
+            logger.debug("Write result: {0!s}".format(result))                
+            self.writeRequestValue = None
+        else:
+            # perform read operation
+            logger.debug("Holding register, perform read operation")
+            self.processCount += 1
+            result = client.read_holding_registers(address=self.address, 
+                                                   count=self.count, 
+                                                   unit=self.unit)
+            if type(result) in [ExceptionResponse, ModbusIOException]:
+                self.errorCount += 1
+                raise DatapointException(result)
+            logger.debug("{0}: {1!s}".format(self.label, result.registers))
+            value = None
+            if self.converter and Converters.Converters[self.converter]['in']:
+                try:
+                    value = Converters.Converters[self.converter]['in'](result.registers)
+                    logger.debug("{0}: converted: {1!s}".format(self.label, value))
+                except Exception as e:
+                    raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
+            else:
+                value = result.registers
+            if self.publishTopic:
+                pubQueue.put(MqttProcessor.PublishItem(self.publishTopic, str(value)))
+            self.lastContact = datetime.datetime.now()
+    
+    def onMessage(self, value):
+        self.writeRequestValue = value
+
+
+class CoilDatapoint(AbstractModbusDatapoint):
+    def __init__(self, label=None, unit=None, address=None, scanRate=None, publishTopic=None, subscribeTopic=None,
+                 feedbackTopic=None):
+        super().__init__(label, unit, address, 1, scanRate, None)
+        self.argList = ['label', 'unit','address','scanRate','publishTopic', 'subscribeTopic', 'feedbackTopic']
+        self.publishTopic = publishTopic
+        self.subscribeTopic = subscribeTopic
+        self.feedbackTopic = feedbackTopic
+        self.writeRequestValue = None
+        self.type = 'coil'
+
+    def __str__(self):
+        return ("{0}, {1}: unit: {2},  address: {3}, scanRate: {4}, "
+                "enqueued: {5}, lastContact: {6}, errorCount: {7}, processCount: {8}, "
+                "publishTopic: {9}, subscribeTopic: {10}, feedbackTopic: {11}"
+                .format(self.type, self.label, self.unit, self.address,
+                        self.scanRate, self.enqueued, self.lastContact,
+                        self.errorCount, self.processCount, 
+                        self.publishTopic, self.subscribeTopic, self.feedbackTopic))
+
+    def onMessage(self, value):
+        self.writeRequestValue = value.decode()
+
+    def process(self, client, pubQueue):
+        logger = logging.getLogger('CoilDatapoint')
+        if self.writeRequestValue:
+            # perform write operation
+            logger.debug("Coil, perform write operation")
+            self.processCount += 1
+            logger.debug("{0}: raw: {1!s}".format(self.label, self.writeRequestValue))
+            value=None
+            if self.writeRequestValue in ['true', 'True', 'yes', 'Yes', 'On', 'on']:
+                value = True
+            elif self.writeRequestValue in ['false', 'False', 'no', 'No', 'Off', 'off']:
+                value = False
+            else:
+                self.writeRequestValue = None
+                raise DatapointException('illegal value {0!s} for coil write'.format(self.writeRequestValue))
+            result = client.write_coil(address=self.address,
+                                       unit=self.unit,
+                                       value=value)
+            logger.debug("Write result: {0!s}".format(result))                
+            self.writeRequestValue = None
+        else:
+            # perform read operation
+            logger.debug("Coil, perform read operation")
+            self.processCount += 1
+            result = client.read_coils(address=self.address, 
+                                       unit=self.unit)
+            if type(result) in [ExceptionResponse, ModbusIOException]:
+                self.errorCount += 1
+                raise DatapointException(result)
+            logger.debug("{0}: {1!s}".format(self.label, result.getBit(0)))
+            value = result.getBit(0)
+            if self.publishTopic:
+                pubQueue.put(MqttProcessor.PublishItem(self.publishTopic, str(value)))
+            self.lastContact = datetime.datetime.now()
+
+
+class ReadOnlyDatapoint(AbstractModbusDatapoint):
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, updateOnly=None, publishTopic=None, converter=None):
+        super().__init__(label, unit, address, count, scanRate, converter)
+        self.argList = self.argList + ['updateOnly', 'publishTopic']
+        self.updateOnly = updateOnly
+        self.lastValue = None
+        self.publishTopic = publishTopic
+
+    def __str__(self):
+        return ("[{0!s}, updateOnly: {1}, publishTopic: {2}, lastValue: {3!s}"
+                .format(super().__str__(), self.updateOnly, self.publishTopic,
+                        self.lastValue))
+
+
+
+class InputRegisterDatapoint(ReadOnlyDatapoint):
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, updateOnly=None, 
+                 publishTopic=None, converter=None):
+        super().__init__(label, unit, address, count, scanRate, updateOnly, publishTopic, converter)
+        self.type = 'input register'
+
+    def process(self, client, pubQueue):
+        logger = logging.getLogger('InputRegisterDatapoint')
+        # perform read operation
+        logger.debug("Input register, perform read operation")
+        self.processCount += 1
+        result = client.read_input_registers(address=self.address,
+                                             count=self.count,
+                                             unit=self.unit)
+        if type(result) in [ExceptionResponse, ModbusIOException]:
+            self.errorCount += 1
+            raise DatapointException(result)
+        if not self.updateOnly or (result.registers != self.lastValue):
+            self.lastValue = result.registers
+            logger.debug("{0}: raw: {1!s}".format(self.label, result.registers))
+            value = None
+            if self.converter and Converters.Converters[self.converter]['in']:
+                try:
+                    value = Converters.Converters[self.converter]['in'](result.registers)
+                    logger.debug("{0}: converted: {1!s}".format(self.label, value))
+                except Exception as e:
+                    raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
+            else:
+                value = result.registers
+            if self.publishTopic:
+                pubQueue.put(MqttProcessor.PublishItem(self.publishTopic, str(value)))
+        self.lastContact = datetime.datetime.now()
+
+
+class DiscreteInputDatapoint(ReadOnlyDatapoint):
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, updateOnly=None, 
+                 publishTopic=None, converter=None, bitCount=8):
+        super().__init__(label, unit, address, count, scanRate, updateOnly, publishTopic, converter)
+        self.argList = self.argList + ['bitCount']
+        self.type = 'discrete input'
+        self.bitCount = bitCount
+        self.lastValues = [None] * self.bitCount
+
+    def __str__(self):
+        return ("[{0!s}, bitCount: {1}"
+                .format(super().__str__(), self.bitCount))
+
+    def process(self, client, pubQueue):
+        logger = logging.getLogger('DiscreteInputDatapoint')
+        # perform read operation
+        logger.debug("Discrete input, perform read operation")
+        self.processCount += 1
+        result = client.read_discrete_inputs(address=self.address,
+                                             count=self.count,
+                                             unit=self.unit)
+        if type(result) in [ExceptionResponse, ModbusIOException]:
+            self.errorCount += 1
+            raise DatapointException(result)
+        logger.debug("{0}: raw: {1!s}".format(self.label, result.bits))
+        for i in range(self.bitCount):
+            if not self.updateOnly or (result.getBit(i) != self.lastValues[i]):
+                self.lastValues[i] = result.getBit(i)
+                logger.debug("{0}, {1}: changed: {2!s}".format(self.label, i, result.getBit(i)))
+                if self.publishTopic:
+                    pubQueue.put(MqttProcessor.PublishItem("{0}/{1}".format(self.publishTopic, i), str(result.getBit(i))))
+        self.lastContact = datetime.datetime.now()
+
+
+
+
+class JsonifyEncoder(json.JSONEncoder):
+    def default(self, o):
+        res = None
+        try:
+            res = o.jsonify()
+        except (TypeError, AttributeError):
+            if type(o) == datetime.timedelta:
+                res = o.total_seconds()
+            else:
+                res = super().default(o)
+        return res
+
+def datapointObjectHook(j):
+    if type(j) == dict and 'type' in j and 'args' in j:
+        klass = eval(j['type'])
+        o = klass(**j['args'])
+        return o
+    else:
+        return j
+
+def saveRegisterList(registerList, registerListFile):
+    js = json.dumps(registerList, cls=JsonifyEncoder, sort_keys=True, indent=4)
+    with open(registerListFile, 'w') as f:
+        f.write(js)
+    
+def loadRegisterList(registerListFile):
+    with open(registerListFile, 'r') as f:
+        js = f.read()
+        registerList = json.loads(js, object_hook=datapointObjectHook)
+        return registerList
+

src/ScanRateConsideringQueueFeeder.py

@@ -0,0 +1,36 @@
+import threading
+import datetime
+from NotificationForwarder import AbstractNotificationReceiver
+import logging
+
+class ScanRateConsideringQueueFeeder(threading.Thread, AbstractNotificationReceiver):
+    def __init__(self, config, registers, queue):
+        super().__init__()
+        self.config = config
+        self.registers = registers
+        self.queue = queue
+        self.delayEvent = threading.Event()
+        self.daemon = True
+        self.logger = logging.getLogger('ScanRateConsideringQueueFeeder')
+
+    def getMinimalScanrate(self):
+            return min([r.scanRate.total_seconds() for r in self.registers if r.scanRate])
+
+    def receiveNotification(self, arg):
+        self.logger.info("ScanRateConsideringQueueFeeder:registersChanged")
+        self.delay = self.getMinimalScanrate()
+
+    def run(self):
+        self.delay = self.getMinimalScanrate()
+        while True:
+            registersToBeHandled = [
+                r for r in self.registers if ((not r.enqueued) and
+                                              (r.scanRate) and
+                                              ((not r.lastContact) or 
+                                               (r.lastContact + r.scanRate < datetime.datetime.now())))
+            ]
+            registersToBeHandled.sort(key=lambda x : x.scanRate)
+            for r in registersToBeHandled:
+                self.queue.put(r)
+                r.enqueued = True
+            self.delayEvent.wait(self.delay)

src/initialRegisterFile.py

@@ -0,0 +1,15 @@
+import datetime
+import RegisterDatapoint
+import pickle
+
+
+datapoints = [
+    RegisterDatapoint.InputRegisterDatapoint('Temperature', 5, 0x0001, 1, datetime.timedelta(seconds=1.0), False, 'Pub/Temperature'),
+    RegisterDatapoint.InputRegisterDatapoint('Humidity', 5, 0x0002, 1, datetime.timedelta(seconds=1.0), True, 'Pub/Humidity'),
+    RegisterDatapoint.DiscreteInputDatapoint('Switches', 4, 0x0000, 1, datetime.timedelta(seconds=1.0), True, 'Pub/Switches'),
+]
+
+
+with open('registers.pkl', 'wb') as f:
+    pickle.dump(datapoints, f)
+

src/loadRegisterFile.py

@@ -0,0 +1,6 @@
+import RegisterDatapoint
+
+registers = RegisterDatapoint.loadRegisterList('registers.json')
+
+for r in registers:
+    print("{0!s}".format(r))

src/master.py

@@ -0,0 +1,57 @@
+import CmdServer
+import MqttProcessor
+import CommunicationProcessor
+import MyPriorityQueue
+from queue import Queue
+import NotificationForwarder
+import Config
+import ScanRateConsideringQueueFeeder
+import datetime
+import RegisterDatapoint
+import pickle
+import logging
+
+
+
+
+if __name__ == "__main__":
+    config = Config.Config()
+
+    logger = logging.getLogger()
+    logger.setLevel(logging.DEBUG)
+    fh = logging.FileHandler(config.logFile)
+    fh.setLevel(logging.DEBUG)
+    ch = logging.StreamHandler()
+    ch.setLevel(logging.ERROR)
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    fh.setFormatter(formatter)
+    ch.setFormatter(formatter)
+    logger.addHandler(fh)
+    logger.addHandler(ch)
+
+    queue = MyPriorityQueue.MyPriorityQueue()
+    pubQueue = Queue()
+    nf = NotificationForwarder.NotificationForwarder()
+    logger.debug('infrastructure prepared')
+
+
+    datapoints = RegisterDatapoint.loadRegisterList(config.registerFile)
+    logger.debug('datapoints read')
+
+    cp = CommunicationProcessor.CommunicationProcessor(config, queue, pubQueue)
+    cp.start()
+    logger.debug('CommunicationProcessor started')
+
+    mp = MqttProcessor.MqttProcessor(config, datapoints, queue, pubQueue)
+    nf.register(mp)
+    mp.start()
+    logger.debug('MqttProcessor started')
+
+    qf = ScanRateConsideringQueueFeeder.ScanRateConsideringQueueFeeder(config, datapoints, queue)
+    nf.register(qf)
+    qf.start()
+    logger.debug('ScanRateConsideringQueueFeeder started')
+
+    cs = CmdServer.CmdServer(config, nf, datapoints)
+    cs.start()
+    logger.debug('CmdServer started')

src/registers.json

@@ -0,0 +1,122 @@
+[
+    {
+        "args": {
+            "address": 1,
+            "converter": "dht20TOFloat",
+            "count": 1,
+            "label": "Temperature",
+            "publishTopic": "Pub/Temperature",
+            "scanRate": 10.0,
+            "unit": 5,
+            "updateOnly": false
+        },
+        "type": "InputRegisterDatapoint"
+    },
+    {
+        "args": {
+            "address": 2,
+            "converter": "dht20TOFloat",
+            "count": 1,
+            "label": "Humidity",
+            "publishTopic": "Pub/Humidity",
+            "scanRate": null,
+            "unit": 5,
+            "updateOnly": false
+        },
+        "type": "InputRegisterDatapoint"
+    },
+    {
+        "args": {
+            "address": 0,
+            "bitCount": 8,
+            "converter": null,
+            "count": 1,
+            "label": "Switches",
+            "publishTopic": "Pub/Switches",
+            "scanRate": null,
+            "unit": 4,
+            "updateOnly": false
+        },
+        "type": "DiscreteInputDatapoint"
+    },
+    {
+        "args": {
+            "address": 40010,
+            "converter": "uint32",
+            "count": 2,
+            "feedbackTopic": "FB/Counter1",
+            "label": "Counter1",
+            "publishTopic": "Pub/Counter1",
+            "scanRate": 1.0,
+            "subscribeTopic": "Sub/Counter1",
+            "unit": 4
+        },
+        "type": "HoldingRegisterDatapoint"
+    },
+    {
+        "args": {
+            "address": 40012,
+            "converter": "uint32",
+            "count": 2,
+            "feedbackTopic": "FB/Counter2",
+            "label": "Counter2",
+            "publishTopic": "Pub/Counter2",
+            "scanRate": null,
+            "subscribeTopic": "Pub/Counter2",
+            "unit": 4
+        },
+        "type": "HoldingRegisterDatapoint"
+    },
+    {
+        "args": {
+            "address": 40014,
+            "converter": "uint32",
+            "count": 2,
+            "feedbackTopic": "FB/Counter3",
+            "label": "Counter3",
+            "publishTopic": "Pub/Counter3",
+            "scanRate": null,
+            "subscribeTopic": "FB/Counter3",
+            "unit": 4
+        },
+        "type": "HoldingRegisterDatapoint"
+    },
+    {
+        "args": {
+            "address": 40016,
+            "converter": "uint32",
+            "count": 2,
+            "feedbackTopic": "FB/Counter4",
+            "label": "Counter4",
+            "publishTopic": "Pub/Counter4",
+            "scanRate": 1.0,
+            "subscribeTopic": "Sub/Counter4",
+            "unit": 4
+        },
+        "type": "HoldingRegisterDatapoint"
+    },
+    {
+        "args": {
+            "address": 0,
+            "feedbackTopic": "FB/Coil1",
+            "label": "Coil1",
+            "publishTopic": "Pub/Coil1",
+            "scanRate": 1.0,
+            "subscribeTopic": "Sub/Coil1",
+            "unit": 4
+        },
+        "type": "CoilDatapoint"
+    },
+    {
+        "args": {
+            "address": 1,
+            "feedbackTopic": "FB/Coil2",
+            "label": "Coil2",
+            "publishTopic": "Pub/Coil2",
+            "scanRate": 1.0,
+            "subscribeTopic": "Sub/Coil2",
+            "unit": 4
+        },
+        "type": "CoilDatapoint"
+    }
+]

src/updateRegisterFile.py

@@ -0,0 +1,24 @@
+import datetime
+import RegisterDatapoint
+import pickle
+import json
+
+
+with open('registers.pkl', 'rb') as f:
+    datapoints = pickle.load(f)
+
+newDatapoints = []
+for dp in datapoints:
+    ndp = type(dp)()
+    for k,v in dp.__dict__.items():
+        if k != 'logger':
+            ndp.__dict__[k] = v
+    newDatapoints.append(ndp)
+
+
+
+js = json.dumps(newDatapoints, cls=RegisterDatapoint.JsonifyEncoder, sort_keys=True, indent=4)
+print(js)
+    
+
+RegisterDatapoint.saveRegisterList(newDatapoints, 'registers.json')