scalefix/cluster.py

import paho.mqtt.subscribe as sub
import json

BROKER = ('homeproxy', 1883)
TOPIC_PREFIX = 'scalefix'
TOPIC_RAW = TOPIC_PREFIX+'/raw/weight'
TOPIC_WEIGHTS = TOPIC_PREFIX+'/weights'
TOPIC_USERS = TOPIC_PREFIX+'/users'

_users={}
_SUFFIX_LAST='/last'
def match_user(weight):
    found_user = 'Gast'
    last_dev = 20
    for username, data in _users.items():
        if username == 'Gast':
            continue
        dev = abs(data['weight'] - weight)
        if dev < 5 and dev < last_dev:
            found_user = username
            last_dev = dev
    return found_user

def handle_weight(client, userdata, message):
    data = json.loads(message.payload)
    weight = data['weight']
    username = match_user(weight)
    print(f'new value for user "{username}": {weight}')
    ext_data = data.copy()
    ext_data['user'] = username

    client.publish(f'{TOPIC_USERS}/{username}/weight', message.payload)
    client.publish(f'{TOPIC_USERS}/{username}{_SUFFIX_LAST}', message.payload, retain=True)
    client.publish(TOPIC_WEIGHTS, json.dumps(ext_data))

def handle_user_last(client, userdata, message):
    global _users
    data = json.loads(message.payload)
    username = message.topic[len(TOPIC_USERS)+1:-len(_SUFFIX_LAST)]
    print(f'got last weight for user "{username}": {data}')
    _users[username] = data

def dispatch(client, userdata, message):
    if message.topic == TOPIC_RAW:
        handle_weight(client, userdata, message)
    elif message.topic.startswith(TOPIC_USERS) and message.topic.endswith(_SUFFIX_LAST):
        handle_user_last(client, userdata, message)

sub.callback(dispatch, [TOPIC_RAW, f'{TOPIC_USERS}/+{_SUFFIX_LAST}'], hostname=BROKER[0], port=BROKER[1])