relais support

config.ini

@@ -6,9 +6,11 @@ port = 1883
 # ca = 
 # cert = 
 # key = 
-subscribeTopic = test1
+relaisSubscribeTopic = IoT/PV/Cmd
 meterPublishTopic = IoT/PV/Meter
 meterPublishPeriod = 15
+relaisPublishTopic = IoT/PV/Relais
+relaisPublishPeriod = 15
 
 [modbus]
 gateway = 172.16.2.42

src/pv_controller/MeterPublish.py

@@ -5,13 +5,13 @@ import json
 import datetime
 
 REGISTERS = [
-  { "addr":0x0048, "attr": "importEnergy", "name":"Import active energy", "unit":"kWh" },
+  { "slave":2, "addr":0x0048, "attr": "importEnergy", "name":"Import active energy", "unit":"kWh" },
-  { "addr":0x004a, "attr": "exportEnergy", "name":"Export active energy", "unit":"kWh" },
+  { "slave":2, "addr":0x004a, "attr": "exportEnergy", "name":"Export active energy", "unit":"kWh" },
-  { "addr":0x000c, "attr": "activePower", "name":"Active power", "unit":"W" },
+  { "slave":2, "addr":0x000c, "attr": "activePower", "name":"Active power", "unit":"W" },
-  { "addr":0x0058, "attr": "positivePower", "name":"Positive power", "unit":"W" },
+  { "slave":2, "addr":0x0058, "attr": "positivePower", "name":"Positive power", "unit":"W" },
-  { "addr":0x005c, "attr": "reversePower", "name":"Reverse power", "unit":"W" },
+  { "slave":2, "addr":0x005c, "attr": "reversePower", "name":"Reverse power", "unit":"W" },
-  { "addr":0x0000, "attr": "voltage", "name":"Voltage", "unit":"V" },
+  { "slave":2, "addr":0x0000, "attr": "voltage", "name":"Voltage", "unit":"V" },
-  { "addr":0x0006, "attr": "current", "name":"Current", "unit":"A" }
+  { "slave":2, "addr":0x0006, "attr": "current", "name":"Current", "unit":"A" },
 ]
 
 
@@ -33,7 +33,7 @@ class MeterPublish(AbstractMqttPublisher):
                 payload['status'] = "Error"
                 payload['timestamp'] = datetime.datetime.isoformat(datetime.datetime.utcnow())
                 for reg in self.registers:
-                    v = self.modbusHandler.readInputRegister(reg['addr'])
+                    v = self.modbusHandler.readInputRegister(reg['slave'], reg['addr'])
                     logger.debug(f"{reg['name']}: {v} {reg['unit']}")
                     payload[reg['attr']] = float(f"{v:0.2f}")
                 payload['status'] = "Ok"

src/pv_controller/ModbusBase.py

@@ -23,11 +23,10 @@ class ModbusHandler:
   def __init__(self, config):
     self.config = config['modbus']
     self.client = ModbusTcpClient(self.config['gateway'])
-
-  def readInputRegister(self, addr):
     self.client.connect()
-    try:
+
-      res = self.client.read_input_registers(addr, 2, slave=2)
+  def readInputRegister(self, slave, addr):
+    res = self.client.read_input_registers(addr, 2, slave=slave)
     if (isinstance(res, ReadInputRegistersResponse)):
       v = BinaryPayloadDecoder.fromRegisters(res.registers, byteorder=Endian.Big, wordorder=Endian.Big).decode_32bit_float()
       return v
@@ -39,17 +38,25 @@ class ModbusHandler:
       msg = f"Unknown type: {type(res)}, Content: {res}"
       logger.warning(msg)
       raise LocalModbusException(msg=msg)
-    finally:
-      self.client.close()
 
-  def writeCoil(self, addr, value):
+  def readHoldingRegister(self, slave, addr):
-    self.client.connect()
+    res = self.client.read_holding_registers(addr, 2, slave=slave)
-    try:
+    if (isinstance(res, ReadHoldingRegistersResponse)):
-      res = self.client.write_coil(addr, value, slave=1)
+      v = res.registers
+      return v
+    elif (isinstance(res, LocalModbusException)):
+      msg = f"Error: {type(res)}, Content: {res}"
+      logger.warning(msg)
+      raise LocalModbusException(msg=msg, cause=res)
+    else:
+      msg = f"Unknown type: {type(res)}, Content: {res}"
+      logger.warning(msg)
+      raise LocalModbusException(msg=msg)
+
+  def writeCoil(self, slave, addr, value):
+    res = self.client.write_coil(addr, value, slave=slave)
     logger.debug(f"write coil result {res}")
     return value
-    finally:
-      self.client.close()
 
 
 #client = ModbusTcpClient("172.16.2.42")

src/pv_controller/RelaisPublish.py

@@ -0,0 +1,36 @@
+from threading import Event
+from loguru import logger
+from MqttBase import AbstractMqttPublisher
+import json
+import datetime
+
+
+class RelaisPublish(AbstractMqttPublisher):
+    def __init__(self, config, modbusHandler):
+        super().__init__(config)
+        self.modbusHandler = modbusHandler
+
+    def localLoop(self):
+        cnt = 0
+        while not self.killBill:
+            cnt += 1
+            topic = self.config["relaisPublishTopic"]
+            payload = str(cnt)
+
+            try:
+                payload = {}
+                payload['status'] = "Error"
+                payload['timestamp'] = datetime.datetime.isoformat(datetime.datetime.utcnow())
+                v = self.modbusHandler.readHoldingRegister(1, 1)
+                logger.debug(f"{v}")
+                payload['state'] = 'Off' if v[0] == 0 else 'On'
+                payload['status'] = "Ok"
+            except Exception as e:
+                logger.error(f"Caught exception: {str(e)}")
+
+            payload['cnt'] = cnt
+            payloadStr = json.dumps(payload)
+            self.client.publish(topic, payloadStr)
+            logger.info(f"mqtt message sent: {topic} -> {payloadStr}")
+
+            self.killEvent.wait(timeout=float(self.config["relaisPublishPeriod"]))

src/pv_controller/RelaisSubscribe.py

@@ -0,0 +1,29 @@
+from MqttBase import AbstractMqttPublisher
+from loguru import logger
+from time import sleep
+
+
+class RelaisSubscribe(AbstractMqttPublisher):
+    def __init__(self, config, modbusHandler):
+        super().__init__(config)
+        self.modbusHandler = modbusHandler
+
+    def localLoop(self):
+        while not self.killBill:
+            sleep(60.0)
+
+    def onMessage(self, topic, payload):
+        logger.info("mqtt message received: {} -> {}".format(topic, str(payload)))
+        if payload == b'On':
+            self.modbusHandler.writeCoil(1, 0, 1)
+        elif payload == b'Off':
+            self.modbusHandler.writeCoil(1, 0, 0)
+        else:
+            logger.warning(f"Illegal command {payload} received")
+
+
+
+    def onConnect(self):
+        logger.info("mqtt connected")
+        self.client.subscribe("{}".format(self.config["relaisSubscribeTopic"]))
+        logger.info("subscribed")

src/pv_controller/TestSubscribe.py

@@ -1,20 +0,0 @@
-from MqttBase import AbstractMqttPublisher
-from loguru import logger
-from time import sleep
-
-
-class TestSubscribe(AbstractMqttPublisher):
-    def __init__(self, config):
-        super().__init__(config)
-
-    def localLoop(self):
-        while not self.killBill:
-            sleep(60.0)
-
-    def onMessage(self, topic, payload):
-        logger.warning("mqtt message received: {} -> {}".format(topic, str(payload)))
-
-    def onConnect(self):
-        logger.info("mqtt connected")
-        self.client.subscribe("{}".format(self.config["subscribeTopic"]))
-        logger.info("subscribed")

src/pv_controller/pvc.py

@@ -1,5 +1,6 @@
 from MeterPublish import MeterPublish
-# from TestSubscribe import TestSubscribe
+from RelaisPublish import RelaisPublish
+from RelaisSubscribe import RelaisSubscribe
 from ModbusBase import ModbusHandler
 from loguru import logger
 import logging
@@ -33,16 +34,20 @@ args = parser.parse_args()
 config = configparser.ConfigParser()
 config.read(args.config)
 
-# testSubscribeThread = TestSubscribe(config)
-# testSubscribeThread.start()
-# logger.info("testSubscribe started")
-
 modbusHandler = ModbusHandler(config)
 
+relaisSubscribeThread = RelaisSubscribe(config, modbusHandler)
+relaisSubscribeThread.start()
+logger.info("relaisSubscribe started")
+
 meterPublishThread = MeterPublish(config, modbusHandler)
 meterPublishThread.start()
 logger.info("meterPublishThread started")
 
+relaisPublishThread = RelaisPublish(config, modbusHandler)
+relaisPublishThread.start()
+logger.info("relaisPublishThread started")
+
 threading.excepthook = exceptHook
 logger.info("Threading excepthook set")
 
@@ -52,13 +57,17 @@ logger.info("pv controller is running")
 deathBell.wait()
 logger.error("pv controller is dying")
 
-# testSubscribeThread.stop()
+relaisSubscribeThread.stop()
 meterPublishThread.stop()
+relaisPublishThread.stop()
 
-# testSubscribeThread.join()
+relaisSubscribeThread.join()
-# logger.error("testSubscribe joined")
+logger.error("relaisSubscribe joined")
 
 meterPublishThread.join()
 logger.error("meterPublishThread joined")
 
+relaisPublishThread.join()
+logger.error("relaisPublishThread joined")
+
 logger.error("pv controller is terminated")