yggdrasil/server/sensors_client.py

import json
from urllib import error, parse, request


class SensorsAPIError(Exception):
    """Raised when the sensors API is unreachable or returns invalid data."""


class SensorsClient:
    def __init__(self, base_url: str, timeout: float = 10.0):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def _request(self, method, path, params=None, payload=None):
        url = f"{self.base_url}{path}"
        if params:
            query = parse.urlencode({k: v for k, v in params.items() if v is not None})
            if query:
                url += f"?{query}"

        data = None
        headers = {"Accept": "application/json"}
        if payload is not None:
            data = json.dumps(payload).encode("utf-8")
            headers["Content-Type"] = "application/json"

        req = request.Request(url, data=data, headers=headers, method=method)
        try:
            with request.urlopen(req, timeout=self.timeout) as response:
                body = response.read()
                return response.status, self._decode(body)
        except error.HTTPError as exc:
            return exc.code, self._decode(exc.read())
        except (error.URLError, TimeoutError, OSError) as exc:
            raise SensorsAPIError(f"sensors API unavailable: {exc}") from exc

    @staticmethod
    def _decode(body: bytes):
        if not body:
            return None
        try:
            return json.loads(body)
        except json.JSONDecodeError as exc:
            raise SensorsAPIError("sensors API returned invalid JSON") from exc

    # --- Sensor data ---

    def list_sensors(self):
        return self._request("GET", "/api/sensors")

    def create_sensor(self, payload):
        return self._request("POST", "/api/sensors", payload=payload)

    def create_readout(self, sensor_id, payload):
        return self._request(
            "POST", f"/api/sensors/{sensor_id}/readouts", payload=payload
        )

    def sensor_types(self, sensor_id):
        return self._request("GET", f"/api/sensors/{sensor_id}/types")

    def sensor_readouts(self, sensor_id, params=None):
        return self._request("GET", f"/api/sensors/{sensor_id}/readouts", params=params)

    def sensor_readouts_range(self, sensor_id, params=None):
        return self._request(
            "GET", f"/api/sensors/{sensor_id}/readouts/range", params=params
        )

    def sensor_readouts_hourly(self, sensor_id, params=None):
        return self._request(
            "GET", f"/api/sensors/{sensor_id}/readouts/hourly", params=params
        )