feat(v2): add websocket client with typed events, reconnect, and tests

README.md

@@ -53,6 +53,32 @@ async def main():
 asyncio.run(main())
 ```
 
+### API v2 WebSocket (real-time updates)
+```python
+import asyncio
+from homewizard_energy import HomeWizardEnergyV2, WebSocketTopic
+
+IP_ADDRESS = "192.168.1.123"
+TOKEN = "your-token"
+
+
+async def main():
+    async with HomeWizardEnergyV2(host=IP_ADDRESS, token=TOKEN) as api:
+        async with api.websocket() as ws:
+            await ws.subscribe(WebSocketTopic.MEASUREMENT)
+
+            # Typed stream: known topics map to existing models
+            # (Measurement, Device, System, Batteries)
+            async for event in ws.events_typed(reconnect=True):
+                if event.type == WebSocketTopic.MEASUREMENT:
+                    print(event.data.power_w)
+
+
+asyncio.run(main())
+```
+
+If you prefer raw payloads, you can still use `ws.events()` / `ws.receive()`.
+
 # Development and contribution
 Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
 

homewizard_energy/__init__.py

@@ -6,12 +6,20 @@
 from .homewizard_energy import HomeWizardEnergy
 from .v1 import HomeWizardEnergyV1
 from .v2 import HomeWizardEnergyV2
+from .v2.websocket import (
+    HomeWizardEnergyWebSocket,
+    HomeWizardEnergyWebSocketEvent,
+    WebSocketTopic,
+)
 
 __all__ = [
     "DisabledError",
     "HomeWizardEnergy",
     "HomeWizardEnergyV1",
     "HomeWizardEnergyV2",
+    "HomeWizardEnergyWebSocket",
+    "HomeWizardEnergyWebSocketEvent",
+    "WebSocketTopic",
     "InvalidStateError",
     "RequestError",
     "UnsupportedError",

homewizard_energy/models.py

@@ -704,14 +704,20 @@ class Permissions(StrEnum):
     permissions: list[Permissions] | None = field(
         default=None,
         metadata={
-            "deserialize": lambda lst: [
-                perm
-                for item in lst
-                if (perm := Batteries.Permissions.__members__.get(item.upper(), None))
-                is not None
-            ]
-            if lst is not None
-            else None
+            "deserialize": lambda lst: (
+                [
+                    perm
+                    for item in lst
+                    if (
+                        perm := Batteries.Permissions.__members__.get(
+                            item.upper(), None
+                        )
+                    )
+                    is not None
+                ]
+                if lst is not None
+                else None
+            )
         },
     )
     power_w: float = field()

homewizard_energy/v2/__init__.py

@@ -36,6 +36,7 @@
     Token,
 )
 from .cacert import CACERT
+from .websocket import HomeWizardEnergyWebSocket
 
 T = TypeVar("T")
 
@@ -320,6 +321,20 @@ async def _request(
 
         return (resp.status, await resp.text())
 
+    def websocket(self) -> HomeWizardEnergyWebSocket:
+        """Create a websocket client for this device.
+
+        Note: The websocket shares this API client's session when available.
+        Keep the parent client/session open while the websocket is active.
+        """
+        return HomeWizardEnergyWebSocket(
+            host=self._host,
+            token=self._token,
+            identifier=self._identifier,
+            clientsession=self._session,
+            timeout=self._request_timeout,
+        )
+
     async def __aenter__(self) -> HomeWizardEnergyV2:
         """Async enter.