697 lines
30 KiB
Python
697 lines
30 KiB
Python
"""Диалог «Тест ввода» — отображение цифровых и аналоговых входов платы.
|
||
|
||
Цифровые входы: команда 91 → INPUT<decimal> (26 бит: 24 входа + 2 хоппера)
|
||
Аналоговые входы: команда 92 → ADC<v1>,<v2>,<v3>,<v4>
|
||
|
||
Опрос запускается автоматически при открытии диалога и работает
|
||
в фоновом потоке, чтобы не блокировать UI.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from PySide6.QtCore import Qt, QThread, QTimer, Signal
|
||
from PySide6.QtWidgets import (
|
||
QComboBox,
|
||
QGridLayout,
|
||
QGroupBox,
|
||
QHBoxLayout,
|
||
QLabel,
|
||
QMessageBox,
|
||
QPushButton,
|
||
QSizePolicy,
|
||
QSpinBox,
|
||
QVBoxLayout,
|
||
)
|
||
|
||
from ...services.com_service import ComConnection
|
||
from ...services.input_mappings_service import (
|
||
InputMapping,
|
||
_MAX_Y,
|
||
get_input_mappings_dir,
|
||
load_all_input_mappings,
|
||
)
|
||
from ...ui.icons import ICON_COLOR_DARK, ICON_COLOR_MUTED, icon
|
||
from .styled_dialog import StyledDialog
|
||
|
||
_NUM_DIGITAL = 26 # 24 входа + 2 хоппера
|
||
|
||
|
||
class _InputPollThread(QThread):
|
||
"""Фоновый поток циклического опроса цифровых входов.
|
||
|
||
Drain приостановлен на время жизни этого потока, поэтому
|
||
можно работать напрямую с serial без lock.
|
||
"""
|
||
|
||
digital_ready = Signal(object) # int — битовая маска 26 бит
|
||
analog_ready = Signal(object) # list[str] — 4 значения ADC
|
||
info_line = Signal(str) # COINS / HOPPER / PRIZES и пр.
|
||
|
||
def __init__(self, connection: ComConnection) -> None:
|
||
super().__init__()
|
||
self._conn = connection
|
||
self._running = True
|
||
self._do_analog = False # однократный запрос ADC по кнопке
|
||
self._do_debounce: int | None = None # однократная установка антидребезга
|
||
self._raw_cmd: str | None = None # произвольная команда (хоппер, монета…)
|
||
|
||
def stop(self) -> None:
|
||
self._running = False
|
||
|
||
def request_analog(self) -> None:
|
||
"""Запросить одноразовый опрос аналоговых входов (из GUI-потока)."""
|
||
self._do_analog = True
|
||
|
||
def request_debounce(self, ms: int) -> None:
|
||
"""Запросить установку антидребезга (из GUI-потока)."""
|
||
self._do_debounce = ms
|
||
|
||
def send_raw(self, cmd: str) -> None:
|
||
"""Поставить произвольную команду на отправку (из GUI-потока)."""
|
||
self._raw_cmd = cmd
|
||
|
||
def _readline(self, ser, deadline) -> str | None:
|
||
"""Прочитать одну строку из serial до deadline.
|
||
|
||
Использует in_waiting для минимальной задержки: если байты
|
||
уже в буфере — забираем сразу, иначе короткий read(1) с
|
||
таймаутом 2 мс, чтобы не блокировать поток.
|
||
"""
|
||
import time
|
||
buf = bytearray()
|
||
while time.monotonic() < deadline:
|
||
# Сначала забираем всё, что уже в буфере драйвера
|
||
avail = ser.in_waiting
|
||
if avail:
|
||
chunk = ser.read(avail)
|
||
for byte in chunk:
|
||
if byte in (0x0D, 0x0A): # \r \n
|
||
if buf:
|
||
return buf.decode("ascii", errors="replace")
|
||
continue
|
||
buf.append(byte)
|
||
else:
|
||
# Буфер пуст — ждём немного
|
||
remaining = max(deadline - time.monotonic(), 0.001)
|
||
ser.timeout = min(remaining, 0.002) # 2 мс
|
||
b = ser.read(1)
|
||
if not b:
|
||
if buf:
|
||
return buf.decode("ascii", errors="replace")
|
||
continue
|
||
if b[0] in (0x0D, 0x0A):
|
||
if buf:
|
||
return buf.decode("ascii", errors="replace")
|
||
continue
|
||
buf.append(b[0])
|
||
if buf:
|
||
return buf.decode("ascii", errors="replace")
|
||
return None
|
||
|
||
def _check_info_line(self, line: str) -> bool:
|
||
"""Проверить, является ли строка информационной (COINS, HOPPER, PRIZES, SERVICE).
|
||
|
||
Если да — отправить сигнал и вернуть True.
|
||
"""
|
||
upper = line.upper()
|
||
for prefix in ("COINS", "HOPPER", "PRIZES", "SERVICE"):
|
||
if upper.startswith(prefix):
|
||
self.info_line.emit(line)
|
||
return True
|
||
return False
|
||
|
||
def run(self) -> None: # noqa: D401
|
||
import time
|
||
|
||
while self._running:
|
||
ser = self._conn.serial
|
||
if ser is None or not ser.is_open:
|
||
self.msleep(100)
|
||
continue
|
||
|
||
old_timeout = ser.timeout
|
||
try:
|
||
# ── Однократная установка антидребезга ──
|
||
debounce = self._do_debounce
|
||
if debounce is not None:
|
||
self._do_debounce = None
|
||
ser.write(f"91{debounce}\r\n".encode())
|
||
self.msleep(20)
|
||
continue
|
||
|
||
# ── Произвольная команда (хоппер, монета…) ──
|
||
raw = self._raw_cmd
|
||
if raw is not None:
|
||
self._raw_cmd = None
|
||
ser.write(f"{raw}\r\n".encode())
|
||
# Читаем ответ (может быть HOPPER OK/ERROR и т.п.)
|
||
deadline = time.monotonic() + 0.5
|
||
while self._running and time.monotonic() < deadline:
|
||
line = self._readline(ser, deadline)
|
||
if line is None:
|
||
break
|
||
self._check_info_line(line)
|
||
continue
|
||
|
||
# ── Однократный опрос ADC ──
|
||
if self._do_analog:
|
||
self._do_analog = False
|
||
ser.reset_input_buffer()
|
||
ser.write(b"92\r\n")
|
||
deadline = time.monotonic() + 0.2
|
||
while self._running and time.monotonic() < deadline:
|
||
line = self._readline(ser, deadline)
|
||
if line is None:
|
||
break
|
||
if line.upper().startswith("ADC"):
|
||
parts = line[3:].split(",")
|
||
self.analog_ready.emit(parts)
|
||
break
|
||
self._check_info_line(line)
|
||
continue
|
||
|
||
# ── Циклический опрос цифровых входов ──
|
||
ser.write(b"91\r\n")
|
||
|
||
deadline = time.monotonic() + 0.08 # 80 мс макс. ожидание
|
||
while self._running and time.monotonic() < deadline:
|
||
line = self._readline(ser, deadline)
|
||
if line is None:
|
||
break
|
||
if line.upper().startswith("INPUT"):
|
||
val = int(line[5:])
|
||
self.digital_ready.emit(val)
|
||
break
|
||
# Информационные строки перехватываем
|
||
self._check_info_line(line)
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
try:
|
||
ser.timeout = old_timeout
|
||
except Exception:
|
||
pass
|
||
# Без msleep — сразу следующий цикл
|
||
|
||
|
||
class InputTestDialog(StyledDialog):
|
||
def __init__(self, connection: ComConnection, parent=None) -> None:
|
||
super().__init__(parent)
|
||
self._conn = connection
|
||
self._closed = False
|
||
self.setWindowTitle("Тест ввода")
|
||
self.setMinimumSize(600, 560)
|
||
|
||
# Состояние программной автовыдачи хоппера (0 = неактивна, 1 или 2 = номер хоппера)
|
||
self._auto_dispense_hopper: int = 0
|
||
self._auto_dispense_timer: QTimer | None = None
|
||
|
||
self._mappings = load_all_input_mappings()
|
||
self._current_mapping: InputMapping | None = None
|
||
|
||
self._setup_ui()
|
||
self._apply_mapping()
|
||
|
||
# Приостанавливаем дренаж — опрос будет единственным читателем порта
|
||
self._conn.pause_drain()
|
||
|
||
# Запуск опроса сразу при открытии
|
||
self._poll_thread = _InputPollThread(connection)
|
||
self._poll_thread.digital_ready.connect(
|
||
self._on_digital, Qt.ConnectionType.QueuedConnection
|
||
)
|
||
self._poll_thread.analog_ready.connect(
|
||
self._on_analog, Qt.ConnectionType.QueuedConnection
|
||
)
|
||
self._poll_thread.info_line.connect(
|
||
self._on_info_line, Qt.ConnectionType.QueuedConnection
|
||
)
|
||
self._poll_thread.start()
|
||
|
||
# ── UI ────────────────────────────────────────────────────────────
|
||
|
||
def _setup_ui(self) -> None:
|
||
root = self.containerLayout()
|
||
root.setSpacing(12)
|
||
|
||
# ── Выбор маппинга ──
|
||
mapping_row = QHBoxLayout()
|
||
mapping_row.addWidget(QLabel("Раскладка:"))
|
||
self._mapping_combo = QComboBox()
|
||
self._mapping_combo.addItem("— Без раскладки —")
|
||
for m in self._mappings:
|
||
self._mapping_combo.addItem(m.project_name)
|
||
self._mapping_combo.currentIndexChanged.connect(self._on_mapping_changed)
|
||
mapping_row.addWidget(self._mapping_combo, 1)
|
||
|
||
reload_btn = QPushButton()
|
||
reload_btn.setIcon(icon("fa5s.sync-alt", color=ICON_COLOR_DARK))
|
||
reload_btn.setToolTip("Перечитать файлы раскладок")
|
||
reload_btn.setFixedWidth(36)
|
||
reload_btn.clicked.connect(self._reload_mappings)
|
||
mapping_row.addWidget(reload_btn)
|
||
|
||
open_dir_btn = QPushButton()
|
||
open_dir_btn.setIcon(icon("fa5s.folder-open", color=ICON_COLOR_DARK))
|
||
open_dir_btn.setToolTip("Открыть папку input_mappings")
|
||
open_dir_btn.setFixedWidth(36)
|
||
open_dir_btn.clicked.connect(self._open_mappings_dir)
|
||
mapping_row.addWidget(open_dir_btn)
|
||
|
||
root.addLayout(mapping_row)
|
||
|
||
# ── Холст кнопок (по координатам из маппинга) ──
|
||
self._canvas_group = QGroupBox("Кнопки")
|
||
self._canvas_grid = QGridLayout(self._canvas_group)
|
||
self._canvas_grid.setSpacing(4)
|
||
self._canvas_labels: dict[int, QLabel] = {} # input_index → QLabel
|
||
root.addWidget(self._canvas_group)
|
||
|
||
# ── Цифровые входы (классическая сетка) ──
|
||
self._digital_group = QGroupBox("Цифровые входы (24 + 2 хоппера)")
|
||
digital_grid = QGridLayout(self._digital_group)
|
||
digital_grid.setSpacing(4)
|
||
|
||
self._digital_labels: list[QLabel] = []
|
||
for i in range(_NUM_DIGITAL):
|
||
lbl = QLabel(f"IN{i}")
|
||
lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
lbl.setFixedSize(52, 28)
|
||
lbl.setObjectName("InputIndicator")
|
||
lbl.setProperty("active", False)
|
||
self._digital_labels.append(lbl)
|
||
row, col = divmod(i, 8)
|
||
digital_grid.addWidget(lbl, row, col)
|
||
|
||
root.addWidget(self._digital_group)
|
||
|
||
# ── Аналоговые входы ──
|
||
analog_group = QGroupBox("Аналоговые входы (4 канала)")
|
||
analog_inner = QVBoxLayout(analog_group)
|
||
analog_values_layout = QHBoxLayout()
|
||
self._analog_labels: list[QLabel] = []
|
||
for i in range(4):
|
||
lbl = QLabel(f"ADC{i}: —")
|
||
lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
lbl.setMinimumWidth(90)
|
||
self._analog_labels.append(lbl)
|
||
analog_values_layout.addWidget(lbl)
|
||
analog_inner.addLayout(analog_values_layout)
|
||
poll_adc_btn = QPushButton("Опросить ADC")
|
||
poll_adc_btn.setIcon(icon("fa5s.sync", color=ICON_COLOR_DARK))
|
||
poll_adc_btn.clicked.connect(self._poll_analog)
|
||
analog_inner.addWidget(poll_adc_btn)
|
||
root.addWidget(analog_group)
|
||
|
||
# ── Монетоприёмник ──
|
||
coin_group = QGroupBox("Монетоприёмник")
|
||
coin_layout = QHBoxLayout(coin_group)
|
||
self._coin_label = QLabel("Монеты: —")
|
||
self._coin_label.setMinimumWidth(120)
|
||
coin_layout.addWidget(self._coin_label)
|
||
coin_confirm_btn = QPushButton("Подтвердить (10)")
|
||
coin_confirm_btn.setIcon(icon("fa5s.check-circle", color=ICON_COLOR_DARK))
|
||
coin_confirm_btn.setToolTip("Отправить подтверждение COIN (команда 10)")
|
||
coin_confirm_btn.clicked.connect(lambda: self._send_raw("10"))
|
||
coin_layout.addWidget(coin_confirm_btn)
|
||
coin_layout.addStretch()
|
||
root.addWidget(coin_group)
|
||
|
||
# ── Хопперы ──
|
||
hopper_group = QGroupBox("Хопперы")
|
||
hopper_layout = QVBoxLayout(hopper_group)
|
||
|
||
# Хоппер 1
|
||
h1_row = QHBoxLayout()
|
||
h1_row.addWidget(QLabel("Хоппер 1:"))
|
||
self._hopper1_label = QLabel("—")
|
||
self._hopper1_label.setMinimumWidth(100)
|
||
h1_row.addWidget(self._hopper1_label)
|
||
h1_on = QPushButton("Вкл")
|
||
h1_on.setIcon(icon("fa5s.play", color=ICON_COLOR_DARK))
|
||
h1_on.setToolTip("Включить мотор (301)")
|
||
h1_on.clicked.connect(lambda: self._send_raw("301"))
|
||
h1_row.addWidget(h1_on)
|
||
h1_off = QPushButton("Выкл")
|
||
h1_off.setIcon(icon("fa5s.stop", color=ICON_COLOR_DARK))
|
||
h1_off.setToolTip("Выключить мотор (300)")
|
||
h1_off.clicked.connect(lambda: self._send_raw("300"))
|
||
h1_row.addWidget(h1_off)
|
||
self._h1_auto_btn = QPushButton("Автовыдача")
|
||
self._h1_auto_btn.setIcon(icon("fa5s.cog", color=ICON_COLOR_DARK))
|
||
self._h1_auto_btn.setToolTip("Программная автовыдача: вкл мотор → ждёт датчик (10 сек)")
|
||
self._h1_auto_btn.clicked.connect(lambda: self._start_auto_dispense(1))
|
||
h1_row.addWidget(self._h1_auto_btn)
|
||
h1_prize = QPushButton("Выдать приз")
|
||
h1_prize.setIcon(icon("fa5s.gift", color=ICON_COLOR_DARK))
|
||
h1_prize.setToolTip("Авторежим выдачи платы (309)")
|
||
h1_prize.clicked.connect(lambda: self._send_raw("309"))
|
||
h1_row.addWidget(h1_prize)
|
||
h1_row.addStretch()
|
||
hopper_layout.addLayout(h1_row)
|
||
|
||
# Хоппер 2
|
||
h2_row = QHBoxLayout()
|
||
h2_row.addWidget(QLabel("Хоппер 2:"))
|
||
self._hopper2_label = QLabel("—")
|
||
self._hopper2_label.setMinimumWidth(100)
|
||
h2_row.addWidget(self._hopper2_label)
|
||
h2_on = QPushButton("Вкл")
|
||
h2_on.setIcon(icon("fa5s.play", color=ICON_COLOR_DARK))
|
||
h2_on.setToolTip("Включить мотор (401)")
|
||
h2_on.clicked.connect(lambda: self._send_raw("401"))
|
||
h2_row.addWidget(h2_on)
|
||
h2_off = QPushButton("Выкл")
|
||
h2_off.setIcon(icon("fa5s.stop", color=ICON_COLOR_DARK))
|
||
h2_off.setToolTip("Выключить мотор (400)")
|
||
h2_off.clicked.connect(lambda: self._send_raw("400"))
|
||
h2_row.addWidget(h2_off)
|
||
self._h2_auto_btn = QPushButton("Автовыдача")
|
||
self._h2_auto_btn.setIcon(icon("fa5s.cog", color=ICON_COLOR_DARK))
|
||
self._h2_auto_btn.setToolTip("Программная автовыдача: вкл мотор → ждёт датчик (10 сек)")
|
||
self._h2_auto_btn.clicked.connect(lambda: self._start_auto_dispense(2))
|
||
h2_row.addWidget(self._h2_auto_btn)
|
||
h2_prize = QPushButton("Выдать приз")
|
||
h2_prize.setIcon(icon("fa5s.gift", color=ICON_COLOR_DARK))
|
||
h2_prize.setToolTip("Авторежим выдачи платы (409)")
|
||
h2_prize.clicked.connect(lambda: self._send_raw("409"))
|
||
h2_row.addWidget(h2_prize)
|
||
h2_row.addStretch()
|
||
hopper_layout.addLayout(h2_row)
|
||
|
||
root.addWidget(hopper_group)
|
||
|
||
# ── Лог событий ──
|
||
self._event_log = QLabel("")
|
||
self._event_log.setWordWrap(True)
|
||
self._event_log.setStyleSheet("color: #888; font-size: 11px;")
|
||
root.addWidget(self._event_log)
|
||
|
||
# ── Антидребезг ──
|
||
debounce_layout = QHBoxLayout()
|
||
debounce_layout.addWidget(QLabel("Антидребезг (мс):"))
|
||
self._debounce_spin = QSpinBox()
|
||
self._debounce_spin.setRange(10, 100)
|
||
self._debounce_spin.setValue(50)
|
||
debounce_layout.addWidget(self._debounce_spin)
|
||
set_debounce_btn = QPushButton("Применить")
|
||
set_debounce_btn.setIcon(icon("fa5s.check", color=ICON_COLOR_DARK))
|
||
set_debounce_btn.clicked.connect(self._set_debounce)
|
||
debounce_layout.addWidget(set_debounce_btn)
|
||
debounce_layout.addStretch()
|
||
root.addLayout(debounce_layout)
|
||
|
||
# ── Закрыть ──
|
||
btn_layout = QHBoxLayout()
|
||
btn_layout.addStretch()
|
||
close_btn = QPushButton("Закрыть")
|
||
close_btn.setIcon(icon("fa5s.times", color=ICON_COLOR_MUTED))
|
||
close_btn.clicked.connect(self.close)
|
||
btn_layout.addWidget(close_btn)
|
||
root.addLayout(btn_layout)
|
||
|
||
# ── Маппинги ──────────────────────────────────────────────────────
|
||
|
||
def _on_mapping_changed(self, index: int) -> None:
|
||
if index <= 0:
|
||
self._current_mapping = None
|
||
else:
|
||
self._current_mapping = self._mappings[index - 1]
|
||
self._apply_mapping()
|
||
|
||
def _apply_mapping(self) -> None:
|
||
"""Перестроить холст кнопок по текущему маппингу."""
|
||
# Очистить старые виджеты
|
||
for lbl in self._canvas_labels.values():
|
||
self._canvas_grid.removeWidget(lbl)
|
||
lbl.deleteLater()
|
||
self._canvas_labels.clear()
|
||
|
||
# Сбросить stretch у всех колонок/строк
|
||
for c in range(self._canvas_grid.columnCount()):
|
||
self._canvas_grid.setColumnStretch(c, 0)
|
||
for r in range(self._canvas_grid.rowCount()):
|
||
self._canvas_grid.setRowStretch(r, 0)
|
||
|
||
mapping = self._current_mapping
|
||
if mapping is None or not mapping.buttons:
|
||
self._canvas_group.hide()
|
||
self._digital_group.show()
|
||
return
|
||
|
||
self._canvas_group.show()
|
||
self._canvas_group.setTitle(f"Кнопки — {mapping.project_name}")
|
||
self._digital_group.hide()
|
||
|
||
used_cols: set[int] = set()
|
||
used_rows: set[int] = set()
|
||
|
||
for btn in mapping.buttons:
|
||
lbl = QLabel(btn.name)
|
||
lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
lbl.setObjectName("InputIndicator")
|
||
lbl.setProperty("active", False)
|
||
lbl.setSizePolicy(
|
||
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred
|
||
)
|
||
lbl.setMinimumHeight(28)
|
||
# Y инвертирован: +Y = вверх → меньший row в grid
|
||
grid_row = _MAX_Y - btn.y
|
||
self._canvas_grid.addWidget(lbl, grid_row, btn.x)
|
||
self._canvas_labels[btn.input_index] = lbl
|
||
used_cols.add(btn.x)
|
||
used_rows.add(grid_row)
|
||
|
||
# Равномерно растянуть используемые колонки
|
||
for c in used_cols:
|
||
self._canvas_grid.setColumnStretch(c, 1)
|
||
for r in used_rows:
|
||
self._canvas_grid.setRowStretch(r, 1)
|
||
|
||
# ── Обновление индикаторов ────────────────────────────────────────
|
||
|
||
def _on_digital(self, val: int) -> None:
|
||
if self._closed:
|
||
return
|
||
|
||
# Проверяем автовыдачу хоппера
|
||
self._check_auto_dispense(val)
|
||
|
||
# Обновляем классическую сетку
|
||
for i, lbl in enumerate(self._digital_labels):
|
||
active = bool(val & (1 << i))
|
||
pressed = not active # pull-up: 1 = не нажата, 0 = нажата
|
||
if lbl.property("active") != pressed:
|
||
lbl.setProperty("active", pressed)
|
||
lbl.style().unpolish(lbl)
|
||
lbl.style().polish(lbl)
|
||
lbl.update()
|
||
|
||
# Обновляем холст маппинга
|
||
for input_idx, lbl in self._canvas_labels.items():
|
||
if input_idx < _NUM_DIGITAL:
|
||
active = bool(val & (1 << input_idx))
|
||
pressed = not active
|
||
if lbl.property("active") != pressed:
|
||
lbl.setProperty("active", pressed)
|
||
lbl.style().unpolish(lbl)
|
||
lbl.style().polish(lbl)
|
||
lbl.update()
|
||
|
||
def _on_analog(self, values: list[str]) -> None:
|
||
if self._closed:
|
||
return
|
||
for i, v in enumerate(values[:4]):
|
||
self._analog_labels[i].setText(f"ADC{i}: {v}")
|
||
|
||
# ── Прочее ────────────────────────────────────────────────────────
|
||
|
||
def _send_raw(self, cmd: str) -> None:
|
||
"""Отправить произвольную команду через поток опроса."""
|
||
if not self._conn.is_connected or self._closed:
|
||
return
|
||
self._poll_thread.send_raw(cmd)
|
||
|
||
def _on_info_line(self, line: str) -> None:
|
||
"""Обработать информационную строку от платы (COINS, HOPPER, …)."""
|
||
if self._closed:
|
||
return
|
||
upper = line.upper()
|
||
if upper.startswith("COINS"):
|
||
self._coin_label.setText(f"Монеты: {line}")
|
||
elif upper.startswith("HOPPER1"):
|
||
self._hopper1_label.setText(line)
|
||
elif upper.startswith("HOPPER2"):
|
||
self._hopper2_label.setText(line)
|
||
elif upper.startswith("HOPPER"):
|
||
# Общий формат без номера
|
||
self._hopper1_label.setText(line)
|
||
elif upper.startswith("PRIZES"):
|
||
self._coin_label.setText(f"Монеты: {line}")
|
||
# Обновляем лог последнего события
|
||
self._event_log.setText(f"← {line}")
|
||
|
||
def _poll_analog(self) -> None:
|
||
"""Запросить однократный опрос аналоговых входов через поток."""
|
||
if not self._conn.is_connected or self._closed:
|
||
return
|
||
self._poll_thread.request_analog()
|
||
|
||
def _set_debounce(self) -> None:
|
||
if not self._conn.is_connected or self._closed:
|
||
return
|
||
val = self._debounce_spin.value()
|
||
self._poll_thread.request_debounce(val)
|
||
|
||
# ── Программная автовыдача хоппера ────────────────────────────────
|
||
|
||
def _start_auto_dispense(self, hopper: int) -> None:
|
||
"""Запустить программную автовыдачу для хоппера 1 или 2.
|
||
|
||
Алгоритм:
|
||
1. Включить мотор (301 / 401)
|
||
2. Следить за битом датчика хоппера в INPUT (бит 24 / 25)
|
||
3. Если бит сработал (0) за 10 сек → выключить мотор → OK
|
||
4. Если 10 сек прошло → выключить мотор → ошибка
|
||
"""
|
||
if not self._conn.is_connected or self._closed:
|
||
return
|
||
if self._auto_dispense_hopper != 0:
|
||
# Уже идёт автовыдача
|
||
return
|
||
|
||
self._auto_dispense_hopper = hopper
|
||
label = self._hopper1_label if hopper == 1 else self._hopper2_label
|
||
btn = self._h1_auto_btn if hopper == 1 else self._h2_auto_btn
|
||
label.setText("Автовыдача…")
|
||
btn.setEnabled(False)
|
||
|
||
# Включаем мотор
|
||
on_cmd = "301" if hopper == 1 else "401"
|
||
self._send_raw(on_cmd)
|
||
|
||
# Таймаут 10 секунд
|
||
self._auto_dispense_timer = QTimer(self)
|
||
self._auto_dispense_timer.setSingleShot(True)
|
||
self._auto_dispense_timer.timeout.connect(self._auto_dispense_timeout)
|
||
self._auto_dispense_timer.start(10_000)
|
||
|
||
def _check_auto_dispense(self, val: int) -> None:
|
||
"""Проверить, сработал ли датчик хоппера (вызывается из _on_digital)."""
|
||
hopper = self._auto_dispense_hopper
|
||
if hopper == 0:
|
||
return
|
||
|
||
# Бит 24 = хоппер 1, бит 25 = хоппер 2
|
||
bit_index = 23 + hopper # 24 или 25
|
||
sensor_active = not bool(val & (1 << bit_index)) # pull-up: 0 = сработал
|
||
|
||
if sensor_active:
|
||
self._finish_auto_dispense(success=True)
|
||
|
||
def _auto_dispense_timeout(self) -> None:
|
||
"""Таймаут 10 сек — датчик не сработал."""
|
||
if self._auto_dispense_hopper != 0:
|
||
self._finish_auto_dispense(success=False)
|
||
|
||
def _finish_auto_dispense(self, *, success: bool) -> None:
|
||
"""Завершить автовыдачу: выключить мотор и показать результат."""
|
||
hopper = self._auto_dispense_hopper
|
||
if hopper == 0:
|
||
return
|
||
|
||
# Останавливаем таймер
|
||
if self._auto_dispense_timer is not None:
|
||
self._auto_dispense_timer.stop()
|
||
self._auto_dispense_timer.deleteLater()
|
||
self._auto_dispense_timer = None
|
||
|
||
# Выключаем мотор
|
||
off_cmd = "300" if hopper == 1 else "400"
|
||
self._send_raw(off_cmd)
|
||
|
||
label = self._hopper1_label if hopper == 1 else self._hopper2_label
|
||
btn = self._h1_auto_btn if hopper == 1 else self._h2_auto_btn
|
||
self._auto_dispense_hopper = 0
|
||
btn.setEnabled(True)
|
||
|
||
if success:
|
||
label.setText("Автовыдача: OK ✓")
|
||
self._event_log.setText(f"← Хоппер {hopper}: приз выдан")
|
||
else:
|
||
label.setText("Автовыдача: ОШИБКА ✗")
|
||
self._event_log.setText(f"← Хоппер {hopper}: таймаут 10 сек")
|
||
QMessageBox.warning(
|
||
self,
|
||
f"Хоппер {hopper}",
|
||
f"Ошибка выдачи: датчик хоппера {hopper} не сработал за 10 секунд.",
|
||
)
|
||
|
||
def _reload_mappings(self) -> None:
|
||
"""Перечитать .mapping файлы из папки."""
|
||
current_name = self._mapping_combo.currentText()
|
||
self._mappings = load_all_input_mappings()
|
||
self._mapping_combo.blockSignals(True)
|
||
self._mapping_combo.clear()
|
||
self._mapping_combo.addItem("— Без раскладки —")
|
||
for m in self._mappings:
|
||
self._mapping_combo.addItem(m.project_name)
|
||
# Попробуем восстановить выбор
|
||
idx = self._mapping_combo.findText(current_name)
|
||
if idx >= 0:
|
||
self._mapping_combo.setCurrentIndex(idx)
|
||
self._mapping_combo.blockSignals(False)
|
||
self._on_mapping_changed(self._mapping_combo.currentIndex())
|
||
|
||
def _open_mappings_dir(self) -> None:
|
||
"""Открыть папку input_mappings/ в проводнике."""
|
||
import os
|
||
import subprocess
|
||
import sys as _sys
|
||
|
||
mappings_dir = get_input_mappings_dir()
|
||
if not mappings_dir.exists():
|
||
mappings_dir.mkdir(parents=True, exist_ok=True)
|
||
if _sys.platform == "win32":
|
||
os.startfile(str(mappings_dir)) # noqa: S606
|
||
elif _sys.platform == "darwin":
|
||
subprocess.Popen(["open", str(mappings_dir)]) # noqa: S603, S607
|
||
else:
|
||
subprocess.Popen(["xdg-open", str(mappings_dir)]) # noqa: S603, S607
|
||
|
||
def closeEvent(self, event) -> None:
|
||
self._closed = True
|
||
# Если автовыдача активна — выключаем мотор
|
||
if self._auto_dispense_hopper != 0:
|
||
off_cmd = "300" if self._auto_dispense_hopper == 1 else "400"
|
||
self._auto_dispense_hopper = 0
|
||
if self._auto_dispense_timer is not None:
|
||
self._auto_dispense_timer.stop()
|
||
self._auto_dispense_timer = None
|
||
# Отправляем напрямую, т.к. поток ещё жив
|
||
self._poll_thread.send_raw(off_cmd)
|
||
self._poll_thread.stop()
|
||
try:
|
||
self._poll_thread.digital_ready.disconnect()
|
||
except RuntimeError:
|
||
pass
|
||
try:
|
||
self._poll_thread.analog_ready.disconnect()
|
||
except RuntimeError:
|
||
pass
|
||
try:
|
||
self._poll_thread.info_line.disconnect()
|
||
except RuntimeError:
|
||
pass
|
||
self._poll_thread.wait(3000)
|
||
# Очищаем буфер serial от возможных остаточных ответов
|
||
try:
|
||
ser = self._conn.serial
|
||
if ser is not None and ser.is_open:
|
||
ser.reset_input_buffer()
|
||
except Exception:
|
||
pass
|
||
# Возобновляем дренаж
|
||
self._conn.resume_drain()
|
||
super().closeEvent(event)
|