Feature/testing framework

PythonTools/OpenIrisClient.py

@@ -0,0 +1,148 @@
+from typing import Optional, Callable
+
+import aiohttp
+
+from constants import WifiPowerPoint
+from .models import (
+    DeviceConfig,
+    CameraConfig,
+    WiFiConfig,
+    TrackerConfig,
+    MDNSConfig,
+)
+
+
+class BaseAPIClient:
+    def __init__(self, tracker_address: str):
+        self.tracker_address = tracker_address
+        self.base_endpoint = "control/builtin/command"
+
+    async def __aenter__(self):
+        self.session = aiohttp.ClientSession()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.session.close()
+
+    async def get(
+        self,
+        command: str,
+        params: Optional[dict] = None,
+        validator: Optional[Callable] = None,
+    ):
+        clean_params = await self._clean_params(params)
+
+        if validator and not validator(clean_params):
+            raise ValueError(f"Params for command {command} are required, none provided")
+
+        async with self.session.get(
+            f"{self.tracker_address}/{self.base_endpoint}/{command}/",
+            params=clean_params,
+        ) as request:
+            await request.read()
+            return request
+
+    async def post(self, command: str, data: dict, validator: Optional[Callable] = None):
+        clean_params = await self._clean_params(data)
+
+        if validator and not validator(clean_params):
+            raise ValueError(f"Params for command {command} are required, none provided")
+
+        async with self.session.post(
+            f"{self.tracker_address}/{self.base_endpoint}/{command}/",
+            params=clean_params,
+        ) as request:
+            await request.read()
+            return request
+
+    async def delete(self, command: str, data: dict, validator: Optional[Callable] = None):
+        clean_params = await self._clean_params(data)
+
+        if validator and not validator(clean_params):
+            raise ValueError(f"Params for command {command} are required, none provided")
+
+        async with self.session.delete(
+            f"{self.tracker_address}/{self.base_endpoint}/{command}/",
+            params=clean_params,
+        ) as request:
+            await request.read()
+            return request
+
+    @staticmethod
+    async def _clean_params(params) -> dict:
+        if params:
+            return {k: v for k, v in params.items() if v not in (None, "")}
+        return {}
+
+
+class OpenIrisClient(BaseAPIClient):
+    async def upsert_wifi_settings(self, network: WiFiConfig):
+
+        params = {
+            "networkName": network.name,
+            "ssid": network.ssid,
+            "password": network.password,
+            "channel": network.channel,
+            "power": network.power,
+            "adhoc": network.adhoc,
+        }
+
+        return await self.post("setWiFi", data=params, validator=all)
+
+    async def remove_wifi_network(self, name: str):
+        params = {
+            "networkName": name,
+        }
+
+        return await self.delete("setWiFi", data=params, validator=all)
+
+    async def get_stored_config(self) -> TrackerConfig:
+        response = await self.get("getStoredConfig")
+        return TrackerConfig(**await response.json())
+
+    async def reset_config(self):
+        return await self.get("resetConfig")
+
+    # we should split this into two separate configs and endpoints and clean them up
+    async def update_device_settings(self, mdns_config: MDNSConfig, device_config: DeviceConfig):
+        params = {
+            "hostname": mdns_config.hostname,
+            "service": mdns_config.service,
+            "ota_login": device_config.OTALogin,
+            "ota_password": device_config.OTAPassword,
+        }
+
+        return await self.get(command="setDevice", params=params)
+
+    async def reboot_device(self):
+        return await self.get(command="rebootDevice")
+
+    async def set_tx_power(self, power_level: WifiPowerPoint):
+        params = {
+            "txPower": power_level,
+        }
+        return await self.get(command="setTxPower", params=params)
+
+    async def get_wifi_strength(self):
+        return await self.get(command="wifiStrength")
+
+    async def update_camera_settings(self, camera_config: CameraConfig):
+        params = {
+            "framesize": camera_config.framesize.value if camera_config.framesize else None,
+            "vflip": int(camera_config.vflip) if camera_config.vflip is not None else None,
+            "hflip": int(camera_config.hflip) if camera_config.hflip is not None else None,
+            "quality": camera_config.quality,
+            "brightness": camera_config.brightness,
+        }
+
+        return await self.get(command="setCamera", params=params)
+
+    async def restart_camera(self, hard_reset: bool = False):
+        params = {"mode": int(hard_reset)}
+        return await self.get(command="restartCamera", params=params)
+
+    async def ping(self):
+        return await self.get(command="ping")
+
+    async def save_config(self):
+        return await self.get(command="save")

PythonTools/conftest.py

@@ -0,0 +1,6 @@
+import pytest_asyncio
+
+
+@pytest_asyncio.fixture()
+async def device_url():
+    return "http://openiristracker.local:81/"

PythonTools/constants.py

@@ -0,0 +1,28 @@
+from enum import IntEnum
+
+
+class FrameSize(IntEnum):
+    FRAMESIZE_96X96 = (1,)  # // 96x96
+    FRAMESIZE_QQVGA = (2,)  # // 160x120
+    FRAMESIZE_QCIF = (3,)  # // 176x144
+    FRAMESIZE_HQVGA = (4,)  # // 240x176
+    FRAMESIZE_240X240 = (5,)  # // 240x240
+    FRAMESIZE_QVGA = (6,)  # // 320x240
+    FRAMESIZE_CIF = (7,)  # // 400x296
+    FRAMESIZE_HVGA = (8,)  # // 480x320
+    FRAMESIZE_VGA = (9,)  # // 640x480
+
+
+class WifiPowerPoint(IntEnum):
+    WIFI_POWER_19_5dBm = (78,)  # 19.5dBm
+    WIFI_POWER_19dBm = (76,)  # 19dBm
+    WIFI_POWER_18_5dBm = (74,)  # 18.5dBm
+    WIFI_POWER_17dBm = (68,)  # 17dBm
+    WIFI_POWER_15dBm = (60,)  # 15dBm
+    WIFI_POWER_13dBm = (52,)  # 13dBm
+    WIFI_POWER_11dBm = (44,)  # 11dBm
+    WIFI_POWER_8_5dBm = (34,)  # 8.5dBm
+    WIFI_POWER_7dBm = (28,)  # 7dBm
+    WIFI_POWER_5dBm = (20,)  # 5dBm
+    WIFI_POWER_2dBm = (8,)  # 2dBm
+    WIFI_POWER_MINUS_1dBm = -4  # -1dBm