#!/usr/bin/env python3 """POG-COM-PI-HAT status LED1 (EtherGold) — BCM2 active-low sink. Patterns (field contract): POST / soft-fault code: N x (100ms ON, 100ms OFF), then 500ms OFF (POST) Soft-fault codes in degraded: 1s OFF between failed phase codes Healthy: 250ms ON / 250ms OFF Degraded HB: 150ms ON / 150ms OFF for 2s, then soft-fault codes, repeat Env: EG_STATUS_LED_GPIO=2 (default) HAT_STATUS_LED=0 disable """ from __future__ import annotations import argparse import os import sys import threading import time from typing import Iterable, Optional, Set # Reuse gpiochip output from Pi HS helper when available. try: from pog422_gpio_hs import GpioOutputLine except Exception: # pragma: no cover - non-Linux / missing module GpioOutputLine = None # type: ignore[misc, assignment] DEFAULT_BCM = 2 POST_PULSE_MS = 100 POST_GAP_MS = 500 SOFT_CODE_GAP_MS = 1000 HEALTHY_MS = 250 DEGRADED_MS = 150 DEGRADED_HB_S = 2.0 def led_enabled() -> bool: v = (os.environ.get("HAT_STATUS_LED") or "1").strip().lower() if v in ("0", "false", "no", "off"): return False return True def resolve_bcm(default: int = DEFAULT_BCM) -> int: raw = (os.environ.get("EG_STATUS_LED_GPIO") or str(default)).strip() try: return int(raw) except ValueError: return default class HatStatusLed: """Background pattern driver for one active-low HAT status LED.""" def __init__(self, bcm: int, *, consumer: str = "eg-status-led") -> None: self.bcm = int(bcm) self._consumer = consumer self._line: Optional[object] = None self._lock = threading.Lock() self._stop = threading.Event() self._thread: Optional[threading.Thread] = None self._mode = "off" # off | post | healthy | degraded self._post_n = 1 self._soft_fails: Set[int] = set() self._disabled = not led_enabled() if self._disabled or GpioOutputLine is None or sys.platform != "linux": self._disabled = True return try: # Idle high = LED off (active-low sink). self._line = GpioOutputLine(self.bcm, consumer=consumer) self._line.set(True) except OSError as exc: print(f"hat_status_led: GPIO{self.bcm} unavailable ({exc}) — LED disabled", flush=True) self._disabled = True self._line = None def _set_lit(self, on: bool) -> None: line = self._line if line is None: return try: # Active-low: ON = drive low. line.set(not bool(on)) except OSError: pass def close(self) -> None: self._stop.set() t = self._thread if t is not None and t.is_alive(): t.join(timeout=1.5) self._thread = None with self._lock: if self._line is not None: try: self._line.set(True) # LED off except OSError: pass try: self._line.close() # type: ignore[attr-defined] except Exception: pass self._line = None def _ensure_thread(self) -> None: if self._disabled: return if self._thread is not None and self._thread.is_alive(): return self._stop.clear() self._thread = threading.Thread(target=self._run, name="hat-status-led", daemon=True) self._thread.start() def set_phase(self, n: int) -> None: """POST pattern for phase N (1-based).""" n = max(1, int(n)) with self._lock: self._mode = "post" self._post_n = n self._ensure_thread() def mark_soft_fail(self, n: int) -> None: with self._lock: self._soft_fails.add(max(1, int(n))) def clear_soft_fail(self, n: int) -> None: with self._lock: self._soft_fails.discard(int(n)) def enter_healthy(self) -> None: with self._lock: if self._soft_fails: self._mode = "degraded" else: self._mode = "healthy" self._ensure_thread() def enter_degraded(self, soft_fails: Optional[Iterable[int]] = None) -> None: with self._lock: if soft_fails is not None: self._soft_fails = {max(1, int(x)) for x in soft_fails} self._mode = "degraded" self._ensure_thread() def _sleep_ms(self, ms: float) -> bool: """Sleep; return False if stop requested.""" return not self._stop.wait(ms / 1000.0) def _pulse_count(self, n: int, *, gap_after_ms: int) -> bool: for _ in range(n): self._set_lit(True) if not self._sleep_ms(POST_PULSE_MS): return False self._set_lit(False) if not self._sleep_ms(POST_PULSE_MS): return False self._set_lit(False) return self._sleep_ms(gap_after_ms) def _heartbeat(self, on_off_ms: int, duration_s: float) -> bool: end = time.monotonic() + duration_s while time.monotonic() < end: self._set_lit(True) if not self._sleep_ms(on_off_ms): return False self._set_lit(False) if not self._sleep_ms(on_off_ms): return False return True def _run(self) -> None: while not self._stop.is_set(): with self._lock: mode = self._mode n = self._post_n fails = sorted(self._soft_fails) if mode == "off": self._set_lit(False) if not self._sleep_ms(200): break elif mode == "post": if not self._pulse_count(n, gap_after_ms=POST_GAP_MS): break elif mode == "healthy": self._set_lit(True) if not self._sleep_ms(HEALTHY_MS): break self._set_lit(False) if not self._sleep_ms(HEALTHY_MS): break elif mode == "degraded": if not self._heartbeat(DEGRADED_MS, DEGRADED_HB_S): break if fails: for i, fn in enumerate(fails): if not self._pulse_count( fn, gap_after_ms=SOFT_CODE_GAP_MS if i < len(fails) - 1 else POST_GAP_MS, ): return else: if not self._sleep_ms(POST_GAP_MS): break else: if not self._sleep_ms(200): break _GLOBAL: Optional[HatStatusLed] = None _GLOBAL_LOCK = threading.Lock() def get_eg_led(*, force_new: bool = False) -> HatStatusLed: global _GLOBAL with _GLOBAL_LOCK: if force_new and _GLOBAL is not None: _GLOBAL.close() _GLOBAL = None if _GLOBAL is None: _GLOBAL = HatStatusLed(resolve_bcm(), consumer="eg-status-led") return _GLOBAL def phase(n: int) -> None: get_eg_led().set_phase(n) def soft_fail(n: int) -> None: get_eg_led().mark_soft_fail(n) def healthy() -> None: get_eg_led().enter_healthy() def degraded(*ns: int) -> None: led = get_eg_led() if ns: led.enter_degraded(ns) else: led.enter_degraded() def shutdown() -> None: global _GLOBAL with _GLOBAL_LOCK: if _GLOBAL is not None: _GLOBAL.close() _GLOBAL = None def main(argv: Optional[list[str]] = None) -> int: ap = argparse.ArgumentParser(description="HAT status LED1 (EtherGold)") ap.add_argument("--bcm", type=int, default=None) ap.add_argument( "cmd", choices=("phase", "soft-fail", "healthy", "degraded", "off", "hold"), ) ap.add_argument("n", nargs="*", type=int, help="phase number(s)") ap.add_argument( "--loop", action="store_true", help="keep process alive (for shell POST before Python claims GPIO)", ) args = ap.parse_args(argv) if args.bcm is not None: os.environ["EG_STATUS_LED_GPIO"] = str(args.bcm) led = get_eg_led(force_new=True) if args.cmd == "phase": led.set_phase(args.n[0] if args.n else 1) elif args.cmd == "soft-fail": for x in args.n or [1]: led.mark_soft_fail(x) led.enter_degraded() elif args.cmd == "healthy": led.enter_healthy() elif args.cmd == "degraded": led.enter_degraded(args.n or None) elif args.cmd in ("off", "hold"): with led._lock: led._mode = "off" led._ensure_thread() if args.loop or args.cmd in ("phase", "healthy", "degraded", "soft-fail", "hold"): try: while True: time.sleep(3600) except KeyboardInterrupt: pass finally: shutdown() else: time.sleep(0.05) shutdown() return 0 if __name__ == "__main__": raise SystemExit(main())