view convert.py @ 4:cd1b8d7bda78

get metrics writing to victoriametrics
author drewp@bigasterisk.com
date Sat, 10 Aug 2024 21:16:19 -0700
parents 579df3a4e62d
children 8390d5d0d512
line wrap: on
line source

'''
Note that these functions need to parse the message['payload'] into a float
'''
converters = []


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]):
    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]):
    return {
        'name': 'air_temperature_f',
        'labels': [{
            'labelName': 'location',
            'labelValue': topicGroups[0],
        }],
        'value': (float(message['payload']) * 9 / 5) + 32,
    }