emcp/backend/app/adapters/device_client.py

from __future__ import annotations

import struct
from abc import ABC, abstractmethod
from datetime import datetime
from random import Random
from typing import Any, Dict, List, Optional, Sequence

from app.core.config import settings

from app.schemas.platform import (
    AlarmEvent,
    ChannelConfigIn,
    DeviceConfigIn,
    DeviceStatus,
    LineAlarmSettingIn,
    LineData,
    RealtimeData,
    SystemConfigIn,
    SwitchControlIn,
    ValueGroup,
)


class DeviceClient(ABC):
    @abstractmethod
    def read_realtime_data(self) -> RealtimeData:
        raise NotImplementedError

    @abstractmethod
    def read_device_status(self) -> DeviceStatus:
        raise NotImplementedError

    @abstractmethod
    def read_alarm_events(self) -> List[AlarmEvent]:
        raise NotImplementedError

    @abstractmethod
    def send_device_config(self, payload: DeviceConfigIn) -> Dict[str, Any]:
        raise NotImplementedError

    @abstractmethod
    def send_channel_config(self, payload: ChannelConfigIn) -> Dict[str, Any]:
        raise NotImplementedError

    @abstractmethod
    def send_line_alarm_setting(self, payload: LineAlarmSettingIn) -> Dict[str, Any]:
        raise NotImplementedError

    @abstractmethod
    def send_ai_alarm_setting(self, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
        raise NotImplementedError

    @abstractmethod
    def send_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]:
        raise NotImplementedError

    @abstractmethod
    def send_switch_control(self, payload: SwitchControlIn) -> Dict[str, Any]:
        raise NotImplementedError


VALUE_GROUP_FIELDS = [
    "Ua",
    "Ub",
    "Uc",
    "Ia",
    "Ib",
    "Ic",
    "Pa",
    "Pb",
    "Pc",
    "Pt",
    "Qa",
    "Qb",
    "Qc",
    "Qt",
    "Sa",
    "Sb",
    "Sc",
    "St",
    "PFa",
    "PFb",
    "PFc",
    "PFt",
    "Uab",
    "Ubc",
    "Uca",
    "frq",
]
LINE_START_REGISTERS = [0x0000, 0x0034, 0x0068, 0x009C]
LINE_REGISTER_COUNT = 0x34
AI_INPUT_START_REGISTER = 0x00D0
AI_INPUT_COUNT = 8
DI_INPUT_REGISTER = 0x00E3
EVENT_FLAG_REGISTER = 0x00E4
EVENT_HOLDING_START_REGISTER = 0x0000
UART_CONFIG_START_REGISTER = 0x0004
UART_CONFIG_REGISTER_COUNT = 16
LINE_ALARM_START_REGISTER = 0x0014
LINE_ALARM_REGISTERS_PER_LINE = 14
AI_CHANNEL_START_REGISTER = 0x004C
AO_CHANNEL_START_REGISTER = 0x006A
AI_ALARM_START_REGISTER = 0x0088
ANALOG_BLOCK_REGISTER_COUNT = 30
DO_STATE_REGISTER = 0x00A6
TIME_SYNC_START_REGISTER = 0x0000


def _clamp_byte(value: Any) -> int:
    try:
        number = int(round(float(value)))
    except (TypeError, ValueError):
        number = 0
    return max(0, min(255, number))


def _normalize_text(value: str) -> str:
    return value.strip().upper().replace(" ", "")


def _serial_parity_flag(value: str) -> str:
    normalized = _normalize_text(value)
    return {"NONE": "N", "N": "N", "ODD": "O", "O": "O", "EVEN": "E", "E": "E"}.get(normalized, "N")


class ModbusTransport(ABC):
    @abstractmethod
    def read_input_registers(self, address: int, count: int) -> List[int]:
        raise NotImplementedError

    @abstractmethod
    def read_holding_registers(self, address: int, count: int) -> List[int]:
        raise NotImplementedError

    @abstractmethod
    def read_discrete_inputs(self, address: int, count: int) -> List[bool]:
        raise NotImplementedError

    @abstractmethod
    def read_coils(self, address: int, count: int) -> List[bool]:
        raise NotImplementedError

    @abstractmethod
    def write_registers(self, address: int, values: Sequence[int]) -> None:
        raise NotImplementedError

    @abstractmethod
    def write_coil(self, address: int, value: bool) -> None:
        raise NotImplementedError


class PymodbusTransport(ModbusTransport):
    def __init__(self) -> None:
        self._client: Any = None

    def _build_client(self) -> Any:
        try:
            from pymodbus.client import ModbusSerialClient, ModbusTcpClient
        except ImportError as exc:
            raise RuntimeError("未安装 pymodbus，无法启用 Modbus 采集") from exc

        transport = settings.modbus_transport.strip().lower()
        if transport == "tcp":
            return ModbusTcpClient(
                host=settings.modbus_tcp_host,
                port=settings.modbus_tcp_port,
                timeout=settings.modbus_timeout_seconds,
            )

        try:
            from pymodbus import FramerType

            return ModbusSerialClient(
                port=settings.modbus_serial_device,
                framer=FramerType.RTU,
                baudrate=settings.modbus_serial_baud,
                parity=_serial_parity_flag(settings.modbus_serial_parity),
                stopbits=settings.modbus_serial_stop_bits,
                bytesize=settings.modbus_serial_data_bits,
                timeout=settings.modbus_timeout_seconds,
            )
        except Exception:
            return ModbusSerialClient(
                method="rtu",
                port=settings.modbus_serial_device,
                baudrate=settings.modbus_serial_baud,
                parity=_serial_parity_flag(settings.modbus_serial_parity),
                stopbits=settings.modbus_serial_stop_bits,
                bytesize=settings.modbus_serial_data_bits,
                timeout=settings.modbus_timeout_seconds,
            )

    def _ensure_client(self) -> Any:
        if self._client is None:
            self._client = self._build_client()
        connect = getattr(self._client, "connect", None)
        if callable(connect):
            result = connect()
            if result is False:
                raise RuntimeError("Modbus 连接失败")
        return self._client

    def _call(self, method_name: str, *args: Any, **kwargs: Any) -> Any:
        client = self._ensure_client()
        method = getattr(client, method_name)
        try:
            return method(*args, slave=settings.modbus_slave_id, **kwargs)
        except TypeError:
            return method(*args, unit=settings.modbus_slave_id, **kwargs)

    def _check_response(self, response: Any, action: str) -> Any:
        if response is None:
            raise RuntimeError(f"Modbus {action} 返回为空")
        is_error = getattr(response, "isError", None)
        if callable(is_error) and response.isError():
            raise RuntimeError(f"Modbus {action} 返回错误: {response}")
        return response

    def read_input_registers(self, address: int, count: int) -> List[int]:
        response = self._check_response(
            self._call("read_input_registers", address, count),
            f"读输入寄存器[{hex(address)}:{count}]",
        )
        return list(getattr(response, "registers", []) or [])

    def read_holding_registers(self, address: int, count: int) -> List[int]:
        response = self._check_response(
            self._call("read_holding_registers", address, count),
            f"读保持寄存器[{hex(address)}:{count}]",
        )
        return list(getattr(response, "registers", []) or [])

    def read_discrete_inputs(self, address: int, count: int) -> List[bool]:
        response = self._check_response(
            self._call("read_discrete_inputs", address, count),
            f"读离散输入[{hex(address)}:{count}]",
        )
        return [bool(value) for value in (getattr(response, "bits", []) or [])[:count]]

    def read_coils(self, address: int, count: int) -> List[bool]:
        response = self._check_response(
            self._call("read_coils", address, count),
            f"读线圈[{hex(address)}:{count}]",
        )
        return [bool(value) for value in (getattr(response, "bits", []) or [])[:count]]

    def write_registers(self, address: int, values: Sequence[int]) -> None:
        self._check_response(
            self._call("write_registers", address, list(values)),
            f"写保持寄存器[{hex(address)}:{len(values)}]",
        )

    def write_coil(self, address: int, value: bool) -> None:
        self._check_response(
            self._call("write_coil", address, bool(value)),
            f"写线圈[{hex(address)}]",
        )


class MockDeviceClient(DeviceClient):
    def __init__(self) -> None:
        self._random = Random(3568)
        self._tick = 0

    def _value_group(self, base_u: float, base_i: float, base_p: float) -> ValueGroup:
        delta = self._tick % 5
        return ValueGroup(
            Ua=base_u + delta,
            Ub=base_u + 2 + delta,
            Uc=base_u - 1 + delta,
            Ia=base_i + 0.1 * delta,
            Ib=base_i - 0.1 + 0.1 * delta,
            Ic=base_i + 0.2 + 0.1 * delta,
            Pa=base_p + delta,
            Pb=base_p - 2 + delta,
            Pc=base_p + 3 + delta,
            Pt=base_p * 3 + delta,
            Qa=12 + delta,
            Qb=11 + delta,
            Qc=13 + delta,
            Qt=36 + delta,
            Sa=base_p + 15 + delta,
            Sb=base_p + 10 + delta,
            Sc=base_p + 16 + delta,
            St=base_p * 3 + 44 + delta,
            PFa=0.98,
            PFb=0.97,
            PFc=0.99,
            PFt=0.98,
            Uab=base_u * 1.73,
            Ubc=base_u * 1.72,
            Uca=base_u * 1.73,
            frq=50.0,
        )

    def read_realtime_data(self) -> RealtimeData:
        self._tick += 1
        line_list = [
            LineData(line_no=1, pri_val=self._value_group(6000, 150, 820), sec_val=self._value_group(57.6, 1.2, 68)),
            LineData(line_no=2, pri_val=self._value_group(5990, 148, 810), sec_val=self._value_group(57.2, 1.1, 66)),
            LineData(line_no=3, pri_val=self._value_group(6010, 151, 830), sec_val=self._value_group(58.1, 1.3, 70)),
            LineData(line_no=4, pri_val=self._value_group(6020, 149, 825), sec_val=self._value_group(58.4, 1.2, 69)),
        ]
        switch = {f"di{i}": int((i + self._tick) % 2 == 0) for i in range(1, 13)}
        switch.update({f"do{i}": int((i + self._tick + 1) % 2 == 0) for i in range(1, 13)})
        ai_collect = {f"ai{i}": round(1 + self._random.random() * 5 + (self._tick % 3) * 0.1, 2) for i in range(1, 13)}
        return RealtimeData(line_list=line_list, switch=switch, ai_collect=ai_collect)

    def read_device_status(self) -> DeviceStatus:
        return DeviceStatus(
            self_check="正常",
            net1="正常",
            net2="正常",
            uart1="正常",
            uart2="正常" if self._tick % 4 else "断开",
        )

    def read_alarm_events(self) -> List[AlarmEvent]:
        if self._tick % 6 != 0:
            return []
        return [
            AlarmEvent(
                alarm_type="line_alarm",
                time=datetime.now(),
                no=str((self._tick // 6) % 4 + 1),
                type="电压",
                content="模拟告警：线路电压越限",
                level="中",
            )
        ]

    def send_device_config(self, payload: DeviceConfigIn) -> Dict[str, Any]:
        return {"send_status": "成功", "target": "device", "items": len(payload.net) + len(payload.uart)}

    def send_channel_config(self, payload: ChannelConfigIn) -> Dict[str, Any]:
        return {
            "send_status": "成功",
            "target": "channel",
            "items": len(payload.ai_channel) + len(payload.ao_channel),
        }

    def send_line_alarm_setting(self, payload: LineAlarmSettingIn) -> Dict[str, Any]:
        return {"send_status": "成功", "target": "line_alarm", "line_no": payload.line_no}

    def send_ai_alarm_setting(self, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
        return {"send_status": "成功", "target": "ai_alarm", "items": len(payload)}

    def send_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]:
        return {"send_status": "成功", "target": "system", "brightness": payload.brightness}

    def send_switch_control(self, payload: SwitchControlIn) -> Dict[str, Any]:
        action_text = "合" if payload.action == 1 else "分"
        return {"control_status": f"执行成功: 开关{payload.ch}{action_text}"}


class CDeviceClient(DeviceClient):
    """基于 Modbus 协议的真实设备客户端。"""

    def __init__(self, transport: Optional[ModbusTransport] = None) -> None:
        self.transport = transport or PymodbusTransport()
        self._do_state = {index: 0 for index in range(1, 13)}
        self._last_event_signature: Optional[tuple[int, int, int, int]] = None

    def _registers_to_float(self, registers: Sequence[int]) -> float:
        ordered = list(registers)
        if settings.modbus_float_word_order.strip().lower() == "little":
            ordered.reverse()

        raw = bytearray()
        for register in ordered:
            high = (int(register) >> 8) & 0xFF
            low = int(register) & 0xFF
            if settings.modbus_float_byte_order.strip().lower() == "little":
                raw.extend([low, high])
            else:
                raw.extend([high, low])
        return float(struct.unpack(">f", bytes(raw))[0])

    def _float_to_registers(self, value: float) -> List[int]:
        raw = struct.pack(">f", float(value))
        words = [bytearray(raw[0:2]), bytearray(raw[2:4])]
        if settings.modbus_float_byte_order.strip().lower() == "little":
            words = [bytearray(reversed(word)) for word in words]
        registers = [(word[0] << 8) | word[1] for word in words]
        if settings.modbus_float_word_order.strip().lower() == "little":
            registers.reverse()
        return registers

    def _bytes_to_registers(self, payload: bytes) -> List[int]:
        if len(payload) % 2:
            payload += b"\x00"
        return [(payload[index] << 8) | payload[index + 1] for index in range(0, len(payload), 2)]

    def _build_value_group(self, registers: Sequence[int]) -> ValueGroup:
        values: Dict[str, float] = {}
        for index, field_name in enumerate(VALUE_GROUP_FIELDS):
            start = index * 2
            values[field_name] = self._registers_to_float(registers[start : start + 2])
        return ValueGroup(**values)

    def _read_line_data(self, line_no: int, start_register: int) -> LineData:
        registers = self.transport.read_input_registers(start_register, LINE_REGISTER_COUNT)
        value_group = self._build_value_group(registers)
        return LineData(line_no=line_no, pri_val=value_group, sec_val=value_group)

    def _read_ai_collect(self) -> Dict[str, float]:
        registers = self.transport.read_input_registers(AI_INPUT_START_REGISTER, AI_INPUT_COUNT * 2)
        values = {
            f"ai{index + 1}": self._registers_to_float(registers[index * 2 : index * 2 + 2])
            for index in range(AI_INPUT_COUNT)
        }
        for index in range(AI_INPUT_COUNT + 1, 13):
            values[f"ai{index}"] = 0.0
        return values

    def _read_di_state(self) -> Dict[str, int]:
        try:
            bits = self.transport.read_discrete_inputs(0x0000, 12)
            return {f"di{index + 1}": int(bits[index]) for index in range(min(12, len(bits)))}
        except Exception:
            register = self.transport.read_input_registers(DI_INPUT_REGISTER, 1)[0]
            return {f"di{index}": int(bool(register & (1 << (index - 1)))) for index in range(1, 13)}

    def _read_do_state(self) -> Dict[str, int]:
        try:
            bits = self.transport.read_coils(0x0000, 12)
            if bits:
                for index, bit in enumerate(bits[:12], start=1):
                    self._do_state[index] = int(bool(bit))
        except Exception:
            pass
        return {f"do{index}": self._do_state.get(index, 0) for index in range(1, 13)}

    def _build_status(self) -> DeviceStatus:
        transport = settings.modbus_transport.strip().lower()
        if transport == "tcp":
            return DeviceStatus(self_check="正常", net1="正常", net2="正常", uart1="断开", uart2="断开")
        return DeviceStatus(self_check="正常", net1="断开", net2="断开", uart1="正常", uart2="正常")

    def _category_map(self, category: str) -> str:
        normalized = category.strip().upper()
        return {
            "UA": "UA",
            "UB": "UB",
            "UC": "UC",
            "UAB": "UAB",
            "UBC": "UBC",
            "UCA": "UCA",
            "P": "P",
            "Q": "Q",
            "F": "F",
            "FRQ": "F",
            "频率": "F",
            "功率": "P",
        }.get(normalized, normalized)

    def _signal_type_code(self, signal_type: str) -> int:
        normalized = _normalize_text(signal_type)
        return 0 if normalized in {"4-20MA", "4~20MA", "4-20MA"} else 1

    def _measure_type_code(self, measure_type: str) -> int:
        normalized = self._category_map(measure_type)
        return {
            "UA": 0,
            "UB": 1,
            "UC": 2,
            "UAB": 3,
            "UBC": 4,
            "UCA": 5,
            "P": 6,
            "Q": 7,
            "F": 8,
        }.get(normalized, 0)

    def _output_node_code(self, output_node: str) -> int:
        digits = "".join(character for character in output_node if character.isdigit())
        return _clamp_byte(digits or 0)

    def _parity_code(self, parity: str) -> int:
        normalized = _normalize_text(parity)
        return {"NONE": 0, "N": 0, "ODD": 1, "O": 1, "EVEN": 2, "E": 2}.get(normalized, 0)

    def _data_bits_code(self, data_bits: int) -> int:
        return 0 if int(data_bits) == 8 else _clamp_byte(data_bits)

    def _stop_bits_code(self, stop_bits: int) -> int:
        return 1 if int(stop_bits) >= 2 else 0

    def _protocol_code(self, protocol: str) -> int:
        return 1 if "MODBUS" in _normalize_text(protocol) else 0

    def _line_rule_payload(self, limit: Any, delay: Any, output_node: str, enabled: bool) -> bytes:
        return bytes(
            [
                _clamp_byte(limit),
                _clamp_byte(delay),
                self._output_node_code(output_node),
                1 if enabled else 0,
            ]
        )

    def read_realtime_data(self) -> RealtimeData:
        line_list = [
            self._read_line_data(line_no=index + 1, start_register=start_register)
            for index, start_register in enumerate(LINE_START_REGISTERS)
        ]
        switch = self._read_di_state()
        switch.update(self._read_do_state())
        return RealtimeData(line_list=line_list, switch=switch, ai_collect=self._read_ai_collect())

    def read_device_status(self) -> DeviceStatus:
        self.transport.read_input_registers(EVENT_FLAG_REGISTER, 1)
        return self._build_status()

    def read_alarm_events(self) -> List[AlarmEvent]:
        event_flag = self.transport.read_input_registers(EVENT_FLAG_REGISTER, 1)[0]
        if event_flag == 0:
            return []

        registers = self.transport.read_holding_registers(EVENT_HOLDING_START_REGISTER, 4)
        signature = tuple(int(value) for value in registers)
        if signature == self._last_event_signature:
            return []
        self._last_event_signature = signature

        timestamp = int(registers[0]) | (int(registers[1]) << 16)
        milliseconds = int(registers[2]) & 0xFFFF
        line_and_type = int(registers[3]) & 0xFFFF
        line_no = (line_and_type >> 8) & 0xFF
        event_type = line_and_type & 0xFF

        event_name = {
            1: "PT断线",
            2: "CT断线",
        }.get(event_type, f"事件{event_type}")
        event_time = datetime.fromtimestamp(timestamp).replace(microsecond=min(milliseconds, 999) * 1000)
        line_text = "总路" if line_no == 0 else f"线路{line_no}"
        return [
            AlarmEvent(
                alarm_type="modbus_event",
                time=event_time,
                no="" if line_no == 0 else str(line_no),
                type=event_name,
                content=f"{line_text}{event_name}",
                level="中",
            )
        ]

    def send_device_config(self, payload: DeviceConfigIn) -> Dict[str, Any]:
        uart_map = {item.port.upper(): item for item in payload.uart}
        raw = bytearray()
        for port_name in ("COM1", "COM2", "COM3", "COM4"):
            item = uart_map.get(port_name)
            baud = item.baud if item is not None else 0
            parity = self._parity_code(item.parity) if item is not None else 0
            data_bits = self._data_bits_code(item.data_bits) if item is not None else 0
            stop_bits = self._stop_bits_code(item.stop_bits) if item is not None else 0
            protocol = self._protocol_code(item.protocol) if item is not None else 0
            raw.extend(struct.pack("<I4B", int(baud), parity, data_bits, stop_bits, protocol))

        registers = self._bytes_to_registers(bytes(raw))
        self.transport.write_registers(UART_CONFIG_START_REGISTER, registers)
        return {
            "send_status": "成功",
            "target": "device",
            "modbus_written": {"uart_registers": len(registers)},
            "ignored_fields": ["password", "hardware_version", "software_version", "net"],
        }

    def send_channel_config(self, payload: ChannelConfigIn) -> Dict[str, Any]:
        def build_channel_payload(items: List[Any]) -> List[int]:
            raw = bytearray()
            channel_map = {int(item.ch): item for item in items}
            for channel_no in range(1, 13):
                item = channel_map.get(channel_no)
                if item is None:
                    raw.extend(b"\x00" * 5)
                    continue
                raw.extend(
                    bytes(
                        [
                            self._signal_type_code(item.singal_type),
                            _clamp_byte(item.line_no),
                            self._measure_type_code(item.type),
                            _clamp_byte(item.limit_high),
                            _clamp_byte(item.limit_low),
                        ]
                    )
                )
            return self._bytes_to_registers(bytes(raw))

        ai_registers = build_channel_payload(payload.ai_channel)
        ao_registers = build_channel_payload(payload.ao_channel)
        self.transport.write_registers(AI_CHANNEL_START_REGISTER, ai_registers)
        self.transport.write_registers(AO_CHANNEL_START_REGISTER, ao_registers)
        return {
            "send_status": "成功",
            "target": "channel",
            "modbus_written": {
                "ai_registers": len(ai_registers),
                "ao_registers": len(ao_registers),
            },
        }

    def send_line_alarm_setting(self, payload: LineAlarmSettingIn) -> Dict[str, Any]:
        over_limit_map = {rule.category: rule for rule in payload.over_limit_alarm}
        fault_map = {rule.category: rule for rule in payload.fault_alarm}

        ordered_rules = [
            over_limit_map.get("电压"),
            over_limit_map.get("电流"),
            over_limit_map.get("差流"),
            over_limit_map.get("功率"),
            over_limit_map.get("频率"),
            fault_map.get("PT断线"),
            fault_map.get("CT断线"),
        ]

        raw = bytearray()
        for rule in ordered_rules:
            if rule is None:
                raw.extend(b"\x00" * 4)
                continue
            raw.extend(self._line_rule_payload(rule.limit, rule.delay, rule.output_node, rule.enabled))

        start_register = LINE_ALARM_START_REGISTER + max(0, payload.line_no - 1) * LINE_ALARM_REGISTERS_PER_LINE
        registers = self._bytes_to_registers(bytes(raw))
        self.transport.write_registers(start_register, registers)
        return {
            "send_status": "成功",
            "target": "line_alarm",
            "line_no": payload.line_no,
            "modbus_written": {"start_register": start_register, "registers": len(registers)},
        }

    def send_ai_alarm_setting(self, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
        raw = bytearray()
        item_map = {int(item.get("channel_no", 0)): item for item in payload}
        for channel_no in range(1, 13):
            item = item_map.get(channel_no)
            if item is None:
                raw.extend(b"\x00" * 5)
                continue
            raw.extend(
                bytes(
                    [
                        _clamp_byte(item.get("limit_high", 0)),
                        _clamp_byte(item.get("limit_low", 0)),
                        _clamp_byte(item.get("delay", 0)),
                        self._output_node_code(str(item.get("output_node", ""))),
                        1 if bool(item.get("enabled", False)) else 0,
                    ]
                )
            )

        registers = self._bytes_to_registers(bytes(raw))
        self.transport.write_registers(AI_ALARM_START_REGISTER, registers)
        return {
            "send_status": "成功",
            "target": "ai_alarm",
            "items": len(payload),
            "modbus_written": {"registers": len(registers)},
        }

    def send_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]:
        time_text = str(payload.time_sync).strip()
        if time_text and time_text.lower() not in {"auto", "manual"}:
            timestamp = int(datetime.fromisoformat(time_text).timestamp())
        else:
            timestamp = int(datetime.now().timestamp())

        registers = [timestamp & 0xFFFF, (timestamp >> 16) & 0xFFFF]
        self.transport.write_registers(TIME_SYNC_START_REGISTER, registers)
        return {
            "send_status": "成功",
            "target": "system",
            "brightness": payload.brightness,
            "screen_saver": payload.screen_saver,
            "modbus_written": {"time_registers": len(registers)},
            "ignored_fields": ["brightness", "screen_saver"],
        }

    def send_switch_control(self, payload: SwitchControlIn) -> Dict[str, Any]:
        coil_address = max(0, payload.ch - 1)
        coil_value = bool(payload.action)
        self.transport.write_coil(coil_address, coil_value)
        self._do_state[payload.ch] = int(coil_value)

        bitmask = 0
        for index, state in self._do_state.items():
            if state:
                bitmask |= 1 << (index - 1)
        self.transport.write_registers(DO_STATE_REGISTER, [bitmask & 0xFFFF])

        action_text = "合" if payload.action == 1 else "分"
        return {
            "control_status": f"执行成功: 开关{payload.ch}{action_text}",
            "modbus_written": {"coil_address": coil_address, "state_register": DO_STATE_REGISTER},
        }