Improvements at cmd interface

src/AbstractModbusDatapoint.py

@@ -1,26 +0,0 @@
-
-class AbstractModbusDatapoint(object):
-    def __init__(self, label, unit, address, count, scanRate):
-        self.label = label
-        self.unit = unit
-        self.address = address
-        self.count = count
-        self.scanRate = scanRate
-        self.type = 'abstract data point'
-        self.command = None
-        self.value = None
-        self.enqueued = False
-        if self.scanRate:
-            self.priority = 1
-        else:
-            self.priority = 0
-
-    def __str__(self):
-        return "{0}, {1}: {2} {3} {4} {5} {6}".format(self.type, self.label, self.unit, self.address, self.count, self.command, self.value)
-    
-    def setCommand(self, cmd):
-        self.command = cmd
-    
-    def setValue(self, value):
-        self.value = value
-

src/CmdServer.py

@@ -3,7 +3,10 @@ import socketserver
 import cmd
 import re
 import io
-
+import datetime
+import pickle
+import sys
+import RegisterDatapoint
 
 class CmdInterpreterException(ValueError): pass
 
@@ -18,9 +21,10 @@ def parseIntArbitraryBase(s):
     return i
 
 class CmdInterpreter(cmd.Cmd):
-    def __init__(self, infile, outfile, notifier, registers):
+    def __init__(self, infile, outfile, config, notifier, registers):
         super().__init__(stdin=infile, stdout=outfile)
         self.use_rawinput = False
+        self.config = config
         self.notifier = notifier
         self.registers = registers
         self.prompt = "test8> "
@@ -45,10 +49,9 @@ class CmdInterpreter(cmd.Cmd):
         self.__println("Bye!")
         return True
     
-    def do_add(self, arg):
+    def do_add_hr(self, arg):
         try:
-            (registerType, label, unit, address, count, scanrate, readTopic, writeTopic, feedbackTopic) = self.splitterRe.split(arg)
+            (label, unit, address, count, scanrate, readTopic, writeTopic, feedbackTopic) = self.splitterRe.split(arg)
-            self.__println("RegisterType:  {0}".format(registerType))
             self.__println("Label:         {0}".format(label))
             self.__println("Unit:          {0}".format(unit))
             self.__println("Address:       {0}".format(address))
@@ -82,20 +85,18 @@ class CmdInterpreter(cmd.Cmd):
                     raise CmdInterpreterException('writeTopic must not be set when scanRate is zero')
                 if feedbackTopic:
                     raise CmdInterpreterException('feedbackTopic must not be set when scanRate is zero')
-            allowedRegisterTypes = ['HoldingRegister']
+            r = RegisterDatapoint.HoldingRegisterDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), readTopic, writeTopic, feedbackTopic)
-            if registerType not in allowedRegisterTypes:
+            self.registers.append(r)
-                raise CmdInterpreterException('Unknown register type {0}, allowed are {1!s}'.format(registerType, allowedRegisterTypes))
-
-
         except ValueError as e:
             self.__println("ERROR: {0!s}, {1!s}".format(e.__class__.__name__, e))
 
-    def help_add(self):
+    def help_add_hr(self):
         # HoldingRegisterDatapoint('Voltage', 1, 0x2000, 2, datetime.timedelta(seconds=10), 'Pub/Voltage', None, None),
-        self.__println("Usage: add <RegisterType> <Label> <Unit> <Address> <Count> <ScanRate>")
+        self.__println("Usage: add <Label> <Unit> <Address> <Count> <ScanRate>")
         self.__println("           <ReadTopic> <WriteTopic> <FeedbackTopic>")
+        self.__println("Adds a holding register")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
         self.__println("---------------------------------------------------------------------")
-        self.__println("<RegisterType>               One of HoldingRegister, ...")
         self.__println("<Label>                      Descriptive label")
         self.__println("<Unit>                       Modbus address of the device")
         self.__println("<Address>                    Register address within the device")
@@ -105,13 +106,49 @@ class CmdInterpreter(cmd.Cmd):
         self.__println("<ReadTopic>                  Topic to publish read data")
         self.__println("<WriteTopic>                 Topic to be subscribe to receive data to be")
         self.__println("                             written")
-        self.__println("<FeedbackTopic>              Topic to publish feedback after a write process")
+        self.__println("<FeedbackTopic>              Topic to publish feedback after a write process,")
         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>.")
         self.__println("For write items the <ScanRate> must be zero, <ReadTopic> must be <None> and ")
         self.__println("<WriteTopic> and <FeedbackTopic> must be set.")
 
+    def do_add_ir(self, arg):
+        try:
+            (label, unit, address, count, scanrate, readTopic) = self.splitterRe.split(arg)
+            self.__println("Label:         {0}".format(label))
+            self.__println("Unit:          {0}".format(unit))
+            self.__println("Address:       {0}".format(address))
+            self.__println("Count:         {0}".format(count))
+            self.__println("ScanRate:      {0}".format(scanrate))
+            self.__println("ReadTopic:     {0}".format(readTopic))
+
+            if readTopic == 'None':
+                readTopic = None
+            unit = parseIntArbitraryBase(unit)
+            address = parseIntArbitraryBase(address)
+            count = parseIntArbitraryBase(count)
+            scanrate = float(scanrate)
+            if scanrate == 0.0:
+                raise CmdInterpreterException('scanRate must not be zero')
+            r = RegisterDatapoint.InputRegisterDatapoint(label, unit, address, count, datetime.timedelta(seconds=scanrate), readTopic)
+            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 <Label> <Unit> <Address> <Count> <ScanRate>")
+        self.__println("           <ReadTopic>")
+        self.__println("Adds an input register")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+        self.__println("---------------------------------------------------------------------")
+        self.__println("<Label>                      Descriptive label")
+        self.__println("<Unit>                       Modbus address of the device")
+        self.__println("<Address>                    Register address within the device")
+        self.__println("<Count>                      Count of registers to be read in words")
+        self.__println("<ScanRate>                   Scanrate in seconds (float)")
+        self.__println("<ReadTopic>                  Topic to publish read data")
+
     def do_list(self, arg):
         for i, r in enumerate(self.registers):
             self.__println("#{0}: {1!s}".format(i, r))
@@ -134,11 +171,43 @@ class CmdInterpreter(cmd.Cmd):
         self.__println("Usage: del <idx>")
         self.__println("Removes an item from the list of datapoints by its index, see list command.")
         self.__println("Be aware: indexes have been changed, rerun list before removing the next item.")
+        self.__println("DO NOT FORGET TO SAVE AFTERWARDS!")
+
+    def do_save(self, arg):
+        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:
+            RegisterDatapoint.checkRegisterList(registers)
+            self.registers = registers
+        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):
-        cmd = CmdInterpreter(io.TextIOWrapper(self.rfile), io.TextIOWrapper(self.wfile), self.server.userData.notifier, self.server.userData.registers)
+        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()
             print("Cmd handle terminated")
@@ -151,7 +220,8 @@ class MyThreadingTCPServer(socketserver.ThreadingTCPServer):
         self.userData = userData
 
 class MyCmdUserData(object):
-    def __init__(self, notifier, registers):
+    def __init__(self, config, notifier, registers):
+        self.config = config
         self.notifier = notifier
         self.registers = registers
 
@@ -159,7 +229,8 @@ class CmdServer(threading.Thread):
     def __init__(self, config, notifier, registers):
         super().__init__()
         self.config = config
-        self.server = MyThreadingTCPServer((config.cmdAddress, config.cmdPort), CmdHandle, MyCmdUserData(notifier, registers))
+        self.server = MyThreadingTCPServer((config.cmdAddress, config.cmdPort), CmdHandle, MyCmdUserData(config, notifier, registers))
+        self.daemon = True
 
     def start(self):
         self.server.serve_forever()

src/CommunicationProcessor.py

@@ -6,11 +6,12 @@ class CommunicationProcessor(threading.Thread):
         super().__init__()
         self.config = config
         self.queue = queue
+        self.daemon = True
 
     def run(self):
         while True:
             r = self.queue.get()
-            # r.process()
+            r.process()
             r.lastContact = datetime.datetime.now()
             print("Dequeued: {0!s}".format(r))
             r.enqueued = False

src/Config.py

@@ -7,4 +7,5 @@ class Config(object):
         self.mqttPassword = ''
         self.cmdAddress = '127.0.0.1'
         self.cmdPort = 9999
+        self.registerFile = 'registers.pkl'
 

src/HoldingRegisterDatapoint.py

@@ -1,43 +0,0 @@
-from AbstractModbusDatapoint import AbstractModbusDatapoint
-
-
-class HoldingRegisterDatapoint(AbstractModbusDatapoint):
-    def __init__(self, label, unit, address, count, scanRate, publishTopic, subscribeTopic, feedbackTopic):
-        super().__init__(label, unit, address, count, scanRate)
-        self.publishTopic = publishTopic
-        self.subscribeTopic = subscribeTopic
-        self.feedbackTopic = feedbackTopic
-        self.writeRequestValue = None
-        self.lastContact = None
-        self.type = 'read holding register'
-
-    def __str__(self):
-        return "[{0!s}, {1} {2} {3}".format(super().__str__(), self.publishTopic, self.subscribeTopic, self.feedbackTopic)
-
-    def process(self):
-        successFull = False
-        giveUp = False
-        if self.writeRequestValue:
-            # perform write operation
-            if successFull:
-                # give feedback
-                self.writeRequestValue = None
-            else:
-                # retries handling
-                if giveUp:
-                    # give negative feedback
-                    self.writeRequestValue = None
-        else:
-            # perform read operation
-            if successFull:
-                self.lastContact = datetime.datetime.now()
-                # publish value
-            else:
-                # retries handling
-                if giveUp:
-                    # backoff and availability handling
-                    # give negative feedback
-                    pass
-    
-    def onMessage(self, value):
-        self.writeRequestValue = value

src/MqttProcessor.py

@@ -22,6 +22,7 @@ class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
         self.client = mqtt.Client(userdata=self)
         self.subscriptions = []
         self.topicRegisterMap ={}
+        self.daemon = True
 
     def __processUpdatedRegisters(self, force=False):
         # print("MqttProcessor.__updateSubscriptions")
@@ -68,8 +69,7 @@ class MqttProcessor(threading.Thread, AbstractNotificationReceiver):
         # print("MqttProcessor.onMessage")
         r = self.topicRegisterMap[topic]
         # print("{0}: {1!s} -> {2!s}".format(topic, payload, r))
-        r.setCommand('w')
+        r.onMessage(payload)
-        r.setValue(payload)
         self.queue.put(r)
 
 

src/RegisterDatapoint.py

@@ -0,0 +1,103 @@
+import datetime
+
+
+
+class AbstractModbusDatapoint(object):
+    def __init__(self, label, unit, address, count, scanRate):
+        self.label = label
+        self.unit = unit
+        self.address = address
+        self.count = count
+        self.scanRate = scanRate
+        self.type = 'abstract data point'
+        self.enqueued = False
+        if self.scanRate:
+            self.priority = 1
+        else:
+            self.priority = 0
+
+    def __str__(self):
+        return "{0}, {1}: Unit: {2},  Address: {3}, Count: {4}".format(self.type, self.label, self.unit, self.address, self.count)
+
+    def process(self):
+        raise NotImplementedError
+
+
+class HoldingRegisterDatapoint(AbstractModbusDatapoint):
+    def __init__(self, label, unit, address, count, scanRate, publishTopic, subscribeTopic, feedbackTopic):
+        super().__init__(label, unit, address, count, scanRate)
+        self.publishTopic = publishTopic
+        self.subscribeTopic = subscribeTopic
+        self.feedbackTopic = feedbackTopic
+        self.writeRequestValue = None
+        self.lastContact = None
+        self.type = 'holding register'
+
+    def __str__(self):
+        return "[{0!s}, Read: {1}, Write: {2}, Feedback: {3}".format(super().__str__(), self.publishTopic, self.subscribeTopic, self.feedbackTopic)
+
+    def process(self):
+        successFull = False
+        giveUp = False
+        if self.writeRequestValue:
+            # perform write operation
+            print("Holding register, perform write operation")
+            if successFull:
+                # give feedback
+                self.writeRequestValue = None
+            else:
+                # retries handling
+                if giveUp:
+                    # give negative feedback
+                    self.writeRequestValue = None
+        else:
+            # perform read operation
+            print("Holding register, perform read operation")
+            if successFull:
+                self.lastContact = datetime.datetime.now()
+                # publish value
+            else:
+                # retries handling
+                if giveUp:
+                    # backoff and availability handling
+                    # give negative feedback
+                    pass
+    
+    def onMessage(self, value):
+        self.writeRequestValue = value
+
+
+
+
+class InputRegisterDatapoint(AbstractModbusDatapoint):
+    def __init__(self, label, unit, address, count, scanRate, publishTopic):
+        super().__init__(label, unit, address, count, scanRate)
+        self.publishTopic = publishTopic
+        self.lastContact = None
+        self.type = 'input register'
+
+    def __str__(self):
+        return "[{0!s}, {1}".format(super().__str__(), self.publishTopic)
+
+    def process(self):
+        successFull = False
+        giveUp = False
+        # perform read operation
+        print("Input register, perform read operation")
+        if successFull:
+            self.lastContact = datetime.datetime.now()
+            # publish value
+        else:
+            # retries handling
+            if giveUp:
+                # backoff and availability handling
+                # give negative feedback
+                pass
+
+
+def checkRegisterList(registers):
+    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))        
+
+

src/ScanRateConsideringQueueFeeder.py

@@ -10,6 +10,7 @@ class ScanRateConsideringQueueFeeder(threading.Thread, AbstractNotificationRecei
         self.registers = registers
         self.queue = queue
         self.delayEvent = threading.Event()
+        self.daemon = True
 
     def getMinimalScanrate(self):
             return min([r.scanRate.total_seconds() for r in self.registers if r.scanRate])
@@ -29,7 +30,6 @@ class ScanRateConsideringQueueFeeder(threading.Thread, AbstractNotificationRecei
             ]
             registersToBeHandled.sort(key=lambda x : x.scanRate)
             for r in registersToBeHandled:
-                r.setCommand('r')
                 self.queue.put(r)
                 r.enqueued = True
             self.delayEvent.wait(self.delay)

src/initialRegisterFile.py

@@ -0,0 +1,21 @@
+import datetime
+import RegisterDatapoint
+import pickle
+
+
+datapoints = [
+    RegisterDatapoint.HoldingRegisterDatapoint('Voltage', 1, 0x2000, 2, datetime.timedelta(seconds=10), 'Pub/Voltage', None, None),
+    RegisterDatapoint.HoldingRegisterDatapoint('Frequency', 1, 0x2020, 2, datetime.timedelta(seconds=10), 'Pub/Frequency', None, None),
+    RegisterDatapoint.HoldingRegisterDatapoint('Current', 1, 0x2060, 2, datetime.timedelta(seconds=10), 'Pub/Current', None, None),
+    RegisterDatapoint.HoldingRegisterDatapoint('Resistance Channel 1', 2, 0x0004, 2, datetime.timedelta(seconds=1), 'Pub/ResistanceChannel1', None, None),
+    RegisterDatapoint.HoldingRegisterDatapoint('Temperature Channel 1', 2, 0x000c, 2, datetime.timedelta(seconds=1), 'Pub/TemperatureChannel1', None, None),
+    RegisterDatapoint.HoldingRegisterDatapoint('Resistance Channel 2', 2, 0x0014, 2, datetime.timedelta(seconds=1), 'Pub/ResistanceChannel2', None, None),
+    RegisterDatapoint.HoldingRegisterDatapoint('Temperature Channel 2', 2, 0x001c, 2, datetime.timedelta(seconds=1), 'Pub/TemperatureChannel2', None, None),
+    RegisterDatapoint.HoldingRegisterDatapoint('Relay1', 5, 0x0001, 1, None, None, 'Sub/Relay1', 'Feedback/Relay1'),
+    RegisterDatapoint.InputRegisterDatapoint('Humidity 1', 6, 0x0001, 1, datetime.timedelta(seconds=1), 'Pub/Humidity1'),
+]
+
+
+with open('registers.pkl', 'wb') as f:
+    pickle.dump(datapoints, f)
+

src/master.py

@@ -6,24 +6,21 @@ import NotificationForwarder
 import Config
 import ScanRateConsideringQueueFeeder
 import datetime
-import HoldingRegisterDatapoint
+import RegisterDatapoint
+import pickle
 
-datapoints = [
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Voltage', 1, 0x2000, 2, datetime.timedelta(seconds=10), 'Pub/Voltage', None, None),
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Frequency', 1, 0x2020, 2, datetime.timedelta(seconds=10), 'Pub/Frequency', None, None),
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Current', 1, 0x2060, 2, datetime.timedelta(seconds=10), 'Pub/Current', None, None),
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Resistance Channel 1', 2, 0x0004, 2, datetime.timedelta(seconds=1), 'Pub/ResistanceChannel1', None, None),
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Temperature Channel 1', 2, 0x000c, 2, datetime.timedelta(seconds=1), 'Pub/TemperatureChannel1', None, None),
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Resistance Channel 2', 2, 0x0014, 2, datetime.timedelta(seconds=1), 'Pub/ResistanceChannel2', None, None),
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Temperature Channel 2', 2, 0x001c, 2, datetime.timedelta(seconds=1), 'Pub/TemperatureChannel2', None, None),
-    HoldingRegisterDatapoint.HoldingRegisterDatapoint('Relay1', 5, 0x0001, 1, None, None, 'Sub/Relay1', 'Feedback/Relay1')
-]
 
-queue = MyPriorityQueue.MyPriorityQueue()
-nf = NotificationForwarder.NotificationForwarder()
-config = Config.Config()
 
 if __name__ == "__main__":
+    queue = MyPriorityQueue.MyPriorityQueue()
+    nf = NotificationForwarder.NotificationForwarder()
+    config = Config.Config()
+
+    datapoints = None
+    with open(config.registerFile, 'rb') as f:
+        datapoints = pickle.load(f)
+    RegisterDatapoint.checkRegisterList(datapoints)
+
     cp = CommunicationProcessor.CommunicationProcessor(config, queue)
     cp.start()