Add sandbox1 and 2

haco/constants.py

@@ -0,0 +1,13 @@
+from fmtr.tools import env, Path, net
+from haco.paths import paths
+
+STATUS = 'status'
+HOSTNAME = net.get_hostname().lower()
+F = net.get_fqdn()
+MAC = net.MAC()
+
+CLIENT_ID = f'{HOSTNAME}-{MAC.hex}'
+
+TOPIC_ROOT = Path(paths.name) / env.CHANNEL
+TOPIC_CLIENT = TOPIC_ROOT / CLIENT_ID
+TOPIC_AVAIL = TOPIC_CLIENT / STATUS

haco/control.py

@@ -13,7 +13,6 @@ if TYPE_CHECKING:
 
 from haco.utils import sanitize_name
 
-
 class Control(dm.Base):
     name: str
     platform: str
@@ -34,8 +33,15 @@ class Control(dm.Base):
     @computed_field
     @property
     def unique_id(self) -> str:
-        name_san = sanitize_name(self.name)
-        return f"{self.device.name_san}-{name_san}"
+        return f"{self.device.name_san}-{self.name_san}"
+
+    @property
+    def name_san(self) -> str:
+        return sanitize_name(self.name)
+
+    @property
+    def topic(self):
+        return self.device.topic / self.name_sanitized
 
     @property
     def announce_topic(self):
@@ -44,7 +50,7 @@ class Control(dm.Base):
     @computed_field
     @property
     def availability_topic(self) -> str:
-        return self.device.availability_topic
+        return self.device.client.will.topic
 
     @property
     def announce(self) -> dict:

haco/device.py

@@ -6,6 +6,7 @@ from typing import List
 from pydantic import Field, computed_field, model_validator
 
 from fmtr.tools import dm
+from haco import constants
 from haco.control import Control
 from haco.utils import sanitize_name
 
@@ -17,13 +18,14 @@ class Device(dm.Base):
 
     controls: List[Control] = Field(default_factory=list, exclude=True, repr=False)
 
+    @cached_property
+    def topic(self):
+        return constants.TOPIC_CLIENT / self.name_san
+
     @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
@@ -42,8 +44,12 @@ class Device(dm.Base):
         return self
 
     @cached_property
-    def announces(self) -> list[dict]:
-        return [control.announce for control in self.controls]
+    def announce(self) -> dict:
+        data = {}
+        for capability in self.capabilities:
+            data |= capability.announce
+
+        return data
 
     @cached_property
     def subscriptions(self) -> dict:

haco/paths.py

@@ -0,0 +1,3 @@
+from fmtr.tools import sets
+
+paths = sets.PackagePaths()

haco/sb1.py

@@ -0,0 +1,89 @@
+import asyncio
+import json
+
+import aiomqtt
+
+MQTT_HOST = "mqtt.service"
+MQTT_PORT = 1883
+MQTT_USERNAME = "user"
+MQTT_PASSWORD = None
+
+DEVICE_ID = "haco_sandbox"
+ENTITY_ID = "demo_switch"
+DEVICE_NAME = "Haco Sandbox Device"
+ENTITY_NAME = "Demo Switch"
+
+# Topics
+CONFIG_TOPIC = f"homeassistant/switch/{ENTITY_ID}/config"
+STATE_TOPIC = f"haco/{DEVICE_ID}/switch/state"
+COMMAND_TOPIC = f"haco/{DEVICE_ID}/switch/set"
+AVAIL_TOPIC = f"haco/{DEVICE_ID}/status"
+
+
+async def publish_discovery(client):
+    """Announce the entity to Home Assistant (retained)."""
+    config = {
+        "name": ENTITY_NAME,
+        "unique_id": ENTITY_ID,
+        "state_topic": STATE_TOPIC,
+        "command_topic": COMMAND_TOPIC,
+        "availability_topic": AVAIL_TOPIC,
+        "payload_on": "ON",
+        "payload_off": "OFF",
+        "payload_available": "online",
+        "payload_not_available": "offline",
+        "device": {
+            "identifiers": [DEVICE_ID],
+            "name": DEVICE_NAME,
+            "manufacturer": "Demo",
+            "model": "Python MQTT Example"
+        }
+    }
+
+    await client.publish(CONFIG_TOPIC, json.dumps(config), retain=True)
+    print(f"📢 Announced entity via {CONFIG_TOPIC}")
+
+
+async def handle_commands(client):
+    """Listen for command messages from HA."""
+    async with client.messages() as messages:
+        await client.subscribe(COMMAND_TOPIC)
+        async for message in messages:
+            payload = message.payload.decode()
+            print(f"📥 Received command: {payload}")
+
+            # Echo back as new state
+            await client.publish(STATE_TOPIC, payload, retain=True)
+
+
+async def main():
+    while True:
+        try:
+            async with aiomqtt.Client(
+                    hostname=MQTT_HOST,
+                    port=MQTT_PORT,
+                    username=MQTT_USERNAME,
+                    password=MQTT_PASSWORD,
+                    will=aiomqtt.Will(
+                        topic=AVAIL_TOPIC,
+                        payload="offline",
+                        retain=True,
+                        qos=1,
+                    ),
+            ) as client:
+
+                # Online now
+                await client.publish(AVAIL_TOPIC, "online", retain=True)
+                await publish_discovery(client)
+                await client.publish(STATE_TOPIC, "OFF", retain=True)
+
+                # Wait for commands
+                await handle_commands(client)
+
+        except aiomqtt.MqttError as e:
+            print(f"⚠️ MQTT disconnected: {e}; retrying in 5s")
+            await asyncio.sleep(5)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

haco/sb2.py

@@ -0,0 +1,108 @@
+import asyncio
+import random
+
+import aiomqtt
+
+from fmtr.tools import Constants, mqtt
+from fmtr.tools.json_tools import to_json
+from haco import constants
+
+
+async def handle_commands(client, control):
+    """Listen for command messages from HA."""
+
+    subs = control.subscriptions
+
+    async with client.messages() as messages:
+
+        for topic_sub in subs.keys():
+            await client.subscribe(topic_sub)
+
+        async for message in messages:
+            payload = message.payload.decode()
+            print(f"{message.topic.value}{Constants.ARROW}{payload}")
+            method = subs[message.topic.value]
+            echo_val = method(payload)
+            echo_val
+
+            # Echo back as new state
+            # await client.publish(STATE_TOPIC, payload, retain=True)
+
+
+async def background_task(client, control):
+    while True:
+        await asyncio.sleep(10)
+        cap = control.capabilities[1]
+        value = random.randint(18, 24)
+        print(f'human is setting temperature to {value}')
+        await client.publish(cap.state.topic, payload=to_json(value), retain=True)
+
+
+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.client = self
+        super().__init__(*args, will=self.will, client_id=client_id, **kwargs)
+
+
+async def main():
+    # Define capabilities
+
+    # device = Device(
+    #     name="Development Device",
+    #     controls=[
+    #         Climate(
+    #             #id="climate_living_room",
+    #             name="Climate Control",
+    #             fan_modes=['a','b','c'],
+    #         )
+    #     ]
+    #
+    # )
+
+    # announce=device.announce
+    # subs=device.subscriptions
+
+    MQTT_HOST = "mqtt.service"
+    MQTT_PORT = 1883
+    MQTT_USERNAME = "user"
+    MQTT_PASSWORD = None
+
+    while True:
+        try:
+            async with ClientHaco(
+                    hostname=MQTT_HOST,
+                    port=MQTT_PORT,
+                    username=MQTT_USERNAME,
+                    password=MQTT_PASSWORD,
+
+                    client_id=constants.CLIENT_ID,
+                    # device=device,
+            ) as client:
+
+                # for topic, data in announce.items():
+                #     data_json = to_json(data)
+                #     print(f'Announcing {topic} with {data_json}')
+                #     await client.publish(topic, payload=data_json, retain=True)
+
+                # Online now
+                await client.publish(client.will.topic, "online", retain=True)
+
+                # await publish_discovery(client)
+                # await client.publish(STATE_TOPIC, "OFF", retain=True)
+
+                # Wait for commands
+                # asyncio.create_task(background_task(client,control=climate))
+                # await handle_commands(client,control=climate)
+                client
+
+        except aiomqtt.MqttError as e:
+            print(f"⚠️ MQTT disconnected: {e}")
+            # await asyncio.sleep(5)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

haco/topics.py

@@ -5,11 +5,14 @@ from typing import ClassVar, TYPE_CHECKING
 
 from pydantic import PrivateAttr
 
-from fmtr.tools import dm
+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
+
 
 class AnnounceTopic(dm.Base):
     IO: ClassVar[str | None] = None
@@ -32,7 +35,7 @@ class AnnounceTopic(dm.Base):
     @property
     def fills(self):
         return dict(
-            path=str(self.capability.control.device.name_san),
+            path=str(self.capability.control.topic),
             capability=self.capability.name,
             io=self.IO
         )

haco/v0/haco/__init__.py

@@ -1,28 +0,0 @@
-from haco import tuya
-from .button import Button
-from .callback import Response
-from .climate import Climate
-from .fan import Fan
-from .number import Number
-from .pulldown import Select
-from .sensor import Sensor
-from .switch import Switch
-from .tasmota import Tasmota
-from .text import Text, Password
-from .update import Update
-
-__all__ = [
-    "Button",
-    "Select",
-    "Sensor",
-    "Tasmota",
-    "Number",
-    "Switch",
-    "Text",
-    "Password",
-    "Update",
-    "Climate",
-    "Fan",
-    "Response",
-    "tuya"
-]

haco/v0/haco/button.py

@@ -1,6 +1,7 @@
-from haco.control import Control
-
+import haco.capabilities
+import haco.topics
 from haco import schema
+from haco.control import Control
 
 
 class Button(Control):
@@ -10,8 +11,8 @@ class Button(Control):
     def get_schema(cls):
         return schema.Schema(
             capabilities=[
-                schema.Capability(
-                    ha=schema.HomeAssistant(announce_data=schema.AnnounceTopic(key='{io_ha}_topic')),
+                haco.capabilities.Capability(
+                    ha=schema.HomeAssistant(announce_data=haco.topics.AnnounceTopic(key='{io_ha}_topic')),
                     tasmota=None
                 )
             ]

haco/v0/haco/control.py

@@ -1,11 +1,12 @@
 import logging
 from functools import partial
 
-import haco.constants
 from haco.callback import Callback
 from haco.schema import Schema, Platform, Tasmota, HomeAssistant, AnnounceTopic, Capability
 from haco.tools import log_publish
 
+import haco.constants
+import haco.utils
 from haco import tools, constants
 
 
@@ -42,7 +43,7 @@ class Control:
 
     @property
     def name_sanitized(self):
-        return tools.sanitize_name(self.name)
+        return haco.utils.sanitize_name(self.name)
 
     def callback(self, trigger=None):
         return partial(self.add_callback, trigger)

haco/v0/haco/device.py

@@ -2,11 +2,12 @@ import json
 from pathlib import Path
 from typing import List
 
-from haco.control import Control
 from haco.device_data import DeviceData
 from haco.tools import log_publish, split_into_chunks, log_subscribe
 
+import haco.utils
 from haco import tools, constants
+from haco.control import Control
 
 
 class Device:
@@ -29,7 +30,7 @@ class Device:
 
     @property
     def name_sanitized(self):
-        return tools.sanitize_name(self.data_announce.name)
+        return haco.utils.sanitize_name(self.data_announce.name)
 
     @property
     def mac_short(self):

haco/v0/haco/sensor.py

@@ -1,9 +1,11 @@
-from haco.control import Control
-
-from haco import schema
 from haco.data.type_sensor import DeviceClassSensor
 from haco.data.uom import Uom
 
+import haco.capabilities
+import haco.topics
+from haco import schema
+from haco.control import Control
+
 
 class Sensor(Control):
     NAME = 'sensor'
@@ -15,9 +17,9 @@ class Sensor(Control):
     def get_schema(cls):
         return schema.Schema(
             capabilities=[
-                schema.Capability(
+                haco.capabilities.Capability(
                     ha=None,
-                    tasmota=schema.Tasmota(announce_data=schema.AnnounceTopic(key='{io_ha}_topic'))
+                    tasmota=schema.Tasmota(announce_data=haco.topics.AnnounceTopic(key='{io_ha}_topic'))
                 )
             ]
         )

haco/v0/haco/tools.py

@@ -62,21 +62,7 @@ def invert_bool(value):
     return {True: 'ON', False: 'OFF', 'ON': True, 'OFF': False}[value]
 
 
-def sanitize_name(s, sep='-'):
-    chars = []
 
-    for c in s.lower():
-        if c in constants.SEPS:
-            chars.append(sep)
-        elif c in constants.CHARS_ALLOWED:
-            chars.append(c)
-
-    sanitized = ''.join(chars)
-
-    if not sanitized:
-        raise ValueError("Sanitized string is empty")
-
-    return sanitized
 
 
 def add_tuya_io(control, control_type, type_id, dp_id):