toros
93b4be9574
pyserial select-based read(4096) на RPi+PL2303 кидает 'readiness to read but no data' и теряет байты в потоке (эхо ввода превращалось в кракозябры). Чтение по in_waiting собирает доступные байты и отдаёт, когда поток утих — стабильно как minicom.
133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
"""Аппаратная абстракция atxpi.
|
||
|
||
Два драйвера:
|
||
- real: питание через существующий модуль atxctl (вызов main.py subprocess'ом),
|
||
консоль через pyserial на /dev/ttyUSB0 (pyserial импортируется лениво).
|
||
- mock: фейки в памяти — для локальной разработки/тестов без железа.
|
||
|
||
Выбор драйвера — переменная окружения ATXPI_DRIVER = "real" | "mock" (по умолчанию real).
|
||
"""
|
||
import os
|
||
import subprocess
|
||
import time
|
||
|
||
DRIVER = os.environ.get("ATXPI_DRIVER", "real")
|
||
|
||
# Путь к скрипту питания (существующий модуль atxctl) и интерпретатору venv.
|
||
ATXCTL_MAIN = os.environ.get("ATXCTL_MAIN", "atxctl/main.py")
|
||
ATXCTL_PYTHON = os.environ.get("ATXCTL_PYTHON", "python3")
|
||
|
||
# Параметры RS232-консоли.
|
||
SERIAL_PORT = os.environ.get("SERIAL_PORT", "/dev/ttyUSB0")
|
||
SERIAL_BAUD = int(os.environ.get("SERIAL_BAUD", "115200"))
|
||
|
||
|
||
class PowerError(RuntimeError):
|
||
"""Ошибка управления питанием."""
|
||
|
||
|
||
def _run_atxctl(*argv) -> str:
|
||
"""Вызвать CLI существующего atxctl/main.py и вернуть stdout."""
|
||
cmd = [ATXCTL_PYTHON, ATXCTL_MAIN, *argv]
|
||
try:
|
||
r = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
|
||
except FileNotFoundError as e:
|
||
raise PowerError(f"не удалось запустить atxctl ({cmd[0]}): {e}")
|
||
except subprocess.TimeoutExpired:
|
||
raise PowerError(f"atxctl превысил таймаут: {argv}")
|
||
if r.returncode != 0:
|
||
raise PowerError(f"atxctl [{argv}] завершился с ошибкой: {r.stderr.strip() or r.stdout.strip()}")
|
||
return r.stdout.strip()
|
||
|
||
|
||
class PowerBackend:
|
||
"""Интерфейс управления питанием."""
|
||
def status(self) -> bool:
|
||
raise NotImplementedError
|
||
def press_power(self) -> None:
|
||
raise NotImplementedError
|
||
def force_off(self) -> None:
|
||
raise NotImplementedError
|
||
def press_reset(self) -> None:
|
||
raise NotImplementedError
|
||
|
||
|
||
class AtxctlPower(PowerBackend):
|
||
"""Реальный драйвер питания — оборачивает atxctl/main.py."""
|
||
def status(self) -> bool:
|
||
out = _run_atxctl("--status")
|
||
return "on" in out.lower()
|
||
def press_power(self) -> None:
|
||
_run_atxctl("--press-power")
|
||
def force_off(self) -> None:
|
||
_run_atxctl("--power-off")
|
||
def press_reset(self) -> None:
|
||
_run_atxctl("--press-reset")
|
||
|
||
|
||
class ConsoleBackend:
|
||
"""Интерфейс RS232-консоли."""
|
||
def open(self) -> None:
|
||
raise NotImplementedError
|
||
def close(self) -> None:
|
||
raise NotImplementedError
|
||
def send(self, data: bytes) -> None:
|
||
raise NotImplementedError
|
||
def read(self, timeout: float = 0.5) -> bytes:
|
||
raise NotImplementedError
|
||
|
||
|
||
class SerialConsole(ConsoleBackend):
|
||
"""Реальный драйвер консоли — pyserial (ленивый импорт).
|
||
|
||
Чтение через `in_waiting` (без select-based read(4096)), т.к. pyserial-чтение
|
||
с select'ом на Raspberry Pi + PL2303 кидает 'readiness to read but no data'
|
||
и теряет байты в потоке. Так читает и minicom.
|
||
"""
|
||
def __init__(self, port: str = SERIAL_PORT, baud: int = SERIAL_BAUD) -> None:
|
||
self.port, self.baud = port, baud
|
||
self._ser = None
|
||
def open(self) -> None:
|
||
import serial # noqa: требуется только для реального железа
|
||
self._ser = serial.Serial(
|
||
self.port, self.baud, timeout=0.05, write_timeout=2, exclusive=True,
|
||
)
|
||
def close(self) -> None:
|
||
if self._ser:
|
||
self._ser.close()
|
||
self._ser = None
|
||
def send(self, data: bytes) -> None:
|
||
if not self._ser:
|
||
raise RuntimeError("консоль не открыта")
|
||
self._ser.write(data)
|
||
def read(self, timeout: float = 0.5) -> bytes:
|
||
"""Собрать доступные байты за `timeout`, отдать, когда поток утих."""
|
||
if not self._ser:
|
||
raise RuntimeError("консоль не открыта")
|
||
collected = b""
|
||
end = time.time() + timeout
|
||
while time.time() < end:
|
||
waiting = self._ser.in_waiting
|
||
if waiting > 0:
|
||
collected += self._ser.read(waiting)
|
||
time.sleep(0.015) # собрать межбайтовые хвосты
|
||
else:
|
||
time.sleep(0.015)
|
||
if collected:
|
||
break # поток данных утих — отдаём как есть
|
||
return collected
|
||
|
||
|
||
def make_power() -> PowerBackend:
|
||
if DRIVER == "mock":
|
||
from .mock import MockPower
|
||
return MockPower()
|
||
return AtxctlPower()
|
||
|
||
|
||
def make_console() -> ConsoleBackend:
|
||
if DRIVER == "mock":
|
||
from .mock import MockConsole
|
||
return MockConsole()
|
||
return SerialConsole()
|