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

Implementation of the 1 hour budgeting system.

Browse Source
This commit is contained in:
Mitja Horvat
2026-06-09 08:07:45 +02:00
parent c5b9036644
commit d3493f50d1
2 changed files with 297 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

249
devices.py
View File

@ -1,9 +1,37 @@
"""Device management: check status and power off kids' devices."""
from datetime import datetime
"""Device management: check status, power off, and budget tracking."""
import json
import os
import subprocess
import threading
from datetime import datetime, timedelta
from pathlib import Path
# ── Config ────────────────────────────────────────────────────────────
BUDGET_SECONDS = int(os.environ.get("BUDGET_MINUTES", "60")) * 60 # default 60 min in seconds
STATE_FILE = Path(__file__).parent / ".device_state.json"
TICK_INTERVAL = 10 # seconds between budget checks
# ── Budget state (persisted to disk) ─────────────────────────────────
def _load_state() -> dict:
if STATE_FILE.exists():
with open(STATE_FILE) as f:
return json.load(f)
return {}
def _save_state(state: dict):
with open(STATE_FILE, "w") as f:
json.dump(state, f)
# ── Helpers ───────────────────────────────────────────────────────────
def _run(cmd: list[str], *, timeout: int = 10) -> subprocess.CompletedProcess | None:
"""Run a command, return CompletedProcess or None on error."""
@ -22,24 +50,24 @@ def tv_check() -> dict:
"""
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)"}
return {"online": False, "detail": "Offline"}
_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 {"online": True, "detail": "Online"}
return {"icon": "📺", "title": "TV", "detail": "Already off"}
return {"online": False, "detail": "Offline"}
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":
if not state["online"]:
return None
_run(["adb", "shell", "input", "keyevent", "26"])
return {"icon": "📺", "title": "TV", "detail": "Screen turned off (power key sent)"}
return {"detail": "Screen turned off (power key sent)"}
# ── Gabi ──────────────────────────────────────────────────────────────
@ -53,41 +81,44 @@ def gabi_check() -> bool:
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"}
return {"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 {"detail": "Session terminated"}
return {"icon": "❌", "title": "Gabi's PC", "detail": "Failed to terminate session"}
return {"detail": "Failed to terminate session"}
# ── Gaja ──────────────────────────────────────────────────────────────
def gaja_check() -> bool:
"""Return True if user 'gaja' has an active login session on her PC."""
def gaja_check() -> dict:
"""Check if Gaja's PC is online and she has an active desktop session.
Returns an action dict describing the current state.
"""
result = _run(["ping", "-c", "1", "-W", "1", "192.168.1.122"])
if not result or result.returncode != 0:
return False
return {"online": False, "detail": "Offline"}
# Check for a session with seat assigned (desktop session, not SSH)
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"]
"loginctl list-sessions --no-pager 2>/dev/null | grep gaja | grep seat"]
)
return check is not None and check.returncode == 0
if check is not None and check.stdout.strip():
return {"online": True, "detail": "Online"}
return {"online": False, "detail": "Offline"}
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"}
state = gaja_check()
if not state["online"]:
return {"detail": "Not online — no action needed"}
_run(
["sshpass", "-p", "Nagaja", "ssh", "-o", "ConnectTimeout=5",
@ -96,40 +127,154 @@ def gaja_turnoff() -> dict | None:
"loginctl terminate-user gaja"],
timeout=5
)
return {"icon": "💻", "title": "Gaja's PC", "detail": "Logged out"}
return {"detail": "Logged out"}
# ── Device registry ───────────────────────────────────────────────────
DEVICES = {
"tv": {"icon": "📺", "name": "TV", "check": tv_check, "turnoff": tv_turnoff},
"gabi": {"icon": "💻", "name": "Gabi's PC", "check": gabi_check, "turnoff": gabi_turnoff},
"gaja": {"icon": "💻", "name": "Gaja's PC", "check": gaja_check, "turnoff": gaja_turnoff},
}
def _get_budget(device: str) -> int:
"""Get remaining budget in seconds."""
state = _load_state()
return state.get(device, {}).get("budget", BUDGET_SECONDS)
def _set_budget(device: str, budget_seconds: int):
state = _load_state()
if device not in state:
state[device] = {"budget": BUDGET_SECONDS}
state[device]["budget"] = budget_seconds
state[device]["last_online"] = datetime.now().isoformat()
_save_state(state)
def _reset_budget(device: str):
_set_budget(device, BUDGET_SECONDS)
def _budget_to_minutes(budget_seconds: int) -> int:
"""Convert budget seconds to minutes for display."""
return max(0, budget_seconds // 60)
# ── Budget timer (runs every 60s in background) ──────────────────────
_timer_running = False
_timer_lock = threading.Lock()
def _budget_tick():
"""Called every 10 seconds by the background timer."""
now = datetime.now()
reset_time = now.replace(hour=7, minute=0, second=0)
# Reset all budgets at 7:00 AM each day
if now >= reset_time and now < reset_time + timedelta(minutes=2):
for dev_id in DEVICES:
_reset_budget(dev_id)
with _timer_lock:
for dev_id, dev_info in DEVICES.items():
state = _load_state()
current_budget = state.get(dev_id, {}).get("budget", BUDGET_SECONDS)
is_online = dev_info["check"]()
if isinstance(is_online, dict):
online = is_online.get("online", False)
else:
online = bool(is_online)
# Expired budget + device back online → shut it down
if current_budget <= 0 and online:
dev_info["turnoff"]()
continue
if online and current_budget > 0:
new_budget = current_budget - TICK_INTERVAL
_set_budget(dev_id, new_budget)
if new_budget <= 0:
dev_info["turnoff"]()
def start_timer():
"""Start the background budget timer (runs every 10s)."""
global _timer_running
with _timer_lock:
if _timer_running:
return
_timer_running = True
def loop():
while _timer_running:
try:
_budget_tick()
except Exception:
pass # Don't crash the timer on device errors
threading.Event().wait(TICK_INTERVAL)
t = threading.Thread(target=loop, daemon=True)
t.start()
def stop_timer():
"""Stop the background timer."""
global _timer_running
with _timer_lock:
_timer_running = False
# ── Orchestrator ──────────────────────────────────────────────────────
def shutdown_all() -> list[dict]:
"""Run all device checks and logouts. Returns ordered list of action dicts."""
"""Indiscriminately turn off all devices. 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()
for dev_id, dev_info in DEVICES.items():
result = dev_info["turnoff"]()
if result:
actions.append(result)
budget_minutes = _budget_to_minutes(_get_budget(dev_id))
actions.append({
"icon": dev_info["icon"],
"title": dev_info["name"],
"detail": f"{result['detail']} ({budget_minutes} min left)",
})
return actions
def status_all() -> list[dict]:
"""Get current status + budget for all devices. Returns ordered list."""
actions = []
for dev_id, dev_info in DEVICES.items():
state = _load_state()
budget_seconds = state.get(dev_id, {}).get("budget", BUDGET_SECONDS)
is_online = dev_info["check"]()
if isinstance(is_online, dict):
online = is_online.get("online", False)
detail = is_online.get("detail", "")
else:
online = bool(is_online)
detail = "Online" if online else "Offline"
budget_minutes = _budget_to_minutes(budget_seconds)
# Budget warning
if budget_minutes <= 0:
icon = "🔴"
elif budget_minutes < 5:
icon = "🟠"
else:
icon = dev_info["icon"]
actions.append({
"icon": icon,
"title": dev_info["name"],
"detail": f"{detail} · {budget_minutes} min left",
})
return actions

99
server.py
View File

@ -4,7 +4,7 @@
import json
from http.server import HTTPServer, BaseHTTPRequestHandler
from devices import shutdown_all
from devices import shutdown_all, status_all, _set_budget, _get_budget, BUDGET_SECONDS, TICK_INTERVAL
class Handler(BaseHTTPRequestHandler):
@ -14,10 +14,18 @@ class Handler(BaseHTTPRequestHandler):
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(HTML.encode())
elif self.path == "/status":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(status_all()).encode())
else:
self.send_error(404)
def do_POST(self):
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length).decode() if content_length > 0 else ""
if self.path == "/run":
try:
actions = shutdown_all()
@ -30,6 +38,24 @@ class Handler(BaseHTTPRequestHandler):
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"status": status, "actions": actions}).encode())
elif self.path == "/budget":
try:
data = json.loads(body) if body else {}
dev_id = data.get("device", "")
delta = int(data.get("delta", 0))
current = _get_budget(dev_id)
new_budget = max(0, min(BUDGET_SECONDS, current + delta))
_set_budget(dev_id, new_budget)
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"status": "ok"}).encode())
except Exception as e:
self.send_response(400)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"status": "error", "detail": str(e)}).encode())
else:
self.send_error(404)
@ -77,12 +103,39 @@ HTML = """\
.action-item .info { flex: 1; }
.action-item .title { font-weight: 600; font-size: .95rem; }
.action-item .detail { color: #aaa; font-size: .82rem; margin-top: .15rem; }
#status {
margin-bottom: 2rem;
}
.status-item {
display: flex; align-items: center;
padding: .4rem .75rem; margin-bottom: .3rem;
border-radius: 6px; background: #0f3460; font-size: .8rem;
gap: .5rem;
}
.status-item .center {
display: flex; align-items: center; gap: .5rem;
flex: 1; min-width: 0;
}
.status-item .icon { font-size: 1.1rem; flex-shrink: 0; }
.status-item .name { font-weight: 500; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
.status-item .budget { color: #aaa; font-size: .75rem; margin-left: auto; flex-shrink: 0; }
.status-item .actions {
display: flex; align-items: center; gap: .25rem;
flex-shrink: 0;
}
.budget-icon {
background: none; border: none;
color: #666; font-size: 1rem; cursor: pointer;
padding: 0; line-height: 1; transition: .15s;
}
.budget-icon:hover { color: #fff; }
</style>
</head>
<body>
<div class="card">
<h1>📺 Kids Devices</h1>
<p>TV + Gabi's computer + Gaja's computer</p>
<div id="status"></div>
<button id="btn" onclick="run()">TURN OFF</button>
<div id="output"></div>
</div>
@ -120,13 +173,57 @@ HTML = """\
btn.disabled = false;
btn.textContent = 'TURN OFF';
}
async function refreshStatus() {
try {
const res = await fetch('https://noom.cc/off/status');
const data = await res.json();
document.getElementById('status').innerHTML = data.map(s => {
const devId = s.title === 'TV' ? 'tv' : s.title.includes("Gabi") ? 'gabi' : 'gaja';
return `<div class="status-item">
<span class="center">
<span class="icon">${s.icon}</span>
<span class="name">${s.title}</span>
<span class="budget">${s.detail}</span>
</span>
<span class="actions">
<button class="budget-icon" onclick="adjustBudget('${devId}', -300)">−</button>
<button class="budget-icon" onclick="adjustBudget('${devId}', 300)">+</button>
</span>
</div>`;
}).join('');
} catch(e) {
// Ignore status errors
}
}
async function adjustBudget(device, delta) {
try {
await fetch('https://noom.cc/off/budget', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({device, delta})
});
} catch(e) {}
refreshStatus();
}
refreshStatus();
setInterval(refreshStatus, 60000);
</script>
</body>
</html>
"""
if __name__ == "__main__":
from devices import start_timer
port = int(__import__("os").environ.get("PORT", "10000"))
start_timer()
server = HTTPServer(("0.0.0.0", port), Handler)
print(f"🚀 Open http://localhost:{port}")
print(f"⏱️ Budget timer running (checks every {TICK_INTERVAL}s, reset at 7:00 AM)")
try:
server.serve_forever()
except KeyboardInterrupt:
from devices import stop_timer
stop_timer()