94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
# hub/ticket/services/base_service.py
|
|
|
|
"""Базовый транспортный сервис Ticket без доменной логики."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
from PySide6.QtCore import QObject, Signal
|
|
|
|
from domain import TicketConnectionStatus, TicketHardwareStatus
|
|
from domain.ticket_constants import STATE_TODO
|
|
|
|
|
|
class BaseService(QObject):
|
|
"""Общий контракт transport-сервисов Ticket."""
|
|
|
|
action_triggered = Signal(object)
|
|
error_occurred = Signal(str)
|
|
com_status_changed = Signal(bool, str)
|
|
buttons_initialized = Signal(bool, int)
|
|
port_disconnected = Signal()
|
|
|
|
def __init__(self, parent: QObject | None = None):
|
|
super().__init__(parent)
|
|
self._button_states: dict[int, int] = {}
|
|
self._status = TicketHardwareStatus()
|
|
|
|
def start(self) -> None:
|
|
raise NotImplementedError
|
|
|
|
def stop(self) -> None:
|
|
raise NotImplementedError
|
|
|
|
def is_running(self) -> bool:
|
|
raise NotImplementedError
|
|
|
|
def get_status(self) -> TicketHardwareStatus:
|
|
"""Вернуть последний известный статус transport-сервиса."""
|
|
return self._status
|
|
|
|
def set_button_state(self, button_id: int, state_code: int) -> None:
|
|
"""Сохранить последнее известное состояние кнопки."""
|
|
self._button_states[int(button_id)] = int(state_code)
|
|
|
|
def remove_button_state(self, button_id: int) -> None:
|
|
"""Удалить состояние кнопки из локального transport-кэша."""
|
|
self._button_states.pop(int(button_id), None)
|
|
|
|
def reset_button_states(self) -> None:
|
|
"""Сбросить все известные состояния кнопок."""
|
|
self._button_states.clear()
|
|
|
|
def _get_button_state(self, button_id: int) -> int:
|
|
return self._button_states.get(int(button_id), STATE_TODO)
|
|
|
|
def _set_connection_status(
|
|
self,
|
|
connection_status: TicketConnectionStatus,
|
|
message: str,
|
|
) -> None:
|
|
if (
|
|
self._status.connection_status == connection_status
|
|
and self._status.message == message
|
|
):
|
|
return
|
|
self._status = replace(
|
|
self._status,
|
|
connection_status=connection_status,
|
|
message=message,
|
|
)
|
|
self.com_status_changed.emit(
|
|
connection_status == TicketConnectionStatus.CONNECTED,
|
|
message,
|
|
)
|
|
|
|
def _set_button_initialization(
|
|
self,
|
|
is_initialized: bool,
|
|
button_count: int,
|
|
) -> None:
|
|
if (
|
|
self._status.buttons_initialized == is_initialized
|
|
and self._status.button_count == button_count
|
|
):
|
|
return
|
|
self._status = replace(
|
|
self._status,
|
|
buttons_initialized=is_initialized,
|
|
button_count=button_count,
|
|
)
|
|
self.buttons_initialized.emit(is_initialized, button_count)
|