284 lines
8.7 KiB
Python
284 lines
8.7 KiB
Python
import asyncio
|
|
import os
|
|
import signal
|
|
from typing import Annotated, Any, cast
|
|
|
|
from fastapi import APIRouter, Form
|
|
from fastapi.responses import HTMLResponse
|
|
|
|
from app.plugins.base import Plugin
|
|
from app.render import render
|
|
from app.utils.subprocess import run_json_async
|
|
|
|
router = APIRouter(prefix="/api/plugins/sleep", tags=["plugins"])
|
|
|
|
WHO = "Dashboard (sleep-inhibit)"
|
|
WHY = "dashboard: keep system awake"
|
|
BLOCK_MODES = ("block", "block-weak")
|
|
|
|
_holder: asyncio.subprocess.Process | None = None
|
|
_toggle_lock = asyncio.Lock()
|
|
|
|
|
|
async def _list() -> tuple[list[dict[str, Any]], str]:
|
|
"""List the currently active systemd inhibitor locks.
|
|
|
|
Runs `systemd-inhibit --json=short --list` with a 5 s timeout (the
|
|
child is killed on timeout). Every failure mode — missing binary,
|
|
timeout, non-zero exit, bad JSON — is returned as a short error
|
|
string rather than raised, so the UI can show a degraded state.
|
|
|
|
Returns:
|
|
(lock entries, "") on success, else ([], error description).
|
|
"""
|
|
data, err = await run_json_async(["systemd-inhibit", "--json=short", "--list"], timeout=5)
|
|
if err:
|
|
return [], err[:200]
|
|
if not isinstance(data, list):
|
|
return [], "unexpected systemd-inhibit output"
|
|
items: list[dict[str, Any]] = [e for e in cast("list[Any]", data) if isinstance(e, dict)]
|
|
return items, ""
|
|
|
|
|
|
def _verdict(inhibitors: list[dict[str, Any]]) -> str:
|
|
"""Whether sleep is currently inhibited by anything.
|
|
|
|
Only locks whose "what" includes "sleep" AND whose mode is block or
|
|
block-weak actually prevent sleep (delay mode does not).
|
|
|
|
Args:
|
|
inhibitors: entries from _list().
|
|
|
|
Returns:
|
|
"blocked" or "ok".
|
|
"""
|
|
for e in inhibitors:
|
|
whats = str(e.get("what", "")).split(":")
|
|
if "sleep" in whats and e.get("mode") in BLOCK_MODES:
|
|
return "blocked"
|
|
return "ok"
|
|
|
|
|
|
def _rows(inhibitors: list[dict[str, Any]]) -> list[dict[str, str | bool]]:
|
|
"""Shape block-mode inhibitor entries into table rows for the UI.
|
|
|
|
Delay-mode locks are skipped (they don't block sleep). The proc cell
|
|
shows "user · pid" when the lock has a live pid. The own flag marks
|
|
the lock held by this dashboard itself.
|
|
|
|
Args:
|
|
inhibitors: entries from _list().
|
|
|
|
Returns:
|
|
One row per block-mode lock: who, proc, what, why, mode, own.
|
|
"""
|
|
rows: list[dict[str, str | bool]] = []
|
|
for e in inhibitors:
|
|
mode = str(e.get("mode", ""))
|
|
if mode not in BLOCK_MODES:
|
|
continue
|
|
user = str(e.get("user", ""))
|
|
pid = e.get("pid")
|
|
if isinstance(pid, int) and pid > 0:
|
|
proc = f"{user} · {pid}" if user else str(pid)
|
|
else:
|
|
proc = user
|
|
rows.append({
|
|
"who": str(e.get("who", "?")),
|
|
"proc": proc,
|
|
"what": str(e.get("what", "")),
|
|
"why": str(e.get("why", "")),
|
|
"mode": mode,
|
|
"own": e.get("who") == WHO,
|
|
})
|
|
return rows
|
|
|
|
|
|
def _reap_dead_holder() -> None:
|
|
"""Forget the holder child if it has already exited on its own.
|
|
|
|
The systemd-inhibit child can die (e.g. the user killed it) without
|
|
going through _release(); checking returncode here keeps "holding" in
|
|
sync with reality.
|
|
"""
|
|
global _holder
|
|
if _holder is not None and _holder.returncode is not None:
|
|
_holder = None
|
|
|
|
|
|
def _context(inhibitors: list[dict[str, Any]], error: str, message: str = "") -> dict[str, Any]:
|
|
"""Build the template context shared by the state and skeleton fragments.
|
|
|
|
Args:
|
|
inhibitors: entries from _list().
|
|
error: error string to display (from _list or a caller), "".
|
|
message: transient success message to display, "".
|
|
|
|
Returns:
|
|
Context with inhibitors rows, verdict, message, error, and
|
|
holding (whether this dashboard holds a lock).
|
|
"""
|
|
_reap_dead_holder()
|
|
return {
|
|
"inhibitors": _rows(inhibitors),
|
|
"verdict": _verdict(inhibitors),
|
|
"message": message,
|
|
"error": error,
|
|
"holding": _holder is not None,
|
|
}
|
|
|
|
|
|
async def _state(message: str = "", error: str = "") -> str:
|
|
"""Render the compact state fragment (polling view).
|
|
|
|
Args:
|
|
message: transient success message, or "".
|
|
error: error to display (overrides the _list error), or "".
|
|
|
|
Returns:
|
|
The rendered sleep_state.html fragment.
|
|
"""
|
|
inhibitors, err = await _list()
|
|
if error:
|
|
err = error
|
|
return render("plugins/sleep_state.html", **_context(inhibitors, err, message))
|
|
|
|
|
|
async def _skeleton(message: str = "", error: str = "") -> str:
|
|
"""Render the full skeleton fragment (initial + post-toggle view).
|
|
|
|
Args:
|
|
message: transient success message, or "".
|
|
error: error to display (overrides the _list error), or "".
|
|
|
|
Returns:
|
|
The rendered sleep_skeleton.html fragment.
|
|
"""
|
|
inhibitors, err = await _list()
|
|
if error:
|
|
err = error
|
|
return render("plugins/sleep_skeleton.html", **_context(inhibitors, err, message))
|
|
|
|
|
|
async def _acquire() -> str:
|
|
"""Start the systemd-inhibit child that holds the dashboard's sleep lock.
|
|
|
|
The child runs `systemd-inhibit --what=sleep --mode=block ... sleep
|
|
infinity` in its own session, so the lock (identified by the WHO
|
|
marker) survives independently of this coroutine and can be reaped
|
|
by _open() on a restart. The whole child group is what _release()
|
|
later kills via os.killpg.
|
|
|
|
Returns:
|
|
"" on success, or a short error string.
|
|
"""
|
|
global _holder
|
|
try:
|
|
_holder = await asyncio.create_subprocess_exec(
|
|
"systemd-inhibit",
|
|
"--what=sleep",
|
|
"--mode=block",
|
|
f"--who={WHO}",
|
|
f"--why={WHY}",
|
|
"sleep", "infinity",
|
|
stdout=asyncio.subprocess.DEVNULL,
|
|
stderr=asyncio.subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
)
|
|
except OSError as e:
|
|
return str(e)[:200]
|
|
return ""
|
|
|
|
|
|
async def _release() -> None:
|
|
"""Release the dashboard's sleep lock by killing the holder child.
|
|
|
|
Clears the holder reference first (so re-entrant calls are safe),
|
|
sends SIGTERM to the child's whole process group, waits up to 3 s,
|
|
and escalates to SIGKILL if it is still alive.
|
|
"""
|
|
global _holder
|
|
p, _holder = _holder, None
|
|
if p is None:
|
|
return
|
|
try:
|
|
os.killpg(p.pid, signal.SIGTERM)
|
|
except (ProcessLookupError, PermissionError):
|
|
pass
|
|
try:
|
|
_ = await asyncio.wait_for(p.wait(), 3)
|
|
except TimeoutError:
|
|
try:
|
|
os.killpg(p.pid, signal.SIGKILL)
|
|
except (ProcessLookupError, PermissionError):
|
|
pass
|
|
_ = await p.wait()
|
|
|
|
|
|
@router.get("/state")
|
|
async def state():
|
|
"""Poll endpoint: return the compact state fragment."""
|
|
return HTMLResponse(await _state())
|
|
|
|
|
|
@router.post("/toggle")
|
|
async def toggle(on: Annotated[str | None, Form()] = None):
|
|
"""Turn the dashboard's sleep lock on or off.
|
|
|
|
Guarded by a module-level lock so rapid double-clicks cannot start
|
|
two holders or race release against acquire. Toggling on acquires
|
|
the lock (errors are shown in the fragment, not raised); toggling
|
|
off releases it.
|
|
|
|
Args:
|
|
on: "on" to acquire, anything else to release.
|
|
|
|
Returns:
|
|
The skeleton fragment with a result message or error.
|
|
"""
|
|
async with _toggle_lock:
|
|
if on and _holder is None:
|
|
err = await _acquire()
|
|
if err:
|
|
return HTMLResponse(await _skeleton(error=err))
|
|
return HTMLResponse(await _skeleton(message="inhibiting sleep"))
|
|
if not on and _holder is not None:
|
|
await _release()
|
|
return HTMLResponse(await _skeleton(message="sleep inhibition released"))
|
|
return HTMLResponse(await _skeleton())
|
|
|
|
|
|
async def _open() -> None:
|
|
"""Reap stale sleep locks left by a previous dashboard instance.
|
|
|
|
On startup, any block lock whose who marker is this dashboard's WHO
|
|
string belongs to a dead instance (the holder child does not survive
|
|
a restart), so it is SIGTERMed by pid. Locks held by other who
|
|
markers are never touched.
|
|
"""
|
|
inhibitors, _err = await _list()
|
|
for e in inhibitors:
|
|
if e.get("who") != WHO:
|
|
continue
|
|
pid = e.get("pid")
|
|
if not isinstance(pid, int) or pid <= 0:
|
|
continue
|
|
try:
|
|
_ = os.kill(pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
|
|
async def _close() -> None:
|
|
"""Shutdown hook: release the lock if the UI left it on."""
|
|
await _release()
|
|
|
|
|
|
plugin = Plugin(
|
|
id="sleep",
|
|
title="Sleep inhibitors",
|
|
description="Active block-mode systemd inhibitor locks.",
|
|
skeleton_fn=_skeleton,
|
|
open_fn=_open,
|
|
close_fn=_close,
|
|
)
|