modbus_device/protocol/models.py - master - DimerTrue/ecg_gui

modbus_device/protocol/models.py
275 строк9 KB
dimer
fix ui bugs; feat: add unit tests modbus
27 июл 2026, 18:01
27 июл 2026, 18:01b5f0b34
# models.py
from dataclasses import dataclass
from typing import Optional, Dict, List, Type, Union
from .enums import IntEnum, ProtocolField, StatusBits
from .configs import PROTOCOL_CONFIGS, STATUS_CONFIG, ERROR_CONFIG


# TODO: нужен ли логгер?
@dataclass
class ModbusParameter:
    address: int
    name: str
    description: str

    unit: str
    scale: float

    min_value: int
    max_value: int

    data_size: int
    command: int

    # bit_definitions хранит ссылку на Класс Enum (например, ErrorBits)
    bit_definitions: Optional[Type[IntEnum]] = None

    _raw_value: int = 0
    _phys_value: Union[int, float] = 0

    def set_value(self, value: int) -> None:
        if self.scale == 0:
            self._raw_value = 0
            self._phys_value = 0
            return

        # raw = max(self.min_value, min(value, self.max_value))
        # self._raw_value = raw
        # self._phys_value = raw * self.scale

        # TODO: Возможна ситуация, когда значения не впишутся в допустимые диапазоны,
        # но необходимо записать реальное значение
        # пока пусть останется
        self._raw_value = value
        raw_normalize = max(self.min_value, min(value, self.max_value))
        self._phys_value = raw_normalize * self.scale

    @property
    def raw_value(self) -> int:
        return self._raw_value

    @property
    def phys_value(self) -> Union[int, float]:
        val = self._raw_value * self.scale

        # Если масштаб 1 и число целое, превращаем в int
        if self.scale == 1.0 and val == int(val):
            return int(val)
        return val

    def get_bits(self) -> List[int]:
        """
        Возвращает список битов в инвертированном виде.
        """
        bits = []

        for i in range(self.data_size):
            bits.append((self._raw_value >> i) & 1)

        # Если нужно обратить
        # return bits[::-1]
        return bits

    def get_active_bit_details(self, config) -> List[Dict[str, str]]:
        """
        Возвращает список активных битов с их именами и описаниями.
        """
        if not self.bit_definitions:
            return []

        bits = self.get_bits()

        messages = []
        # Итерация по всем значениям enum
        for i, el in enumerate(self.bit_definitions):
            if bits[int(el)] & 1:
                try:
                    status = config[el]
                    messages.append(status)
                except:
                    continue

        return messages

    def __str__(self) -> str:
        """Краткое представление: [Addr] Name: Value Unit"""
        val = self._phys_value

        # Если значение целое (тип int или float .0), форматируем как целое
        if isinstance(val, int) or (isinstance(val, float) and val.is_integer()):
            # :>10d - выравнивание вправо, ширина 10, целое число
            val_str = f"{int(val):>10d}"
        else:
            # :>10.2f - выравнивание вправо, ширина 10, 2 знака после запятой
            val_str = f"{val:>10.2f}"

        return f"[{self.address:>3}] {self.name:<30}: {val_str} {self.unit}"

    def detailed_str(self, config=None) -> str:
        """Подробное представление регистра"""
        lines = [
            f"--- {self.name} (Addr: {self.address}) ---",
            f"  Description : {self.description}",
            f"  Raw Value   : {self.raw_value}",
            f"  Phys Value  : {self.phys_value} {self.unit}",
            f"  Range       : [{self.min_value} .. {self.max_value}]",
            f"  Scale       : {self.scale}",
            f"  Command     : {self.command}",
        ]

        # Добавляем инфо о битах, если они есть
        if self.bit_definitions and config is not None:
            bits_info = self.get_active_bit_details(config)
            if bits_info:
                lines.append("  Active Bits:")
                for b in bits_info:
                    lines.append(f"    - [{b['name']}: {b['description']}]")
            else:
                lines.append("  Active Bits : None")

        return "\n".join(lines)


# TODO: date, timestamp?
class TelemetryFrame:
    def __init__(self) -> None:
        self.parameters = {}

        for field in ProtocolField:
            if field in PROTOCOL_CONFIGS:
                try:
                    reg = self._create_register(field)
                    self.parameters[field] = reg
                    # print(f"OK: {reg.name} (Addr: {reg.address})")
                except Exception as e:
                    # print(f"Error creating {field}: {e}")
                    pass

    def _create_register(self, field: ProtocolField) -> ModbusParameter:
        """Фабричная функция для создания объекта регистра из конфига"""
        if field not in PROTOCOL_CONFIGS:
            raise ValueError(f"Field {field} not found in PROTOCOL_CONFIGS")

        config = PROTOCOL_CONFIGS[field]

        return ModbusParameter(
            address=field.value,
            name=config["name"],
            description=config["description"],
            unit=config["unit"],
            scale=config["scale"],
            min_value=config["min_value"],
            max_value=config["max_value"],
            data_size=config["data_size"],
            command=config["modbus_command"],
            bit_definitions=config["bit_definitions"],
        )

    def get(self, field: ProtocolField) -> Optional[ModbusParameter]:
        return self.parameters.get(field)

    def get_by_address(self, address: int) -> Optional[ModbusParameter]:
        for param in self.parameters.values():
            if param.address == address:
                return param
        return None

    def get_by_name(self, name: str) -> Optional[ModbusParameter]:
        for param in self.parameters.values():
            if param.name == name:
                return param
        return None

    def get_telemetry_dict(self):
        if not self.parameters:
            # print("Empty Telemetry Frame")
            return None

        param_dict = {
            str(param.name): str(param.raw_value) for param in self.parameters.values()
        }
        return param_dict

    @staticmethod
    def get_bits(value):
        bits = []

        for i in range(16):
            bits.append((value >> i) & 1)

        return bits

    @staticmethod
    def bits_to_int(bits: List) -> int:
        binary_string = "".join(str(bit) for bit in bits)
        return int(binary_string, 2)

    def _get_status(self) -> Optional[str]:
        return TelemetryFrame.get_status(self.get(ProtocolField.STATUS_READ)._raw_value)

    @staticmethod
    def get_status(status_read_value) -> Optional[str]:
        # Берем биты STATUS_BIT0-STATUS_BIT2
        status_bits = TelemetryFrame.get_bits(status_read_value)[3:6][::-1]

        # Преобразуем биты в число
        status_value = TelemetryFrame.bits_to_int(status_bits)

        try:
            status = StatusBits(status_value)
            status_description = STATUS_CONFIG[StatusBits(status)].get("name")

            return status_description
        except:
            # print(f"Invalid status ({status_value})!")
            return None

    def _get_errors(self) -> List[str]:
        errors = self.get(ProtocolField.ERRORS).raw_value

        return TelemetryFrame.get_errors(errors)

    @staticmethod
    def get_errors(errors: int) -> List[str]:
        """
        Получает список ошибок из числа
        """
        errors_names = []
        for bit_pos, info in ERROR_CONFIG.items():
            is_active = bool(errors & (1 << bit_pos))
            if is_active:
                errors_names.append(info["name"])

        return errors_names

    def __str__(self) -> str:
        """Печать всего фрейма."""
        if not self.parameters:
            return "Empty Telemetry Frame"

        # Заголовок
        header = "=" * 50 + "\nTELEMETRY FRAME\n" + "=" * 50

        # Список строк для каждого параметра
        param_lines = [str(param) for param in self.parameters.values()]

        return f"{header}\n" + "\n".join(param_lines)

    def _print_detailed(self) -> None:
        """Метод для подробного вывода всех регистров"""
        print("\n" + "-" * 50 + "\nDETAILED VIEW:\n" + "-" * 50)

        # Генерируем подробные строки и соединяем их
        detailed_output = "\n\n".join(
            param.detailed_str() for param in self.parameters.values()
        )
        print(detailed_output)

    def _print_reg_with_bits(self, field: "ProtocolField") -> None:
        """Вывод конкретного регистра с битами"""
        reg = self.parameters.get(field)
        if not reg:
            print(f"Field {field} not found")
            return

        print(reg.detailed_str())