import datetime import logging import typing from typing import Any, Optional, Callable from paho.mqtt.client import Client as Client _logger = logging.getLogger("door_pi_control").getChild("util") class Value: def __init__(self, client: Client, topic: str, start_value: Any = None, *, persistent: bool = False, translate: typing.Union[ None, typing.Dict[Any, str], typing.Callable] = None, max_update: float = 0.1): self.client = client self.topic = topic self.persistent = persistent self.value = start_value self.last_update = datetime.datetime.now() - datetime.timedelta( seconds=max_update) self.last_update_value = None self.max_update = max_update if translate is None: self.translate: Callable = str elif callable(translate): self.translate = translate else: self.translate = lambda x: translate.get(x, "None") if start_value is not None: self.update(start_value) def update(self, value: Any, *, force: bool = False, no_update: bool = False) -> None: if value != self.value or value != self.last_update_value: self.value = value if value is not None and \ (force or (not no_update and value != self.last_update_value and (datetime.datetime.now() - self.last_update ).total_seconds() >= self.max_update)): self.last_update = datetime.datetime.now() self.last_update_value = value if self.client is not None: self.client.publish( self.topic, self.translate(value), qos=2, retain=self.persistent) def force_update(self, value) -> None: self.update(value, force=True) def __call__(self, value=None, **kwargs) -> typing.Optional[Any]: if value is not None: self.update(value, **kwargs) return value return self.value def str(self): return self.translate(self.value) def reconnecting_client(host: str, *, keepalive: int = 60): client = Client() client.on_connect = lambda client, userdata, flags, rc: \ _logger.debug("Connected to mqtt host") client.connect_async(host, keepalive=keepalive) client.loop_start() return client