templates/python/set_devices/qt_ports/emulator_serial.py

"""Async COM adapter for direct RTU or automatic emulator discovery.

Panels retain their register/TMS decoders. Only this boundary translates them
to SET v2 on the wire. No protocol cache survives closing/reopening a port.
"""
import time

from .qt_compat import QObject, QTimer, Signal, QSerialPort  # Общий порт не зависит от импортов SETGUI и поддерживает Qt6/Qt5.
from set_devices.emulator_protocol import EmulatorProtocol  # Автомат SET v2/RTU находится рядом с остальными переносимыми моделями templates.


def response_timeout(port, milliseconds):
    return port.response_timeout(milliseconds) if isinstance(port, EmulatorSerialPort) else milliseconds


def cancel_pending(port):
    if isinstance(port, EmulatorSerialPort):
        port.cancel_pending()


class EmulatorSerialPort(QObject):
    readyRead = Signal()
    errorOccurred = Signal(object)
    protocolChanged = Signal(str)
    wireEvent = Signal(str, bytes)
    PROBE_MS = 300

    def __init__(self, parent=None, port=None, *, protocol_mode='auto', rtu_suffix=b'', mcu="STM32F407"):  # Профиль MCU фиксирует проверяемую модель прошивки, сохраняя все старые параметры порта.
        super().__init__(parent)
        self._port = port if port is not None else QSerialPort(self)
        self._port.readyRead.connect(self._receive)
        self._port.errorOccurred.connect(self.errorOccurred.emit)
        self._probe = QTimer(self)
        self._probe.setSingleShot(True)
        self._probe.timeout.connect(self._probe_timeout)
        self._protocol_mode = protocol_mode
        self._rtu_suffix = bytes(rtu_suffix)
        self._mcu = mcu  # Аппаратная привязка сохраняется между закрытием и повторным открытием того же COM-адаптера.
        self.protocol = EmulatorProtocol(protocol_mode, mcu=mcu)  # Первичная конфигурация использует ожидаемую модель выбранного MCU.
        self._rx = bytearray()
        self._error = ''
        self._deadline = 0

    def __getattr__(self, name):
        port = self.__dict__.get('_port')
        if port is None:
            raise AttributeError(name)
        return getattr(port, name)

    def open(self, mode):
        self._probe.stop()
        self._deadline = 0
        self.protocol = EmulatorProtocol(self._protocol_mode, mcu=self._mcu)  # Повторное открытие начинает новый handshake с той же аппаратной привязкой.
        self._rx.clear()
        self._error = ''
        return self._port.open(mode)

    def configure_protocol(self, mode, *, rtu_suffix=b''):
        """Choose the wire format before opening a new connection."""
        if self.isOpen():
            raise ValueError('Сначала отключите COM-порт для смены протокола')
        protocol = EmulatorProtocol(mode, mcu=self._mcu)  # Смена wire-формата не должна сбрасывать выбранный MCU на F407.
        self.cancel_pending()
        self._protocol_mode = mode
        self._rtu_suffix = bytes(rtu_suffix)
        self.protocol = protocol
        self._rx.clear()

    def close(self):
        self.cancel_pending()
        self._port.close()
        self._rx.clear()

    def set_mcu(self, mcu):  # Профиль разрешается менять только до открытия физического соединения.
        if self.isOpen():  # Открытая транзакция должна завершиться с первоначальной идентичностью устройства.
            raise ValueError("Сначала отключите COM-порт для смены MCU")  # Предотвращаем подмену цели незавершённого обмена.
        protocol = EmulatorProtocol(self._protocol_mode, mcu=mcu)  # Неизвестный профиль отвергается до изменения сохранённых параметров.
        self._mcu = mcu  # Следующее открытие порта проверит DeviceInfo именно этой платы.
        self.protocol = protocol  # Старое состояние обнаружения не переносится на другой MCU.

    def cancel_pending(self):
        self._probe.stop()
        self._deadline = 0
        self.protocol.cancel()

    def response_timeout(self, milliseconds):
        return milliseconds + (2 * self.PROBE_MS if self.protocol.mode == 'auto' else 0)

    def errorString(self):
        return self._error or self._port.errorString()

    def clear(self, *args):
        self.cancel_pending()
        self._rx.clear()
        return self._port.clear(*args)

    def bytesAvailable(self):
        return len(self._rx)

    @property
    def supports_wave_blocks(self):
        return self.protocol.mode == 'v2'

    def readAll(self):
        result = bytes(self._rx)
        self._rx.clear()
        return result

    def write(self, data):
        self._error = ''
        self._rx.clear()
        try:
            packet = self.protocol.begin(bytes(data))
            return len(data) if self._send(packet) else -1
        except ValueError as error:
            self._fail(str(error))
            return -1

    def _send(self, packet):
        self._probe.stop()
        self._deadline = 0
        if self.protocol.phase in ('ping', 'info'):
            self._deadline = time.monotonic() + self.PROBE_MS / 1000
            self._probe.start(self.PROBE_MS)
        if self.protocol.mode == 'rtu':
            packet += self._rtu_suffix
        written = self._port.write(packet)
        if written > 0:
            self.wireEvent.emit('TX', packet[:written])
        if written != len(packet):
            self._fail('Не удалось передать запрос целиком')
            return False
        return True

    def _receive(self):
        raw = bytes(self._port.readAll())
        if not raw:
            return
        self.wireEvent.emit('RX', raw)
        try:
            old_mode = self.protocol.mode
            outgoing, incoming = self.protocol.feed(raw)
            if self.protocol.mode != old_mode:
                self.protocolChanged.emit('SET v2')
            if outgoing and not self._send(outgoing):
                return
            if incoming:
                self._rx.extend(incoming)
                self.readyRead.emit()
        except ValueError as error:
            self._fail(str(error))

    def _probe_timeout(self):
        if not self._deadline:
            return
        if time.monotonic() < self._deadline:
            self._probe.start(max(1, int((self._deadline-time.monotonic())*1000)+1))
            return
        try:
            packet = self.protocol.fallback()
            self.protocolChanged.emit('RTU')
            self._send(packet)
        except ValueError as error:
            self._fail(str(error))

    def _fail(self, message):
        self._error = message
        self.cancel_pending()
        self.errorOccurred.emit(QSerialPort.SerialPortError.ReadError)

    def waitForReadyRead(self, milliseconds):
        # Used only in WaveTransfer's worker. No GUI event loop is blocked.
        deadline = time.monotonic() + milliseconds / 1000
        while not self._rx and not self._error and time.monotonic() < deadline:
            if self._deadline and time.monotonic() >= self._deadline:
                self._probe_timeout()
            remaining = max(1, int((deadline-time.monotonic())*1000))
            self._port.waitForReadyRead(min(remaining, 20))
        if self._error:
            raise OSError(self._error)
        return bool(self._rx)