diff --git a/devices.py b/devices.py new file mode 100644 index 0000000..709b191 --- /dev/null +++ b/devices.py @@ -0,0 +1,135 @@ +"""Device management: check status and power off kids' devices.""" + +from datetime import datetime + +import subprocess + + +def _run(cmd: list[str], *, timeout: int = 10) -> subprocess.CompletedProcess | None: + """Run a command, return CompletedProcess or None on error.""" + try: + return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + except (FileNotFoundError, subprocess.TimeoutExpired): + return None + + +# ── TV ──────────────────────────────────────────────────────────────── + +def tv_check() -> dict: + """Check if the TV is reachable and powered on. + + Returns an action dict describing the current state. + """ + result = _run(["ping", "-c", "1", "-W", "1", "192.168.1.12"]) + if not result or result.returncode != 0: + return {"icon": "❌", "title": "TV", "detail": "Unreachable (ping failed)"} + + _run(["adb", "connect", "192.168.1.12"]) + result = _run(["adb", "shell", "dumpsys", "power"]) + if result and "mWakefulness=Awake" in result.stdout: + return {"icon": "📺", "title": "TV", "detail": "Screen is on"} + + return {"icon": "📺", "title": "TV", "detail": "Already off"} + + +def tv_turnoff() -> dict | None: + """Send power key to the TV if it's on. Returns action dict or None.""" + state = tv_check() + if state["detail"] != "Screen is on": + return None + + _run(["adb", "shell", "input", "keyevent", "26"]) + return {"icon": "📺", "title": "TV", "detail": "Screen turned off (power key sent)"} + + +# ── Gabi ────────────────────────────────────────────────────────────── + +def gabi_check() -> bool: + """Return True if user 'gabi' has an active login session.""" + result = _run(["loginctl", "list-users"]) + return result is not None and "gabi" in result.stdout + + +def gabi_turnoff() -> dict | None: + """Log out Gabi if she's logged in. Returns action dict or None.""" + if not gabi_check(): + return {"icon": "💻", "title": "Gabi's PC", "detail": "Not logged in — no action needed"} + + result = _run(["doas", "loginctl", "terminate-user", "gabi"]) + if result and result.returncode == 0: + return {"icon": "💻", "title": "Gabi's PC", "detail": "Session terminated"} + + return {"icon": "❌", "title": "Gabi's PC", "detail": "Failed to terminate session"} + + +# ── Gaja ────────────────────────────────────────────────────────────── + +def gaja_check() -> bool: + """Return True if user 'gaja' has an active login session on her PC.""" + result = _run(["ping", "-c", "1", "-W", "1", "192.168.1.122"]) + if not result or result.returncode != 0: + return False + + check = _run( + ["sshpass", "-p", "Nagaja", "ssh", "-o", "ConnectTimeout=5", + "-o", "StrictHostKeyChecking=no", "-F", "/dev/null", + "gaja@192.168.1.122", + "loginctl list-sessions --no-pager 2>/dev/null | grep -q gaja"] + ) + return check is not None and check.returncode == 0 + + +def gaja_turnoff() -> dict | None: + """Log out Gaja if she's logged in. Returns action dict or None.""" + # Ping check + result = _run(["ping", "-c", "1", "-W", "1", "192.168.1.122"]) + if not result or result.returncode != 0: + return {"icon": "❌", "title": "Gaja's PC", "detail": "Unreachable (ping failed)"} + + if not gaja_check(): + return {"icon": "💻", "title": "Gaja's PC", "detail": "Not logged in — no action needed"} + + _run( + ["sshpass", "-p", "Nagaja", "ssh", "-o", "ConnectTimeout=5", + "-o", "StrictHostKeyChecking=no", "-F", "/dev/null", + "gaja@192.168.1.122", + "loginctl terminate-user gaja"], + timeout=5 + ) + return {"icon": "💻", "title": "Gaja's PC", "detail": "Logged out"} + + +# ── Orchestrator ────────────────────────────────────────────────────── + +def shutdown_all() -> list[dict]: + """Run all device checks and logouts. Returns ordered list of action dicts.""" + actions = [] + + # Time window + now = datetime.now() + time_start = now.replace(hour=0, minute=0, second=0) + time_stop = now.replace(hour=23, minute=59, second=59) + + if now < time_start: + diff = int((time_start - now).total_seconds()) + actions.append({"icon": "⏳", "title": "Time Window", "detail": f"Before active hours — {diff}s until start"}) + elif now > time_stop: + diff = int((now - time_stop).total_seconds()) + actions.append({"icon": "⏳", "title": "Time Window", "detail": f"After active hours — {diff}s past end"}) + else: + actions.append({"icon": "🟢", "title": "Active Hours", "detail": f"Executing for the next {time_stop - now}"}) + # Check → act for each device + for check_fn, action_fn in [ + (tv_check, tv_turnoff), + (gabi_check, gabi_turnoff), + (gaja_check, gaja_turnoff), + ]: + status = check_fn() + if isinstance(status, dict): + actions.append(status) # e.g. unreachable + elif status is True: + result = action_fn() + if result: + actions.append(result) + + return actions diff --git a/off.sh b/off.sh deleted file mode 100644 index 0848495..0000000 --- a/off.sh +++ /dev/null @@ -1,57 +0,0 @@ -TIME_START=00:00 -TIME_STOP=23:59 - -tv_check() -{ - ping -c 1 -W 1 192.168.1.12 2>&1 > /dev/null || { echo "TV is unreachable."; return; } - adb connect 192.168.1.12 - - if adb shell dumpsys power | grep mWakefulness=Awake - then - adb shell input keyevent 26 - else - echo "TV is turning OFF." - fi -} - -gabi_check() -{ - loginctl list-users | grep gabi || { echo 'Gabi is not online.'; return; } - doas loginctl terminate-user gabi -} - -gaja_check() -{ - ping -c 1 -W 1 192.168.1.122 2>&1 >/dev/null || { echo "Gaja's PC is unreachable."; return; } - - # Check if Gaja has any sessions - sshpass -p Nagaja ssh -o ConnectTimeout=5 -o StrictHostKeyChecking=no -F /dev/null gaja@192.168.1.122 \ - "loginctl list-sessions --no-pager 2>/dev/null | grep -q gaja" 2>&1 - if [ $? -ne 0 ]; then - echo "Gaja is not online." - return - fi - - # Terminate all of Gaja's sessions (SSH drops = success) - sshpass -p Nagaja ssh -o ConnectTimeout=5 -o StrictHostKeyChecking=no -F /dev/null gaja@192.168.1.122 \ - "loginctl terminate-user gaja" 2>/dev/null - echo "Gaja's PC: logged out." -} - -S=$(date -d "$TIME_START" +%s) -T=$(date -d "$TIME_STOP" +%s) -C=$(date +%s) -if [ $C -lt $S ] -then - echo "PRE: $((S - C)) seconds to go." -elif [ $C -gt $T ] -then - echo "POST: $((C - T)) seconds passed." -else - echo "ACTIVE: Executing for the next $((T - C)) seconds." - gabi_check - gaja_check - tv_check -fi - -exit 0 diff --git a/server.py b/server.py index d317db9..6f1b7d7 100755 --- a/server.py +++ b/server.py @@ -2,80 +2,9 @@ """Simple web server: one button to turn off all kids' devices.""" import json -import re -import subprocess -from pathlib import Path from http.server import HTTPServer, BaseHTTPRequestHandler -SCRIPT = Path(__file__).parent / "off.sh" - - -def parse_output(text: str) -> list[dict]: - """Parse off.sh output into structured action items.""" - actions = [] - lines = text.strip().splitlines() - - for line in lines: - line = line.strip() - if not line: - continue - - # Time window messages - m = re.match(r"PRE:\s*(.+)seconds?\s*to\s*go\.", line) - if m: - actions.append({"icon": "⏳", "title": "Time Window", "detail": f"Before active hours — {m.group(1)}s until start"}) - continue - - m = re.match(r"POST:\s*(.+)seconds?\s*passed\.", line) - if m: - actions.append({"icon": "⏳", "title": "Time Window", "detail": f"After active hours — {m.group(1)}s past end"}) - continue - - m = re.match(r"ACTIVE:\s*(.+)", line) - if m: - actions.append({"icon": "🟢", "title": "Active Hours", "detail": m.group(1)}) - continue - - # TV checks - if "TV is unreachable" in line: - actions.append({"icon": "❌", "title": "TV", "detail": "Unreachable (ping failed)"}) - continue - - if "TV is turning OFF" in line: - actions.append({"icon": "📺", "title": "TV", "detail": "Already off — no action needed"}) - continue - - if "mWakefulness=Awake" in line: - actions.append({"icon": "📺", "title": "TV", "detail": "Screen turned off (power key sent)"}) - continue - - # User checks - if "Gabi is not online" in line: - actions.append({"icon": "💻", "title": "Gabi's PC", "detail": "Not logged in — no action needed"}) - continue - - m = re.match(r"terminate-user gabi\b", line) - if m or "session terminated" in line.lower(): - actions.append({"icon": "💻", "title": "Gabi's PC", "detail": "Session terminated"}) - continue - - if "Gaja is not online" in line: - actions.append({"icon": "💻", "title": "Gaja's PC", "detail": "Not logged in — no action needed"}) - continue - - if "Gaja's PC is unreachable" in line: - actions.append({"icon": "❌", "title": "Gaja's PC", "detail": "Unreachable (ping failed)"}) - continue - - if "Gaja's PC: logged out" in line: - actions.append({"icon": "💻", "title": "Gaja's PC", "detail": "Logged out"}) - continue - - # If no structured items found, fallback to raw - if not actions: - actions.append({"icon": "📜", "title": "Output", "detail": text.strip() or "(empty)"}) - - return actions +from devices import shutdown_all class Handler(BaseHTTPRequestHandler): @@ -91,26 +20,12 @@ class Handler(BaseHTTPRequestHandler): def do_POST(self): if self.path == "/run": try: - result = subprocess.run( - ["bash", str(SCRIPT)], - capture_output=True, - text=True, - timeout=30, - ) - raw_output = (result.stdout + result.stderr).strip() or "Script executed (no output)." - status = "ok" if result.returncode == 0 else f"exit {result.returncode}" - except FileNotFoundError: - raw_output = f"Script not found at {SCRIPT}" - status = "error" - except subprocess.TimeoutExpired: - raw_output = "Script timed out after 30 seconds." - status = "error" + actions = shutdown_all() + status = "ok" except Exception as e: - raw_output = str(e) + actions = [{"icon": "❌", "title": "Error", "detail": str(e)}] status = "error" - actions = parse_output(raw_output) if status == "ok" else [{"icon": "❌", "title": "Error", "detail": raw_output}] - self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() @@ -167,7 +82,7 @@ HTML = """\

📺 Kids Devices

-

TV + Gabi's computer

+

TV + Gabi's computer + Gaja's computer