101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
import json
|
|
import pathlib
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from .constants import FIRSTBOOT_DIR, FIRSTBOOT_DONE, FIRSTBOOT_ERROR, FIRSTBOOT_LOG, FIRSTBOOT_STATE, WEB_ROOT
|
|
from .helpers import ensure_dir, sha256_file
|
|
|
|
DEFAULT_STEPS = [
|
|
"Preparing system",
|
|
"Generating security keys",
|
|
"Securing the dashboard",
|
|
"Updating software (this can take a while)",
|
|
"Final checks",
|
|
"Starting Pi-Kit",
|
|
]
|
|
|
|
|
|
def _tail_text(path: pathlib.Path, max_lines: int = 200) -> str:
|
|
if not path.exists():
|
|
return ""
|
|
try:
|
|
text = path.read_text(errors="ignore")
|
|
except Exception:
|
|
return ""
|
|
lines = text.splitlines()
|
|
if len(lines) > max_lines:
|
|
lines = lines[-max_lines:]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _normalize_steps(raw_steps: Optional[List[Dict[str, Any]]], state: str) -> List[Dict[str, Any]]:
|
|
steps: List[Dict[str, Any]] = []
|
|
if raw_steps:
|
|
for entry in raw_steps:
|
|
label = (entry or {}).get("label") or (entry or {}).get("name")
|
|
if not label:
|
|
continue
|
|
status = (entry or {}).get("status") or "pending"
|
|
steps.append({"label": str(label), "status": str(status)})
|
|
|
|
if not steps:
|
|
steps = [{"label": label, "status": "pending"} for label in DEFAULT_STEPS]
|
|
|
|
if state == "done":
|
|
for step in steps:
|
|
step["status"] = "done"
|
|
return steps
|
|
|
|
|
|
def _current_step(steps: List[Dict[str, Any]]) -> Optional[str]:
|
|
for step in steps:
|
|
if step.get("status") in ("current", "running", "error"):
|
|
return step.get("label")
|
|
return None
|
|
|
|
|
|
def _load_state_file() -> Dict[str, Any]:
|
|
if FIRSTBOOT_STATE.exists():
|
|
try:
|
|
return json.loads(FIRSTBOOT_STATE.read_text())
|
|
except Exception:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def read_firstboot_status() -> Dict[str, Any]:
|
|
ensure_dir(FIRSTBOOT_DIR)
|
|
state_file = _load_state_file()
|
|
|
|
if FIRSTBOOT_ERROR.exists():
|
|
state = "error"
|
|
elif FIRSTBOOT_DONE.exists():
|
|
state = "done"
|
|
else:
|
|
state = state_file.get("state") or "running"
|
|
if state not in ("running", "done", "error"):
|
|
state = "running"
|
|
|
|
steps = _normalize_steps(state_file.get("steps"), state)
|
|
current_step = state_file.get("current_step") or _current_step(steps)
|
|
|
|
ca_path = WEB_ROOT / "assets" / "pikit-ca.crt"
|
|
ca_hash = sha256_file(ca_path) if ca_path.exists() else None
|
|
|
|
return {
|
|
"state": state,
|
|
"steps": steps,
|
|
"current_step": current_step,
|
|
"log_tail": _tail_text(FIRSTBOOT_LOG, 200),
|
|
"error_present": FIRSTBOOT_ERROR.exists(),
|
|
"error_path": "/api/firstboot/error",
|
|
"ca_hash": ca_hash,
|
|
"ca_url": "/assets/pikit-ca.crt",
|
|
}
|
|
|
|
|
|
def read_firstboot_error(max_lines: int = 200) -> Dict[str, Any]:
|
|
if not FIRSTBOOT_ERROR.exists():
|
|
return {"present": False, "text": ""}
|
|
return {"present": True, "text": _tail_text(FIRSTBOOT_ERROR, max_lines)}
|