data-collector-mcp/data_collector_mcp/s7_server.py

from __future__ import annotations

from typing import Any, NotRequired, TypedDict

from pydantic import BaseModel

from .collector_api import (
    create_s7_devices as api_create_s7_devices,
    create_s7_points as api_create_s7_points,
    edit_s7_device as api_edit_s7_device,
    edit_s7_point as api_edit_s7_point,
)
from .gateway_api import (
    s7_connect_scan as api_s7_connect_scan,
    s7_point_collect_test as api_s7_point_collect_test,
    s7_raw_read as api_s7_raw_read,
)
from .mcp_app import mcp


class S7DeviceCreateItem(BaseModel):
    name: str
    ip: str
    rock: int
    slot: int
    port: int = 102
    device_type: int = 1
    tsap_conn_type: str | None = None
    is_persistent: bool = False
    device_group_id: int = 0
    group_id: int | None = None
    timeout: int = 3
    alarm_interval: int = 90
    collect_interval: int = 5


class S7PointCreateItem(TypedDict):
    device_id: int
    name: str
    address: str
    data_type: NotRequired[str]
    type: NotRequired[str]
    register_type: NotRequired[int]
    register_area: NotRequired[str]
    point_id: NotRequired[str]
    scale_ratio: NotRequired[float]
    value_offset: NotRequired[float]
    group_id: NotRequired[int]
    invalid_values: NotRequired[str]
    valid_range_start: NotRequired[float | None]
    valid_range_end: NotRequired[float | None]
    describe: NotRequired[str]


@mcp.tool(
    name="s7.raw_read",
    description=(
        "通过采集网关对 Siemens S7 TCP 设备执行原始字节读取。调用 "
        "{base_url}/api/dc-gateway/s7/read。请求字段使用网关/汇采一致命名："
        "ip 为 PLC 地址，port 默认 102，rock 为机架号，slot 为槽号，"
        "device_type 为设备类型，默认 S7-1200，可传 S7-1200、S7-Smart200。"
        "tsap_conn_type 可选 PG、OP、BASIC；未传时 S7-1200 使用 PG，"
        "device_type=S7-Smart200 时使用 OP。read 必须包含 area、start、size；"
        "area 可用 DB、M、I、Q，S7-Smart200 还支持 V；area=DB 时 read.db 必须大于 0。"
        "该接口不解析业务值，只返回语义化 communication：S7_CONNECT、S7_CONNECTED、"
        "S7_READ、原始十六进制字节、S7_DISCONNECT 或 S7_ERROR。"
        "响应透传上游 JSON，code=0 表示业务成功，code!=0 时 msg 为失败原因。"
    ),
)
def s7_raw_read(
    project_key: str,
    ip: str,
    read: dict[str, Any],
    rock: int = 0,
    slot: int = 1,
    device_type: str = "S7-1200",
    port: int = 102,
    tsap_conn_type: str | None = None,
) -> dict[str, Any]:
    return api_s7_raw_read(
        project_key,
        ip=ip,
        rock=rock,
        slot=slot,
        read=read,
        device_type=device_type,
        port=port,
        tsap_conn_type=tsap_conn_type,
    )


@mcp.tool(
    name="s7.point_collect_test",
    description=(
        "通过采集网关读取 Siemens S7 TCP 点位并转换为业务值。调用 "
        "{base_url}/api/dc-gateway/s7/read_points。请求字段使用网关/汇采一致命名："
        "ip 为 PLC 地址，port 默认 102，rock 为机架号，slot 为槽号，"
        "device_type 为设备类型，默认 S7-1200，可传 S7-1200、S7-Smart200。"
        "tsap_conn_type 可选 PG、OP、BASIC；未传时 S7-1200 使用 PG，"
        "device_type=S7-Smart200 时,tsap_conn_type使用 OP。points 为点位数组，每个点位必须包含 "
        "area、start、type；area 可用 DB、M、I、Q，S7-Smart200 还支持 V；"
        "area=DB 时 db 必须大于 0。"
        "type 可用 bool、byte、int8、int16、uint16、int32、uint32、int64、uint64、"
        "float32、float64；bool 点可传 bit 读取指定 0..7 位，非 bool 点不能传 bit。"
        "S7 多字节数值按大端解析，响应 data.points[].value 为转换后的业务值，"
        "data.communication 为语义化连接/读取/断开记录。响应透传上游 JSON，code=0 表示业务成功。"
    ),
)
def s7_point_collect_test(
    project_key: str,
    ip: str,
    rock: int,
    slot: int,
    points: list[dict[str, Any]],
    device_type: str = "S7-1200",
    port: int = 102,
    tsap_conn_type: str | None = None,
) -> dict[str, Any]:
    return api_s7_point_collect_test(
        project_key,
        ip=ip,
        rock=rock,
        slot=slot,
        points=points,
        device_type=device_type,
        port=port,
        tsap_conn_type=tsap_conn_type,
    )


@mcp.tool(
    name="s7.connect_scan",
    description=(
        "通过采集网关扫描 Siemens S7 TCP 设备可用连接组合。调用 "
        "{base_url}/api/dc-gateway/s7/connect_scan。只需要传 PLC 的 ip，网关固定使用端口 102，"
        "扫描 rock=0/1/2、slot=0/1/2、tsap_conn_type=PG/OP/BASIC 共 27 种组合。"
        "响应 data.available 只返回可连接成功的组合，每项包含 rock、slot、tsap_conn_type；"
        "如果没有可用组合则 available 为空数组。响应透传上游 JSON，code=0 表示扫描完成。"
    ),
)
def s7_connect_scan(project_key: str, ip: str) -> dict[str, Any]:
    return api_s7_connect_scan(project_key, ip=ip)


def _resolve_collector_s7_tsap_conn_type(device_type: int, tsap_conn_type: str | None) -> str:
    normalized_tsap = str(tsap_conn_type or "").strip().upper()
    if normalized_tsap:
        return normalized_tsap
    if int(device_type) == 3:
        return "OP"
    return "PG"


@mcp.tool(
    name="collector.s7_device_create",
    description=(
        "汇采-批量创建 S7 设备。依次调用 {data_collector_base_url}/api/collector/device，"
        "与 Modbus 创建设备共用地址，但请求体固定 type=s7，入参使用 S7 字段。"
        "全部设备创建调用完成后会内部调用设备列表并匹配每个设备 id；如果匹配到多个设备，"
        "选择 id 最大的。devices 每项必须传 name 名称、ip IP 地址、rock 机架号(轨道号)、slot 槽号；"
        "port 默认 102，device_type 默认 1。tsap_conn_type 可选 PG、OP、BASIC；"
        "未传时 device_type=1 使用 PG，device_type=3 时使用 OP。"
        "默认参数: is_persistent=false, "
        "device_group_id=0, timeout=3, alarm_interval=90, collect_interval=5。"
        "device_type: 1=S7-1200, 2=S7-1500, 3=S7-Smart200。"
        "tsap_conn_type 可用 PG、OP、BASIC。返回批量结果，state=0 表示全部创建并匹配成功。"
    ),
)
def collector_s7_device_create(
    project_key: str,
    devices: list[S7DeviceCreateItem],
) -> dict[str, Any]:
    return api_create_s7_devices(project_key, [_dump_model_or_dict(item) for item in devices])


@mcp.tool(
    name="collector.s7_device_edit",
    description=(
        "汇采-编辑 S7 设备。调用 {data_collector_base_url}/api/collector/s7/device/update。"
        "必须传 ori_id 原设备 id、name、ip、rock、slot；port 默认 102，device_type 默认 1，"
        "tsap_conn_type 可选 PG、OP、BASIC；未传时 device_type=1 使用 PG，device_type=3 时使用 OP。"
        "编辑前设备不能处于已连接状态；若已连接，请先调用 "
        "collector.device_disconnect，并传 device_type=s7。该接口是全量更新语义。"
        "默认参数: is_persistent=false, device_group_id=0, timeout=3, alarm_interval=90, "
        "collect_interval=5。device_type: 1=S7-1200, 2=S7-1500, 3=S7-Smart200。"
        "tsap_conn_type 可用 PG、OP、BASIC。响应透传上游 JSON，state=0 表示业务成功。"
    ),
)
def collector_s7_device_edit(
    project_key: str,
    ori_id: int,
    name: str,
    ip: str,
    rock: int,
    slot: int,
    port: int = 102,
    device_type: int = 1,
    tsap_conn_type: str | None = None,
    is_persistent: bool = False,
    device_group_id: int = 0,
    timeout: int = 3,
    alarm_interval: int = 90,
    collect_interval: int = 5,
) -> dict[str, Any]:
    return api_edit_s7_device(
        project_key,
        {
            "ori_id": ori_id,
            "name": name,
            "ip": ip,
            "rock": rock,
            "slot": slot,
            "port": port,
            "device_type": device_type,
            "tsap_conn_type": _resolve_collector_s7_tsap_conn_type(device_type, tsap_conn_type),
            "is_persistent": is_persistent,
            "device_group_id": device_group_id,
            "timeout": timeout,
            "alarm_interval": alarm_interval,
            "collect_interval": collect_interval,
        },
    )


@mcp.tool(
    name="collector.s7_point_create",
    description=(
        "汇采-批量创建 S7 采集点位。依次调用 {data_collector_base_url}/api/collector/s7/point/add。"
        "points 每项必须传 device_id、name、address、data_type 或 type，"
        "以及 register_type 或 register_area。每个点位会补齐默认值: point_id='', "
        "scale_ratio=1, value_offset=0, group_Id=0, invalid_values='', valid_range_start=null, "
        "valid_range_end=null。register_type: 1=I输入区, 2=Q输出区, 3=M存储区, 4=DB数据块, "
        "5=V区, 6=AI模拟输入；register_area 可用 I、Q、M、DB、V、AI。"
        "data_type 可用 bool, uint8, int8, uint16, int16, uint32, int32, float32, float64。"
        "S7 地址格式：I/Q/M/V 布尔点通常为 byte.bit，如 10.2；DB 布尔点为 db.byte.bit，"
        "如 1.10.2；非布尔点 I/Q/M/V 通常为 byte 地址，如 10，DB 为 db.byte，如 1.10。"
        "响应透传上游 JSON，state=0 表示业务成功。"
    ),
)
def collector_s7_point_create(
    project_key: str,
    points: list[S7PointCreateItem],
) -> dict[str, Any]:
    return api_create_s7_points(project_key, points)


def _dump_model_or_dict(item: Any) -> dict[str, Any]:
    if isinstance(item, BaseModel):
        return item.model_dump(exclude_none=True)
    return dict(item)


@mcp.tool(
    name="collector.s7_point_edit",
    description=(
        "汇采-编辑 S7 采集点位。调用 {data_collector_base_url}/api/collector/s7/point/update。"
        "必须传 ori_id 原点位 id、device_id 所属设备 id、name、address、data_type，"
        "以及 register_type 或 register_area。ori_id 对应 collector.device_points 返回的 data.point[].id。"
        "该接口是全量更新语义，未传字段可能被默认值覆盖。默认参数: point_id='', "
        "scale_ratio=1, value_offset=0, group_Id=0, invalid_values='', valid_range_start=null, "
        "valid_range_end=null。register_type: 1=I, 2=Q, 3=M, 4=DB, 5=V, 6=AI。"
        "data_type 可用 bool, uint8, int8, uint16, int16, uint32, int32, float32, float64。"
        "响应透传上游 JSON，state=0 表示业务成功。"
    ),
)
def collector_s7_point_edit(
    project_key: str,
    ori_id: int,
    device_id: int,
    name: str,
    address: str,
    data_type: str,
    register_type: int = 0,
    register_area: str = "",
    point_id: str = "",
    scale_ratio: float = 1,
    value_offset: float = 0,
    group_id: int = 0,
    invalid_values: str = "",
    valid_range_start: float | None = None,
    valid_range_end: float | None = None,
    describe: str = "",
) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "ori_id": ori_id,
        "device_id": device_id,
        "name": name,
        "address": address,
        "data_type": data_type,
        "point_id": point_id,
        "scale_ratio": scale_ratio,
        "value_offset": value_offset,
        "group_id": group_id,
        "invalid_values": invalid_values,
        "valid_range_start": valid_range_start,
        "valid_range_end": valid_range_end,
        "describe": describe,
    }
    if register_type:
        payload["register_type"] = register_type
    else:
        payload["register_area"] = register_area
    return api_edit_s7_point(project_key, payload)