add option to send speed tests and get results

Author: alexandrehassan

sagemcom_api/client.py

@@ -40,7 +40,7 @@
     UnknownException,
     UnknownPathException,
 )
-from .models import Device, DeviceInfo, PortMapping
+from .models import Device, DeviceInfo, PortMapping, SpeedTestResult
 
 
 class SagemcomClient:
@@ -168,7 +168,7 @@ def __get_response_value(self, response, index=0):
 
         return value
 
-    async def __api_request_async(self, actions, priority=False):
+    async def __api_request_async(self, actions, priority=False, **request_kwargs):
         """Build request to the internal JSON-req API."""
         self.__generate_request_id()
         self.__generate_nonce()
@@ -188,7 +188,9 @@ async def __api_request_async(self, actions, priority=False):
         }
 
         async with self.session.post(
-            api_host, data="req=" + json.dumps(payload, separators=(",", ":"))
+            api_host,
+            data="req=" + json.dumps(payload, separators=(",", ":")),
+            **request_kwargs,
         ) as response:
 
             if response.status == 400:
@@ -408,3 +410,35 @@ 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, timeout=100)
+
+    async def get_speed_test_results(self):
+        """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
+                )
+            ]
+            return results
+        return []

sagemcom_api/models.py

@@ -2,6 +2,7 @@
 
 import dataclasses
 from dataclasses import dataclass
+import time
 from typing import Any, List, Optional
 
 
@@ -162,3 +163,30 @@ 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: str
+    selected_server_address: str
+    block_traffic: bool
+    latency: str
+    upload: str
+    download: str
+
+    def __post_init__(self):
+        """Process data after init."""
+        # Convert timestamp to datetime object.
+        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}, "
+            f"upload: {self.upload}, download: {self.download}"
+        )