publish cache 4

config/config.yaml

@@ -1,6 +1,6 @@
 global:
-  scan_interval: 1
+  scan_interval: 0.25
-  log_level: DEBUG
+  log_level: INFO
 
 mqtt:
   broker: emqx01-anonymous-cluster-internal.broker.svc.cluster.local
@@ -123,10 +123,11 @@ output:
   - name: pv_control
     publish_topic: IoT/PV/Control/State
     scan_rate: 1
+    raw_output: true # use only for output device with only one register, name this register 'output'
     slave_id: 1
     registers:
       - address: 0x0001
-        attribute: state
+        attribute: output
         name: State
         unit: "-"
         register_type: holding
@@ -136,10 +137,11 @@ output:
     enabled: true
     publish_topic: IoT/Car/Control/State
     scan_rate: 1
+    raw_output: true # use only for output device with only one register, name this register 'output'
     slave_id: 5
     registers:
       - address: 0x0001
-        attribute: state
+        attribute: output
         name: State
         unit: "-"
         register_type: holding
@@ -228,3 +230,17 @@ output:
         register_type: input
         data_type: float32
         adaptor: floatAdaptor
+  - name: car_feedback
+    enabled: true
+    publish_topic: IoT/Car/Feedback/State
+    scan_rate: 1
+    raw_output: true # use only for output device with only one register, name this register 'output'
+    slave_id: 7
+    registers:
+      - address: 0x0010
+        attribute: output
+        name: State
+        unit: "-"
+        register_type: holding
+        data_type: int32
+        adaptor: onOffAdaptor

src/pv_controller/FromDevices.py

@@ -9,7 +9,7 @@ def floatAdaptor(i):
     return float(f"{i:0.2f}") if i else 0.0
 
 def onOffAdaptor(i):
-    return bool(i)
+    return 'on' if bool(i) else 'off'
 
 
 
@@ -49,12 +49,13 @@ class FromDevices(AbstractMqttPublisher):
                         payload[registers.attribute] = value
 
                     payload['status'] = "Ok"
+                    payload['cnt'] = cnt
+                
+                    payloadStr = json.dumps(payload) if not device.raw_output else str(payload['output'])
+                    self.publish_with_cache(device.publish_topic, payloadStr)
+                    logger.debug(f"mqtt message sent: {device.publish_topic} -> {payloadStr}")
                 except Exception as e:
                     logger.error(f"Caught exception: {str(e)}")
 
-                payload['cnt'] = cnt
-                payloadStr = json.dumps(payload)
-                self.client.publish(device.publish_topic, payloadStr)
-                logger.debug(f"mqtt message sent: {device.publish_topic} -> {payloadStr}")
 
             self.killEvent.wait(timeout=float(self.config.global_.scan_interval))

src/pv_controller/MqttBase.py

@@ -25,6 +25,8 @@ class AbstractMqttPublisher(threading.Thread):
         logger.info(f"mqtt client id: {client_id}")
         self.client = mqtt.Client(client_id=client_id, userdata=self)
 
+        self.cache = {}
+
         # consider this flag in the localLoop
         self.killBill = False
         self.killEvent = threading.Event()
@@ -62,3 +64,10 @@ class AbstractMqttPublisher(threading.Thread):
     def onMessage(self, topic, payload):
         logger.warning("mqtt unexpected message received: {} -> {}".format(topic, str(payload)))
 
+    def publish_with_cache(self, topic, payload):
+        if topic in self.cache and self.cache[topic] == payload:
+            logger.debug(f"mqtt message unchanged, not publishing: {topic} -> {payload}")
+            return
+        self.cache[topic] = payload
+        self.client.publish(topic, payload)
+

src/pv_controller/config.py

@@ -24,6 +24,7 @@ class OutputConfig(BaseModel):
     enabled: bool = Field(default=True)
     scan_rate: Optional[int] = Field(default=60)
     publish_topic: str
+    raw_output: Optional[bool] = Field(default=False)
     slave_id: int
     registers: List[RegisterConfig]
 
@@ -50,7 +51,7 @@ class ModbusConfig(BaseModel):
 
 class GlobalConfig(BaseModel):
     """Global settings"""
-    scan_interval: int
+    scan_interval: float
     log_level: str