hookvision-rk/hookvision/modbus_client.py

import logging
from pymodbus.client import ModbusTcpClient
import threading
import time
from typing import Callable, List

logger = logging.getLogger(__name__)

class ModbusClient:
    def __init__(self, config, on_data_callback: Callable[[List[bool]], None]):
        self.config = config
        self.on_data_callback = on_data_callback
        self.client: ModbusTcpClient = None
        self.running = False
        self.thread = None

    def connect(self) -> bool:
        try:
            self.client = ModbusTcpClient(self.config.ip, port=self.config.port)
            self.client.connect()
            self.running = True
            self.thread = threading.Thread(target=self._poll_loop, daemon=True)
            self.thread.start()
            logger.info(f"Modbus连接成功 {self.config.ip}:{self.config.port}")
            return True
        except Exception as e:
            logger.error(f"Modbus连接失败: {e}")
            return False

    def disconnect(self):
        self.running = False
        if self.client:
            self.client.close()
            logger.info("Modbus已断开")

    def _poll_loop(self):
        while self.running:
            try:
                result = self.client.read_coils(0, 100)
                if result and not result.isError():
                    bools = [result.bits[i] for i in range(len(result.bits))]
                    self.on_data_callback(bools)
            except Exception as e:
                logger.error(f"Modbus读取错误: {e}")
            time.sleep(self.config.frequency / 1000.0)