toros
12e6a453de
- core/hardware: драйверы real (atxctl/pyserial) + mock, ATXPI_DRIVER - core/power, console, mock: операции питания и RS232-консоли - cli/atxpi: status/on/off/reset/cycle/console (stdlib, без deps) - install.sh: one-shot curl|bash, идемпотентный, хост-агностичный - deploy: systemd-юнит, шаблон env, update.sh - README, AGENTS.md: документация и проектные решения - AGENTS.md и доки очищены от внутренних хостов/адресов (публичный репо)
72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
"""Mock-драйвер for локальной разработки/тестов без железа.
|
|
|
|
Питание — фейковое состояние, сохраняемое в /tmp (чтобы CLI-вызовы в разных
|
|
процессах видели согласованное состояние). Консоль — буфер с «эхо».
|
|
"""
|
|
import io
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
from .hardware import PowerBackend, ConsoleBackend
|
|
|
|
_TMPDIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tmp")
|
|
os.makedirs(_TMPDIR, exist_ok=True)
|
|
_MOCK_STATE = os.environ.get("ATXPI_MOCK_STATE", os.path.join(_TMPDIR, "atxpi_mock_power.json"))
|
|
|
|
|
|
class MockPower(PowerBackend):
|
|
def __init__(self, initial: bool = True) -> None:
|
|
self._path = _MOCK_STATE
|
|
if os.path.exists(self._path):
|
|
try:
|
|
self._on = bool(json.load(open(self._path)).get("on", initial))
|
|
except Exception:
|
|
self._on = initial
|
|
else:
|
|
self._on = initial
|
|
self._save()
|
|
|
|
def _save(self) -> None:
|
|
with open(self._path, "w") as f:
|
|
json.dump({"on": self._on}, f)
|
|
|
|
def status(self) -> bool:
|
|
return self._on
|
|
|
|
def press_power(self) -> None:
|
|
self._on = not self._on
|
|
self._save()
|
|
|
|
def force_off(self) -> None:
|
|
self._on = False
|
|
self._save()
|
|
|
|
def press_reset(self) -> None:
|
|
# сброс не меняет состояние питания
|
|
pass
|
|
|
|
|
|
class MockConsole(ConsoleBackend):
|
|
"""Буфер-эхо плюс немного «вывода загрузки», чтобы read не был пустым."""
|
|
def __init__(self) -> None:
|
|
self._buf = io.BytesIO()
|
|
self._buf.write(b"ATXPI mock console ready\r\nPress Enter for login:\r\n")
|
|
|
|
def open(self) -> None:
|
|
pass
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def send(self, data: bytes) -> None:
|
|
# «эхо»: то, что отправили, возвращается при чтении.
|
|
self._buf.write(data)
|
|
|
|
def read(self, timeout: float = 0.5) -> bytes:
|
|
time.sleep(min(timeout, 0.1))
|
|
self._buf.seek(0)
|
|
data = self._buf.read(4096)
|
|
self._buf = io.BytesIO() # одно чтение = всё и обнулить
|
|
return data
|