just a test, not proceeded, switched to forwarding via ssh

ENV

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

pyserialext/Makefile

@@ -0,0 +1,14 @@
+CFLAGS=
+LIBS=-lwiringPi
+
+writec.so:	writec.o 
+	$(LD) -shared $(LIBS) -o $@ $^
+
+writec.o:	writec.c
+
+.c.o:
+	$(CC) $(CFLAGS) -c $<
+
+.PHONY: clean
+clean:
+	-rm -f *.so *.o

pyserialext/RS485Ext.py

@@ -0,0 +1,16 @@
+import serial.rs485
+import serial.serialutil
+import ctypes
+
+
+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
+

pyserialext/writec.c

@@ -0,0 +1,26 @@
+#include <unistd.h>
+#include <sys/ioctl.h>
+#include <stdint.h>
+#include <wiringPi.h>
+
+
+const uint8_t DE_PIN = 0;
+
+int init() {
+  wiringPiSetup();
+  pinMode(DE_PIN, OUTPUT);
+  digitalWrite(DE_PIN, LOW);
+}
+
+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;
+}
+

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/test8.py

@@ -1,105 +0,0 @@
-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

@@ -1,14 +0,0 @@
-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

@@ -4,9 +4,9 @@ import cmd
 import re
 import io
 import datetime
+import pickle
+import sys
 import RegisterDatapoint
-import logging
-import Converters
 
 class CmdInterpreterException(ValueError): pass
 
@@ -30,15 +30,6 @@ class CmdInterpreter(cmd.Cmd):
         self.prompt = "test8> "
         self.intro = "test8 admin interface"
         self.splitterRe = re.compile('\s+')
-        self.logger = logging.getLogger('CmdInterpreter')
-
-    def onecmd(self, line):
-        try:
-            return super().onecmd(line)
-        except Exception as e:
-            msg = 'Caught exception in cmd "{0}": {1!s}'.format(line, e)
-            self.__println(msg)
-            self.logger.error(msg)
 
     def __print(self, text):
         self.stdout.write(text)
@@ -47,14 +38,20 @@ class CmdInterpreter(cmd.Cmd):
         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 __listConverterNames(self):
+    def do_quit(self, arg):
-        return [ name for name in Converters.Converters ]
+        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)
+            (label, unit, address, count, 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))
@@ -63,7 +60,6 @@ class CmdInterpreter(cmd.Cmd):
             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
@@ -71,30 +67,33 @@ class CmdInterpreter(cmd.Cmd):
                 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=label, 
+            if scanrate == 0:
-                                                           unit=unit, 
+                if readTopic:
-                                                           address=address, 
+                    raise CmdInterpreterException('readTopic must not be set when scanRate is zero')
-                                                           count=count, 
+                if not writeTopic:
-                                                           scanRate=datetime.timedelta(seconds=scanrate), 
+                    raise CmdInterpreterException('writeTopic must be set when scanRate is zero')
-                                                           publishTopic=readTopic, 
+                if not feedbackTopic:
-                                                           subscribe=writeTopic, 
+                    raise CmdInterpreterException('feedbackTopic must be set when scanRate is zero')
-                                                           feedbackTopic=feedbackTopic, 
+            else:
-                                                           converter=converter)
+                if not readTopic:
+                    raise CmdInterpreterException('readTopic must be set when scanRate is zero')
+                if writeTopic:
+                    raise CmdInterpreterException('writeTopic must not be set when scanRate is zero')
+                if feedbackTopic:
+                    raise CmdInterpreterException('feedbackTopic must not be set when scanRate is zero')
+            r = RegisterDatapoint.HoldingRegisterDatapoint(label, unit, address, count, 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_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("Usage: add <Label> <Unit> <Address> <Count> <ScanRate>")
-        self.__println("              <ReadTopic> <WriteTopic> <FeedbackTopic>")
+        self.__println("           <ReadTopic> <WriteTopic> <FeedbackTopic>")
-        self.__println("              <Converter>")
         self.__println("Adds a holding register")
         self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
         self.__println("---------------------------------------------------------------------")
@@ -108,59 +107,15 @@ class CmdInterpreter(cmd.Cmd):
         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, one of {0}".format(', '.join(self.__listConverterNames())))
+        self.__println("")
-
+        self.__println("For read items the <ScanRate> must be non-zero, a <ReadTopic> must be set and")
-    
+        self.__println("<WriteTopic> and <FeedbackTopic> must be <None>.")
-    def do_add_coil(self, arg):
+        self.__println("For write items the <ScanRate> must be zero, <ReadTopic> must be <None> and ")
-        try:
+        self.__println("<WriteTopic> and <FeedbackTopic> must be set.")
-            (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=label, 
-                                                unit=unit, 
-                                                address=address, 
-                                                scanRate=datetime.timedelta(seconds=scanrate), 
-                                                publishTopic=readTopic, 
-                                                subscribeTopic=writeTopic, 
-                                                feedbackTopic=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)
+            (label, unit, address, count, scanrate, updateOnly, readTopic) = self.splitterRe.split(arg)
             self.__println("Label:         {0}".format(label))
             self.__println("Unit:          {0}".format(unit))
             self.__println("Address:       {0}".format(address))
@@ -168,12 +123,9 @@ class CmdInterpreter(cmd.Cmd):
             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']:
@@ -184,20 +136,16 @@ class CmdInterpreter(cmd.Cmd):
             address = parseIntArbitraryBase(address)
             count = parseIntArbitraryBase(count)
             scanrate = float(scanrate)
-            r = RegisterDatapoint.InputRegisterDatapoint(label=label, 
+            if scanrate == 0.0:
-                                                         unit=unit, 
+                raise CmdInterpreterException('scanRate must not be zero')
-                                                         address=address, 
+            r = RegisterDatapoint.InputRegisterDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), updateOnly, readTopic)
-                                                         count=count, scanRate=datetime.timedelta(seconds=scanrate), 
-                                                         updateOnly=updateOnly, 
-                                                         publishTopic=readTopic, 
-                                                         converter=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("Usage: add <Label> <Unit> <Address> <Count> <ScanRate>")
-        self.__println("              <UpdateOnly> <ReadTopic> <Converter>")
+        self.__println("           <UpdateOnly> <ReadTopic>")
         self.__println("Adds an input register")
         self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
         self.__println("---------------------------------------------------------------------")
@@ -208,11 +156,10 @@ class CmdInterpreter(cmd.Cmd):
         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, one of {0}".format(', '.join(self.__listConverterNames())))
 
     def do_add_di(self, arg):
         try:
-            (label, unit, address, count, scanrate, updateOnly, readTopic, bitCount) = self.splitterRe.split(arg)
+            (label, unit, address, count, scanrate, updateOnly, readTopic) = self.splitterRe.split(arg)
             self.__println("Label:         {0}".format(label))
             self.__println("Unit:          {0}".format(unit))
             self.__println("Address:       {0}".format(address))
@@ -220,7 +167,6 @@ class CmdInterpreter(cmd.Cmd):
             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
@@ -234,22 +180,16 @@ class CmdInterpreter(cmd.Cmd):
             address = parseIntArbitraryBase(address)
             count = parseIntArbitraryBase(count)
             scanrate = float(scanrate)
-            bitCount = int(bitCount)
+            if scanrate == 0.0:
-            r = RegisterDatapoint.DiscreteInputDatapoint(label=label, 
+                raise CmdInterpreterException('scanRate must not be zero')
-                                                         unit=unit, 
+            r = RegisterDatapoint.DiscreteInputDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), updateOnly, readTopic)
-                                                         address=address, 
-                                                         count=count, 
-                                                         scanRate=datetime.timedelta(seconds=scanrate), 
-                                                         updateOnly=updateOnly, 
-                                                         publishTopic=readTopic, 
-                                                         bitCount=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("Usage: add <Label> <Unit> <Address> <Count> <ScanRate>")
-        self.__println("              <UpdateOnly> <ReadTopic> <bitCount>")
+        self.__println("           <UpdateOnly> <ReadTopic>")
         self.__println("Adds a discrete input")
         self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
         self.__println("---------------------------------------------------------------------")
@@ -260,7 +200,6 @@ class CmdInterpreter(cmd.Cmd):
         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):
@@ -274,8 +213,7 @@ class CmdInterpreter(cmd.Cmd):
     def do_reset(self, arg):
         for r in self.registers:
             r.errorCount = 0
-            r.writeCount = 0
+            r.processCount = 0
-            r.readCount = 0
     
     def help_reset(self):
         self.__println("Usage: reset")
@@ -284,20 +222,21 @@ class CmdInterpreter(cmd.Cmd):
 
     def do_stats(self, arg):
         for i, r in enumerate(self.registers):
-            processCount = r.readCount + r.writeCount
+            if r.processCount == 0:
-            if processCount == 0:
                 ratio = -1
             else:
-                ratio = float(r.errorCount) / float(processCount)
+                ratio = float(r.errorCount) / float(r.processCount)
-            self.__println("#{0:2d}: {1:15s} ({2:2d}, {3:5d}), rc: {4:7d}, wc: {5:7d}, pc: {6:7d}, ec: {7:7d}, q: {8:1.4f}"
+            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.readCount, r.writeCount,
+                           .format(i, r.label, r.unit, r.address,
-                                   processCount, r.errorCount, ratio))
+                                   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:
@@ -349,6 +288,9 @@ class CmdInterpreter(cmd.Cmd):
         self.__println("                                      considered)")
         self.__println("<value>                    New value")
 
+
+
+
     def do_del(self, arg):
         try:
             i = int(arg)
@@ -364,53 +306,46 @@ class CmdInterpreter(cmd.Cmd):
         self.__println("Be aware: indexes have been changed, rerun list before removing the next item.")
         self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
 
-    def __notify(self):
-        self.notifier.notify()
-
-    def do_notify(self, arg):
-        self.__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 __save(self):
-        RegisterDatapoint.saveRegisterList(self.registers, self.config.registerFile)
-
     def do_save(self, arg):
-        self.__save()
+        with open(self.config.registerFile, 'wb') as f:
+            pickle.dump(self.registers, f)
 
     def help_save(self):
         self.__println("Usage: save")
         self.__println("Saves a modified register list into the register file.")
 
     def do_load(self, arg):
+        registers = None
+        with open(self.config.registerFile, 'rb') as f:
+            registers = pickle.load(f)
         try:
-            registers = RegisterDatapoint.loadRegisterList(self.config.registerFile)
+            RegisterDatapoint.checkRegisterList(registers)
             self.registers = registers
-        except Exception as e:
+        except ValueError 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.")
 
+    #def do_shutdown(self, arg):
+    #    sys.exit()
+
+    #def help_shutdown(self):
+    #    self.__println("Usage: shutdown")
+    #    self.__println("Shuts down the application")
+
 
 
 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")
+            print("Cmd handle terminated")
         except ConnectionAbortedError as e:
-            logger.info("Cmd handle externally interrupted")
+            print("Cmd handle externally interrupted")
 
 class MyThreadingTCPServer(socketserver.ThreadingTCPServer):
     def __init__(self, host, handler, userData):
@@ -428,9 +363,9 @@ class CmdServer(threading.Thread):
         super().__init__()
         self.config = config
         self.server = MyThreadingTCPServer((config.cmdAddress, config.cmdPort), CmdHandle, MyCmdUserData(config, notifier, registers))
-        # self.daemon = True
+        self.daemon = True
 
-    def run(self):
+    def start(self):
         self.server.serve_forever()
 
 

src/CommunicationProcessor.py

@@ -3,8 +3,7 @@ import datetime
 # import RS485Ext
 import RegisterDatapoint
 from pymodbus.client.sync import ModbusSerialClient
-# import wiringpi
+import wiringpi
-import Pins
 import MyRS485
 import time
 import logging
@@ -18,9 +17,9 @@ class CommunicationProcessor(threading.Thread):
         self.config = config
         self.queue = queue
         self.pubQueue = pubQueue
-        # wiringpi.wiringPiSetup()
+        wiringpi.wiringPiSetup()
-        # wiringpi.pinMode(ERROR_PIN, wiringpi.OUTPUT)
+        wiringpi.pinMode(ERROR_PIN, wiringpi.OUTPUT)
-        # self.daemon = True
+        self.daemon = True
         if self.config.modbusDebug:
             logging.getLogger('pymodbus').setLevel(logging.DEBUG)
         else:
@@ -42,14 +41,12 @@ class CommunicationProcessor(threading.Thread):
         while True:
             r = self.queue.get()
             try:
-                # wiringpi.digitalWrite(ERROR_PIN, wiringpi.LOW)
+                wiringpi.digitalWrite(ERROR_PIN, wiringpi.LOW)
-                Pins.pinsWrite('ERROR', False)
                 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)
+                wiringpi.digitalWrite(ERROR_PIN, wiringpi.HIGH)
-                Pins.pinsWrite('ERROR', True)
                 self.logger.error("ERROR when processing '{0}': {1!s}".format(r.label, e))
                 if client.socket is None:
                     self.logger.error("renew socket")

src/Config.py

@@ -9,9 +9,7 @@ class Config(object):
         self.mqttPassword = ''
         self.cmdAddress = '127.0.0.1'
         self.cmdPort = 9999
-        self.registerFile = 'registers.json'
+        self.registerFile = 'registers.pkl'
         self.serialPort = '/dev/ttyAMA0'
         self.serialBaudRate = 9600
         self.interCommDelay = 0.025
-        self.heartbeatTopic = 'IoT/Heartbeat/Modbus2'
-        self.heartbeatPeriod = 10.0

src/Converters.py

@@ -1,18 +0,0 @@
-# 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/Heartbeat.py

@@ -1,20 +0,0 @@
-import threading
-import MqttProcessor
-import logging
-import time
-
-class Heartbeat(threading.Thread):
-    def __init__(self, config, pubQueue):
-        super().__init__()
-        self.config = config
-        self.pubQueue = pubQueue
-        # self.daemon = True
-        self.logger = logging.getLogger('Heartbeat')
-
-    def run(self):
-        cnt = 0
-        while True:
-            cnt += 1
-            pubItem = MqttProcessor.PublishItem(self.config.heartbeatTopic, str(cnt))
-            self.pubQueue.put(pubItem)
-            time.sleep(self.config.heartbeatPeriod)

src/MqttProcessor.py

@@ -2,16 +2,13 @@ import threading
 import paho.mqtt.client as mqtt
 from NotificationForwarder import AbstractNotificationReceiver
 import logging
-import Pins
+
 
 class PublishItem(object):
     def __init__(self, topic, payload):
         self.topic = topic
         self.payload = payload
 
-    def __str__(self):
-        return 'Topic: {0}, Payload: {1}'.format(self.topic, self.payload)
-
 def mqttOnConnectCallback(client, userdata, flags, rc):
     userdata.onConnect()
 
@@ -31,13 +28,13 @@ class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
         self.client = mqtt.Client(userdata=self)
         self.subscriptions = []
         self.topicRegisterMap ={}
-        # self.daemon = True
+        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]
+        subscribeTopics = [ r.subscribeTopic for r in self.registers if r.subscribeTopic]
         self.logger.debug("Topics: {0!s}".format(subscribeTopics))
 
         for subscribeTopic in subscribeTopics:
@@ -52,7 +49,7 @@ class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
                 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 }
+        self.topicRegisterMap = { r.subscribeTopic: r for r in self.registers if r.subscribeTopic }
 
     def receiveNotification(self, arg):
         self.logger.info("MqttProcessor:registersChanged")
@@ -70,9 +67,7 @@ class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
         while True:
             pubItem = self.pubQueue.get()
             if isinstance(pubItem, PublishItem):
-                self.logger.debug('Publishing {0!s}'.format(pubItem))
                 self.client.publish(pubItem.topic, pubItem.payload)
-                Pins.pinsWrite('MSG', False)
             else:
                 self.logger.error("Invalid object in publish queue")
 
@@ -85,7 +80,6 @@ class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
         self.logger.error("Disconnected from MQTT broker: {0}".format(rc))
 
     def onMessage(self, topic, payload):
-        Pins.pinsWrite('MSG', True)
         # print("MqttProcessor.onMessage")
         r = self.topicRegisterMap[topic]
         self.logger.debug("{0}: {1!s} -> {2!s}".format(topic, payload, r))

src/MyRS485.py

@@ -1,6 +1,5 @@
 import serial
-# import wiringpi
+import wiringpi
-import Pins
 import array
 import fcntl
 import termios
@@ -10,18 +9,16 @@ DE_PIN = 0
 class MyRS485(serial.Serial):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        # wiringpi.wiringPiSetup()
+        wiringpi.wiringPiSetup()
-        # wiringpi.pinMode(DE_PIN, wiringpi.OUTPUT)
+        wiringpi.pinMode(DE_PIN, wiringpi.OUTPUT)
         self.buf = array.array('h', [0])
 
     def write(self, b):
-        # wiringpi.digitalWrite(DE_PIN, wiringpi.HIGH)
+        wiringpi.digitalWrite(DE_PIN, wiringpi.HIGH)
-        Pins.pinsWrite('DE', True)
         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)
+        wiringpi.digitalWrite(DE_PIN, wiringpi.LOW)
-        Pins.pinsWrite('DE', False)
 

src/MySSLThreadingTCPServer.py

@@ -0,0 +1,28 @@
+from socketserver import TCPServer, ThreadingMixIn
+import ssl
+
+class MySSL_TCPServer(TCPServer):
+    def __init__(self,
+                 server_address,
+                 RequestHandlerClass,
+                 certfile,
+                 keyfile,
+                 ssl_version=ssl.PROTOCOL_TLSv1,
+                 bind_and_activate=True):
+        TCPServer.__init__(self, server_address, RequestHandlerClass, bind_and_activate)
+        self.certfile = certfile
+        self.keyfile = keyfile
+        self.ssl_version = ssl_version
+
+    def get_request(self):
+        newsocket, fromaddr = self.socket.accept()
+        connstream = ssl.wrap_socket(newsocket,
+                                 server_side=True,
+                                 certfile = self.certfile,
+                                 keyfile = self.keyfile,
+                                 ssl_version = self.ssl_version)
+        return connstream, fromaddr
+
+class MySSL_ThreadingTCPServer(ThreadingMixIn, MySSL_TCPServer): pass
+
+  

src/Pins.py

@@ -1,24 +0,0 @@
-import wiringpi
-
-
-PINS = {
-    'DE': 0,
-    'ERROR': 29,
-    'MSG': 28
-}
-
-
-
-def pinsInit():
-    wiringpi.wiringPiSetup()
-    for pin in PINS.values():
-        wiringpi.pinMode(pin, wiringpi.OUTPUT)
-
-
-def pinsWrite(pinName, v):
-    if v:
-        pinState = wiringpi.HIGH
-    else:
-        pinState = wiringpi.LOW
-    wiringpi.digitalWrite(PINS[pinName], pinState)
-

src/RegisterDatapoint.py

@@ -3,29 +3,23 @@ from pymodbus.pdu import ExceptionResponse
 from pymodbus.exceptions import ModbusIOException
 import MqttProcessor
 import logging
-import json
+import pickle
-import Converters
+
 
 class DatapointException(Exception): pass
 
 class AbstractModbusDatapoint(object):
-    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, converter=None):
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None):
-        self.argList = ['label', 'unit', 'address', 'count', 'scanRate', 'converter']
         self.label = label
         self.unit = unit
         self.address = address
         self.count = count
-        self.converter = converter
+        self.scanRate = scanRate
-        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.readCount = 0
+        self.processCount = 0
-        self.writeCount = 0
         if self.scanRate:
             self.priority = 1
         else:
@@ -33,27 +27,19 @@ class AbstractModbusDatapoint(object):
 
     def __str__(self):
         return ("{0}, {1}: unit: {2},  address: {3}, count: {4}, scanRate: {5}, "
-                "enqueued: {6}, lastContact: {7}, errorCount: {8}, readCount: {9}, "
+                "enqueued: {6}, lastContact: {7}, errorCount: {8}, processCount: {9}"
-                "writeCount: {10}, converter: {11}"
                 .format(self.type, self.label, self.unit, self.address, self.count,
                         self.scanRate, self.enqueued, self.lastContact,
-                        self.errorCount, self.readCount, self.writeCount, self.converter))
+                        self.errorCount, self.processCount))
-
-    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):
+                 publishTopic=None, subscribeTopic=None, feedbackTopic=None):
-        super().__init__(label, unit, address, count, scanRate, converter)
+        super().__init__(label, unit, address, count, scanRate)
-        self.argList = self.argList + ['publishTopic', 'subscribeTopic', 'feedbackTopic']
         self.publishTopic = publishTopic
         self.subscribeTopic = subscribeTopic
         self.feedbackTopic = feedbackTopic
@@ -67,117 +53,32 @@ class HoldingRegisterDatapoint(AbstractModbusDatapoint):
                         self.writeRequestValue))
 
     def process(self, client, pubQueue):
-        logger = logging.getLogger('HoldingRegisterDatapoint')
+        self.logger = logging.getLogger('HoldingRegisterDatapoint')
         if self.writeRequestValue:
             # perform write operation
-            logger.debug("Holding register, perform write operation")
+            self.logger.debug("Holding register, perform write operation")
-            self.writeCount += 1
-            values = None
-            logger.debug("{0}: raw: {1!s}".format(self.label, self.writeRequestValue))
-            try:
-                if self.converter and Converters.Converters[self.converter]['out']:
-                    values = Converters.Converters[self.converter]['out'](self.writeRequestValue)
-                    logger.debug("{0}: converted: {1!s}".format(self.label, values))
-                else:
-                    values = [int(self.writeRequestValue)]
-            except Exception as e:
-                raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
-            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.logger.debug("Holding register, perform read operation")
-            self.readCount += 1
+            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))
+            self.logger.debug("{0}: {1!s}".format(self.label, result.registers))
-            value = None
+            pubQueue.put(MqttProcessor.PublishItem(self.publishTopic, str(result.registers)))
-            try:
-                if self.converter and Converters.Converters[self.converter]['in']:
-                    value = Converters.Converters[self.converter]['in'](result.registers)
-                    logger.debug("{0}: converted: {1!s}".format(self.label, value))
-                else:
-                    value = result.registers
-            except Exception as e:
-                raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
-            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}, readCount: {8}, "
-                "writeCount: {9}, publishTopic: {10}, subscribeTopic: {11}, feedbackTopic: {12}"
-                .format(self.type, self.label, self.unit, self.address,
-                        self.scanRate, self.enqueued, self.lastContact,
-                        self.errorCount, self.readCount, self.writeCount, 
-                        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.writeCount += 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.readCount += 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):
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, updateOnly=None, publishTopic=None):
-        super().__init__(label, unit, address, count, scanRate, converter)
+        super().__init__(label, unit, address, count, scanRate)
-        self.argList = self.argList + ['updateOnly', 'publishTopic']
         self.updateOnly = updateOnly
         self.lastValue = None
         self.publishTopic = publishTopic
@@ -190,16 +91,15 @@ class ReadOnlyDatapoint(AbstractModbusDatapoint):
 
 
 class InputRegisterDatapoint(ReadOnlyDatapoint):
-    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, updateOnly=None, 
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, updateOnly=None, publishTopic=None):
-                 publishTopic=None, converter=None):
+        super().__init__(label, unit, address, count, scanRate, updateOnly, publishTopic)
-        super().__init__(label, unit, address, count, scanRate, updateOnly, publishTopic, converter)
         self.type = 'input register'
 
     def process(self, client, pubQueue):
-        logger = logging.getLogger('InputRegisterDatapoint')
+        self.logger = logging.getLogger('InputRegisterDatapoint')
         # perform read operation
-        logger.debug("Input register, perform read operation")
+        self.logger.debug("Input register, perform read operation")
-        self.readCount += 1
+        self.processCount += 1
         result = client.read_input_registers(address=self.address,
                                              count=self.count,
                                              unit=self.unit)
@@ -208,85 +108,68 @@ class InputRegisterDatapoint(ReadOnlyDatapoint):
             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))
+            self.logger.debug("{0}: {1!s}".format(self.label, result.registers))        
-            value = None
+            pubQueue.put(MqttProcessor.PublishItem(self.publishTopic, str(result.registers)))
-            try:
-                if self.converter and Converters.Converters[self.converter]['in']:
-                    value = Converters.Converters[self.converter]['in'](result.registers)
-                    logger.debug("{0}: converted: {1!s}".format(self.label, value))
-                else:
-                    value = result.registers
-            except Exception as e:
-                raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
-            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, 
+    def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, updateOnly=None, publishTopic=None):
-                 publishTopic=None, converter=None, bitCount=8):
+        super().__init__(label, unit, address, count, scanRate, updateOnly, publishTopic)
-        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')
+        self.logger = logging.getLogger('DiscreteInputDatapoint')
         # perform read operation
-        logger.debug("Discrete input, perform read operation")
+        self.logger.debug("Discrete input, perform read operation")
-        self.readCount += 1
+        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))
+        if not self.updateOnly or (result.bits != self.lastValue):
-        for i in range(self.bitCount):
+            self.lastValue = result.bits
-            if not self.updateOnly or (result.getBit(i) != self.lastValues[i]):
+            self.logger.debug("{0}: {1!s}".format(self.label, result.bits))        
-                self.lastValues[i] = result.getBit(i)
+            pubQueue.put(MqttProcessor.PublishItem(self.publishTopic, str(result.bits)))
-                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()
 
 
 
+def loadRegisterList(registerList):
+    # Load, check and auto-update registers file
 
-class JsonifyEncoder(json.JSONEncoder):
+    with open(registerList, 'rb') as f:
-    def default(self, o):
+        datapoints = pickle.load(f)
-        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):
+    checkRegisterList(datapoints)
-    if type(j) == dict and 'type' in j and 'args' in j:
+
-        klass = eval(j['type'])
+    newDatapoints = []
-        o = klass(**j['args'])
+    for dp in datapoints:
-        return o
+        ndp = type(dp)()
-    else:
+        for k,v in dp.__dict__.items():
-        return j
+            ndp.__dict__[k] = v
+        newDatapoints.append(ndp)
+        logging.getLogger('loadRegisterList').debug("Datapoint loaded: {0!s}".format(ndp))
+
+    checkRegisterList(newDatapoints, reset=True)
+
+    with open(registerList, 'wb') as f:
+        pickle.dump(newDatapoints, f)
+
+    return newDatapoints
+
+
+def checkRegisterList(registers, reset=False):
+    for r in registers:
+        if not isinstance(r, AbstractModbusDatapoint):
+            raise ValueError('Entry in register list {0!s} is not derived from class AbstractModbusDatapoint'.format(r))        
+        else:
+            if reset:
+                r.errorCount = 0
+                r.processCount = 0
+                r.enqueued = False
 
-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

@@ -10,7 +10,7 @@ class ScanRateConsideringQueueFeeder(threading.Thread, AbstractNotificationRecei
         self.registers = registers
         self.queue = queue
         self.delayEvent = threading.Event()
-        # self.daemon = True
+        self.daemon = True
         self.logger = logging.getLogger('ScanRateConsideringQueueFeeder')
 
     def getMinimalScanrate(self):

src/loadRegisterFile.py

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

src/master.py

@@ -10,8 +10,8 @@ import datetime
 import RegisterDatapoint
 import pickle
 import logging
-import Pins
+
-import Heartbeat
+
 
 
 if __name__ == "__main__":
@@ -29,8 +29,6 @@ if __name__ == "__main__":
     logger.addHandler(fh)
     logger.addHandler(ch)
 
-    Pins.pinsInit()
-
     queue = MyPriorityQueue.MyPriorityQueue()
     pubQueue = Queue()
     nf = NotificationForwarder.NotificationForwarder()
@@ -49,15 +47,11 @@ if __name__ == "__main__":
     mp.start()
     logger.debug('MqttProcessor started')
 
-    cs = CmdServer.CmdServer(config, nf, datapoints)
-    cs.start()
-    logger.debug('CmdServer started')
-
-    hb = Heartbeat.Heartbeat(config, pubQueue)
-    hb.start()
-    logger.debug('Heartbeat 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

@@ -1,122 +0,0 @@
-[
-    {
-        "args": {
-            "address": 1,
-            "converter": "dht20TOFloat",
-            "count": 1,
-            "label": "Temperature",
-            "publishTopic": "IoT/Measurment/Modbus2/Laundry/Temperature",
-            "scanRate": 60.0,
-            "unit": 5,
-            "updateOnly": false
-        },
-        "type": "InputRegisterDatapoint"
-    },
-    {
-        "args": {
-            "address": 2,
-            "converter": "dht20TOFloat",
-            "count": 1,
-            "label": "Humidity",
-            "publishTopic": "IoT/Measurment/Modbus2/Laundry/Humidity",
-            "scanRate": 60.0,
-            "unit": 5,
-            "updateOnly": false
-        },
-        "type": "InputRegisterDatapoint"
-    },
-    {
-        "args": {
-            "address": 0,
-            "bitCount": 8,
-            "converter": null,
-            "count": 1,
-            "label": "Switches",
-            "publishTopic": "Pub/Switches",
-            "scanRate": 0.5,
-            "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

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