models/device_state.py - master - DimerTrue/ecg_gui

models/device_state.py
246 строк8 KB
dimer
add status connection bar
21 июл 2026, 15:22
21 июл 2026, 15:22245fab6
# device_state.py
"""
API-модуль для изоляции GUI от реализации device
"""

from core.app_logger import AppLogger

logger = AppLogger.get_logger(__name__)

from dataclasses import dataclass
from typing import Optional, Any, List, Tuple
from enum import Enum
from modbus_device.protocol import (
    TelemetryFrame,
    ProtocolField,
    ERROR_CONFIG,
    PROTOCOL_CONFIGS,
)
from models.common import MessageType

STATUS_CORRECT = "Ожидание"

# CONNECTION_STATE_ON = "Подключено                "


@dataclass
class ConnectionStatus:
    CONNECTED = "Подключено"
    DISCONNECTED = "Нет связи: переподключение"
    REBOOT_REQUIRED = "Необходимо включить устройство!"
    EXIT_REQUIRED = "Перезапустите приложение!"


@dataclass
class DeviceMessage:
    """Сообщение для GUI от modbus_worker"""

    slave_id: int
    message: str
    message_type: MessageType = MessageType.INFO


class WriteCommandType(Enum):
    """
    Регистры, доступные для записи
    """

    WRITE_VOLTAGE_MIN = 1
    WRITE_VOLTAGE_MAX = 2
    WRITE_CURRENT_MAX = 3
    WRITE_STATUS = 4
    # В силу особенностей прошивки после baudrate необходимо отправлять id
    WRITE_RS485 = 5


@dataclass
class WriteDataCommand:
    """
    Структура передаваемых данных с GUI на modbus_worker
    Возможна передача нескольких значений
    """

    slave_id: int
    command_type: WriteCommandType
    values: List[Any]

    @staticmethod
    def validate(raw_value: Any, register: ProtocolField) -> Tuple[bool, str, int]:
        """ "Валидация входящих числовых значений"""
        try:
            if isinstance(raw_value, str):
                # Заменяем запятую на точку, если пользователь ввел по-русски
                raw_value = raw_value.replace(",", ".")
                float_val = float(raw_value)
            else:
                float_val = float(raw_value)

            modbus_param = PROTOCOL_CONFIGS.get(register)

            if modbus_param is None:
                logger.warning(f"Не существует параметра {modbus_param}")

                return False, f"Не существует параметра {modbus_param}", 0

            scale = modbus_param.get("scale", 1)
            min_value = int(modbus_param.get("min_value", 0))
            max_value = int(modbus_param.get("max_value", 32767))

            value = int(round(float_val / scale))

            message = ""
            if value < min_value:
                if scale == 1:
                    message = f"Значение переменной должно быть больше {min_value}!"
                else:
                    message = f"Значение переменной должно быть больше {min_value * scale:.3f}!"
                return (False, message, value)
            elif value > max_value:
                if scale == 1:
                    message = f"Значение переменной должно быть меньше {max_value}!"
                else:
                    message = f"Значение переменной должно быть меньше {max_value * scale:.3f}!"
                return (False, message, value)

            return True, "Успех!", value
        except ValueError:
            logger.exception(f"Некорректное значение в поле {raw_value}!")

            return (
                False,
                "Некорректное значение в поле! Проверьте, что вводите только числовые значения.",
                0,
            )
        except Exception as e:
            logger.exception(f"Ошибка валидации!")
            return False, f"Ошибка валидации: {str(e)}!", 0


@dataclass
class WriteRegisterConfig:
    """
    Конфигурация параметра, который можно записать
    Допустима ситуация записи нескольких регистров, как в случае baudrate
    из-за особенностей прошивки
    """

    name: str
    address: List[ProtocolField]
    max_value: List[str]
    min_value: List[str]


@dataclass
class DeviceState:
    """Полное состояние устройства, полученное по Modbus"""

    # Параметры
    current_ecg: float = 0.0
    voltage_ecg: float = 0.0
    power_ecg: float = 0.0

    current_battery: float = 0.0
    voltage_battery: float = 0.0
    power_battery: float = 0.0

    current_load: float = 0.0
    voltage_load: float = 0.0
    power_load: float = 0.0

    current_output_max: float = 0.0
    voltage_output_min: float = 0.0
    voltage_output_max: float = 0.0

    # Температуры
    temp_pcb: float = 0.0
    temp_battery: float = 0.0
    temp_tubes: float = 0.0
    temp_enclosure: float = 0.0

    # Топливо и Воздух
    pressure_fuel: float = 0.0
    flow_fuel: float = 0.0
    pressure_atm: float = 0.0
    humidity_air: float = 0.0

    # Статусы и Ошибки
    status_read: int = 0
    status_write: int = 0
    error_bits: int = 0

    charge_battery: int = 0

    # Системная информация
    firmware_serial_raw: int = 0
    generator_serial: int = 0
    slave_id: int = 1
    baudrate: int = 9600
    # Для графиков (счетчик или время)
    timestamp: float = 0.0

    device_status: str = "Ожидание"
    device_power_state: str = "Откл"

    firmware_serial: int = 0
    gas: str = "???"

    def calculate_powers(self) -> None:
        """Пересчитывает мощности на основе токов и напряжений"""
        self.power_ecg = self.current_ecg * self.voltage_ecg if self.voltage_ecg else 0
        self.power_battery = (
            self.current_battery * self.voltage_battery if self.voltage_battery else 0
        )
        self.power_load = (
            (self.current_load * self.voltage_load) if self.voltage_load else 0
        )

    @staticmethod
    def get_errors_config() -> List[Tuple[str, str]]:
        """
        Возвращает список кортежей (Имя_Ошибки, Описание) для инициализации UI.
        Нужно для изоляции конкретики протокола от UI
        """
        errors_config = []

        for info in ERROR_CONFIG.values():
            errors_config.append((info["name"], info["description"]))

        return errors_config

    @staticmethod
    def decode_errors(errors: int) -> List[str]:
        """
        Расшифровывает сырое число ошибок в список имен активных ошибок.
        """
        return TelemetryFrame.get_errors(errors)

    @staticmethod
    def get_status(status_read_value) -> Optional[str]:
        return TelemetryFrame.get_status(status_read_value)

    @staticmethod
    def get_power_state(status_read_value) -> str:
        # Получаем состояние бита ТЭ
        state = int(TelemetryFrame.get_bits(status_read_value)[1])

        if state == 0:
            return "Откл"
        elif state == 1:
            return "Вкл"
        else:
            return "Неизв"

    @staticmethod
    def get_serial_and_gase(firmware_serial_raw) -> Tuple[str, str]:
        """У регистра 7 бит - газ, а от 0 до 6 биты - номер прошивки"""
        gas_bit = (firmware_serial_raw >> 7) & 1

        gas = "???"
        if gas_bit == 0:
            gas = "Метан"
        elif gas_bit:
            gas = "Пропан"

        firmware_serial = firmware_serial_raw & 0x7F

        return (gas, firmware_serial)