instrument-data-mcp/instrument_config_mcp/config_api.py

from __future__ import annotations

from typing import Any

from .auth import find_project_config, resolve_project_token, _request_json

CONFIG_OPERATOR = "Dataturing"


def _post_config(project_key: str, path: str, payload: dict[str, Any]) -> Any:
    project = find_project_config(project_key)
    authorization = resolve_project_token(project)
    response_payload = _request_json(
        "POST",
        f"{project['base_url']}{path}",
        authorization,
        json_payload=payload,
    )

    if not isinstance(response_payload, dict):
        raise ValueError(
            f"config API returned invalid payload for {path}: {response_payload}"
        )

    state = response_payload.get("state")
    if str(state) not in {"0", "0.0"}:
        state_info = str(response_payload.get("state_info") or "").strip()
        raise ValueError(
            f"config API failed for {path}: {state_info or response_payload}"
        )

    return response_payload


def list_locations(
    project_key: str,
    keyword: str | None = None,
    page_size: int = 100,
    page_num: int = 1,
) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/location/list",
        {
            "operator": CONFIG_OPERATOR,
            "page_size": page_size,
            "page_num": page_num,
            "keyword": keyword or "",
        },
    )


def list_system_tree(project_key: str) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/system/tree",
        {"operator": CONFIG_OPERATOR},
    )


def list_systems(
    project_key: str,
    page_size: int = 100,
    page_num: int = 1,
    system_type_id: int = 0,
    show_below: bool = True,
) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/system/list",
        {
            "operator": CONFIG_OPERATOR,
            "page_size": page_size,
            "page_num": page_num,
            "system_type_id": system_type_id,
            "show_below": show_below,
        },
    )


def list_device_types(project_key: str) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/devicetype/list",
        {
            "operator": CONFIG_OPERATOR,
            "page_size": -1,
        },
    )


def list_meter_types(project_key: str) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/metertype/list",
        {
            "operator": CONFIG_OPERATOR,
            "page_size": -1,
        },
    )


def search_devices(
    project_key: str,
    page_size: int = 100,
    page_num: int = 1,
    keyword: str | None = None,
    location_id: int = 0,
    show_below: bool = True,
    system_ids: list[int] | None = None,
    device_type_ids: list[int] | None = None,
) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/device/list",
        {
            "operator": CONFIG_OPERATOR,
            "page_size": page_size,
            "page_num": page_num,
            "location_id": location_id,
            "show_below": show_below,
            "keyword": keyword or "",
            "system_ids": system_ids or [],
            "device_type_ids": device_type_ids or [],
        },
    )


def search_meters(
    project_key: str,
    page_size: int = 100,
    page_num: int = 1,
    keyword: str | None = None,
    location_id: int = 0,
    show_below: bool = True,
    meter_type_id: int = 0,
    measurement_location_ids: list[int] | None = None,
    measurement_system_ids: list[int] | None = None,
    measurement_device_type_ids: list[int] | None = None,
    status: int | None = None,
) -> Any:
    payload: dict[str, Any] = {
        "operator": CONFIG_OPERATOR,
        "page_size": page_size,
        "page_num": page_num,
        "location_id": location_id,
        "show_below": show_below,
        "keyword": keyword or "",
        "meter_type_id": meter_type_id,
        "measurement_location_ids": measurement_location_ids or [],
        "measurement_system_ids": measurement_system_ids or [],
        "measurement_device_type_ids": measurement_device_type_ids or [],
    }
    if status is not None:
        payload["status"] = status
    return _post_config(project_key, "/api/configapi/meter/list", payload)


def search_points(
    project_key: str, id: int, page_size: int = 100, page_num: int = 1
) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/meter/search_point",
        {
            "operator": CONFIG_OPERATOR,
            "page_size": page_size,
            "page_num": page_num,
            "id": id,
        },
    )


def list_topologies_with_group(
    project_key: str, group_ids: list[int] | None = None
) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/topo/list_with_group",
        {
            "operator": CONFIG_OPERATOR,
            "group_ids": group_ids or [],
        },
    )


def get_topology(project_key: str, id: int) -> Any:
    return _post_config(
        project_key,
        "/api/configapi/topo/get",
        {
            "operator": CONFIG_OPERATOR,
            "id": id,
        },
    )