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

modbus_device/device/simulated_device.py
253 строки11 KB
dimer
test build
29 июл 2026, 10:52
29 июл 2026, 10:52b04ab63
# simulated_device.py
from ..protocol import (
    TelemetryFrame,
    ProtocolField,
    PROTOCOL_CONFIGS,
)
from ..protocol.enums import Baudrate
from .interface import ModbusDeviceInterface
import random
import threading
import time
from typing import Optional


# Симулированное Modbus устройство
class SimulatedModbusDevice(ModbusDeviceInterface):
    def __init__(self, ser: str = "SIMULATED", slave_id: int = 1):
        self.ser = ser
        self.slave_id = slave_id
        self.telemetry = TelemetryFrame()
        self._connected = False
        self._running = False
        self._simulator_thread = None
        self.timeout = 0.2
        self.update_interval = 0.1

    def connect(self) -> bool:
        """Запуск симулятора"""
        if self._running:
            return True

        self._running = True
        self._connected = True
        self._initialize_data()

        # Запускаем поток обновления данных
        self._simulator_thread = threading.Thread(target=self._update_data, daemon=True)
        self._simulator_thread.start()

        return True

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

    def _initialize_data(self):
        """Инициализация начальных значений"""
        for field in ProtocolField:
            if field in self.telemetry.parameters:
                reg = self.telemetry.parameters[field]

                # Логика инициализации в зависимости от адреса
                if field == ProtocolField.CURRENT_ECG:
                    initial_value = 1300  # 1.3 A
                elif field == ProtocolField.VOLTAGE_ECG:
                    initial_value = 12000  # 12 V
                elif field == ProtocolField.CURRENT_BATTERY:
                    initial_value = 1200  # 1.2 A
                elif field == ProtocolField.VOLTAGE_BATTERY:
                    initial_value = 5200  # 5.2 V
                elif field == ProtocolField.CURRENT_LOAD:
                    initial_value = 300  # 0.3 A
                elif field == ProtocolField.VOLTAGE_LOAD:
                    initial_value = 5100  # 5.1 V
                elif field == ProtocolField.TEMP_PCB:
                    initial_value = 350  # 35.0 °C
                elif field == ProtocolField.TEMP_BATTERY:
                    initial_value = 280  # 28.0 °C
                elif field == ProtocolField.TEMP_TUBES:
                    initial_value = 450  # 45.0 °C
                elif field == ProtocolField.PRESSURE_FUEL:
                    initial_value = 1100  # 1.1 Bar
                elif field == ProtocolField.FLOW_FUEL:
                    initial_value = 500  # 5.00 ml/min
                elif field == ProtocolField.TEMP_ENCLOSURE:
                    initial_value = 250  # 25.0 °C
                elif field == ProtocolField.PRESSURE_ATM:
                    initial_value = 1013  # 1.013 kPa
                elif field == ProtocolField.HUMIDITY_AIR:
                    initial_value = 600  # 60.0 %
                elif field == ProtocolField.STATUS_READ:
                    initial_value = 54
                elif field == ProtocolField.STATUS_WRITE:
                    initial_value = 2
                elif field == ProtocolField.VOLTAGE_OUTPUT_MIN_W:
                    initial_value = 5000  # 5 V
                elif field == ProtocolField.VOLTAGE_OUTPUT_MAX_W:
                    initial_value = 24000  # 24 V
                elif field == ProtocolField.CURRENT_OUTPUT_MAX_W:
                    initial_value = 1000  # 1 A
                elif field == ProtocolField.CURRENT_OUTPUT_MAX_R:
                    initial_value = 1000  # 1 A
                elif field == ProtocolField.VOLTAGE_OUTPUT_MIN_R:
                    initial_value = 1000  # 21.00 V
                elif field == ProtocolField.VOLTAGE_OUTPUT_MAX_R:
                    initial_value = 24000  # 24 V
                elif field == ProtocolField.FIRMWARE_SERIAL:
                    initial_value = 171
                elif field == ProtocolField.ERRORS:
                    initial_value = 5
                elif field == ProtocolField.GENERATOR_SERIAL:
                    initial_value = 24456
                elif field == ProtocolField.SLAVE_ID:
                    initial_value = self.slave_id
                elif field == ProtocolField.BAUDRATE:
                    initial_value = 19200
                else:
                    initial_value = 0

                reg.set_value(initial_value)

    def _update_data(self):
        """Фоновое обновление симулированных данных - полностью независимая генерация"""
        while self._running:
            try:
                # Генерация случайных значений для каждого параметра
                for field, config in PROTOCOL_CONFIGS.items():
                    if "min_value" not in config or "max_value" not in config:
                        continue

                    if field == ProtocolField.FIRMWARE_SERIAL:
                        continue
                    elif field == ProtocolField.GENERATOR_SERIAL:
                        continue
                    elif field == ProtocolField.SLAVE_ID:
                        continue
                    elif field == ProtocolField.BAUDRATE:
                        continue
                    if field == ProtocolField.CURRENT_ECG:
                        delta = 500
                    elif field == ProtocolField.VOLTAGE_ECG:
                        delta = 500
                    elif field == ProtocolField.CURRENT_BATTERY:
                        delta = 500
                    elif field == ProtocolField.VOLTAGE_BATTERY:
                        delta = 500
                    elif field == ProtocolField.CURRENT_LOAD:
                        delta = 500
                    elif field == ProtocolField.VOLTAGE_LOAD:
                        delta = 500
                    elif field == ProtocolField.TEMP_PCB:
                        delta = 500
                    elif field == ProtocolField.TEMP_BATTERY:
                        delta = 500
                    elif field == ProtocolField.TEMP_TUBES:
                        delta = 500
                    elif field == ProtocolField.PRESSURE_FUEL:
                        delta = 500
                    elif field == ProtocolField.FLOW_FUEL:
                        delta = 500
                    elif field == ProtocolField.TEMP_ENCLOSURE:
                        delta = 500
                    elif field == ProtocolField.PRESSURE_ATM:
                        delta = 100
                    elif field == ProtocolField.HUMIDITY_AIR:
                        delta = 100
                    elif field == ProtocolField.STATUS_READ:
                        delta = 10
                    elif field == ProtocolField.STATUS_WRITE:
                        delta = 3
                    elif field == ProtocolField.ERRORS:
                        delta = 100
                    elif field == ProtocolField.VOLTAGE_OUTPUT_MIN_W:
                        continue
                    elif field == ProtocolField.VOLTAGE_OUTPUT_MAX_W:
                        continue
                    elif field == ProtocolField.CURRENT_OUTPUT_MAX_W:
                        continue
                    elif field == ProtocolField.CURRENT_OUTPUT_MAX_R:
                        continue
                    elif field == ProtocolField.VOLTAGE_OUTPUT_MIN_R:
                        continue
                    elif field == ProtocolField.VOLTAGE_OUTPUT_MAX_R:
                        continue

                    current_value = self.telemetry.parameters[field].raw_value

                    # Генерируем новое значение в пределах заданного диапазона из конфига
                    min_val = max(config["min_value"], current_value - delta)
                    max_val = min(config["max_value"], current_value + delta)

                    # Убеждаемся, что min_val <= max_val
                    if min_val > max_val:
                        min_val, max_val = max_val, min_val

                    random_value = random.randint(min_val, max_val)

                    # Устанавливаем значение
                    param_reg = self.telemetry.parameters[field]
                    param_reg.set_value(random_value)

                time.sleep(self.update_interval)

            except Exception as e:
                time.sleep(1)

    # def read(
    #     self, register_address: int = 1, values_count: int = 28
    # ) -> Optional[TelemetryFrame]:

    def read(
        self, register_address: int = 1, values_count: int = len(ProtocolField)
    ) -> Optional[TelemetryFrame]:
        """Чтение текущих симулированных данных"""
        if not self._connected:
            raise Exception("Simulator not connected")
        return self.telemetry

    def write(self, register_address: int, value: int) -> bool:
        """Запись данных на устройство"""
        try:
            # Поле в которое производится запись
            field_enum = ProtocolField(register_address)
            # Поле, в которое фактически сохраняется значение (симуляция поведения реального устройства)
            field_save_value = None

            if field_enum == ProtocolField.VOLTAGE_OUTPUT_MIN_W:
                field_save_value = ProtocolField.VOLTAGE_OUTPUT_MIN_R
            elif field_enum == ProtocolField.VOLTAGE_OUTPUT_MAX_W:
                field_save_value = ProtocolField.VOLTAGE_OUTPUT_MAX_R
            elif field_enum == ProtocolField.CURRENT_OUTPUT_MAX_W:
                field_save_value = ProtocolField.CURRENT_OUTPUT_MAX_R
            elif field_enum == ProtocolField.BAUDRATE:
                field_save_value = ProtocolField.BAUDRATE
                value = list(Baudrate)[value - 1]
            else:
                field_save_value = field_enum

            register = self.telemetry.parameters[field_save_value]
            register.set_value(value)

            return True

        except KeyError:
            return False
        except Exception as e:
            return False

    def disconnect(self):
        """Остановка симулятора"""
        self._running = False
        self._connected = False
        if self._simulator_thread:
            self._simulator_thread.join(timeout=2)

    def set_timeout_device(self, timeout: float) -> None:
        pass

    def is_connected(self) -> bool:
        return self._connected

    def __str__(self):
        return f"SimulatedModbusDevice(port=SIM, slave_id={self.slave_id})"