OpenIris/PythonTools/OpenIrisClient.py
2023-04-09 16:29:19 +02:00

161 lines
4.8 KiB
Python

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,
"vflip": camera_config.vflip,
"hflip": camera_config.hflip,
"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")