publish cache

config/config.yaml

@@ -231,7 +231,7 @@ output:
         data_type: float32
         adaptor: floatAdaptor
   - name: car_feedback
-    enabled: false
+    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'

src/pv_controller/FromDevices.py

@@ -52,7 +52,7 @@ class FromDevices(AbstractMqttPublisher):
                     payload['cnt'] = cnt
                 
                     payloadStr = json.dumps(payload) if not device.raw_output else str(payload['output'])
-                    self.client.publish(device.publish_topic, payloadStr)
+                    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)}")

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)
+