data-collector-mcp/data_collector_mcp/s7_server.py

from __future__ import annotations

from typing import Annotated, Any

from pydantic import BaseModel, ConfigDict, Field

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


JsonScalar = str | int | float | bool | None

ProjectKeyParam = Annotated[str, Field(description="项目标识，来自 project.list 返回的 project_key。")]
S7IpParam = Annotated[str, Field(description="PLC 地址。")]
S7RockParam = Annotated[int, Field(description="机架号。")]
S7SlotParam = Annotated[int, Field(description="槽号。")]
S7PortParam = Annotated[int, Field(description="S7 TCP 端口，默认 102。")]
S7DeviceTypeParam = Annotated[str, Field(description="网关设备类型，可用 S7-1200、S7-1500、S7-Smart200。")]
S7TsapConnTypeParam = Annotated[str | None, Field(description="TSAP 连接类型，可用 PG、OP、BASIC；不传时默认 PG。")]
S7DataTypeParam = Annotated[
    str,
    Field(description="数据类型；支持 bool、uint8、int8、uint16、int16、uint32、int32、float32、float64 及常见别名。"),
]


class FlexibleModel(BaseModel):
    model_config = ConfigDict(extra="allow")


class S7DeviceCreateItem(BaseModel):
    name: str = Field(description="设备名称。")
    ip: str = Field(description="PLC 地址。")
    rock: int = Field(description="机架号。")
    slot: int = Field(description="槽号。")
    port: int = Field(default=102, description="S7 TCP 端口。")
    device_type: int = Field(default=1, description="设备类型：1=S7-1200，2=S7-1500，3=S7-Smart200。")
    tsap_conn_type: str | None = Field(default=None, description="TSAP 连接类型，可用 PG、OP、BASIC；不传时默认 PG。")
    is_persistent: bool = Field(default=False, description="是否持久化连接。")
    device_group_id: int = Field(default=0, description="设备分组 id。")
    group_id: int | None = Field(default=None, description="设备分组 id；会转换为 device_group_id。")
    timeout: int = Field(default=3, description="连接超时时间。")
    alarm_interval: int = Field(default=90, description="告警间隔，单位秒。")
    collect_interval: int = Field(default=5, description="采集周期，单位秒。")


class S7PointCreateItem(BaseModel):
    device_id: int = Field(description="所属 S7 设备 id，来自 collector.device_list 或设备创建结果。")
    name: str = Field(description="点位名称。")
    address: str = Field(description="S7 地址；DB 非 bool 通常为 db.byte，DB bool 为 db.byte.bit。")
    data_type: str | None = Field(default=None, description="数据类型；也可用 type 指定。")
    type: str | None = Field(default=None, description="数据类型别名；会转换为 data_type。")
    register_type: int = Field(description="寄存器区域：1=I，2=Q，3=M，4=DB，5=V，6=AI。")
    point_id: str = Field(default="", description="外部点位编码。")
    scale_ratio: float = Field(default=1, description="缩放系数。")
    value_offset: float = Field(default=0, description="值偏移。")
    group_id: int = Field(default=0, description="点位分组 id。")
    invalid_values: str = Field(default="", description="无效值列表，多个值用逗号分隔。")
    valid_range_start: float | None = Field(default=None, description="合法范围最小值。")
    valid_range_end: float | None = Field(default=None, description="合法范围最大值。")
    describe: str = Field(default="", description="点位描述。")


class S7RawReadSpec(FlexibleModel):
    area: str = Field(description="读取区域，可用 DB、M、I、Q、V。")
    start: int = Field(description="起始字节地址。")
    size: int = Field(description="读取字节数。")
    db: int | None = Field(default=None, description="DB 块编号；area=DB 时使用。")


class S7PointCollectSpec(FlexibleModel):
    area: str = Field(description="读取区域，可用 DB、M、I、Q、V。")
    start: int = Field(description="起始字节地址。")
    type: S7DataTypeParam
    db: int | None = Field(default=None, description="DB 块编号；area=DB 时使用。")
    bit: int | None = Field(default=None, description="bool 点读取的位索引，范围 0..7。")
    name: str | None = Field(default=None, description="点位名称；测试读取时可不传。")


class GatewayBaseOutput(FlexibleModel):
    code: int | str | None = Field(default=None, description="网关状态码；0 表示成功。")
    msg: str | None = Field(default=None, description="网关状态说明或错误信息。")


class S7CommunicationData(FlexibleModel):
    communication: list[str] = Field(default_factory=list, description="本次请求期间的连接、读取和断开过程记录。")


class S7RawReadOutput(GatewayBaseOutput):
    data: S7CommunicationData | None = Field(default=None, description="S7 原始字节读取结果。")


class S7GatewayPoint(FlexibleModel):
    name: str | None = Field(default=None, description="点位名称。")
    area: str | None = Field(default=None, description="读取区域。")
    start: int | None = Field(default=None, description="起始字节地址。")
    type: str | None = Field(default=None, description="数据类型。")
    value: JsonScalar = Field(default=None, description="转换后的点位值。")


class S7PointCollectData(S7CommunicationData):
    points: list[S7GatewayPoint] = Field(default_factory=list, description="转换后的 S7 点位值列表。")


class S7PointCollectOutput(GatewayBaseOutput):
    data: S7PointCollectData | None = Field(default=None, description="S7 点位读取结果。")


class S7ConnectScanItem(FlexibleModel):
    rock: int | None = Field(default=None, description="可用机架号。")
    slot: int | None = Field(default=None, description="可用槽号。")
    tsap_conn_type: str | None = Field(default=None, description="可用 TSAP 连接类型。")
    device_type: str | None = Field(default=None, description="可用设备类型。")


class S7ConnectScanData(FlexibleModel):
    available: list[S7ConnectScanItem] = Field(default_factory=list, description="扫描到的可用连接组合。")


class S7ConnectScanOutput(GatewayBaseOutput):
    data: S7ConnectScanData | None = Field(default=None, description="S7 连接扫描结果。")


class CollectorBaseOutput(FlexibleModel):
    state: int | str | None = Field(default=None, description="汇采状态码；0 表示成功。")
    state_info: str | None = Field(default=None, description="汇采状态说明。")
    data: Any = Field(default=None, description="汇采接口返回数据。")


class CollectorPointEditOutput(FlexibleModel):
    state: int | str | None = Field(default=None, description="汇采状态码；0 表示成功。")
    state_info: str | None = Field(default=None, description="汇采状态说明。")


class BatchError(FlexibleModel):
    index: int | None = Field(default=None, description="输入数组中的下标。")
    name: str | None = Field(default=None, description="输入项名称。")
    stage: str | None = Field(default=None, description="失败阶段。")
    error: str | None = Field(default=None, description="失败原因。")


class DeviceCreateSummary(FlexibleModel):
    total: int | None = Field(default=None, description="输入设备总数。")
    created: int | None = Field(default=None, description="汇采创建设备成功数量。")
    matched: int | None = Field(default=None, description="创建后从设备列表匹配到设备 id 的数量。")
    failed: int | None = Field(default=None, description="失败数量。")


class DeviceCreateResult(FlexibleModel):
    index: int | None = Field(default=None, description="输入 devices 数组中的下标。")
    name: str | None = Field(default=None, description="设备名称。")
    device_id: int | None = Field(default=None, description="匹配到的汇采设备 id。")
    create_response: dict[str, Any] | None = Field(default=None, description="汇采创建设备接口原始响应。")
    matched_device: dict[str, Any] | None = Field(default=None, description="设备列表中匹配到的设备详情。")


class S7DeviceCreateOutput(FlexibleModel):
    state: int | str | None = Field(default=None, description="批量创建状态；0 表示全部成功，1 表示存在失败。")
    summary: DeviceCreateSummary | None = Field(default=None, description="批量创建设备汇总。")
    results: list[DeviceCreateResult] = Field(default_factory=list, description="每个设备的创建和匹配结果。")
    errors: list[BatchError] = Field(default_factory=list, description="失败项列表。")


class PointCreateSummary(FlexibleModel):
    total: int | None = Field(default=None, description="输入点位总数。")
    success: int | None = Field(default=None, description="汇采创建点位成功数量。")
    failed: int | None = Field(default=None, description="失败数量。")


class PointCreateResult(FlexibleModel):
    index: int | None = Field(default=None, description="输入 points 数组中的下标。")
    name: str | None = Field(default=None, description="点位名称。")
    device_id: int | None = Field(default=None, description="所属 S7 设备 id。")
    response: dict[str, Any] | None = Field(default=None, description="汇采创建点位接口原始响应。")


class S7PointCreateOutput(FlexibleModel):
    state: int | str | None = Field(default=None, description="批量创建状态；0 表示全部成功，1 表示存在失败。")
    summary: PointCreateSummary | None = Field(default=None, description="批量创建点位汇总。")
    results: list[PointCreateResult] = Field(default_factory=list, description="每个点位的创建结果。")
    errors: list[BatchError] = Field(default_factory=list, description="失败项列表。")


@mcp.tool(
    name="s7.raw_read",
    description=(
        "采集网关-对 Siemens S7 TCP 设备执行原始字节读取，不解析业务值。"
    ),
)
def s7_raw_read(
    project_key: ProjectKeyParam,
    ip: S7IpParam,
    read: Annotated[S7RawReadSpec, Field(description="原始字节读取参数。")],
    rock: S7RockParam = 0,
    slot: S7SlotParam = 1,
    device_type: S7DeviceTypeParam = "S7-1200",
    port: S7PortParam = 102,
    tsap_conn_type: S7TsapConnTypeParam = None,
) -> S7RawReadOutput:
    return api_s7_raw_read(
        project_key,
        ip=ip,
        rock=rock,
        slot=slot,
        read=_dump_model_or_dict(read),
        device_type=device_type,
        port=port,
        tsap_conn_type=tsap_conn_type,
    )


@mcp.tool(
    name="s7.point_collect_test",
    description=(
        "采集网关-读取 Siemens S7 TCP 点位并转换为业务值。"
    ),
)
def s7_point_collect_test(
    project_key: ProjectKeyParam,
    ip: S7IpParam,
    rock: S7RockParam,
    slot: S7SlotParam,
    points: Annotated[list[S7PointCollectSpec], Field(description="要读取的 S7 点位列表。")],
    device_type: S7DeviceTypeParam = "S7-1200",
    port: S7PortParam = 102,
    tsap_conn_type: S7TsapConnTypeParam = None,
) -> S7PointCollectOutput:
    return api_s7_point_collect_test(
        project_key,
        ip=ip,
        rock=rock,
        slot=slot,
        points=[_dump_model_or_dict(item) for item in points],
        device_type=device_type,
        port=port,
        tsap_conn_type=tsap_conn_type,
    )


@mcp.tool(
    name="s7.connect_scan",
    description=(
        "采集网关-扫描 Siemens S7 TCP 设备可用连接组合。"
    ),
)
def s7_connect_scan(project_key: ProjectKeyParam, ip: S7IpParam) -> S7ConnectScanOutput:
    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
    return "PG"


@mcp.tool(
    name="collector.s7_device_create",
    description=(
        "汇采-批量创建 S7 设备。创建后会读取设备列表并匹配返回设备 id。"
    ),
)
def collector_s7_device_create(
    project_key: ProjectKeyParam,
    devices: Annotated[list[S7DeviceCreateItem], Field(description="要创建的 S7 设备列表。")],
) -> S7DeviceCreateOutput:
    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 设备。只有明确需要 OP/BASIC 时才显式传 tsap_conn_type。"
        "编辑前设备不能处于已连接状态；若已连接，请先调用 "
        "collector.device_disconnect，并传 device_type=s7。"
    ),
)
def collector_s7_device_edit(
    project_key: ProjectKeyParam,
    id: Annotated[int, Field(description="要编辑的原汇采设备 id。")],
    name: Annotated[str, Field(description="设备名称。")],
    ip: S7IpParam,
    rock: S7RockParam,
    slot: S7SlotParam,
    port: S7PortParam = 102,
    device_type: Annotated[int, Field(description="设备类型：1=S7-1200，2=S7-1500，3=S7-Smart200。")]= 1,
    tsap_conn_type: Annotated[str | None, Field(description="TSAP 连接类型，可用 PG、OP、BASIC；不传时默认 PG。")]= None,
    is_persistent: Annotated[bool, Field(description="是否持久化连接。")]= False,
    device_group_id: Annotated[int, Field(description="设备分组 id。")]= 0,
    timeout: Annotated[int, Field(description="连接超时时间。")]= 3,
    alarm_interval: Annotated[int, Field(description="告警间隔，单位秒。")]= 90,
    collect_interval: Annotated[int, Field(description="采集周期，单位秒。")]= 5,
) -> CollectorBaseOutput:
    return api_edit_s7_device(
        project_key,
        {
            "id": 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 采集点位。数据类型会在调用汇采接口前规范化。"
        "address 地址格式：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。"
    ),
)
def collector_s7_point_create(
    project_key: ProjectKeyParam,
    points: Annotated[list[S7PointCreateItem], Field(description="要创建的 S7 采集点位列表。")],
) -> S7PointCreateOutput:
    return api_create_s7_points(project_key, [_dump_model_or_dict(item) for item in 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 采集点位。数据类型会在调用汇采接口前规范化。"
    ),
)
def collector_s7_point_edit(
    project_key: ProjectKeyParam,
    id: Annotated[int, Field(description="要编辑的原采集点位 id，对应 collector.device_points 返回的点位 id。")],
    device_id: Annotated[int, Field(description="所属 S7 设备 id。")],
    name: Annotated[str, Field(description="点位名称。")],
    address: Annotated[str, Field(description="S7 地址；DB 非 bool 通常为 db.byte，DB bool 为 db.byte.bit。")],
    data_type: S7DataTypeParam,
    register_type: Annotated[int, Field(description="寄存器区域：1=I，2=Q，3=M，4=DB，5=V，6=AI。")],
    point_id: Annotated[str, Field(description="外部点位编码。")]= "",
    scale_ratio: Annotated[float, Field(description="缩放系数。")]= 1,
    value_offset: Annotated[float, Field(description="值偏移。")]= 0,
    group_id: Annotated[int, Field(description="点位分组 id。")]= 0,
    invalid_values: Annotated[str, Field(description="无效值列表，多个值用逗号分隔。")]= "",
    valid_range_start: Annotated[float | None, Field(description="合法范围最小值。")]= None,
    valid_range_end: Annotated[float | None, Field(description="合法范围最大值。")]= None,
    describe: Annotated[str, Field(description="点位描述。")]= "",
) -> CollectorPointEditOutput:
    payload: dict[str, Any] = {
        "id": 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,
    }
    payload["register_type"] = register_type
    return api_edit_s7_point(project_key, payload)