"""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