rt/haco
1
0
Fork 0
mirror of https://github.com/fmtr/haco.git synced 2026-10-25 18:40:15 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add prototype doc

Browse source
This commit is contained in:
Edward Brown 2026-04-29 10:47:13 +01:00
parent 8dc96d708d
commit f42becceb0
24 changed files with 657 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

32
README.md
View file

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

99
docs/controls.md Normal file
View file

@ -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`.

7
docs/index.md Normal file
View file

@ -0,0 +1,7 @@
!!! tip "How Do I Get Started?"
See the [Quick Start](quick-start.md) section.
--8<--
README.md
--8<--

48
docs/quick-start.md Normal file
View file

@ -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).

11
haco/base.py
View file

@ -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

26
haco/binary_sensor.py
View file

@ -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)
]

19
haco/button.py
View file

@ -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)
]

37
haco/capabilities.py
View file

@ -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

26
haco/client.py
View file

@ -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:

53
haco/climate.py
View file

@ -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"),

6
haco/constants.py
View file

@ -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'

70
haco/control.py
View file

@ -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

46
haco/device.py
View file

@ -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

14
haco/meta.json
View file

@ -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

19
haco/number.py
View file

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

9
haco/obs.py
View file

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

5
haco/paths.py
View file

@ -1,3 +1,8 @@
"""
Path definitions for the haco library.
"""
from corio import sets
paths = sets.PackagePaths()

19
haco/select.py
View file

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

27
haco/sensor.py
View file

@ -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)
]

18
haco/switch.py
View file

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

19
haco/text.py
View file

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

58
haco/topics.py
View file

@ -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}

2
haco/uom.py
View file

@ -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
"""

38
haco/utils.py
View file

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