modbus_device/device/manager.py - master - DimerTrue/ecg_gui

modbus_device/device/manager.py
210 строк8 KB
dimer
fix ui bugs; feat: add unit tests modbus
27 июл 2026, 18:01
27 июл 2026, 18:01b5f0b34
# manager.py
import serial
from typing import Optional, Tuple
from dataclasses import dataclass
from .real_device import RealModbusDevice
from .simulated_device import SimulatedModbusDevice
from ..protocol import TelemetryFrame, ProtocolField, WriteResult


@dataclass
class DeviceConfig:
    """Конфигурация одного устройства"""

    name: str
    id: int
    simulator: bool
    read_interval_ms: int


class DeviceModbusRTU:
    def __init__(
        self,
        ser: Optional[serial.Serial],
        slave_id: int = 1,
        use_simulator: bool = False,
    ):
        """
        Инициализация устройства Modbus RTU

        Args:
            use_simulator: True - использовать симулятор, False - реальное устройство
            ser Optional используется потом, что в случае simulator device должна быть возможность работать без COM-порта
        """
        self.ser = ser
        self.use_simulator = use_simulator
        self.slave_id = slave_id

        # Создаем нужное устройство
        if use_simulator:
            self.device = SimulatedModbusDevice(slave_id=slave_id)
            # self.device = SimulatedModbusDevice(ser, slave_id)
        else:
            self.device = RealModbusDevice(ser, slave_id)

    def connect(self) -> bool:
        """Подключение к устройству (реальному или симулированному)"""
        return self.device.connect()

    def ping(self, timeout: float) -> bool:
        """Проверка соединения с устройством"""
        return self.device.ping(timeout)

    def read(
        self, register_address: int = 1, values_count: int = len(ProtocolField)
    ) -> Optional[TelemetryFrame]:
        """Чтение данных с устройства"""
        return self.device.read(register_address, values_count)

    def write(self, register_address: int, value: int) -> bool:
        """Запись данных на устройство"""
        return self.device.write(register_address, value)

    def safe_write(self, register_address: int, value: int) -> WriteResult:
        """Запись данных на устройство c учетом косяка с CRC в прошивке"""
        if self.use_simulator:
            self.device.write(register_address, value)
            return WriteResult.OK
        else:
            return self.device.safe_write(register_address, value)

    def write_status_write(
        self,
        charge_dc_dc: bool = False,
        power_off: bool = False,
        power_on: bool = False,
    ) -> bool:
        """
        Управление устройством через регистр статуса записи
        Биты:
        - бит 0: заряд АКБ извне через DC-DC
        - бит 1: выключение
        - бит 2: команда на включение
        """
        status_value = 0

        # Невалидная команда. Нельзя одновременно включить и выключить устройство
        if power_off and power_on:
            power_on = False
            # return False

        # TODO: пока не обрабатываем
        # if charge_dc_dc:
        #     status_value |= 1 << 0  # Установить бит 0

        if power_off:
            status_value |= 1 << 1  # Установить бит 1

        if power_on:
            status_value |= 1 << 2  # Установить бит 2

        return self.write(ProtocolField.STATUS_WRITE, status_value)

    def write_baudrate(
        self,
        new_baudrate_number: int,
        new_slave_id: int,
    ) -> bool:
        """
        В силу особенностей прошивки, для успешной отправки скорости
        необходимо сразу же слать адрес. Сам по себе адрес можно записывать и отдельно.
        """
        res_baud = self.write(ProtocolField.BAUDRATE, new_baudrate_number)

        if not res_baud:
            return False

        res_id = self.write(ProtocolField.SLAVE_ID, new_slave_id)

        if not res_id:
            return False

        return True

    def safe_write_status_write(
        self,
        charge_dc_dc: bool = False,
        power_off: bool = False,
        power_on: bool = False,
    ) -> WriteResult:
        """
        Управление устройством через регистр статуса записи
        Реализация c учетом косяка с CRC в прошивке
        Биты:
        - бит 0: заряд АКБ извне через DC-DC
        - бит 1: выключение
        - бит 2: команда на включение
        """
        status_value = 0

        # Невалидная команда. Нельзя одновременно включить и выключить устройство
        if power_off and power_on:
            power_on = False
            # return False

        # TODO: пока не обрабатываем
        # if charge_dc_dc:
        #     status_value |= 1 << 0  # Установить бит 0

        if power_off:
            status_value |= 1 << 1  # Установить бит 1

        if power_on:
            status_value |= 1 << 2  # Установить бит 2

        if self.use_simulator:
            self.write(ProtocolField.STATUS_WRITE, status_value)
            return WriteResult.OK
        else:
            return self.safe_write(ProtocolField.STATUS_WRITE, status_value)

    def safe_write_baudrate(
        self,
        new_baudrate_number: int,
        new_slave_id: int,
    ) -> WriteResult:
        """
        В силу особенностей прошивки, для успешной отправки скорости
        необходимо сразу же слать адрес. Сам по себе адрес можно записывать и отдельно.
        Реализация c учетом косяка с CRC в прошивке
        """

        if self.use_simulator:
            self.write_baudrate(new_baudrate_number, new_slave_id)
            return WriteResult.OK
        else:
            result = self.safe_write(ProtocolField.BAUDRATE, new_baudrate_number)

            if result == WriteResult.ERROR:
                return WriteResult.ERROR

            result = self.safe_write(ProtocolField.SLAVE_ID, new_slave_id)

            return result

    def get_device_info(self) -> str:
        """Получение информации об устройстве"""
        device_type = "SIMULATOR" if self.use_simulator else "REAL DEVICE"
        return f"{device_type}: {self.device}"

    def __str__(self):
        lines = [
            f"--- Device info ---",
            f" Mode         : {'SIMULATOR' if self.use_simulator else 'REAL'}",
            f" Port         : {'None' if self.use_simulator else self.ser.port}",
            f" Slave ID     : {self.slave_id}",
            f" Baudrate     : {'None' if self.use_simulator else self.ser.baudrate}",
            f" Connected    : {self.is_connected()}",
        ]
        return "\n".join(lines)

    def set_timeout(self, timeout: float = 0.2):
        return self.device.set_timeout_device(timeout)

    def is_connected(self) -> bool:
        """Проверка подключения"""
        return self.device.is_connected()

    def disconnect(self):
        """Отключение от устройства"""
        self.device.disconnect()