From 583e64ccf68df3675a50294ea9ab3810e463e0b5 Mon Sep 17 00:00:00 2001 From: Edward Brown Date: Sat, 15 Nov 2025 10:34:48 +0000 Subject: [PATCH] Add prototype rewrite data structure, imports, typing etc --- haco/__init__.py | 0 haco/capabilities.py | 39 +++++++++++++++++++++ haco/climate.py | 32 +++++++++++++++++ haco/control.py | 82 ++++++++++++++++++++++++++++++++++++++++++++ haco/device.py | 54 +++++++++++++++++++++++++++++ haco/topics.py | 75 ++++++++++++++++++++++++++++++++++++++++ haco/utils.py | 27 +++++++++++++++ 7 files changed, 309 insertions(+) create mode 100644 haco/__init__.py create mode 100644 haco/capabilities.py create mode 100644 haco/climate.py create mode 100644 haco/control.py create mode 100644 haco/device.py create mode 100644 haco/topics.py create mode 100644 haco/utils.py diff --git a/haco/__init__.py b/haco/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/haco/capabilities.py b/haco/capabilities.py new file mode 100644 index 0000000..b5911da --- /dev/null +++ b/haco/capabilities.py @@ -0,0 +1,39 @@ +from typing import Optional + +from pydantic import Field, model_validator + +from fmtr.tools import dm +from fmtr.tools.iterator_tools import strip_none +from haco.control import Control +from haco.topics import AnnounceTopicState, AnnounceTopicCommand + + +class Capability(dm.Base): + name: str + alias: str = None + + state: AnnounceTopicState | None = Field(default_factory=AnnounceTopicState) + command: AnnounceTopicCommand | None = Field(default_factory=AnnounceTopicCommand) + + control: Optional[Control] = Field(None, exclude=True, repr=False) + + @property + def announce(self) -> dict: + data = {} + for topic in strip_none(self.state, self.command): + data |= topic.announce + return data + + @property + def subscriptions(self): + + if not self.command: + return {} + + return self.command.subscriptions + + @model_validator(mode="after") + def attach_parent(self): + for topic in strip_none(self.state, self.command): + topic.capability = self + return self diff --git a/haco/climate.py b/haco/climate.py new file mode 100644 index 0000000..28c0b97 --- /dev/null +++ b/haco/climate.py @@ -0,0 +1,32 @@ +from typing import Literal, Optional, List + +from haco.capabilities import Capability +from haco.control import Control + + +class Climate(Control): + platform: Literal["climate"] = "climate" + temperature_unit: Literal["C", "F"] = "C" + temperature_min: float = 0 + temperature_max: float = 40 + humidity_min: float = 0 + humidity_max: float = 100 + precision: int = 1 + modes: Optional[List[str]] = None + preset_modes: Optional[List[str]] = None + fan_modes: Optional[List[str]] = None + swing_modes: Optional[List[str]] = None + + def temperature_command(self, value): + return value + + @classmethod + def get_capabilities(self): + return [ + Capability(name="current_temperature", command=None), + Capability(name="temperature"), + Capability(name="current_humidity"), + Capability(name="target_humidity"), + Capability(name="mode"), + Capability(name="fan_mode"), + ] diff --git a/haco/control.py b/haco/control.py new file mode 100644 index 0000000..d4dd59f --- /dev/null +++ b/haco/control.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from functools import cached_property +from typing import Optional, TYPE_CHECKING + +from pydantic import computed_field, model_validator, PrivateAttr + +from fmtr.tools import dm + +if TYPE_CHECKING: + from haco.capabilities import Capability + from haco.device import Device + +from haco.utils import sanitize_name + + +class Control(dm.Base): + name: str + platform: str + + icon: Optional[str] = None + + _device: Device = PrivateAttr(default=None) + + @computed_field + @property + def device(self) -> Device: # todo. quite ugly. perhaps just use @dataclass for topics? + return self._device + + @device.setter + def device(self, value: Device): + self._device = value + + @computed_field + @property + def unique_id(self) -> str: + name_san = sanitize_name(self.name) + return f"{self.device.name_san}-{name_san}" + + @property + def announce_topic(self): + return f"homeassistant/{self.platform}/{self.unique_id}/config" + + @computed_field + @property + def availability_topic(self) -> str: + return self.device.availability_topic + + @property + def announce(self) -> dict: + + data = self.model_dump(mode="json", exclude_none=True) + for capability in self.capabilities: + data |= capability.announce + + data = {self.announce_topic: data} + return data + + @property + def subscriptions(self) -> dict: + + data = {} + + for capability in self.capabilities: + data |= capability.subscriptions + + return data + + @model_validator(mode="after") + def attach_parent(self): + self.capabilities = self.get_capabilities() + for capability in self.capabilities: + capability.control = self + return self + + @classmethod + def get_capabilities(cls): + raise NotImplementedError() + + @cached_property + def capabilities(self) -> list[Capability]: + return self.get_capabilities() diff --git a/haco/device.py b/haco/device.py new file mode 100644 index 0000000..8cd3698 --- /dev/null +++ b/haco/device.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from functools import cached_property +from typing import List + +from pydantic import Field, computed_field, model_validator + +from fmtr.tools import dm +from haco.control import Control +from haco.utils import sanitize_name + + +class Device(dm.Base): + name: str + manufacturer: str = "Demo" + model: str = "Python MQTT Example" + + controls: List[Control] = Field(default_factory=list, exclude=True, repr=False) + + @property + def name_san(self) -> str: + return sanitize_name(self.name) + + @property + def availability_topic(self) -> str: + return f'status/{self.name_san}/availability' + + @computed_field + @property + def identifiers(self) -> list[str]: + return [self.name_san] + + @computed_field + @property + def connections(self) -> list[list[str]]: + return [["mac", "AA:BB:CC:DD:EE:FF"]] + + @model_validator(mode="after") + def attach_parent(self): + for control in self.controls: + control.device = self + return self + + @cached_property + def announces(self) -> list[dict]: + return [control.announce for control in self.controls] + + @cached_property + def subscriptions(self) -> dict: + data = {} + for control in self.controls: + data |= control.subscriptions + + return data diff --git a/haco/topics.py b/haco/topics.py new file mode 100644 index 0000000..2c19a44 --- /dev/null +++ b/haco/topics.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from functools import cached_property +from typing import ClassVar, TYPE_CHECKING + +from pydantic import PrivateAttr + +from fmtr.tools import dm + +if TYPE_CHECKING: + from haco.capabilities import Capability + + +class AnnounceTopic(dm.Base): + IO: ClassVar[str | None] = None + key: str = '{capability}_{io}_topic' + value: str = '{path}/{capability}/{io}' + + _capability: Capability = PrivateAttr(default=None) + + @property + def capability(self) -> Capability: # todo. quite ugly. perhaps just use @dataclass for topics? + return self._capability + + @capability.setter + def capability(self, value: Capability): + self._capability = value + + def fill(self, mask): + return mask.format(**self.fills) + + @property + def fills(self): + return dict( + path=str(self.capability.control.device.name_san), + capability=self.capability.name, + io=self.IO + ) + + @cached_property + def topic_key(self): + return self.fill(self.key) + + @cached_property + def topic(self): + return self.fill(self.value) + + @property + def announce(self): + data = {self.topic_key: self.topic} + return data + + @property + def callback_name(self): + return f'{self.capability.name}_{self.IO}' + + +class AnnounceTopicState(AnnounceTopic): + IO = 'state' + + @property + def subscriptions(self): + return {} + + +class AnnounceTopicCommand(AnnounceTopic): + IO = 'command' + + @property + def subscriptions(self): + method = getattr(self.capability.control, self.callback_name, None) + if not method: + return {} + + return {self.topic: method} diff --git a/haco/utils.py b/haco/utils.py new file mode 100644 index 0000000..99c0be2 --- /dev/null +++ b/haco/utils.py @@ -0,0 +1,27 @@ +import string + +CHARS_ALLOWED = string.ascii_lowercase + string.digits +SEPS = '_- /' +BRANCH = 'release' + +CAPABILITY_DEFAULT = 'default' +RECONNECT_DELAY_SEC = 5 +ON = 'ON' +OFF = 'OFF' + + +def sanitize_name(s, sep='-'): + chars = [] + + for c in s.lower(): + if c in SEPS: + chars.append(sep) + elif c in CHARS_ALLOWED: + chars.append(c) + + sanitized = ''.join(chars) + + if not sanitized: + raise ValueError("Sanitized string is empty") + + return sanitized