Rebase PR #253 onto main: add keep_keys, get_logs, speed test, LoginConnectionException, SpeedTestResult

Author: iMicknl

sagemcom_api/client.py

@@ -44,6 +44,7 @@
     AuthenticationException,
     BadRequestException,
     InvalidSessionException,
+    LoginConnectionException,
     LoginRetryErrorException,
     LoginTimeoutException,
     MaximumSessionCountException,
@@ -53,7 +54,7 @@
     UnknownPathException,
     UnsupportedHostException,
 )
-from .models import Device, DeviceInfo, PortMapping
+from .models import Device, DeviceInfo, PortMapping, SpeedTestResult
 
 
 async def retry_login(invocation: Mapping[str, Any]) -> None:
@@ -77,6 +78,7 @@ def __init__(
         session: ClientSession | None = None,
         ssl: bool | None = False,
         verify_ssl: bool | None = True,
+        keep_keys: bool = False,
     ):
         """Create a SagemCom client.
 
@@ -85,11 +87,13 @@ def __init__(
         :param password: the password for your Sagemcom router
         :param authentication_method: the auth method of your Sagemcom router
         :param session: use a custom session, for example to configure the timeout
+        :param keep_keys: return response keys as originally written (no snake_case conversion)
         """
         self.host = host
         self.username = username
         self.authentication_method = authentication_method
         self.password = password
+        self.keep_keys = keep_keys
         self._current_nonce = None
         self._password_hash = self.__generate_hash(password)
         self.protocol = "https" if ssl else "http"
@@ -191,7 +195,7 @@ def __get_response(self, response, index=0):
 
         return value
 
-    def __get_response_value(self, response, index=0):
+    def __get_response_value(self, response, index=0, keep_keys: bool | None = None):
         """Retrieve response value from value."""
         try:
             value = self.__get_response(response, index)["value"]
@@ -200,8 +204,9 @@ def __get_response_value(self, response, index=0):
         except IndexError:
             value = None
 
-        # Rewrite result to snake_case
-        if value is not None:
+        # Rewrite result to snake_case unless keep_keys is requested
+        should_keep = keep_keys if keep_keys is not None else self.keep_keys
+        if value is not None and not should_keep:
             value = humps.decamelize(value)
 
         return value
@@ -335,6 +340,8 @@ async def login(self):
             raise LoginTimeoutException(
                 "Login request timed-out. This could be caused by using the wrong encryption method, or using a (non) SSL connection."
             ) from exception
+        except (ClientConnectorError, ClientOSError) as exception:
+            raise LoginConnectionException("Unable to connect to the device. Please check the host address.") from exception
 
         data = self.__get_response(response)
 
@@ -552,6 +559,22 @@ async def get_port_mappings(self) -> list[PortMapping]:
         max_tries=1,
         on_backoff=retry_login,
     )
+    async def get_logs(self) -> str:
+        """Retrieve system logs from Sagemcom F@st device."""
+        actions = {
+            "id": 0,
+            "method": "getVendorLogDownloadURI",
+            "xpath": urllib.parse.quote("Device/DeviceInfo/VendorLogFiles/VendorLogFile[@uid='1']"),
+        }
+
+        response = await self.__api_request_async([actions], False)
+        log_path = response["reply"]["actions"][0]["callbacks"][0]["parameters"]["uri"]
+
+        log_uri = f"{self.protocol}://{self.host}{log_path}"
+        log_response = await self.session.get(log_uri)
+
+        return await log_response.text()
+
     async def reboot(self):
         """Reboot Sagemcom F@st device."""
         action = {
@@ -565,3 +588,32 @@ async def reboot(self):
         data = self.__get_response_value(response)
 
         return data
+
+    async def run_speed_test(self, block_traffic: bool = False):
+        """Run Speed Test on Sagemcom F@st device."""
+        actions = [
+            {
+                "id": 0,
+                "method": "speedTestClient",
+                "xpath": "Device/IP/Diagnostics/SpeedTest",
+                "parameters": {"BlockTraffic": block_traffic},
+            }
+        ]
+        return await self.__api_request_async(actions, False)
+
+    async def get_speed_test_results(self) -> list[SpeedTestResult]:
+        """Retrieve Speed Test results from Sagemcom F@st device."""
+        ret = await self.get_value_by_xpath("Device/IP/Diagnostics/SpeedTest")
+        history = ret["speed_test"]["history"]
+        if history:
+            timestamps = (int(k) for k in history["timestamp"].split(","))
+            server_address = history["selected_server_address"].split(",")
+            block_traffic = history["block_traffic"].split(",")
+            latency = history["latency"].split(",")
+            upload = (float(k) for k in history["upload"].split(","))
+            download = (float(k) for k in history["download"].split(","))
+            results = [
+                SpeedTestResult(*data) for data in zip(timestamps, server_address, block_traffic, latency, upload, download, strict=True)
+            ]
+            return results
+        return []

sagemcom_api/exceptions.py

@@ -43,6 +43,10 @@ class LoginTimeoutException(BaseSagemcomException):
     """Raised when a timeout is encountered during login."""
 
 
+class LoginConnectionException(BaseSagemcomException):
+    """Raised when a connection error is encountered during login."""
+
+
 class NonWritableParameterException(BaseSagemcomException):
     """Raised when provided parameter is not writable."""
 

sagemcom_api/models.py

@@ -1,6 +1,7 @@
 """Models for the Sagemcom F@st client."""
 
 import dataclasses
+import time
 from dataclasses import dataclass
 from typing import Any
 
@@ -164,3 +165,26 @@ def __init__(self, **kwargs):
     def id(self):
         """Return unique ID for port mapping."""
         return self.uid
+
+
+@dataclass
+class SpeedTestResult:
+    """Representation of a speedtest result."""
+
+    timestamp: int | str
+    selected_server_address: str
+    block_traffic: bool
+    latency: str
+    upload: float
+    download: float
+
+    def __post_init__(self):
+        """Process data after init."""
+        # Convert timestamp to human-readable string.
+        if isinstance(self.timestamp, int):
+            self.timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.timestamp))
+        self.block_traffic = bool(self.block_traffic)
+
+    def __str__(self) -> str:
+        """Return string representation of speedtest result."""
+        return f"timestamp: {self.timestamp}, latency: {self.latency}, upload: {self.upload}, download: {self.download}"