Mitja Horvat
7071cb7c34
- curfew_status: rewrite next-transition calculation to scan up to 7 days ahead; old branching showed wrong 'blocked until' for Friday nights (assumed next day = weekday schedule) and Sunday nights (off by 2 days). Verified against a minute-by-minute reference for a full week: 10080/10080. - server: ThreadingHTTPServer so slow /status (pings/ssh) no longer blocks the TURN OFF button and /budget requests. - devices: atomic state file writes (tmp + os.replace) and tolerant _load_state so a crash mid-write can't silently kill the whole system. - server: /budget takes _timer_lock (no lost updates vs the budget tick) and rejects unknown device ids instead of creating junk state entries. - status: devices carry an explicit id; web UI keys off it instead of parsing titles. - gabi_check: exact username match (no more 'gabriel' false positives). - gaja: password readable from GAJA_PASSWORD env var (hardcoded value remains as fallback). - tui: drop unused import.
121 lines
3.0 KiB
Python
121 lines
3.0 KiB
Python
"""Terminal TUI for device status monitoring."""
|
|
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
from devices import status_all
|
|
|
|
|
|
# ANSI escape codes
|
|
CLEAR = "\033[2J\033[H"
|
|
BOLD = "\033[1m"
|
|
RESET = "\033[0m"
|
|
GREEN = "\033[92m"
|
|
RED = "\033[91m"
|
|
YELLOW = "\033[93m"
|
|
CYAN = "\033[96m"
|
|
GRAY = "\033[90m"
|
|
|
|
|
|
def _get_terminal_size():
|
|
"""Get terminal width and height."""
|
|
try:
|
|
return os.get_terminal_size()
|
|
except OSError:
|
|
return (80, 24)
|
|
|
|
|
|
def _color(text: str, color_code: str) -> str:
|
|
"""Wrap text in ANSI color code."""
|
|
return f"{color_code}{text}{RESET}"
|
|
|
|
|
|
def _separator(width: int = 60) -> str:
|
|
"""Create a separator line."""
|
|
return "─" * width
|
|
|
|
|
|
def render(devices_status, curfew, refresh_interval: int = 10):
|
|
"""Render the TUI screen."""
|
|
width, height = _get_terminal_size()
|
|
lines = []
|
|
|
|
# Header
|
|
lines.append(_color("⚡ KIDS DEVICES — STATUS", BOLD))
|
|
lines.append("")
|
|
|
|
# Curfew status
|
|
if curfew["in_curfew"]:
|
|
curfew_icon = _color("🔴 BLOCKED", RED)
|
|
else:
|
|
curfew_icon = _color("🟢 ALLOWED", GREEN)
|
|
lines.append(f" Curfew: {curfew_icon} | {curfew['message']}")
|
|
lines.append("")
|
|
|
|
# Device list
|
|
for device in devices_status:
|
|
icon = device["icon"]
|
|
name = device["title"]
|
|
detail = device["detail"]
|
|
|
|
# Parse budget from detail
|
|
parts = detail.split(" · ")
|
|
status_text = parts[0] if parts else "Unknown"
|
|
budget_text = parts[1] if len(parts) > 1 else ""
|
|
|
|
# Color the status
|
|
if "Online" in status_text:
|
|
status_color = GREEN
|
|
elif "Offline" in status_text:
|
|
status_color = GRAY
|
|
else:
|
|
status_color = YELLOW
|
|
|
|
status_colored = _color(status_text, status_color)
|
|
|
|
# Color the budget
|
|
if "🔴" in device["icon"]:
|
|
budget_color = RED
|
|
elif "🟠" in device["icon"]:
|
|
budget_color = YELLOW
|
|
else:
|
|
budget_color = CYAN
|
|
|
|
budget_colored = _color(budget_text, budget_color) if budget_text else ""
|
|
|
|
lines.append(f" {icon} {name:<12} {status_colored} · {budget_colored}")
|
|
|
|
# Footer with timestamp
|
|
now = time.strftime("%H:%M:%S")
|
|
lines.append(f" {_color('Last update:', GRAY)} {now} | Refresh: {refresh_interval}s")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def run_tui(refresh_interval: int = 10):
|
|
"""Run the TUI in a loop."""
|
|
# Run in a separate thread to not block the server
|
|
while True:
|
|
try:
|
|
devices_status, curfew = status_all()
|
|
screen = render(devices_status, curfew, refresh_interval)
|
|
sys.stdout.write(CLEAR + screen)
|
|
sys.stdout.flush()
|
|
except Exception:
|
|
pass # Don't crash TUI on errors
|
|
|
|
time.sleep(refresh_interval)
|
|
|
|
|
|
def start_tui(refresh_interval: int = 10):
|
|
"""Start the TUI in a background thread."""
|
|
tui_thread = threading.Thread(
|
|
target=run_tui,
|
|
args=(refresh_interval,),
|
|
daemon=True,
|
|
)
|
|
tui_thread.start()
|
|
return tui_thread
|