7 Commits

Author SHA1 Message Date
d9b1000665 publish cache retained
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
2025-12-06 14:46:39 +01:00
7794fabaf3 publish cache 4
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
2025-12-06 14:41:39 +01:00
6e6ff4c229 publish cache 3
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
2025-12-06 14:39:53 +01:00
dc2175c298 publish cache 2
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
2025-12-06 14:38:01 +01:00
6d8c5c25db publish cache
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
2025-12-06 14:34:52 +01:00
ca08059e13 add car feedback 6
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
2025-12-06 14:26:52 +01:00
6796bdd905 add car feedback 5
All checks were successful
ci/woodpecker/tag/woodpecker Pipeline was successful
2025-12-06 14:24:12 +01:00
4 changed files with 14 additions and 5 deletions

View File

@@ -1,6 +1,6 @@
global: global:
scan_interval: 1 scan_interval: 0.25
log_level: DEBUG log_level: INFO
mqtt: mqtt:
broker: emqx01-anonymous-cluster-internal.broker.svc.cluster.local broker: emqx01-anonymous-cluster-internal.broker.svc.cluster.local
@@ -231,7 +231,7 @@ output:
data_type: float32 data_type: float32
adaptor: floatAdaptor adaptor: floatAdaptor
- name: car_feedback - name: car_feedback
enabled: false enabled: true
publish_topic: IoT/Car/Feedback/State publish_topic: IoT/Car/Feedback/State
scan_rate: 1 scan_rate: 1
raw_output: true # use only for output device with only one register, name this register 'output' raw_output: true # use only for output device with only one register, name this register 'output'

View File

@@ -52,7 +52,7 @@ class FromDevices(AbstractMqttPublisher):
payload['cnt'] = cnt payload['cnt'] = cnt
payloadStr = json.dumps(payload) if not device.raw_output else str(payload['output']) 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}") logger.debug(f"mqtt message sent: {device.publish_topic} -> {payloadStr}")
except Exception as e: except Exception as e:
logger.error(f"Caught exception: {str(e)}") logger.error(f"Caught exception: {str(e)}")

View File

@@ -25,6 +25,8 @@ class AbstractMqttPublisher(threading.Thread):
logger.info(f"mqtt client id: {client_id}") logger.info(f"mqtt client id: {client_id}")
self.client = mqtt.Client(client_id=client_id, userdata=self) self.client = mqtt.Client(client_id=client_id, userdata=self)
self.cache = {}
# consider this flag in the localLoop # consider this flag in the localLoop
self.killBill = False self.killBill = False
self.killEvent = threading.Event() self.killEvent = threading.Event()
@@ -62,3 +64,10 @@ class AbstractMqttPublisher(threading.Thread):
def onMessage(self, topic, payload): def onMessage(self, topic, payload):
logger.warning("mqtt unexpected message received: {} -> {}".format(topic, str(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, retain=True)

View File

@@ -51,7 +51,7 @@ class ModbusConfig(BaseModel):
class GlobalConfig(BaseModel): class GlobalConfig(BaseModel):
"""Global settings""" """Global settings"""
scan_interval: int scan_interval: float
log_level: str log_level: str