pinkfluid/off
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move the off.sh script to a pure Python implementation -- devices.py.

Browse Source
This commit is contained in:
Mitja Horvat
2026-06-08 22:59:39 +02:00
parent b15f87e3d5
commit c5b9036644
3 changed files with 140 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
devices.py Normal file
View File

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

57
off.sh
View File

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

95
server.py
View File

@ -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 = """\
<body>
<div class="card">
<h1>📺 Kids Devices</h1>
<p>TV + Gabi's computer</p>
<p>TV + Gabi's computer + Gaja's computer</p>
<button id="btn" onclick="run()">TURN OFF</button>
<div id="output"></div>
</div>