mirrors/pyhOn
1
0
Fork 0
mirror of https://github.com/Andre0512/pyhOn.git synced 2024-12-22 18:55:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

subscribe to updates

Browse source
This commit is contained in:
Andre Basche 2024-03-25 02:14:17 +01:00
parent f108005a4d
commit 7e59f76784
2 changed files with 18 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
pyhon/appliance.py
View file

@ -3,7 +3,7 @@ import logging
import re import re
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path from pathlib import Path
from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload, Callable
from pyhon import diagnose, exceptions from pyhon import diagnose, exceptions
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
@ -43,6 +43,7 @@ class HonAppliance:
self._additional_data: Dict[str, Any] = {} self._additional_data: Dict[str, Any] = {}
self._last_update: Optional[datetime] = None self._last_update: Optional[datetime] = None
self._default_setting = HonParameter("", {}, "") self._default_setting = HonParameter("", {}, "")
self._notify_function: Optional[Callable[[Any], None]] = None
try: try:
self._extra: Optional[ApplianceBase] = importlib.import_module( self._extra: Optional[ApplianceBase] = importlib.import_module(
@ -312,3 +313,11 @@ class HonAppliance:
elif isinstance(target, HonParameterEnum): elif isinstance(target, HonParameterEnum):
target.values = main.values target.values = main.values
target.value = main.value target.value = main.value
def subscribe(self, notify_function: Callable[[Any], None]) -> None:
self._notify_function = notify_function
def notify(self) -> None:
self.sync_params_to_command("settings")
if self._notify_function:
self._notify_function(self.attributes)

16
pyhon/connection/mqtt.py
View file

@ -17,16 +17,15 @@ appliances: list[HonAppliance] = []
def on_lifecycle_stopped(lifecycle_stopped_data: mqtt5.LifecycleStoppedData) -> None: def on_lifecycle_stopped(lifecycle_stopped_data: mqtt5.LifecycleStoppedData) -> None:
print("Lifecycle Stopped") _LOGGER.info("Lifecycle Stopped: %s", str(lifecycle_stopped_data))
print(lifecycle_stopped_data)
def on_lifecycle_connection_success( def on_lifecycle_connection_success(
lifecycle_connect_success_data: mqtt5.LifecycleConnectSuccessData, lifecycle_connect_success_data: mqtt5.LifecycleConnectSuccessData,
) -> None: ) -> None:
print("Lifecycle Connection Success") _LOGGER.info(
print(lifecycle_connect_success_data.connack_packet) "Lifecycle Connection Success: %s", str(lifecycle_connect_success_data)
print(lifecycle_connect_success_data.negotiated_settings) )
def on_publish_received(data: mqtt5.PublishReceivedData) -> None: def on_publish_received(data: mqtt5.PublishReceivedData) -> None:
@ -40,9 +39,10 @@ def on_publish_received(data: mqtt5.PublishReceivedData) -> None:
) )
for parameter in payload["parameters"]: for parameter in payload["parameters"]:
appliance.attributes["parameters"][parameter["parName"]].update(parameter) appliance.attributes["parameters"][parameter["parName"]].update(parameter)
print(parameter) appliance.notify()
_LOGGER.debug("%s - %s", topic, payload)
else: else:
print(topic, payload) _LOGGER.debug("%s - %s", topic, payload)
async def create_mqtt_client(api: "HonAPI") -> mqtt5.Client: async def create_mqtt_client(api: "HonAPI") -> mqtt5.Client:
@ -64,7 +64,7 @@ async def create_mqtt_client(api: "HonAPI") -> mqtt5.Client:
def subscribe(client: mqtt5.Client, appliance: HonAppliance) -> None: def subscribe(client: mqtt5.Client, appliance: HonAppliance) -> None:
for topic in appliance.info.get("topics", {}).get("subscribe", []): for topic in appliance.info.get("topics", {}).get("subscribe", []):
client.subscribe(mqtt5.SubscribePacket([mqtt5.Subscription(topic)])).result(10) client.subscribe(mqtt5.SubscribePacket([mqtt5.Subscription(topic)])).result(10)
_LOGGER.error("Subscribed to topic %s", topic) _LOGGER.info("Subscribed to topic %s", topic)
async def start(api: "HonAPI", app: list[HonAppliance]) -> mqtt5.Client: async def start(api: "HonAPI", app: list[HonAppliance]) -> mqtt5.Client: