Simplify implementation using dataclasses etc

haco/base.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, fields
+from typing import Any
+
+
+@dataclass(kw_only=True)
+class Base:
+    def model_dump(self) -> dict[str, Any]:
+        data = {}
+        for fld in fields(self):
+            meta = fld.metadata
+            if meta.get('exclude', False):
+                continue
+
+            value = getattr(self, fld.name)
+
+            if hasattr(value, self.model_dump.__name__):
+                value = value.model_dump()
+            elif isinstance(value, list):
+                value = [v.model_dump() if hasattr(v, self.model_dump.__name__) else v for v in value]
+
+            if value is not None:
+                data[fld.name] = value
+        return data

haco/capabilities.py

@@ -1,39 +1,36 @@
-from typing import Optional
+from __future__ import annotations
 
-from pydantic import Field, model_validator
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
 
-from fmtr.tools import dm
-from fmtr.tools.iterator_tools import strip_none
-from haco.control import Control
+from haco.base import Base
 from haco.topics import AnnounceTopicState, AnnounceTopicCommand
 
+if TYPE_CHECKING:
+    from haco.control import Control
 
-class Capability(dm.Base):
+
+@dataclass(kw_only=True)
+class Capability(Base):
     name: str
-    alias: str = None
 
-    state: AnnounceTopicState | None = Field(default_factory=AnnounceTopicState)
-    command: AnnounceTopicCommand | None = Field(default_factory=AnnounceTopicCommand)
+    state: AnnounceTopicState | None = field(default_factory=AnnounceTopicState)
+    command: AnnounceTopicCommand | None = field(default_factory=AnnounceTopicCommand)
+    parent: Control | None = None
+    announce = None
 
-    control: Optional[Control] = Field(None, exclude=True, repr=False)
+    def set_parent(self, control):
+        self.parent = control
+        for topic in (t for t in (self.state, self.command) if t):
+            topic.set_parent(self)
+        self.announce = self.get_announce()
 
-    @property
-    def announce(self) -> dict:
+    def get_announce(self) -> dict:
         data = {}
-        for topic in strip_none(self.state, self.command):
+        for topic in (t for t in (self.state, self.command) if t):
             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
+    def control(self):
+        return self.parent

haco/client.py

@@ -0,0 +1,16 @@
+from fmtr.tools import mqtt
+from haco import constants
+
+
+class ClientHaco(mqtt.Client):
+
+    def __init__(self, *args, device=None, will=None, client_id=None, **kwargs):
+        self.will = will or mqtt.Will(topic=str(constants.TOPIC_AVAIL), payload="offline", retain=True, qos=1, )
+        client_id = client_id or constants.CLIENT_ID
+        self.device = device
+        self.device.set_parent(self)
+        super().__init__(*args, will=self.will, client_id=client_id, **kwargs)
+
+    @property
+    def topic(self):
+        return constants.TOPIC_CLIENT

haco/climate.py

@@ -1,9 +1,11 @@
-from typing import Literal, Optional, List
+from dataclasses import dataclass
+from typing import List, Optional, Literal
 
-from haco.capabilities import Capability
-from haco.control import Control
+from capabilities import Capability
+from control import Control
 
 
+@dataclass(kw_only=True)
 class Climate(Control):
     platform: Literal["climate"] = "climate"
     temperature_unit: Literal["C", "F"] = "C"
@@ -21,7 +23,7 @@ class Climate(Control):
         return value
 
     @classmethod
-    def get_capabilities(self):
+    def get_capabilities(cls):
         return [
             Capability(name="current_temperature", command=None),
             Capability(name="temperature"),

haco/constants.py

@@ -11,3 +11,7 @@ CLIENT_ID = f'{HOSTNAME}-{MAC.hex}'
 TOPIC_ROOT = Path(paths.name) / env.CHANNEL
 TOPIC_CLIENT = TOPIC_ROOT / CLIENT_ID
 TOPIC_AVAIL = TOPIC_CLIENT / STATUS
+MQTT_HOST = "mqtt.service"
+MQTT_PORT = 1883
+MQTT_USERNAME = "user"
+MQTT_PASSWORD = None

haco/control.py

@@ -1,36 +1,52 @@
 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 dataclasses import dataclass, field
+from typing import TYPE_CHECKING
 
+from haco.base import Base
+from haco.capabilities import Capability
 from haco.utils import sanitize_name
 
-class Control(dm.Base):
+if TYPE_CHECKING:
+    from haco.device import Device
+
+
+@dataclass(kw_only=True)
+class Control(Base):
     name: str
     platform: str
 
-    icon: Optional[str] = None
+    device: Device = field(default=None)
+    capabilities: None | list[Capability] = field(default=None, metadata=dict(exclude=True))
+    announce: dict | None = field(default=None, metadata=dict(exclude=True))
+
+    def __post_init__(self):
+        self.caps = self.get_capabilities()
+
+    def set_parent(self, device):
+        self.device = device
+        for cap in self.caps:
+            cap.set_parent(self)
+        self.announce = self.get_announce()
+
 
-    _device: Device = PrivateAttr(default=None)
 
-    @computed_field
     @property
-    def device(self) -> Device:  # todo. quite ugly. perhaps just use @dataclass for topics?
-        return self._device
+    def parent(self):
+        return self.device
 
-    @device.setter
-    def device(self, value: Device):
-        self._device = value
+    def get_announce(self):
+        data = self.model_dump()
+        for cap in self.caps:
+            data |= cap.announce
+
+        data = {self.announce_topic: data}
+        return data
+
+    @classmethod
+    def get_capabilities(cls):
+        raise NotImplementedError()
 
-    @computed_field
     @property
     def unique_id(self) -> str:
         return f"{self.device.name_san}-{self.name_san}"
@@ -41,27 +57,16 @@ class Control(dm.Base):
 
     @property
     def topic(self):
-        return self.device.topic / self.name_sanitized
+        return self.device.topic / self.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.client.will.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:
 
@@ -71,18 +76,3 @@ class Control(dm.Base):
             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

@@ -1,26 +1,38 @@
 from __future__ import annotations
 
+from dataclasses import dataclass, field
 from functools import cached_property
-from typing import List
+from typing import List, TYPE_CHECKING
 
-from pydantic import Field, computed_field, model_validator
+from pydantic import computed_field
 
-from fmtr.tools import dm
-from haco import constants
+from haco.base import Base
 from haco.control import Control
 from haco.utils import sanitize_name
 
+if TYPE_CHECKING:
+    from haco.client import ClientHaco
 
-class Device(dm.Base):
+
+@dataclass(kw_only=True)
+class Device(Base):
     name: str
     manufacturer: str = "Demo"
     model: str = "Python MQTT Example"
 
-    controls: List[Control] = Field(default_factory=list, exclude=True, repr=False)
+    controls: List[Control] = field(default_factory=list, metadata=dict(exclude=True))
+    parent: ClientHaco | None = field(metadata=dict(exclude=True))
+    announce: dict | None = field(default=None, metadata=dict(exclude=True))
+
+    def set_parent(self, client):
+        self.parent = client
+        for control in self.controls:
+            control.set_parent(self)
+        self.announce = self.get_announce()
 
     @cached_property
     def topic(self):
-        return constants.TOPIC_CLIENT / self.name_san
+        return self.client.topic / self.name_san
 
     @property
     def name_san(self) -> str:
@@ -37,20 +49,16 @@ class Device(dm.Base):
     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 announce(self) -> dict:
+    def get_announce(self):
         data = {}
-        for capability in self.capabilities:
+        for capability in self.controls:
             data |= capability.announce
-
         return data
 
+    @property
+    def client(self):
+        return self.parent
+
     @cached_property
     def subscriptions(self) -> dict:
         data = {}

haco/sb1.py

@@ -3,10 +3,7 @@ import json
 
 import aiomqtt
 
-MQTT_HOST = "mqtt.service"
-MQTT_PORT = 1883
-MQTT_USERNAME = "user"
-MQTT_PASSWORD = None
+from haco.constants import MQTT_HOST, MQTT_PORT, MQTT_USERNAME, MQTT_PASSWORD
 
 DEVICE_ID = "haco_sandbox"
 ENTITY_ID = "demo_switch"

haco/sb3.py

@@ -0,0 +1,20 @@
+from haco.client import ClientHaco
+from haco.climate import Climate
+from haco.constants import MQTT_HOST
+from haco.device import Device
+
+client = ClientHaco(
+
+    hostname=MQTT_HOST,
+
+    device=Device(
+        parent=None,
+        name='dev-device',
+        controls=[
+            Climate(name='Dev Climate A', platform='climate'),
+            Climate(name='Dev Climate B', platform='climate'),
+        ]
+    )
+)
+
+client

haco/topics.py

@@ -1,33 +1,25 @@
 from __future__ import annotations
 
+from dataclasses import dataclass, field
 from functools import cached_property
 from typing import ClassVar, TYPE_CHECKING
 
-from pydantic import PrivateAttr
-
-from fmtr.tools import dm, env, Path
-from haco.paths import paths
-
 if TYPE_CHECKING:
-    from haco.capabilities import Capability
-
-TOPIC_ROOT = Path(paths.name) / env.CHANNEL
+    from capabilities import Capability
 
 
-class AnnounceTopic(dm.Base):
+@dataclass(kw_only=True)
+class AnnounceTopic:
     IO: ClassVar[str | None] = None
     key: str = '{capability}_{io}_topic'
     value: str = '{path}/{capability}/{io}'
 
-    _capability: Capability = PrivateAttr(default=None)
+    parent: Capability | None = None
+    announce: dict | None = field(default=None, metadata=dict(exclude=True))
 
-    @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 set_parent(self, control):
+        self.parent = control
+        self.announce = self.get_announce()
 
     def fill(self, mask):
         return mask.format(**self.fills)
@@ -48,31 +40,38 @@ class AnnounceTopic(dm.Base):
     def topic(self):
         return self.fill(self.value)
 
-    @property
-    def announce(self):
-        data = {self.topic_key: self.topic}
-        return data
+    def get_announce(self):
+        return {self.topic_key: self.topic}
 
     @property
     def callback_name(self):
         return f'{self.capability.name}_{self.IO}'
 
+    @property
+    def subscriptions(self):
+        return {}
 
+    @cached_property
+    def capability(self):
+        return self.parent
+
+
+@dataclass(kw_only=True)
 class AnnounceTopicState(AnnounceTopic):
-    IO = 'state'
+    IO: ClassVar[str] = 'state'
 
     @property
     def subscriptions(self):
         return {}
 
 
+@dataclass(kw_only=True)
 class AnnounceTopicCommand(AnnounceTopic):
-    IO = 'command'
+    IO: ClassVar[str] = '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

@@ -1,4 +1,5 @@
 import string
+from dataclasses import dataclass
 
 CHARS_ALLOWED = string.ascii_lowercase + string.digits
 SEPS = '_- /'
@@ -25,3 +26,8 @@ def sanitize_name(s, sep='-'):
         raise ValueError("Sanitized string is empty")
 
     return sanitized
+
+
+@dataclass
+class Metadata:
+    exclude: bool = False