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

feat: complete device management system with float budget tracking, curfew enforcement, TUI, and parallel checks

Browse Source 
- Float-based budget tracking with precise elapsed time calculation
- Curfew enforcement (devices turn off when blocked)
- Terminal TUI with live status refresh
- Parallel device checks using ThreadPoolExecutor
- Simplified config with time range format (e.g. '15:00-20:30')
- Optimistic UI updates for budget buttons
- Bug fixes: curfew shutdown, last_check reset on offline
This commit is contained in:
Mitja Horvat
2026-06-16 22:37:48 +02:00
parent f3b5cd40fb
commit 2e47aa84f9
8 changed files with 76 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
devices.py
View File

@ -3,6 +3,7 @@
import json
import subprocess
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from pathlib import Path
@ -12,10 +13,10 @@ import config
BUDGET_S = config.BUDGET_S
STATE_FILE = Path(__file__).parent / ".device_state.json"
TICK_INTERVAL = config.TICK_INTERVAL
ALLOWED_WEEKDAY_START = config.ALLOWED_WEEKDAY_START
ALLOWED_WEEKDAY_END = config.ALLOWED_WEEKDAY_END
ALLOWED_WEEKEND_START = config.ALLOWED_WEEKEND_START
ALLOWED_END_MINUTE = config.ALLOWED_END_MINUTE
ALLOWED_WEEKDAY_START = config.WEEKDAY_START
ALLOWED_WEEKDAY_END = config.WEEKDAY_END
ALLOWED_WEEKEND_START = config.WEEKEND_START
ALLOWED_WEEKEND_END = config.WEEKEND_END
# ── Budget state (persisted to disk) ─────────────────────────────────
@ -143,47 +144,49 @@ DEVICES = {
}
def _get_budget(device: str) -> int:
"""Get remaining budget in seconds."""
def _get_budget(device: str) -> float:
"""Get remaining budget in seconds (float for precision)."""
state = _load_state()
return state.get(device, {}).get("budget", BUDGET_S)
value = state.get(device, {}).get("budget", float(BUDGET_S))
return float(value) # ensure always float
def _set_budget(device: str, budget_seconds: int):
def _set_budget(device: str, budget_seconds: float):
"""Set budget and track last check time for accurate elapsed calculation."""
state = _load_state()
if device not in state:
state[device] = {"budget": BUDGET_S}
state[device] = {"budget": float(BUDGET_S)}
state[device]["budget"] = budget_seconds
state[device]["last_online"] = datetime.now().isoformat()
state[device]["last_check"] = time.time() # track for elapsed calculation
_save_state(state)
def _reset_budget(device: str):
_set_budget(device, BUDGET_S)
_set_budget(device, float(BUDGET_S))
def _budget_to_minutes(budget_seconds: int) -> int:
"""Convert budget seconds to minutes for display."""
return max(0, budget_seconds // 60)
def _budget_to_minutes(budget_seconds: float) -> int:
"""Convert budget seconds to minutes for display (floor)."""
return max(0, int(budget_seconds) // 60)
def is_curfew_allowed() -> bool:
"""Check if current time is within allowed hours (devices allowed).
Allowed hours:
Mon-Fri: ALLOWED_WEEKDAY_START:00 to ALLOWED_WEEKDAY_END:ALLOWED_END_MINUTE
Sat-Sun: ALLOWED_WEEKEND_START:00 to ALLOWED_WEEKDAY_END:ALLOWED_END_MINUTE
Mon-Fri: WEEKDAY_START to WEEKDAY_END (minutes since midnight)
Sat-Sun: WEEKEND_START to WEEKEND_END (minutes since midnight)
"""
now = datetime.now()
weekday = now.weekday() # 0=Mon, 6=Sun
current_minutes = now.hour * 60 + now.minute
if weekday < 5: # Mon-Fri
start = ALLOWED_WEEKDAY_START * 60
end = (ALLOWED_WEEKDAY_END * 60) + ALLOWED_END_MINUTE
start = ALLOWED_WEEKDAY_START
end = ALLOWED_WEEKDAY_END
else: # Sat-Sun
start = ALLOWED_WEEKEND_START * 60
end = (ALLOWED_WEEKDAY_END * 60) + ALLOWED_END_MINUTE
start = ALLOWED_WEEKEND_START
end = ALLOWED_WEEKEND_END
return start <= current_minutes < end
@ -209,11 +212,11 @@ def curfew_status() -> dict:
current_minutes = now.hour * 60 + now.minute
if weekday < 5: # Mon-Fri
start = ALLOWED_WEEKDAY_START * 60
end = (ALLOWED_WEEKDAY_END * 60) + ALLOWED_END_MINUTE
start = ALLOWED_WEEKDAY_START
end = ALLOWED_WEEKDAY_END
else: # Sat-Sun
start = ALLOWED_WEEKEND_START * 60
end = (ALLOWED_WEEKDAY_END * 60) + ALLOWED_END_MINUTE
start = ALLOWED_WEEKEND_START
end = ALLOWED_WEEKEND_END
in_curfew = not (start <= current_minutes < end) # True when devices are BLOCKED
@ -237,15 +240,15 @@ def curfew_status() -> dict:
minutes_left = (24 * 60) - current_minutes + start
next_time = _minutes_to_time(start)
else: # Sunday after allowed ends → Monday
minutes_left = (3 * 24 * 60) - current_minutes + ALLOWED_WEEKDAY_START * 60
next_time = _minutes_to_time(ALLOWED_WEEKDAY_START * 60)
minutes_left = (3 * 24 * 60) - current_minutes + ALLOWED_WEEKDAY_START
next_time = _minutes_to_time(ALLOWED_WEEKDAY_START)
message = f"Blocked until {next_time} ({minutes_left} min)"
else:
# Devices allowed — show time until allowed hours end
minutes_left = end - current_minutes
next_time = _minutes_to_time(end)
message = f"Allowed until {_minutes_to_time(end)} ({minutes_left} min)"
message = f"Allowed until {next_time} ({minutes_left} min)"
return {
"in_curfew": in_curfew,
@ -271,10 +274,6 @@ def _budget_tick():
for dev_id in DEVICES:
_reset_budget(dev_id)
# Only count down budget during allowed curfew hours
if not is_curfew_allowed():
return
# Check all devices in parallel
online_status = {}
with ThreadPoolExecutor(max_workers=3) as executor:
@ -292,22 +291,39 @@ def _budget_tick():
# Process results sequentially (budget operations need to be thread-safe)
with _timer_lock:
if not is_curfew_allowed():
# Curfew in effect — turn off all online devices
for dev_id, online in online_status.items():
if online:
DEVICES[dev_id]["turnoff"]()
return
current_time = time.time() # float timestamp for precise elapsed calculation
for dev_id, online in online_status.items():
dev_info = DEVICES[dev_id]
state = _load_state()
current_budget = state.get(dev_id, {}).get("budget", BUDGET_S)
current_budget = state.get(dev_id, {}).get("budget", float(BUDGET_S))
last_check = state.get(dev_id, {}).get("last_check", current_time)
# Expired budget + device back online → shut it down
if current_budget <= 0 and online:
dev_info["turnoff"]()
_set_budget(dev_id, 0.0)
continue
if online and current_budget > 0:
new_budget = current_budget - TICK_INTERVAL
# Calculate actual elapsed time since last check
elapsed = current_time - last_check
if elapsed < 1: # ignore very small deltas (< 1s)
continue
new_budget = current_budget - elapsed
_set_budget(dev_id, new_budget)
if new_budget <= 0:
dev_info["turnoff"]()
else:
# Always update last_check even when offline
_set_budget(dev_id, current_budget)
def start_timer():
@ -365,7 +381,7 @@ def _check_device(dev_id: str, dev_info: dict) -> dict:
detail = "Online" if online else "Offline"
state = _load_state()
budget_seconds = state.get(dev_id, {}).get("budget", BUDGET_S)
budget_seconds = state.get(dev_id, {}).get("budget", float(BUDGET_S))
budget_minutes = _budget_to_minutes(budget_seconds)
# Budget warning