diff --git a/README.md b/README.md index f3bb28c..408d50d 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,30 @@ # `haco` Home Assistant Control Objects -`haco` is a Python/Berry library to greatly simplify the process of -exposing Home -Assistant controls (e.g. Pull-down Lists, Number Sliders, Sensors, etc.) from a device - and handling the -communication between both sides. +`haco` is a to greatly simplify the process of exposing Home Assistant controls (e.g. Pull-down Lists, Number Sliders, Sensors, etc.) from a device - and handling the communication between both sides. -It is in very early release. Documentation is (may) be forthcoming. +While the current implementation is in Python, the capability definitions for controls are designed to be shared across different languages. Support for additional languages (namely native support for Tasmota Berry) is planned. + +### Why MQTT? + +You might wonder why we use MQTT instead of directly using Home Assistant's APIs. The reason is flexibility: using MQTT allows +`haco` to target any device or environment that has an MQTT client, not just those that can easily make HTTP requests or run a full HA integration. This makes it ideal for lightweight embedded systems and cross-platform utility. + +### Why Would I Want Such a Thing? + +Integrating custom hardware or scripts with Home Assistant often involves a lot of boilerplate: defining MQTT topics, crafting JSON payloads for discovery, and managing state updates. +`haco` handles the heavy lifting, letting you define your device and its controls in clean, declarative Python. + +If you're building a custom IoT device, a script to monitor your system, or just want a simpler way to bridge your code to Home Assistant, then you might find +`haco` useful. + +### How it Works + +`haco` uses MQTT to communicate with Home Assistant. It automatically generates the discovery configurations required for Home Assistant to "see" your device and its entities. All you need to do is define a +`Device` with some `Controls` and start the `ClientHaco`. + +## Documentation + +[See Documentation](docs/index.md) + +- [Quick Start](docs/quick-start.md) +- [Control Types](docs/controls.md) diff --git a/docs/controls.md b/docs/controls.md new file mode 100644 index 0000000..07beb5c --- /dev/null +++ b/docs/controls.md @@ -0,0 +1,99 @@ +`haco` provides several control types that correspond to Home Assistant platforms. Each control typically has a +`state` and a `command` method that you can override to handle communication. + +# Supported Controls + +## Button + +A button in Home Assistant that can be pressed to trigger an action in your script. + +- **Platform**: `button` +- **Overridable Methods**: + - `command(self, value)`: Called when the button is pressed in Home Assistant. + +```python +from haco import Button + +class MyButton(Button): + def command(self, value): + print("Button was pressed!") +``` + +## Switch + +A toggle switch that can be turned on or off. + +- **Platform**: `switch` +- **Overridable Methods**: + - `state(self, value)`: Should return the current state of the switch (`True` for on, `False` for off). + - `command(self, value)`: Called when the switch is toggled in Home Assistant. + +```python +from haco import Switch + +class MySwitch(Switch): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._on = False + + def state(self, value): + return self._on + + def command(self, value): + self._on = value + print(f"Switch is now {'ON' if self._on else 'OFF'}") +``` + +## Sensor + +A read-only sensor that displays a value in Home Assistant. + +- **Platform**: `sensor` +- **Overridable Methods**: + - `state(self, value)`: Should return the current value of the sensor. + +```python +from haco import Sensor + +class MySensor(Sensor): + def state(self, value): + return 42 # Or some dynamic value +``` + +## Number + +A slider or input box for numeric values. + +- **Platform**: `number` +- **Overridable Methods**: + - `state(self, value)`: Should return the current numeric value. + - `command(self, value)`: Called when the value is changed in Home Assistant. + +## Select + +A dropdown list of options. + +- **Platform**: `select` +- **Parameters**: `options` (list of strings) +- **Overridable Methods**: + - `state(self, value)`: Should return the currently selected option. + - `command(self, value)`: Called when a new option is selected in Home Assistant. + +# Customizing Controls + +Most controls accept common parameters in their constructor: + +- `name`: The display name in Home Assistant. +- `icon`: An optional MDI icon (e.g., `mdi:lightbulb` or just `lightbulb`). + +Example: + +```python +btn = Button(name="Panic Button", icon="alert") +``` + +# Advanced: Custom Capabilities + +For more complex platforms like `climate`, +`haco` uses "Capabilities". A capability groups a state and a command topic. You can see how these are implemented in +`haco/climate.py`. diff --git a/docs/index.md b/docs/index.md new file mode 100644 index 0000000..2ffdb68 --- /dev/null +++ b/docs/index.md @@ -0,0 +1,7 @@ +!!! tip "How Do I Get Started?" + + See the [Quick Start](quick-start.md) section. + +--8<-- +README.md +--8<-- diff --git a/docs/quick-start.md b/docs/quick-start.md new file mode 100644 index 0000000..28be7b8 --- /dev/null +++ b/docs/quick-start.md @@ -0,0 +1,48 @@ +The easiest way to get started with `haco` is to define a simple device and run the client. + +# Installation + +`haco` can be installed via pip: + +```bash +pip install haco +``` + +!!! tip "MQTT Broker Required" + + `haco` communicates with Home Assistant via MQTT. Ensure you have an MQTT broker (like Mosquitto) running and accessible. + +# Basic Usage + +Here is a minimal example that creates a device with a single button: + +```python +import asyncio +from haco import Button, Device, ClientHaco + +async def main(): + # 1. Define your controls + btn = Button(name="My Awesome Button") + + # 2. Group them into a device + device = Device(name="My Cool Device", controls=[btn]) + + # 3. Create and start the client + # Replace 'mqtt.service' with your MQTT broker's address + client = ClientHaco(hostname='mqtt.service', device=device) + + print("Starting haco client...") + await client.start() + +if __name__ == "__main__": + asyncio.run(main()) +``` + +# What Happens Next? + +1. **Discovery**: When the script runs, it sends discovery messages to Home Assistant. +2. **Entity Created + **: A new device named "My Cool Device" will appear in Home Assistant with a button entity named "My Awesome Button". +3. **Control**: When you press the button in the Home Assistant UI, + `haco` will receive the command. You can then handle this command in your Python script by overriding the + `command` method or providing a callback (see [Control Types](controls.md) for more details). diff --git a/haco/base.py b/haco/base.py index 2828af6..a24c157 100644 --- a/haco/base.py +++ b/haco/base.py @@ -6,9 +6,20 @@ from typing import Any @dataclass(kw_only=True) class Base: + """ + + Base class for haco objects. Provides model dumping functionality. + + """ + DATA = dict() def model_dump(self) -> dict[str, Any]: + """ + + Dump the object to a dictionary, excluding fields marked with `exclude=True` in metadata. + + """ data = {} | self.DATA for fld in fields(self): meta = fld.metadata diff --git a/haco/binary_sensor.py b/haco/binary_sensor.py index 15b5f24..480fa20 100644 --- a/haco/binary_sensor.py +++ b/haco/binary_sensor.py @@ -45,20 +45,40 @@ class DeviceClass(StrEnum): @dataclass(kw_only=True) class BinarySensor(Control): + """ + + A binary sensor for reporting on/off states to Home Assistant. + + """ DATA = dict( platform='binary_sensor' ) device_class: DeviceClass | None = None - def command(self, value): + def command(self, value: bool): + """ + + Binary sensors do not support commands. + + """ raise NotImplementedError() - def state(self, value): + def state(self, value: bool | None) -> bool: + """ + + Return the current state of the binary sensor. + + """ raise NotImplementedError() @classmethod - def get_capabilities(cls): + def get_capabilities(cls) -> list[Capability]: + """ + + Get the capabilities for the binary sensor (state only). + + """ return [ Capability(name=None, command=None) ] diff --git a/haco/button.py b/haco/button.py index 9364177..2329a64 100644 --- a/haco/button.py +++ b/haco/button.py @@ -6,15 +6,30 @@ from haco.control import Control @dataclass(kw_only=True) class Button(Control): + """ + + A button control that can be pressed in Home Assistant. + + """ DATA = dict( platform='button' ) - def command(self, value): + def command(self, value: None): + """ + + Handle button press commands from Home Assistant. + + """ raise NotImplementedError() @classmethod - def get_capabilities(cls): + def get_capabilities(cls) -> list[Capability]: + """ + + Get the capabilities for the button (command only). + + """ return [ Capability(name=None, state=None) ] diff --git a/haco/capabilities.py b/haco/capabilities.py index e30e62d..cf6506c 100644 --- a/haco/capabilities.py +++ b/haco/capabilities.py @@ -4,6 +4,7 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Type from corio.iterator_tools import strip_none + from haco.base import Base from haco.topics import AnnounceTopicState, AnnounceTopicCommand from haco.utils import Converters @@ -14,6 +15,11 @@ if TYPE_CHECKING: @dataclass(kw_only=True) class Capability(Base): + """ + + A capability groups a state and a command topic. + + """ name: str | None = None converters: Type[Converters] | None = field(default=None, metadata=dict(exclude=True)) @@ -25,7 +31,12 @@ class Capability(Base): parent: Control | None = None announce = None - def set_parent(self, control): + def set_parent(self, control: Control): + """ + + Set the parent control for the capability and its topics. + + """ self.parent = control for topic in self.topics: topic.set_parent(self) @@ -33,21 +44,41 @@ class Capability(Base): self.converters = self.converters or self.control.converters def get_announce(self) -> dict: + """ + + Get the discovery announcement data for the capability's topics. + + """ data = {} for topic in self.topics: data |= topic.get_announce() return data def get_subscriptions(self) -> dict: + """ + + Get the MQTT topic subscriptions for the capability's command topics. + + """ data = {} for topic in self.topics: data |= topic.get_subscriptions() return data @property - def topics(self): + def topics(self) -> list: + """ + + The state and command topics for the capability. + + """ return strip_none(self.state, self.command) @property - def control(self): + def control(self) -> Control: + """ + + The parent control of the capability. + + """ return self.parent diff --git a/haco/client.py b/haco/client.py index ab319c1..02d1654 100644 --- a/haco/client.py +++ b/haco/client.py @@ -1,7 +1,9 @@ -import aiomqtt import asyncio +from pathlib import Path +import aiomqtt from corio import mqtt + from haco import constants from haco.constants import SUBSCRIBE from haco.obs import logger @@ -9,6 +11,11 @@ from haco.utils import get_prefix class ClientHaco(mqtt.Client): + """ + + MQTT Client for haco, handling communication between Home Assistant and a device. + + """ ONLINE = 'online' OFFLINE = 'offline' @@ -20,10 +27,20 @@ class ClientHaco(mqtt.Client): super().__init__(*args, will=self.will, identifier=identifier, **kwargs) @property - def topic(self): + def topic(self) -> Path: + """ + + The MQTT topic prefix for the client. + + """ return constants.TOPIC_CLIENT async def handle(self): + """ + + Handle incoming MQTT messages by delegating them to the device's controls. + + """ for topic_sub in self.device.subscriptions.keys(): logger.info(f"{get_prefix(SUBSCRIBE)}: {topic_sub}") @@ -36,6 +53,11 @@ class ClientHaco(mqtt.Client): async def start(self): + """ + + Start the MQTT client and handle reconnection. + + """ while True: try: async with self: diff --git a/haco/climate.py b/haco/climate.py index f18d880..69635da 100644 --- a/haco/climate.py +++ b/haco/climate.py @@ -7,6 +7,11 @@ from haco.control import Control @dataclass(kw_only=True) class Climate(Control): + """ + + A climate control for managing HVAC systems. + + """ DATA = dict( platform='climate' ) @@ -23,27 +28,61 @@ class Climate(Control): fan_modes: list[str] | None = None swing_modes: list[str] | None = None + def temperature_command(self, value: float) -> float: + """ - def temperature_command(self, value): + Handle target temperature commands. + + """ raise NotImplementedError() - def mode_command(self, value): + def mode_command(self, value: str) -> str: + """ + + Handle HVAC mode commands. + + """ raise NotImplementedError() - def fan_mode_command(self, value): + def fan_mode_command(self, value: str) -> str: + """ + + Handle fan mode commands. + + """ raise NotImplementedError() - def fan_mode_state(self, value): + def fan_mode_state(self, value: str | None) -> str: + """ + + Return the current fan mode. + + """ raise NotImplementedError() - def power_command(self, value): + def power_command(self, value: bool) -> bool: + """ + + Handle power toggle commands. + + """ raise NotImplementedError() - def power_state(self, value): + def power_state(self, value: bool | None) -> bool: + """ + + Return the current power state. + + """ raise NotImplementedError() @classmethod - def get_capabilities(cls): + def get_capabilities(cls) -> list[Capability]: + """ + + Get the capabilities for the climate control. + + """ return [ Capability(name="current_temperature", command=None), Capability(name="temperature"), diff --git a/haco/constants.py b/haco/constants.py index 0e39799..989496f 100644 --- a/haco/constants.py +++ b/haco/constants.py @@ -1,6 +1,12 @@ +""" + +Global constants for the haco library. + +""" from __future__ import annotations from corio import env, Path, net + from haco.paths import paths STATUS = 'status' diff --git a/haco/control.py b/haco/control.py index c4e9636..b635c3c 100644 --- a/haco/control.py +++ b/haco/control.py @@ -4,7 +4,9 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, TypeVar, Generic, Type, ClassVar from corio import Constants +from corio import Path from corio.json_tools import to_json + from haco.base import Base from haco.capabilities import Capability from haco.constants import PREFIX_MDI, ANNOUNCE @@ -19,6 +21,11 @@ DeviceT = TypeVar("DeviceT", bound="Device") @dataclass(kw_only=True) class Control(Base, Generic[DeviceT]): + """ + + Base class for Home Assistant controls. + + """ converters: ClassVar[Type[Converters]] = ConvertersBool name: str @@ -39,8 +46,12 @@ class Control(Base, Generic[DeviceT]): if not self.icon.startswith(PREFIX_MDI): self.icon = f"{PREFIX_MDI}{self.icon}" + def set_parent(self, device: DeviceT): + """ - def set_parent(self, device): + Set the parent device for the control and its capabilities. + + """ self.device = device for cap in self.capabilities: @@ -51,10 +62,20 @@ class Control(Base, Generic[DeviceT]): self.subscriptions = self.get_subscriptions() @property - def parent(self): + def parent(self) -> DeviceT: + """ + + The parent device of the control. + + """ return self.device - def get_announce(self): + def get_announce(self) -> dict: + """ + + Get the discovery announcement data for the control. + + """ data = self.model_dump() for cap in self.capabilities: data |= cap.get_announce() @@ -62,7 +83,12 @@ class Control(Base, Generic[DeviceT]): return data @classmethod - def get_capabilities(cls): + def get_capabilities(cls) -> list[Capability]: + """ + + Get the capabilities of the control. + + """ return [ Capability(name=None) ] @@ -72,20 +98,45 @@ class Control(Base, Generic[DeviceT]): @property def name_san(self) -> str: + """ + + Sanitized name of the control. + + """ return sanitize_name(self.name) @property - def topic(self): + def topic(self) -> Path: + """ + + The MQTT topic prefix for the control. + + """ return self.device.topic / self.name_san @property - def announce_topic(self): + def announce_topic(self) -> str: + """ + + The MQTT topic for discovery announcement. + + """ return f"homeassistant/{self.DATA['platform']}/{self.unique_id}/config" def get_availability_topic(self) -> str: + """ + + Get the availability topic for the control (uses the client's will topic). + + """ return self.device.client.will.topic def get_subscriptions(self) -> dict: + """ + + Get the MQTT topic subscriptions for the control's capabilities. + + """ data = {} @@ -95,6 +146,11 @@ class Control(Base, Generic[DeviceT]): return data async def initialise(self): + """ + + Initialise the control by announcing it and setting its initial state. + + """ await self.announce() for capability in self.capabilities: if not capability.state: @@ -104,7 +160,7 @@ class Control(Base, Generic[DeviceT]): async def announce(self): """ - Entities must be able to announce themselves to Home Assistant, e.g. when a pulldown changes its options, etc. + Announce the control to Home Assistant via MQTT discovery. """ topic = self.announce_topic diff --git a/haco/device.py b/haco/device.py index e0587ac..e66cffd 100644 --- a/haco/device.py +++ b/haco/device.py @@ -2,6 +2,7 @@ from __future__ import annotations from dataclasses import dataclass, field from functools import cached_property +from pathlib import Path from typing import List, TYPE_CHECKING from haco.base import Base @@ -14,6 +15,11 @@ if TYPE_CHECKING: @dataclass(kw_only=True) class Device(Base): + """ + + A Home Assistant device, containing one or more controls. + + """ name: str manufacturer: str | None = None model: str | None = None @@ -26,8 +32,12 @@ class Device(Base): parent: ClientHaco | None = field(metadata=dict(exclude=True), init=False) subscriptions: dict | None = field(default=None, metadata=dict(exclude=True), init=False) + def set_parent(self, client: ClientHaco): + """ - def set_parent(self, client): + Set the parent client for the device and its controls. + + """ self.parent = client self.identifiers = [self.name_san] @@ -37,27 +47,57 @@ class Device(Base): self.subscriptions = self.get_subscriptions() @cached_property - def topic(self): + def topic(self) -> Path: + """ + + The MQTT topic prefix for the device. + + """ return self.client.topic / self.name_san @property def name_san(self) -> str: + """ + + Sanitized name of the device. + + """ return sanitize_name(self.name) async def announce(self): + """ + + Announce all controls of the device to Home Assistant. + + """ for control in self.controls: await control.announce() async def initialise(self): + """ + + Initialise all controls of the device. + + """ for control in self.controls: await control.initialise() @property - def client(self): + def client(self) -> ClientHaco: + """ + + The parent client of the device. + + """ return self.parent def get_subscriptions(self) -> dict: + """ + + Get all MQTT topic subscriptions for the device's controls. + + """ data = {} for control in self.controls: data |= control.subscriptions diff --git a/haco/meta.json b/haco/meta.json index b35929e..eddd78f 100644 --- a/haco/meta.json +++ b/haco/meta.json @@ -10,7 +10,19 @@ "is_client": false, "scripts": [], "services": [], - "docs": {}, + "docs": { + "nav": [ + { + "Home": "index.md" + }, + { + "Quick Start": "quick-start.md" + }, + { + "Control Types": "controls.md" + } + ] + }, "setup": {}, "is_pypi": true, "is_dockerhub": false diff --git a/haco/number.py b/haco/number.py index e60574a..8604d9b 100644 --- a/haco/number.py +++ b/haco/number.py @@ -14,6 +14,11 @@ class Mode(StrEnum): @dataclass(kw_only=True) class Number(Control): + """ + + A number control (slider or box) for numeric input in Home Assistant. + + """ DATA = dict( platform='number' ) @@ -25,8 +30,18 @@ class Number(Control): step: float = 1. uom: Uom | None = None - def command(self, value): + def command(self, value: float) -> float: + """ + + Handle numeric value commands from Home Assistant. + + """ raise NotImplementedError() - def state(self, value): + def state(self, value: float | None) -> float: + """ + + Return the current numeric state. + + """ raise NotImplementedError() diff --git a/haco/obs.py b/haco/obs.py index 5bf7afa..f8832b6 100644 --- a/haco/obs.py +++ b/haco/obs.py @@ -1,6 +1,11 @@ -from corio import logging, debug, Constants -from haco.paths import paths +""" +Observability and logging setup for the haco library. + +""" +from corio import logging, debug, Constants + +from haco.paths import paths debug.trace() diff --git a/haco/paths.py b/haco/paths.py index 47ebd8b..aa2a0b4 100644 --- a/haco/paths.py +++ b/haco/paths.py @@ -1,3 +1,8 @@ +""" + +Path definitions for the haco library. + +""" from corio import sets paths = sets.PackagePaths() diff --git a/haco/select.py b/haco/select.py index 716bd7c..1a1db28 100644 --- a/haco/select.py +++ b/haco/select.py @@ -6,6 +6,11 @@ from haco.utils import ConvertersString @dataclass(kw_only=True) class Select(Control): + """ + + A select control (dropdown) for choosing from a list of options in Home Assistant. + + """ DATA = dict( platform='select' ) @@ -13,8 +18,18 @@ class Select(Control): options: list[str] - def command(self, value): + def command(self, value: str) -> str: + """ + + Handle option selection commands from Home Assistant. + + """ raise NotImplementedError() - def state(self, value): + def state(self, value: str | None) -> str: + """ + + Return the currently selected option. + + """ raise NotImplementedError() diff --git a/haco/sensor.py b/haco/sensor.py index a430d32..91dc1df 100644 --- a/haco/sensor.py +++ b/haco/sensor.py @@ -1,5 +1,6 @@ from dataclasses import dataclass from enum import StrEnum +from typing import Any from haco.capabilities import Capability from haco.control import Control @@ -68,6 +69,11 @@ class DeviceClass(StrEnum): @dataclass(kw_only=True) class Sensor(Control): + """ + + A sensor control for reporting numeric or string values to Home Assistant. + + """ DATA = dict( platform='sensor' ) @@ -76,14 +82,29 @@ class Sensor(Control): unit_of_measurement: Uom | None = None device_class: DeviceClass | None = None - def command(self, value): + def command(self, value: Any): + """ + + Sensors do not support commands. + + """ raise NotImplementedError() - def state(self, value): + def state(self, value: Any) -> Any: + """ + + Return the current state of the sensor. + + """ raise NotImplementedError() @classmethod - def get_capabilities(cls): + def get_capabilities(cls) -> list[Capability]: + """ + + Get the capabilities for the sensor (state only). + + """ return [ Capability(name=None, command=None) ] diff --git a/haco/switch.py b/haco/switch.py index 6c9ab22..09eb5c4 100644 --- a/haco/switch.py +++ b/haco/switch.py @@ -5,13 +5,27 @@ from haco.control import Control @dataclass(kw_only=True) class Switch(Control): + """ + + A switch control that can be toggled on or off. + + """ DATA = dict( platform='switch' ) + def command(self, value: bool) -> bool: + """ - def command(self, value): + Handle toggle commands from Home Assistant. + + """ raise NotImplementedError() - def state(self, value): + def state(self, value: bool | None) -> bool: + """ + + Return the current state of the switch. + + """ raise NotImplementedError() diff --git a/haco/text.py b/haco/text.py index e9132bb..d511a74 100644 --- a/haco/text.py +++ b/haco/text.py @@ -12,6 +12,11 @@ class Mode(StrEnum): @dataclass(kw_only=True) class Text(Control): + """ + + A text control for text input in Home Assistant. + + """ DATA = dict( platform='text' ) @@ -22,8 +27,18 @@ class Text(Control): max: int | None = None mode: Mode = Mode.TEXT - def command(self, value): + def command(self, value: str) -> str: + """ + + Handle text input commands from Home Assistant. + + """ raise NotImplementedError() - def state(self, value): + def state(self, value: str | None) -> str: + """ + + Return the current text state. + + """ raise NotImplementedError() diff --git a/haco/topics.py b/haco/topics.py index d67c44f..eaed85b 100644 --- a/haco/topics.py +++ b/haco/topics.py @@ -5,6 +5,7 @@ from functools import cached_property from typing import ClassVar, TYPE_CHECKING, Callable from corio import aio, Constants + from haco.obs import logger from haco.utils import get_prefix @@ -14,6 +15,11 @@ if TYPE_CHECKING: @dataclass(kw_only=True) class AnnounceTopic: + """ + + Base class for MQTT topics used for discovery and communication. + + """ IO: ClassVar[str | None] = None KEY_DEFAULTS: ClassVar[dict] = {True: '{capability}_{io}_topic', False: '{io}_topic'} @@ -28,6 +34,11 @@ class AnnounceTopic: control_method: Callable | None = field(default=None, metadata=dict(exclude=True)) def set_parent(self, capability: Capability): + """ + + Set the parent capability for the topic and link it to the control's method. + + """ self.parent = capability if self.key is None: self.key = self.KEY_DEFAULTS[bool(self.capability.name)] @@ -56,7 +67,12 @@ class AnnounceTopic: def topic(self): return self.fill(self.value) - def get_announce(self): + def get_announce(self) -> dict: + """ + + Get the discovery announcement data for the topic. + + """ return {self.topic_key: self.topic} @property @@ -70,7 +86,12 @@ class AnnounceTopic: def callback_class_method_name(self): return f'{self.capability.control.__class__.__name__}.{self.callback_name}' - def get_subscriptions(self): + def get_subscriptions(self) -> dict: + """ + + Get the MQTT topic subscriptions for this topic (empty for base class). + + """ return {} @cached_property @@ -83,12 +104,27 @@ class AnnounceTopic: @dataclass(kw_only=True) class AnnounceTopicState(AnnounceTopic): + """ + + MQTT topic for reporting state to Home Assistant. + + """ IO: ClassVar[str] = 'state' - def get_subscriptions(self): + def get_subscriptions(self) -> dict: + """ + + Get the MQTT topic subscriptions for this state topic (empty as it is output only). + + """ return {} async def handle(self, value=None): + """ + + Handle state reporting. Calls the linked control method and publishes the result. + + """ if not self.control_method: logger.error(f'Incomplete base class: {self.callback_class_method_name}') @@ -115,6 +151,11 @@ class AnnounceTopicState(AnnounceTopic): @dataclass(kw_only=True) class AnnounceTopicCommand(AnnounceTopic): + """ + + MQTT topic for receiving commands from Home Assistant. + + """ IO: ClassVar[str] = 'command' @property @@ -122,6 +163,11 @@ class AnnounceTopicCommand(AnnounceTopic): return self.capability.state async def handle(self, message): + """ + + Handle incoming command messages. Decodes, converts, calls the linked control method, and updates state. + + """ value = message.payload.decode('utf-8') value = self.capability.converters.command(value) @@ -149,6 +195,10 @@ class AnnounceTopicCommand(AnnounceTopic): if topic_state: await topic_state.handle(value_raw) + def get_subscriptions(self) -> dict: + """ - def get_subscriptions(self): + Get the MQTT topic subscriptions for this command topic. + + """ return {self.topic: self} diff --git a/haco/uom.py b/haco/uom.py index 3986da8..dc579ac 100644 --- a/haco/uom.py +++ b/haco/uom.py @@ -4,6 +4,8 @@ from enum import StrEnum class Uom(StrEnum): """ + Units of measurement for sensors. + Sourced from: https://github.com/home-assistant/core/blob/03b1c7ad1deb4b9494df01b27ef8e1d7dfe179ba/homeassistant/const.py#L489 """ diff --git a/haco/utils.py b/haco/utils.py index 5de1833..5c99117 100644 --- a/haco/utils.py +++ b/haco/utils.py @@ -18,7 +18,13 @@ BOOL_MAP = {True: ON, False: OFF} from_bool = BOOL_MAP.get -def sanitize_name(s, sep='-'): + +def sanitize_name(s: str, sep: str = '-') -> str: + """ + + Sanitize a string for use in MQTT topics and Home Assistant identifiers. + + """ chars = [] for c in s.lower(): @@ -36,28 +42,58 @@ def sanitize_name(s, sep='-'): class Converters: + """ + + Base class for message converters. + + """ command: Callable = None state: Callable = None class ConvertersBool(Converters): + """ + + Converters for boolean values. + + """ command: Callable = to_bool state: Callable = from_bool class ConvertersString(Converters): + """ + + Converters for string values. + + """ command: Callable = str state: Callable = str class ConvertersNumeric(Converters): + """ + + Converters for numeric values. + + """ command: Callable = json.from_json state: Callable = json.to_json @dataclass class Metadata: + """ + + Metadata for dataclass fields. + + """ exclude: bool = False def get_prefix(io: str) -> str: + """ + + Get a short prefix for IO operations (e.g. 'ST' for state, 'CO' for command). + + """ return io[:2].upper()