templates/python/set_devices/qt_ports/mock_port.py

# Qt-порт виртуального устройства для разработки интерфейса без COM-подключения. Сигналы и
# ответы повторяют контракт обычного порта, чтобы UI мог переключать источник без изменения
# обработчиков.

"""@file mock_port.py
@brief Qt-таймер, имитирующий подключённый контроллер без COM-порта.

Реализует mock-устройство для тестирования и разработки без физического
оборудования. Использует детерминированные значения для надёжного тестирования.
"""

from __future__ import annotations

from .qt_compat import QElapsedTimer, QObject, QTimer, Signal

from set_devices.ds18b20 import MockSensorBus
from set_devices.eeprom import CatalogEntry, MemoryInfo
from set_devices.models import MockSignalSource, SignalCatalog
from set_devices.panel import MockPanel


class MockDevicePort(QObject):
    """@brief Асинхронный mock-порт с интерфейсом реального порта.

    Qt-таймер является платформенным планировщиком, а генерация значений
    делегируется переносимому ``MockSignalSource``.
    """

    connected_changed = Signal(bool, str)
    snapshot_received = Signal(object)
    sensors_received = Signal(object)
    memory_info_received = Signal(object)
    catalog_received = Signal(object)
    sensor_list_received = Signal(object)
    panel_received = Signal(object)
    tx_logged = Signal(str)
    rx_logged = Signal(str)
    error_occurred = Signal(str)

    def __init__(self, catalog: SignalCatalog, parent: QObject | None = None) -> None:
        super().__init__(parent)
        self._source = MockSignalSource(catalog)
        self._bus = MockSensorBus()
        # Каталог mock-прибора: два датчика уже разложены по позициям одной
        # сборки (assembly_serial=7), третий остаётся без записи.
        self._catalog = [
            CatalogEntry(rom=self._bus.roms()[0], position=1, assembly_serial=7),
            CatalogEntry(rom=self._bus.roms()[2], position=258, assembly_serial=7),
        ]
        self._panel = MockPanel(self._panel_rows)
        self._elapsed = QElapsedTimer()
        self._timer = QTimer(self)
        self._timer.setInterval(250)
        self._timer.timeout.connect(self._publish_snapshot)

    def open(self) -> None:
        """@brief Запускает генерацию четырёх снимков в секунду.

        Повторный вызов безопасен и не создаёт второй таймер.
        """
        if self._timer.isActive():
            return
        self._elapsed.start()
        self._timer.start()
        self.connected_changed.emit(True, "MOCK")
        self.rx_logged.emit("Mock-контроллер подключён")
        self._publish_snapshot()

    def close(self) -> None:
        """@brief Останавливает генератор и публикует отключение.

        Повторное закрытие не создаёт ложное событие изменения состояния.
        """
        was_active = self._timer.isActive()
        self._timer.stop()
        if was_active:
            self.connected_changed.emit(False, "MOCK")

    def scan_sensors(self) -> None:
        """@brief Публикует список ROM mock-шины 1-Wire.

        Поиск выполняется мгновенно: модель шины известна заранее.
        """
        self.tx_logged.emit("SENSOR_SCAN mock")
        self.sensor_list_received.emit(self._bus.roms())
        self.publish_sensors()

    def publish_sensors(self) -> None:
        """@brief Публикует очередной набор измерений DS18B20."""
        self.sensors_received.emit(self._bus.snapshot(self._timestamp_s()))

    def scan_memory(self) -> None:
        """@brief Публикует состояние mock-памяти и каталога позиций."""
        self.tx_logged.emit("EEPROM_SCAN mock")
        self.memory_info_received.emit(
            MemoryInfo(
                present=True,
                persistent=True,
                count=len(self._catalog),
                capacity=32,
                i2c_address=0x50,
                page_size=64,
                size=32768,
                base_address=1024,
                bus_errors=0,
                blob_size=332,
            )
        )

    def read_catalog(self) -> None:
        """@brief Публикует записи каталога mock-прибора."""
        self.tx_logged.emit("EEPROM_READ mock")
        self.catalog_received.emit(list(self._catalog))

    def set_user_bytes(self, rom: bytes, user_byte1: int, user_byte2: int) -> None:
        """@brief Применяет пользовательские байты к mock-датчику.

        @param rom Идентификатор датчика на mock-шине.
        @param user_byte1 Новое значение TH.
        @param user_byte2 Новое значение TL.
        """
        try:
            self._bus.set_user_bytes(rom, user_byte1, user_byte2)
        except (KeyError, ValueError) as error:
            self.error_occurred.emit(str(error))
            return
        self.tx_logged.emit(f"SET_USER_BYTES mock {user_byte1:#04x} {user_byte2:#04x}")
        self.publish_sensors()

    def set_resolution(self, rom: bytes, bits: int) -> None:
        """@brief Меняет разрешение mock-датчика.

        @param rom Идентификатор датчика на mock-шине.
        @param bits Разрешение 9..12 бит.
        """
        try:
            self._bus.set_resolution(rom, bits)
        except (KeyError, ValueError) as error:
            self.error_occurred.emit(str(error))
            return
        self.tx_logged.emit(f"SET_RESOLUTION mock {bits} бит")
        self.publish_sensors()

    def read_panel(self) -> None:
        """@brief Публикует снимок экрана mock-панели."""
        self.panel_received.emit(self._panel.screen())

    def press_panel_key(self, key: int, hold: bool = False) -> None:
        """@brief Применяет нажатие кнопки к mock-панели.

        @param key Код кнопки из PanelKey.
        @param hold Длинное нажатие.
        """
        self._panel.press(key, hold)
        self.tx_logged.emit(f"UI_KEY mock {int(key)}{' hold' if hold else ''}")
        self.read_panel()

    def _panel_rows(self) -> list[tuple[str, str]]:
        """Строки экрана датчиков mock-панели: номер и температура."""
        rows: list[tuple[str, str]] = []
        for index, reading in enumerate(self._bus.snapshot(self._timestamp_s())):
            temperature = reading.temperature
            rows.append((
                f"{index + 1} {reading.serial}",
                "---" if temperature is None else f"{temperature:+.1f}",
            ))
        return rows

    def set_output(self, key: str, value: bool) -> None:
        """@brief Применяет управляющую дискрету к mock-модели.

        @param key Ключ разрешённого выхода.
        @param value Новое состояние; ошибка неизвестного ключа идёт в signal.
        """
        try:
            self._source.set_output(key, value)
        except KeyError as error:
            self.error_occurred.emit(str(error))
            return
        self.tx_logged.emit(f"{key} = {int(value)}")
        self._publish_snapshot()

    def _timestamp_s(self) -> float:
        """Возвращает монотонное время mock-порта в секундах."""
        return self._elapsed.elapsed() / 1000.0

    def _publish_snapshot(self) -> None:
        """@brief Создаёт и публикует атомарный снимок вне UI-виджетов."""
        timestamp_s = self._timestamp_s()
        self.snapshot_received.emit(self._source.snapshot(timestamp_s))
        self.sensors_received.emit(self._bus.snapshot(timestamp_s))