70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
"""Последовательный клиент CAN-логгера поверх уже подключённого Dima-адаптера."""
|
|
import secrets
|
|
from .qt_compat import QObject, QTimer, Signal
|
|
from set_devices.ump_logger_can import Response, build_request
|
|
|
|
|
|
class UmpCanClient(QObject):
|
|
completed = Signal(str, object)
|
|
failed = Signal(str)
|
|
|
|
def __init__(self, send, parent=None):
|
|
super().__init__(parent)
|
|
self._send = send
|
|
self._token = secrets.randbelow(255) + 1
|
|
self._pending = None
|
|
self._operation = ''
|
|
self._timer = QTimer(self)
|
|
self._timer.setSingleShot(True)
|
|
self._timer.timeout.connect(self._timeout)
|
|
|
|
@property
|
|
def busy(self):
|
|
return self._pending is not None
|
|
|
|
def cancel(self):
|
|
# Поздний ответ после переключения транспорта не должен продолжить
|
|
# старую цепочку и отправить команду в новое подключение.
|
|
self._timer.stop()
|
|
self._pending = None
|
|
self._operation = ''
|
|
|
|
def request(self, operation, mode, function, address, value, timeout_ms=2000):
|
|
if self.busy:
|
|
self.failed.emit('CAN-логгер занят предыдущим запросом')
|
|
return
|
|
self._token = self._token % 255 + 1
|
|
try:
|
|
can_id, data = build_request(mode, self._token, function, address, value)
|
|
self._pending = Response(mode, self._token, function, address, value)
|
|
self._operation = operation
|
|
self._timer.start(timeout_ms)
|
|
self._send(can_id, data)
|
|
except (ValueError, RuntimeError, OSError) as error:
|
|
self.cancel()
|
|
self.failed.emit(str(error))
|
|
|
|
def receive(self, can_id, data):
|
|
if self._pending is None:
|
|
return
|
|
try:
|
|
words = self._pending.feed(can_id, data)
|
|
except ValueError as error:
|
|
self.cancel()
|
|
self.failed.emit(str(error))
|
|
return
|
|
if words is None:
|
|
return
|
|
operation = self._operation
|
|
self.cancel()
|
|
self.completed.emit(operation, words)
|
|
|
|
def _timeout(self):
|
|
writing = self._pending is not None and self._pending.function == 6
|
|
self.cancel()
|
|
text = ('Таймаут CAN-логгера: проверьте адаптер, скорость, номер платы '
|
|
'и прошивку с поддержкой CAN-логгера.')
|
|
if writing:
|
|
text += ' Подтверждение записи потеряно; команда могла выполниться.'
|
|
self.failed.emit(text)
|