decrease log level from DEBUG to INFO

.gitlab-ci.yml

@@ -0,0 +1,29 @@
+stages:
+        - check
+        - deploy
+
+build:
+        stage: check
+        image: registry.gitlab.com/wolutator/base-build-env:latest
+        tags:
+                - hottis
+                - linux
+                - docker
+        script:
+                - for I in src/*.py; do python -m py_compile $I; done
+                - for I in src/*.py; do python -m pycodestyle --max-line-length=120 $I; done
+
+deploy:
+        stage: deploy
+        tags:
+                - hottis
+                - linux
+                - rpi
+                - modbus
+        only:
+                - deploy
+        script:
+                - sudo service modbusMaster stop
+                - cp src/*.py /opt/services/modbusMaster
+                - sudo service modbusMaster start
+

ENV

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

pyserialext/Makefile

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

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

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

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

src/CmdServer.py

@@ -6,8 +6,12 @@ import io
 import datetime
 import RegisterDatapoint
 import logging
+import Converters
+
+
+class CmdInterpreterException(ValueError):
+    pass
 
-class CmdInterpreterException(ValueError): pass
 
 def parseIntArbitraryBase(s):
     i = 0
@@ -19,6 +23,7 @@ def parseIntArbitraryBase(s):
         i = int(s, 10)
     return i
 
+
 class CmdInterpreter(cmd.Cmd):
     def __init__(self, infile, outfile, config, notifier, registers):
         super().__init__(stdin=infile, stdout=outfile)
@@ -28,9 +33,17 @@ class CmdInterpreter(cmd.Cmd):
         self.registers = registers
         self.prompt = "test8> "
         self.intro = "test8 admin interface"
-        self.splitterRe = re.compile('\s+')
+        self.splitterRe = re.compile(r'\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)
 
@@ -38,20 +51,14 @@ class CmdInterpreter(cmd.Cmd):
         self.stdout.write(text)
         self.stdout.write("\n\r")
 
-    def do_notify(self, arg):
-        self.notifier.notify()
+    def __listConverterNames(self):
+        return [name for name in Converters.Converters]
 
-    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)
+            (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))
@@ -74,7 +81,15 @@ class CmdInterpreter(cmd.Cmd):
             address = parseIntArbitraryBase(address)
             count = parseIntArbitraryBase(count)
             scanrate = float(scanrate)
-            r = RegisterDatapoint.HoldingRegisterDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), readTopic, writeTopic, feedbackTopic, converter)
+            r = RegisterDatapoint.HoldingRegisterDatapoint(label=label,
+                                                           unit=unit,
+                                                           address=address,
+                                                           count=count,
+                                                           scanRate=datetime.timedelta(seconds=scanrate),
+                                                           publishTopic=readTopic,
+                                                           subscribe=writeTopic,
+                                                           feedbackTopic=feedbackTopic,
+                                                           converter=converter)
             self.registers.append(r)
         except ValueError as e:
             self.__println("ERROR: {0!s}, {1!s}".format(e.__class__.__name__, e))
@@ -97,37 +112,38 @@ 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")
+        self.__println("<Converter>                  Converter for data, one of {0}"
+                       .format(', '.join(self.__listConverterNames())))
 
-    
     def do_add_coil(self, arg):
         try:
-            (label, unit, address, scanrate, readTopic, writeTopic, feedbackTopic) = self.splitterRe.split(arg)
+            (label, unit, address, scanrate, 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)
+            scanrate = 0.0
+            r = RegisterDatapoint.CoilDatapoint(label=label,
+                                                unit=unit,
+                                                address=address,
+                                                scanRate=datetime.timedelta(seconds=scanrate),
+                                                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("                <WriteTopic> <FeedbackTopic>")
         self.__println("Adds a coil")
         self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
         self.__println("---------------------------------------------------------------------")
@@ -136,11 +152,10 @@ class CmdInterpreter(cmd.Cmd):
         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)
@@ -167,7 +182,13 @@ class CmdInterpreter(cmd.Cmd):
             address = parseIntArbitraryBase(address)
             count = parseIntArbitraryBase(count)
             scanrate = float(scanrate)
-            r = RegisterDatapoint.InputRegisterDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), updateOnly, readTopic, converter)
+            r = RegisterDatapoint.InputRegisterDatapoint(label=label,
+                                                         unit=unit,
+                                                         address=address,
+                                                         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))
@@ -185,7 +206,8 @@ 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")
+        self.__println("<Converter>                  Converter for data, one of {0}"
+                       .format(', '.join(self.__listConverterNames())))
 
     def do_add_di(self, arg):
         try:
@@ -212,7 +234,14 @@ class CmdInterpreter(cmd.Cmd):
             count = parseIntArbitraryBase(count)
             scanrate = float(scanrate)
             bitCount = int(bitCount)
-            r = RegisterDatapoint.DiscreteInputDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), updateOnly, readTopic, None, bitCount)
+            r = RegisterDatapoint.DiscreteInputDatapoint(label=label,
+                                                         unit=unit,
+                                                         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))
@@ -235,7 +264,7 @@ class CmdInterpreter(cmd.Cmd):
     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("-----------")
@@ -244,8 +273,9 @@ class CmdInterpreter(cmd.Cmd):
     def do_reset(self, arg):
         for r in self.registers:
             r.errorCount = 0
-            r.processCount = 0
-    
+            r.writeCount = 0
+            r.readCount = 0
+
     def help_reset(self):
         self.__println("Usage: reset")
         self.__println("-----------")
@@ -253,21 +283,21 @@ class CmdInterpreter(cmd.Cmd):
 
     def do_stats(self, arg):
         for i, r in enumerate(self.registers):
-            if r.processCount == 0:
+            processCount = r.readCount + r.writeCount
+            if 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))
+                ratio = float(r.errorCount) / float(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}"
+                           .format(i, r.label, r.unit, r.address, r.readCount, r.writeCount,
+                                   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:
@@ -294,10 +324,10 @@ class CmdInterpreter(cmd.Cmd):
                 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))
@@ -319,9 +349,6 @@ class CmdInterpreter(cmd.Cmd):
         self.__println("                                      considered)")
         self.__println("<value>                    New value")
 
-
-
-
     def do_del(self, arg):
         try:
             i = int(arg)
@@ -337,9 +364,26 @@ 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 do_save(self, arg):
+    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()
+
     def help_save(self):
         self.__println("Usage: save")
         self.__println("Saves a modified register list into the register file.")
@@ -356,11 +400,10 @@ class CmdInterpreter(cmd.Cmd):
         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, 
+        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()
@@ -368,25 +411,28 @@ class CmdHandle(socketserver.StreamRequestHandler):
         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
+        self.server = MyThreadingTCPServer((config.cmdAddress, config.cmdPort),
+                                           CmdHandle,
+                                           MyCmdUserData(config, notifier, registers))
+        # self.daemon = True
 
-    def start(self):
+    def run(self):
         self.server.serve_forever()
-
-

src/CommunicationProcessor.py

@@ -3,7 +3,8 @@ import datetime
 # import RS485Ext
 import RegisterDatapoint
 from pymodbus.client.sync import ModbusSerialClient
-import wiringpi
+# import wiringpi
+import Pins
 import MyRS485
 import time
 import logging
@@ -11,15 +12,16 @@ 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
+        # wiringpi.wiringPiSetup()
+        # wiringpi.pinMode(ERROR_PIN, wiringpi.OUTPUT)
+        # self.daemon = True
         if self.config.modbusDebug:
             logging.getLogger('pymodbus').setLevel(logging.DEBUG)
         else:
@@ -32,7 +34,6 @@ class CommunicationProcessor(threading.Thread):
         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()
@@ -41,19 +42,17 @@ 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")
                     client.socket = self.__getSerial()
             finally:
                 time.sleep(self.config.interCommDelay)
-
-
-
-

src/Config.py

@@ -13,3 +13,5 @@ class Config(object):
         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 +1,30 @@
-# in:  from Modbus to MQTT, input is a list of 16bit integers, output shall be the desired format 
+# 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 
+# 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
 
 
+def fix1twos(x):
+    x = x[0]
+    r = x
+    if x & 0x8000:
+        r = ((x - 1) ^ 0xffff) * -1
+    return r / 10
+
+
 Converters = {
     "dht20TOFloat": {
-        "in": lambda x : float(x[0]) / 10.0,
+        "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)))
+        "in": lambda x: unpack('L', pack('HH', *x))[0],
+        "out": lambda x: unpack('HH', pack('L', int(x)))
+    },
+    "fix1twos": {
+        "in": lambda x: fix1twos(x),
+        "out": None
     }
 }

src/Heartbeat.py

@@ -0,0 +1,21 @@
+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,6 +2,7 @@ import threading
 import paho.mqtt.client as mqtt
 from NotificationForwarder import AbstractNotificationReceiver
 import logging
+import Pins
 
 
 class PublishItem(object):
@@ -9,15 +10,22 @@ class PublishItem(object):
         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()
 
+
 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__()
@@ -27,14 +35,15 @@ class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
         self.pubQueue = pubQueue
         self.client = mqtt.Client(userdata=self)
         self.subscriptions = []
-        self.topicRegisterMap ={}
-        self.daemon = True
+        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]
+        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:
@@ -49,7 +58,8 @@ 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 hasattr(r, 'subscribeTopic')
+                                 and r.subscribeTopic}
 
     def receiveNotification(self, arg):
         self.logger.info("MqttProcessor:registersChanged")
@@ -67,11 +77,12 @@ 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")
 
-
     def onConnect(self):
         # print("MqttProcessor.onConnect")
         self.__processUpdatedRegisters(force=True)
@@ -80,9 +91,9 @@ 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))
         r.onMessage(payload)
         self.queue.put(r)
-

src/MyPriorityQueue.py

@@ -6,12 +6,18 @@ class MyPriorityQueueItem(object):
         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)

src/MyRS485.py

@@ -1,24 +1,27 @@
 import serial
-import wiringpi
+# import wiringpi
+import Pins
 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)
+        # wiringpi.wiringPiSetup()
+        # 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/NotificationForwarder.py

@@ -3,6 +3,7 @@ class AbstractNotificationReceiver(object):
     def receiveNotification(self, arg):
         raise NotImplementedError
 
+
 class NotificationForwarder(object):
     def __init__(self):
         self.receivers = []

src/Pins.py

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

@@ -6,7 +6,10 @@ import logging
 import json
 import Converters
 
-class DatapointException(Exception): pass
+
+class DatapointException(Exception):
+    pass
+
 
 class AbstractModbusDatapoint(object):
     def __init__(self, label=None, unit=None, address=None, count=None, scanRate=None, converter=None):
@@ -24,7 +27,8 @@ class AbstractModbusDatapoint(object):
         self.enqueued = False
         self.lastContact = None
         self.errorCount = 0
-        self.processCount = 0
+        self.readCount = 0
+        self.writeCount = 0
         if self.scanRate:
             self.priority = 1
         else:
@@ -32,24 +36,23 @@ class AbstractModbusDatapoint(object):
 
     def __str__(self):
         return ("{0}, {1}: unit: {2},  address: {3}, count: {4}, scanRate: {5}, "
-                "enqueued: {6}, lastContact: {7}, errorCount: {8}, processCount: {9}, "
-                "converter: {10}"
+                "enqueued: {6}, lastContact: {7}, errorCount: {8}, readCount: {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.processCount, self.converter))
+                        self.errorCount, self.readCount, self.writeCount, self.converter))
 
     def jsonify(self):
-        return {'type':self.__class__.__name__, 
-                'args': { k: getattr(self, k) for k in self.argList }
-               }
+        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, 
+    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']
@@ -70,56 +73,55 @@ class HoldingRegisterDatapoint(AbstractModbusDatapoint):
         if self.writeRequestValue:
             # perform write operation
             logger.debug("Holding register, perform write operation")
-            self.processCount += 1
+            self.writeCount += 1
             values = None
             logger.debug("{0}: raw: {1!s}".format(self.label, self.writeRequestValue))
-            if self.converter and Converters.Converters[self.converter]['out']:
-                try:
+            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))
-                except Exception as e:
-                    raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
-            else:
-                values = [int(self.writeRequestValue)]
+                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))                
+                                            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, 
+            self.readCount += 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:
+            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))
-                except Exception as e:
-                    raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
-            else:
-                value = result.registers
+                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,
+    def __init__(self, label=None, unit=None, address=None, scanRate=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.argList = ['label', 'unit', 'address', 'scanRate', 'subscribeTopic', 'feedbackTopic']
         self.subscribeTopic = subscribeTopic
         self.feedbackTopic = feedbackTopic
         self.writeRequestValue = None
@@ -127,12 +129,12 @@ class CoilDatapoint(AbstractModbusDatapoint):
 
     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}"
+                "enqueued: {5}, lastContact: {6}, errorCount: {7}, readCount: {8}, "
+                "writeCount: {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))
+                        self.errorCount, self.readCount, self.writeCount,
+                        self.subscribeTopic, self.feedbackTopic))
 
     def onMessage(self, value):
         self.writeRequestValue = value.decode()
@@ -142,9 +144,9 @@ class CoilDatapoint(AbstractModbusDatapoint):
         if self.writeRequestValue:
             # perform write operation
             logger.debug("Coil, perform write operation")
-            self.processCount += 1
+            self.writeCount += 1
             logger.debug("{0}: raw: {1!s}".format(self.label, self.writeRequestValue))
-            value=None
+            value = None
             if self.writeRequestValue in ['true', 'True', 'yes', 'Yes', 'On', 'on']:
                 value = True
             elif self.writeRequestValue in ['false', 'False', 'no', 'No', 'Off', 'off']:
@@ -155,26 +157,18 @@ class CoilDatapoint(AbstractModbusDatapoint):
             result = client.write_coil(address=self.address,
                                        unit=self.unit,
                                        value=value)
-            logger.debug("Write result: {0!s}".format(result))                
+            logger.debug("Write result: {0!s}".format(result))
+            if self.feedbackTopic:
+                pubQueue.put(MqttProcessor.PublishItem(self.feedbackTopic, str(value)))
             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()
+            # no write op, strange
+            logger.debug("Coil, process call but no write value available, strange")
 
 
 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, converter=None):
         super().__init__(label, unit, address, count, scanRate, converter)
         self.argList = self.argList + ['updateOnly', 'publishTopic']
         self.updateOnly = updateOnly
@@ -187,9 +181,8 @@ class ReadOnlyDatapoint(AbstractModbusDatapoint):
                         self.lastValue))
 
 
-
 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, converter=None):
         super().__init__(label, unit, address, count, scanRate, updateOnly, publishTopic, converter)
         self.type = 'input register'
@@ -198,7 +191,7 @@ class InputRegisterDatapoint(ReadOnlyDatapoint):
         logger = logging.getLogger('InputRegisterDatapoint')
         # perform read operation
         logger.debug("Input register, perform read operation")
-        self.processCount += 1
+        self.readCount += 1
         result = client.read_input_registers(address=self.address,
                                              count=self.count,
                                              unit=self.unit)
@@ -209,21 +202,21 @@ class InputRegisterDatapoint(ReadOnlyDatapoint):
             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:
+            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))
-                except Exception as e:
-                    raise DatapointException("Exception caught when trying to converter modbus data: {0!s}".format(e))
-            else:
-                value = result.registers
+                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, converter=None, bitCount=8):
         super().__init__(label, unit, address, count, scanRate, updateOnly, publishTopic, converter)
         self.argList = self.argList + ['bitCount']
@@ -239,7 +232,7 @@ class DiscreteInputDatapoint(ReadOnlyDatapoint):
         logger = logging.getLogger('DiscreteInputDatapoint')
         # perform read operation
         logger.debug("Discrete input, perform read operation")
-        self.processCount += 1
+        self.readCount += 1
         result = client.read_discrete_inputs(address=self.address,
                                              count=self.count,
                                              unit=self.unit)
@@ -252,12 +245,11 @@ class DiscreteInputDatapoint(ReadOnlyDatapoint):
                 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))))
+                    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
@@ -270,6 +262,7 @@ class JsonifyEncoder(json.JSONEncoder):
                 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'])
@@ -278,14 +271,15 @@ def datapointObjectHook(j):
     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

@@ -3,6 +3,7 @@ import datetime
 from NotificationForwarder import AbstractNotificationReceiver
 import logging
 
+
 class ScanRateConsideringQueueFeeder(threading.Thread, AbstractNotificationReceiver):
     def __init__(self, config, registers, queue):
         super().__init__()
@@ -10,11 +11,11 @@ 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):
-            return min([r.scanRate.total_seconds() for r in self.registers if r.scanRate])
+        return min([r.scanRate.total_seconds() for r in self.registers if r.scanRate])
 
     def receiveNotification(self, arg):
         self.logger.info("ScanRateConsideringQueueFeeder:registersChanged")
@@ -26,10 +27,10 @@ class ScanRateConsideringQueueFeeder(threading.Thread, AbstractNotificationRecei
             registersToBeHandled = [
                 r for r in self.registers if ((not r.enqueued) and
                                               (r.scanRate) and
-                                              ((not r.lastContact) or 
+                                              ((not r.lastContact) or
                                                (r.lastContact + r.scanRate < datetime.datetime.now())))
             ]
-            registersToBeHandled.sort(key=lambda x : x.scanRate)
+            registersToBeHandled.sort(key=lambda x: x.scanRate)
             for r in registersToBeHandled:
                 self.queue.put(r)
                 r.enqueued = True

src/initialRegisterFile.py

@@ -4,12 +4,14 @@ 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'),
+    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

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

src/master.py

@@ -10,17 +10,17 @@ import datetime
 import RegisterDatapoint
 import pickle
 import logging
-
-
+import Pins
+import Heartbeat
 
 
 if __name__ == "__main__":
     config = Config.Config()
 
     logger = logging.getLogger()
-    logger.setLevel(logging.DEBUG)
+    logger.setLevel(logging.INFO)
     fh = logging.FileHandler(config.logFile)
-    fh.setLevel(logging.DEBUG)
+    fh.setLevel(logging.INFO)
     ch = logging.StreamHandler()
     ch.setLevel(logging.ERROR)
     formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
@@ -29,12 +29,13 @@ if __name__ == "__main__":
     logger.addHandler(fh)
     logger.addHandler(ch)
 
+    Pins.pinsInit()
+
     queue = MyPriorityQueue.MyPriorityQueue()
     pubQueue = Queue()
     nf = NotificationForwarder.NotificationForwarder()
     logger.debug('infrastructure prepared')
 
-
     datapoints = RegisterDatapoint.loadRegisterList(config.registerFile)
     logger.debug('datapoints read')
 
@@ -47,11 +48,15 @@ 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/modbusMaster.service

@@ -0,0 +1,19 @@
+[Unit]
+Description=ModbusMaster
+Wants=network-online.target
+After=network-online.target
+
+
+[Service]
+Type=simple
+GuessMainPID=yes
+ExecStart=/usr/bin/python master.py
+ExecStop=kill -SIGINT $mainpid
+Restart=on-failure
+WorkingDirectory=/opt/services/modbusMaster
+
+[Install]
+Alias=ModbusMaster
+WantedBy=multi-user.target
+
+

src/registers.json

@@ -5,8 +5,8 @@
             "converter": "dht20TOFloat",
             "count": 1,
             "label": "Temperature",
-            "publishTopic": "Pub/Temperature",
-            "scanRate": 10.0,
+            "publishTopic": "IoT/Measurment/Modbus2/Laundry/Temperature",
+            "scanRate": 60.0,
             "unit": 5,
             "updateOnly": false
         },
@@ -18,8 +18,8 @@
             "converter": "dht20TOFloat",
             "count": 1,
             "label": "Humidity",
-            "publishTopic": "Pub/Humidity",
-            "scanRate": null,
+            "publishTopic": "IoT/Measurment/Modbus2/Laundry/Humidity",
+            "scanRate": 60.0,
             "unit": 5,
             "updateOnly": false
         },
@@ -32,77 +32,21 @@
             "converter": null,
             "count": 1,
             "label": "Switches",
-            "publishTopic": "Pub/Switches",
-            "scanRate": null,
+            "publishTopic": "IoT/Status/Modbus2/Switches",
+            "scanRate": 0.5,
             "unit": 4,
-            "updateOnly": false
+            "updateOnly": true
         },
         "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",
+            "feedbackTopic": "IoT/Feedback/Modbus2/Coils/0",
+            "label": "Coil0",
+            "publishTopic": "IoT/Status/Modbus2/Coils/0",
+            "scanRate": 0.0,
+            "subscribeTopic": "IoT/Action/Modbus2/Coils/0",
             "unit": 4
         },
         "type": "CoilDatapoint"
@@ -110,13 +54,63 @@
     {
         "args": {
             "address": 1,
-            "feedbackTopic": "FB/Coil2",
-            "label": "Coil2",
-            "publishTopic": "Pub/Coil2",
-            "scanRate": 1.0,
-            "subscribeTopic": "Sub/Coil2",
+            "feedbackTopic": "IoT/Feedback/Modbus2/Coils/1",
+            "label": "Coil1",
+            "publishTopic": "IoT/Status/Modbus2/Coils/1",
+            "scanRate": 0.0,
+            "subscribeTopic": "IoT/Action/Modbus2/Coils/1",
             "unit": 4
         },
         "type": "CoilDatapoint"
+    },
+    {
+        "args": {
+            "address": 2,
+            "feedbackTopic": "IoT/Feedback/Modbus2/Coils/2",
+            "label": "Coil2",
+            "publishTopic": "IoT/Status/Modbus2/Coils/2",
+            "scanRate": 0.0,
+            "subscribeTopic": "IoT/Action/Modbus2/Coils/2",
+            "unit": 4
+        },
+        "type": "CoilDatapoint"
+    },
+    {
+        "args": {
+            "address": 3,
+            "feedbackTopic": "IoT/Feedback/Modbus2/Coils/3",
+            "label": "Coil3",
+            "publishTopic": "IoT/Status/Modbus2/Coils/3",
+            "scanRate": 0.0,
+            "subscribeTopic": "IoT/Action/Modbus2/Coils/3",
+            "unit": 4
+        },
+        "type": "CoilDatapoint"
+    },
+    {
+        "args": {
+            "address": 1,
+            "converter": "fix1twos",
+            "count": 1,
+            "label": "wago1",
+            "publishTopic": "IoT/Measurement/Modbus2/Wago1",
+            "scanRate": 1.0,
+            "unit": 11,
+            "updateOnly": false
+        },
+        "type": "InputRegisterDatapoint"
+    },
+    {
+        "args": {
+            "address": 0,
+            "converter": "fix1twos",
+            "count": 1,
+            "label": "Freezer",
+            "publishTopic": "IoT/Measurement/Modbus2/Freezer",
+            "scanRate": 1.0,
+            "unit": 11,
+            "updateOnly": false
+        },
+        "type": "InputRegisterDatapoint"
     }
-]
+]

src/updateRegisterFile.py

@@ -10,15 +10,12 @@ with open('registers.pkl', 'rb') as f:
 newDatapoints = []
 for dp in datapoints:
     ndp = type(dp)()
-    for k,v in dp.__dict__.items():
+    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')
+RegisterDatapoint.saveRegisterList(newDatapoints, 'registers.json')