ignis-client-python/ignis_client/sync.py

from __future__ import annotations

from typing import Any

import httpx

from .models import (
    SCENES,
    BlinkResponse,
    CommandRequest,
    DeleteStatusResponse,
    DeviceControlResponse,
    DeviceStatusResponse,
    GroupControlResponse,
    GroupStatusResponse,
    IgnisError,
    KeyActionRequest,
    RescanResponse,
    ScheduleCreateResponse,
    ScheduleCronRequest,
    ScheduleOnceRequest,
    ScheduleTasksResponse,
    ServerInfoResponse,
)


class IgnisClient:
    def __init__(self, base_url: str, api_key: str, *, timeout: float = 10.0):
        self._base = base_url.rstrip("/")
        self._api_key = api_key
        self._client = httpx.Client(
            base_url=self._base,
            headers={"X-API-Key": self._api_key},
            timeout=httpx.Timeout(timeout),
            follow_redirects=True,
        )

    def close(self):
        self._client.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

    def _request(self, method: str, path: str, **kwargs) -> Any:
        resp = self._client.request(method, path, **kwargs)
        if resp.status_code >= 400:
            try:
                detail = resp.json().get("detail", resp.text)
            except Exception:
                detail = resp.text or f"HTTP {resp.status_code}"
            raise IgnisError(resp.status_code, detail)
        return resp.json()

    def _get(self, path: str, **kwargs) -> Any:
        return self._request("GET", path, **kwargs)

    def _post(self, path: str, **kwargs) -> Any:
        return self._request("POST", path, **kwargs)

    def _delete(self, path: str, **kwargs) -> Any:
        return self._request("DELETE", path, **kwargs)

    # ------------------------------------------------------------------ Auth

    def auth_me(self) -> dict[str, Any]:
        return self._get("/auth/me")

    # ------------------------------------------------------------------ System

    def system_info(self) -> ServerInfoResponse:
        data = self._get("/system/info")
        return ServerInfoResponse(**data)

    # ------------------------------------------------------------------ Devices

    def list_devices(self) -> dict[str, dict[str, str]]:
        return self._get("/devices")

    def list_groups(self) -> dict[str, dict[str, Any]]:
        return self._get("/devices/groups")

    def list_scenes(self) -> dict[str, int]:
        return dict(SCENES)

    def create_group(self, group_id: str, name: str, macs: list[str]) -> dict[str, Any]:
        return self._post(
            "/devices/groups", json={"id": group_id, "name": name, "macs": macs}
        )

    def delete_group(self, group_id: str) -> dict[str, Any]:
        return self._delete(f"/devices/groups/{group_id}")

    def rescan(self) -> RescanResponse:
        data = self._post("/devices/rescan")
        return RescanResponse(**data)

    # ------------------------------------------------------------------ Control

    def control_device(
        self, device_id: str, payload: CommandRequest
    ) -> DeviceControlResponse:
        data = self._post(
            f"/control/device/{device_id}",
            json=payload.model_dump(exclude_none=True),
        )
        return DeviceControlResponse(**data)

    def control_group(
        self, group_id: str, payload: CommandRequest
    ) -> GroupControlResponse:
        data = self._post(
            f"/control/group/{group_id}",
            json=payload.model_dump(exclude_none=True),
        )
        return GroupControlResponse(**data)

    def blink_device(self, device_id: str) -> BlinkResponse:
        data = self._post(f"/control/device/{device_id}/blink")
        return BlinkResponse(**data)

    def device_status(self, device_id: str) -> DeviceStatusResponse:
        data = self._get(f"/control/device/{device_id}/status")
        return DeviceStatusResponse(**data)

    def group_status(self, group_id: str) -> GroupStatusResponse:
        data = self._get(f"/control/group/{group_id}/status")
        return GroupStatusResponse(**data)

    # --------------------------------------------------------------- Schedules

    def create_once_schedule(
        self, payload: ScheduleOnceRequest
    ) -> ScheduleCreateResponse:
        data = self._post(
            "/schedules/once",
            json=payload.model_dump(exclude_none=True),
        )
        return ScheduleCreateResponse(**data)

    def create_cron_schedule(
        self, payload: ScheduleCronRequest
    ) -> ScheduleCreateResponse:
        data = self._post(
            "/schedules/cron",
            json=payload.model_dump(exclude_none=True),
        )
        return ScheduleCreateResponse(**data)

    def list_schedules(self) -> ScheduleTasksResponse:
        data = self._get("/schedules/tasks")
        return ScheduleTasksResponse(**data)

    def delete_schedule(self, job_id: str) -> DeleteStatusResponse:
        data = self._delete(f"/schedules/{job_id}")
        return DeleteStatusResponse(**data)

    # ------------------------------------------------------------------- Stats

    def stats_summary(self, days: int = 7) -> dict[str, Any]:
        return self._get("/stats/summary", params={"days": days})

    def stats_log(self, limit: int = 50) -> list[dict[str, Any]]:
        return self._get("/stats/log", params={"limit": limit})

    # ---------------------------------------------------------------- API Keys

    def list_api_keys(self) -> list[dict[str, Any]]:
        return self._get("/api-keys")

    def create_api_key(self, name: str, *, is_admin: bool = False) -> dict[str, Any]:
        return self._post("/api-keys", params={"name": name, "is_admin": is_admin})

    def revoke_api_key(self, key_or_id: str) -> dict[str, Any]:
        return self._post(
            "/api-keys/revoke", json=KeyActionRequest(key=key_or_id).model_dump()
        )

    def activate_api_key(self, key_or_id: str) -> dict[str, Any]:
        return self._post(
            "/api-keys/activate", json=KeyActionRequest(key=key_or_id).model_dump()
        )