diff --git a/config/config.yaml b/config/config.yaml index ded1b87..7668524 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -1,6 +1,6 @@ global: scan_interval: 1 - log_level: INFO + log_level: DEBUG mqtt: broker: emqx01-anonymous-cluster-internal.broker.svc.cluster.local diff --git a/src/pv_controller/FromDevices.py b/src/pv_controller/FromDevices.py index 2518d65..ab509a2 100644 --- a/src/pv_controller/FromDevices.py +++ b/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)}") diff --git a/src/pv_controller/MqttBase.py b/src/pv_controller/MqttBase.py index ba11816..1fc825c 100644 --- a/src/pv_controller/MqttBase.py +++ b/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) +