Cloud connection status and WiFi signal strenght

CHANGELOG.md

@@ -5,6 +5,13 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.9.3] - 2026-05-29
+
+### Added
+
+- `get_cloud_status()` for cloud connection status
+- `get_wifi_rssi()` for WiFi signal strength
+
 ## [0.9.1] - 2026-05-11
 
 ## Changed

README.md

@@ -13,6 +13,7 @@ This client uses an undocumented local HTTP API. It provides live readings for p
 
 > **Disclaimer:** Use at your own risk. No liability for damages or malfunctions.
 
+
 ## Features
 
 - **Async/await support** for non-blocking operations
@@ -24,6 +25,10 @@ This client uses an undocumented local HTTP API. It provides live readings for p
 - **Command-line interface** for direct device interaction and testing
 - **Secure by default** - WiFi passwords excluded unless explicitly requested
 - **Comprehensive error handling** with detailed logging
+- **Cloud connection** status
+- **WiFi RSSI** signal
+
+Each method queries the device live and returns the current value.
 - **SSL/HTTPS support** for secure communication
 
 ## Prerequisites

docs/api-reference.md

@@ -24,6 +24,8 @@ PooldoseClient(host, timeout=30, *, websession=None, include_sensitive_data=Fals
 ### Methods
 
 - `async connect()` → `RequestStatus` - Connect to device and initialize all components
+- `async get_cloud_status()` → `Optional[bool]` — Retrieve the current cloud connection status
+- `async get_wifi_rssi()` → `Optional[int]` — Retrieve the current WiFi RSSI (signal strength)
 - `static_values()` → `tuple[RequestStatus, StaticValues | None]` - Get static device information
 - `async instant_values()` → `tuple[RequestStatus, InstantValues | None]` - Get current sensor readings and device state
 - `async instant_values_structured()` → `tuple[RequestStatus, dict[str, Any]]` - Get structured data organized by type

examples/demo.py

@@ -48,8 +48,16 @@ async def main() -> None:
     if client_status != RequestStatus.SUCCESS:
         print(f"Error connecting to PooldoseClient: {client_status}")
         return
+
     print("Connected to Pooldose device.")
 
+    # Fetch and display cloud status and WiFi RSSI (real client only, but mock can return fixed values if desired)
+    cloud_status = await client.request_handler.get_cloud_status()
+    print(f"\nCloud-Status (wdp_status): {cloud_status}")
+
+    wifi_rssi = await client.request_handler.get_wifi_rssi()
+    print(f"WiFi RSSI: {wifi_rssi}")
+
     # Fetch and display static values
     print("\nFetching static values...")
     static_values_status, static_values = client.static_values()

pyproject.toml

@@ -6,7 +6,7 @@ authors = [{ name = "Lukas Maertin", email = "pypi@lukas-maertin.de" }]
 license = "MIT"
 readme = "README.md"
 requires-python = ">=3.11"
-dependencies = ["aiohttp", "aiofiles", "getmac"]
+dependencies = ["aiohttp", "aiofiles", "getmac", "websockets"]
 
 [project.urls]
 Homepage = "https://github.com/lmaertin/python-pooldose"

requirements.txt

@@ -3,4 +3,5 @@ aiofiles
 types-aiofiles
 getmac
 pytest
-pytest-asyncio
+pytest-asyncio
+websockets

src/pooldose/__init__.py

@@ -1,5 +1,5 @@
 """Async API client for SEKO Pooldose."""
 from .client import PooldoseClient
 
-__version__ = "0.9.1"
+__version__ = "0.9.3"
 __all__ = ["PooldoseClient"]

src/pooldose/request_handler.py

@@ -1,3 +1,5 @@
+#
+
 """Request Handler for async API client for SEKO Pooldose."""
 
 import asyncio
@@ -8,6 +10,8 @@
 from typing import Any, Optional, Tuple, Union, List, Dict
 
 import aiohttp
+import websockets
+import websockets.exceptions
 
 from pooldose.type_definitions import (
     AccessPointDict,
@@ -541,3 +545,57 @@ async def reboot_device(self):
         except (aiohttp.ClientError, asyncio.TimeoutError) as err:
             _LOGGER.warning("Error sending reboot command: %s", err)
             return RequestStatus.UNKNOWN_ERROR, False
+
+    async def _get_websocket_data(self, topic_filter: Union[str, List[str]], value_path: List[str]) -> Optional[Any]:
+        """
+        Open a WebSocket connection and extract a value from the first matching topic.
+
+        Args:
+            topic_filter: Topic name or list of topic names to match.
+            value_path: List of keys to traverse in the data dict to extract the value.
+
+        Returns:
+            The extracted value or None if not found or error.
+        """
+        url = f"ws://{self.host}:1334"
+        try:
+            async with websockets.connect(url) as ws:
+                while True:
+                    msg = await ws.recv()
+                    try:
+                        data = json.loads(msg)
+                    except json.JSONDecodeError:
+                        continue
+                    topic = data.get("topic")
+                    if (isinstance(topic_filter, str) and topic == topic_filter) or (
+                        isinstance(topic_filter, list) and topic in topic_filter
+                    ):
+                        val = data.get("data", {})
+                        for key in value_path:
+                            if isinstance(val, dict):
+                                val = val.get(key)
+                            else:
+                                return None
+                        return val
+        except (OSError, websockets.exceptions.WebSocketException) as err:
+            _LOGGER.error("WebSocket error: %s", err)
+            return None
+
+
+    async def get_cloud_status(self) -> Optional[bool]:
+        """
+        Retrieve the current cloud connection status (wdp_status) live via WebSocket.
+
+        Returns:
+            Optional[bool]: Cloud connection status (True/False/None)
+        """
+        return await self._get_websocket_data(["wdp_status", "wdp_connection"], ["connection"])
+
+    async def get_wifi_rssi(self) -> Optional[int]:
+        """
+        Retrieve the current WiFi RSSI (signal strength) live via WebSocket.
+
+        Returns:
+            Optional[int]: WiFi RSSI (int/None)
+        """
+        return await self._get_websocket_data("wifi_station", ["rssi"])

tests/test_request_handler.py

@@ -1,6 +1,7 @@
 """Tests for RequestHandler for Async API client for SEKO Pooldose."""
 
 from unittest.mock import AsyncMock, MagicMock, patch
+import json
 import asyncio
 import aiohttp
 import pytest
@@ -576,3 +577,48 @@ async def test_get_wifi_station_timeout_error(self):
 
             assert status == RequestStatus.UNKNOWN_ERROR
             assert data is None
+
+class TestWebsocketParsing:
+    """Tests for the get_cloud_status() and get_wifi_rssi() parsing."""
+
+    @pytest.mark.asyncio
+    async def test_get_cloud_status_success(self):
+        """Test successful retrieval of cloud connection via WebSocket."""
+        handler = RequestHandler("localhost")
+        ws_mock = AsyncMock()
+        ws_instance = AsyncMock()
+        wdp_msg = json.dumps({"topic": "wdp_status", "data": {"connection": True}})
+        ws_instance.recv = AsyncMock(side_effect=[wdp_msg])
+        ws_mock.__aenter__.return_value = ws_instance
+        with patch("websockets.connect", return_value=ws_mock):
+            status = await handler.get_cloud_status()
+        assert status is True
+
+    @pytest.mark.asyncio
+    async def test_get_cloud_status_error(self):
+        """Test error case when WebSocket connection fails for cloud status."""
+        handler = RequestHandler("localhost")
+        with patch("websockets.connect", side_effect=OSError("Verbindungsfehler")):
+            status = await handler.get_cloud_status()
+        assert status is None
+
+    @pytest.mark.asyncio
+    async def test_get_wifi_rssi_success(self):
+        """Test successful retrieval of WiFi RSSI via WebSocket."""
+        handler = RequestHandler("localhost")
+        ws_mock = AsyncMock()
+        ws_instance = AsyncMock()
+        wifi_msg = json.dumps({"topic": "wifi_station", "data": {"rssi": -42}})
+        ws_instance.recv = AsyncMock(side_effect=[wifi_msg])
+        ws_mock.__aenter__.return_value = ws_instance
+        with patch("websockets.connect", return_value=ws_mock):
+            rssi = await handler.get_wifi_rssi()
+        assert rssi == -42
+
+    @pytest.mark.asyncio
+    async def test_get_wifi_rssi_error(self):
+        """Test error case when WebSocket connection fails for WiFi RSSI."""
+        handler = RequestHandler("localhost")
+        with patch("websockets.connect", side_effect=OSError("Verbindungsfehler")):
+            rssi = await handler.get_wifi_rssi()
+        assert rssi is None