Add prototype rewrite data structure, imports, typing etc

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

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"),
+        ]

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()

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

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}

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