data-collector-mcp/data_collector_mcp/gateway_api.py

from __future__ import annotations

from typing import Any

from .auth import find_project_config
from .http_client import request_json
from .protocols import MODBUS_SPEC, S7_SPEC


def _request_gateway(project_key: str, path: str | None, payload: dict[str, Any], protocol: str) -> dict[str, Any]:
    project = find_project_config(project_key)
    if not path:
        raise ValueError(f"{protocol} gateway path is not configured")

    response_payload = request_json(
        "POST",
        f"{project['base_url']}{path}",
        json_payload=payload,
    )
    if not isinstance(response_payload, dict):
        raise ValueError(f"gateway API returned invalid payload: {response_payload}")
    return response_payload


def modbus_point_collect_test(
    project_key: str,
    *,
    ip: str,
    port: int,
    slave_id: int,
    points: list[dict[str, Any]],
    device_type: str = "ModbusTCP",
    word_byte_order: str = "ABCD",
    address_base: int = 0,
) -> dict[str, Any]:
    payload = {
        "device_type": device_type,
        "ip": ip,
        "port": port,
        "word_byte_order": word_byte_order,
        "address_base": address_base,
        "slave_id": slave_id,
        "points": points,
    }
    return _request_gateway(project_key, MODBUS_SPEC.point_test_path, payload, "modbus point test")


def s7_raw_read(
    project_key: str,
    *,
    ip: str,
    rock: int,
    slot: int,
    read: dict[str, Any],
    device_type: str = "S7TCP",
    port: int = 102,
    tsap_conn_type: str = "PG",
) -> dict[str, Any]:
    payload = {
        "device_type": device_type,
        "ip": ip,
        "port": port,
        "rock": rock,
        "slot": slot,
        "tsap_conn_type": tsap_conn_type,
        "read": read,
    }
    return _request_gateway(project_key, S7_SPEC.raw_read_path, payload, "s7 raw read")


def s7_point_collect_test(
    project_key: str,
    *,
    ip: str,
    rock: int,
    slot: int,
    points: list[dict[str, Any]],
    device_type: str = "S7TCP",
    port: int = 102,
    tsap_conn_type: str = "PG",
) -> dict[str, Any]:
    payload = {
        "device_type": device_type,
        "ip": ip,
        "port": port,
        "rock": rock,
        "slot": slot,
        "tsap_conn_type": tsap_conn_type,
        "points": points,
    }
    return _request_gateway(project_key, S7_SPEC.point_test_path, payload, "s7 point test")


def s7_connect_scan(project_key: str, *, ip: str) -> dict[str, Any]:
    return _request_gateway(project_key, S7_SPEC.connect_scan_path, {"ip": ip}, "s7 connect scan")