Mercurial > code > home > repos > mqtt_metrics
view convert.py @ 5:8390d5d0d512
one mqtt message can convert to multiple measurements
author | drewp@bigasterisk.com |
---|---|
date | Sat, 10 Aug 2024 23:02:51 -0700 |
parents | cd1b8d7bda78 |
children | bc2a93b306e9 |
line wrap: on
line source
''' Note that these functions need to parse the message['payload'] into a float ''' converters = [] def f_from_c(c): return (c * 9 / 5) + 32 def topic(topic_pattern: str): def decorator(func): func.topic_pattern = topic_pattern converters.append(func) return func return decorator @topic(r'([^-]+)-air-quality/sensor/particulate_matter__10_0__m_concentration/state') def pm(message, topicGroups: tuple[str]) -> list[dict]: return [{ 'name': 'air_quality_pm', 'labels': [{ 'labelName': 'location', 'labelValue': topicGroups[0], }, { 'labelName': 'size', 'labelValue': '10', }], 'value': float(message['payload']), }] @topic(r'([^-]+)-air-quality/sensor/air_temperature_c/state') def air_temp(message, topicGroups: tuple[str]) -> list[dict]: return [{ 'name': 'air_temperature_f', 'labels': [{ 'labelName': 'location', 'labelValue': topicGroups[0], }], 'value': f_from_c(float(message['payload'])), }]