Implement event support for gateway

examples/automation_02/main.py

@@ -1,9 +1,8 @@
 import asyncio
 import logging
 
-from pyown.client import Client
+from pyown.client import Client, SessionType
 from pyown.items.automation import Automation, WhatAutomation
-from pyown.protocol import SessionType
 
 log = logging.getLogger(__name__)
 

examples/gateway_01/README.md

@@ -0,0 +1,3 @@
+# light_01
+
+This example demonstrates how to create a simple light source and control it.

examples/gateway_01/main.py

@@ -0,0 +1,62 @@
+import asyncio
+import logging
+
+from pyown.client import Client
+from pyown.items import Gateway
+
+
+async def run(host: str, port: int, password: str):
+    client = Client(
+        host=host,
+        port=port,
+        password=password,
+    )
+
+    await client.start()
+
+    gateway = Gateway(
+        client=client
+    )
+
+    # get ip address of the gateway
+    ip = await gateway.get_ip()
+    print(ip)
+
+    # get the model of the gateway
+    model = await gateway.get_model()
+    print(model.name)
+
+    # get datetime of the gateway
+    datetime = await gateway.get_datetime()
+    print(datetime)
+
+    # get the kernel version of the gateway
+    kernel = await gateway.get_kernel_version()
+    print(kernel)
+
+    await client.close()
+
+
+def main(host: str, port: int, password: str):
+    # Set the logging level to DEBUG
+    logging.basicConfig(
+        level=logging.DEBUG,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    # Run the asyncio event loop
+    asyncio.run(run(host, port, password))
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--host", type=str, help="The host to connect to", default="192.168.1.35")
+    parser.add_argument("--port", type=int, help="The port to connect to", default=20000)
+    parser.add_argument("--password", type=str, help="The password to authenticate with", default="12345")
+
+    args = parser.parse_args()
+
+    main(args.host, args.port, args.password)

examples/gateway_02/README.md

@@ -0,0 +1,3 @@
+# light_01
+
+This example demonstrates how to create a simple light source and control it.

examples/gateway_02/main.py

@@ -0,0 +1,53 @@
+import asyncio
+import logging
+
+from black import datetime
+
+from pyown.client import Client, SessionType
+from pyown.items import Gateway,  WhatGateway
+
+
+async def on_time_change(gateway: Gateway, time: datetime.time):
+    print(f"Time of the gateway is now {time}")
+
+
+async def run(host: str, port: int, password: str):
+    client = Client(
+        host=host,
+        port=port,
+        password=password,
+        session_type=SessionType.EventSession
+    )
+
+    Gateway.register_callback(
+        WhatGateway.TIME,
+        on_time_change
+    )
+
+    await client.start()
+    await client.loop()
+
+
+def main(host: str, port: int, password: str):
+    # Set the logging level to DEBUG
+    logging.basicConfig(
+        level=logging.WARN,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    # Run the asyncio event loop
+    asyncio.run(run(host, port, password))
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--host", type=str, help="The host to connect to", default="192.168.1.35")
+    parser.add_argument("--port", type=int, help="The port to connect to", default=20000)
+    parser.add_argument("--password", type=str, help="The password to authenticate with", default="12345")
+
+    args = parser.parse_args()
+
+    main(args.host, args.port, args.password)

pyown/client/client.py

@@ -139,7 +139,7 @@ async def loop(self, *, client: BaseClient | None = None):
             for item_obj in BaseItem.__subclasses__():
                 if message.who == item_obj.who:
                     try:
-                        tasks = item_obj.call_callbacks(item, message)
+                        tasks = await item_obj.call_callbacks(item, message)
                     except InvalidMessage as e:
                         log.warning(f"Message not supported {e.message}")
                     else:

pyown/items/__init__.py

@@ -1 +1,3 @@
-from .lighting import Light, Dimmer
+from .lighting import *
+from .automation import *
+from .gateway import *

pyown/items/automation/automation.py

@@ -93,7 +93,7 @@ def on_status_change(cls, callback: Callable[[Self, WhatAutomation], Coroutine[N
         cls._event_callbacks.setdefault(AutomationEvents.ALL, []).append(callback)
 
     @classmethod
-    def call_callbacks(cls, item: Self, message: BaseMessage) -> list[Task]:
+    async def call_callbacks(cls, item: Self, message: BaseMessage) -> list[Task]:
         tasks: list[Task] = []
 
         if isinstance(message, NormalMessage):

pyown/items/base.py

@@ -76,7 +76,7 @@ def _create_tasks(funcs: list[CoroutineCallback], *args: Any) -> list[Task]:
 
     @classmethod
     @abstractmethod
-    def call_callbacks(cls, item: Self, message: BaseMessage) -> list[Task]:
+    async def call_callbacks(cls, item: Self, message: BaseMessage) -> list[Task]:
         """
         Calls the registered callbacks for the event.
         Used internally by the client to dispatch the events to the correct callbacks.

pyown/items/gateway/__init__.py

@@ -0,0 +1 @@
+from .gateway import *

pyown/items/gateway/gateway.py

@@ -2,12 +2,13 @@
 import ipaddress
 from asyncio import Task
 from enum import StrEnum
-from typing import Self
+from typing import Self, Any
 
 from ..base import BaseItem, CoroutineCallback
+from ...client import BaseClient
 from ...exceptions import InvalidMessage
 from ...messages import DimensionResponse, BaseMessage, DimensionWriting
-from ...tags import Who, What, Value
+from ...tags import Who, What, Value, Where
 
 __all__ = [
     "Gateway",
@@ -20,9 +21,9 @@ class GatewayModel(StrEnum):
     """
     This enum is used to define the various models of gateways that are supported by the library.
 
-    This is not a complete list of all the gateways, because there are many different models of gateways that are not
-    listed in the official documentation. So, if you have a gateway that is not listed here, you can send an issue
-    on GitHub.
+    This is not a complete list of all the gateways because there are many different models of gateways that are not
+    listed in the official documentation.
+    So, if you have a gateway not listed here, you can send an issue on GitHub.
 
     Attributes:
         MHServer:
@@ -38,6 +39,8 @@ class GatewayModel(StrEnum):
     F452V = "7"
     MHServer2 = "11"
     H4684 = "12"
+    HL4684 = "23"
+
 
 
 class WhatGateway(What, StrEnum):
@@ -78,6 +81,14 @@ class Gateway(BaseItem):
 
     _event_callbacks: dict[WhatGateway, list[CoroutineCallback]] = {}
 
+    def __init__(self, client: BaseClient, where: Where | str = ""):
+        """
+        Initializes the item.
+        Args:
+            client: The client to use to communicate with the server.
+        """
+        super().__init__(client, Where(""))
+
     async def _single_dim_req(self, what: WhatGateway) -> DimensionResponse:
         messages = [msg async for msg in self.send_dimension_request(what)]
 
@@ -89,6 +100,10 @@ async def _single_dim_req(self, what: WhatGateway) -> DimensionResponse:
 
     @staticmethod
     def _parse_own_timezone(t: Value) -> datetime.timezone:
+        if t.string == "":
+            # return UTC if the timezone is not set
+            return datetime.timezone.utc
+
         sign = t.string[0]
         hours = int(t.string[1:3])
 
@@ -368,7 +383,7 @@ async def get_datetime(self, *, message: EventMessage = None) -> datetime.dateti
         s = int(resp.values[2].string)
         t = resp.values[3]
 
-        #w = int(resp.values[4].string)
+        # w = int(resp.values[4].string)
         d = int(resp.values[5].string)
         mo = int(resp.values[6].string)
         y = int(resp.values[7].string)
@@ -450,6 +465,80 @@ async def get_distribution_version(self, *, message: EventMessage = None) -> str
 
         return f"{v}.{r}.{b}"
 
+    # this does not follow the same pattern as the other item class because that would add too much complexity,
+    # and event messages for the gateway are very rarely sent
+    @classmethod
+    def register_callback(cls, what: WhatGateway, callback: CoroutineCallback):
+        """
+        Register a callback for a specific event.
+
+        Args:
+            what: The event to register the callback for.
+            callback: The callback to call when the event occurs.
+
+        Returns:
+            None
+        """
+        if what not in cls._event_callbacks:
+            cls._event_callbacks[what] = []
+
+        cls._event_callbacks[what].append(callback)
+
     @classmethod
-    def call_callbacks(cls, item: Self, message: BaseMessage) -> list[Task]:
-        raise NotImplementedError
+    async def call_callbacks(cls, item: Self, message: BaseMessage) -> list[Task]:
+        tasks = []
+
+        if isinstance(message, DimensionWriting):
+            # convert the DimensionWriting message to a DimensionResponse message
+            # noinspection PyTypeChecker
+            message = DimensionResponse(
+                (
+                    message.who,
+                    message.where,
+                    message.dimension,
+                    *message.values
+                )  # type: ignore[arg-type]
+            )
+
+        if isinstance(message, DimensionResponse):
+            what = WhatGateway(message.dimension.string)
+            callbacks = cls._event_callbacks.get(what, [])
+
+            # noinspection PyUnusedLocal
+            args: Any = None
+            match what:
+                case WhatGateway.TIME:
+                    args = await item.get_time(message=message)
+                case WhatGateway.DATE:
+                    args = await item.get_date(message=message)
+                case WhatGateway.IP_ADDRESS:
+                    args = await item.get_ip(message=message)
+                case WhatGateway.NET_MASK:
+                    args = await item.get_netmask(message=message)
+                case WhatGateway.MAC_ADDRESS:
+                    args = await item.get_macaddress(message=message)
+                case WhatGateway.DEVICE_TYPE:
+                    args = await item.get_model(message=message)
+                case WhatGateway.FIRMWARE_VERSION:
+                    args = await item.get_firmware(message=message)
+                case WhatGateway.UPTIME:
+                    args = await item.get_uptime(message=message)
+                case WhatGateway.DATE_TIME:
+                    args = await item.get_datetime(message=message)
+                case WhatGateway.KERNEL_VERSION:
+                    args = await item.get_kernel_version(message=message)
+                case WhatGateway.DISTRIBUTION_VERSION:
+                    args = await item.get_distribution_version(message=message)
+                case _:
+                    return []
+
+            tasks += cls._create_tasks(
+                callbacks,
+                item,
+                *args
+            )
+        else:
+            raise InvalidMessage("The message is not a DimensionResponse message.")
+
+        return tasks
+

pyown/items/lighting/base.py

@@ -227,7 +227,7 @@ def on_temporization_change(cls, callback: Callable[[Self, int, int, int], Corou
         cls._event_callbacks.setdefault(LightEvents.LIGHT_TEMPORIZATION, []).append(callback)
 
     @classmethod
-    def call_callbacks(cls, item: BaseItem, message: BaseMessage) -> list[Task]:
+    async def call_callbacks(cls, item: BaseItem, message: BaseMessage) -> list[Task]:
         tasks: list[Task] = []
 
         if isinstance(message, DimensionResponse):

pyown/items/utils.py

@@ -2,6 +2,7 @@
 
 from .automation import Automation
 from .base import BaseItem
+from .gateway import Gateway
 from .lighting import Light
 from ..tags import Who
 
@@ -13,5 +14,6 @@
 ITEM_TYPES: Final[dict[Who, Type[BaseItem]]] = {
     Who.LIGHTING: Light,
     Who.AUTOMATION: Automation,
+    Who.GATEWAY: Gateway
 }
 """A dictionary that maps the Who tag to the corresponding item class."""