michael/klangkiste
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upload

Browse Source
This commit is contained in:
mboehmlaender
2026-01-12 15:25:34 +00:00
parent 83bd89a25a
commit fc5a281e85
85 changed files with 3748 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
box/__pycache__/api.cpython-312.pyc
BIN
View File
Binary file not shown.
box/api.py
+141
View File
@@ -0,0 +1,141 @@
"""HTTP API for the Klangkiste Box (FastAPI)."""
from __future__ import annotations
from typing import Any, Callable
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from core.box_controller import BoxController
from storage.box_storage import BoxStorage
from wifi.manager import WifiManager
from spotify.manager import SpotifyManager
class CommandRequest(BaseModel):
command: str
payload: dict[str, Any] = {}
def create_app(
controller: BoxController,
persist_resume: Callable[[], None],
wifi_manager: WifiManager,
spotify_manager: SpotifyManager,
storage: BoxStorage,
) -> FastAPI:
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/status")
def status() -> dict[str, Any]:
state = controller.state
return {
"box_id": state.box_id,
"network_status": wifi_manager.network_status,
"connected_ssid": wifi_manager.connected_ssid,
"wifi_profiles_count": wifi_manager.profiles_count,
"spotify_status": spotify_manager.status,
"server_status": _server_status(storage),
"playback_status": {
"state": state.state.value,
"active_uid": state.active_uid,
"playback_root": state.playback_path,
"current_file": state.current_media,
"file_index": state.file_index,
"position": state.position,
"duration": state.current_duration,
"volume": state.volume,
"last_error": state.last_error,
},
}
@app.post("/command")
def command(request: CommandRequest) -> dict[str, Any]:
cmd = request.command
payload = request.payload or {}
print(f"API command: {cmd} payload={payload}")
if cmd == "nfc_on":
uid = payload.get("uid")
if not isinstance(uid, str):
raise HTTPException(status_code=400, detail="uid is required")
controller.handle_nfc_on(uid)
elif cmd == "nfc_off":
uid = payload.get("uid")
if not isinstance(uid, str):
raise HTTPException(status_code=400, detail="uid is required")
controller.handle_nfc_off(uid)
elif cmd == "play_pause":
controller.play_pause()
elif cmd == "next":
controller.next_track()
elif cmd == "prev":
controller.prev_track()
elif cmd == "vol_up":
controller.volume_up()
elif cmd == "vol_down":
controller.volume_down()
elif cmd == "stop":
if controller.state.active_uid:
controller.handle_nfc_off(controller.state.active_uid)
else:
controller.stop()
elif cmd == "trigger_error":
error_type = payload.get("type")
if not isinstance(error_type, str):
raise HTTPException(status_code=400, detail="type is required")
controller.trigger_error(error_type)
elif cmd == "wifi_add_profile":
ssid = payload.get("ssid")
password = payload.get("password")
priority = payload.get("priority", 0)
if not isinstance(ssid, str) or not isinstance(password, str):
raise HTTPException(status_code=400, detail="ssid and password required")
if not isinstance(priority, int):
priority = 0
wifi_manager.add_profile(ssid, password, priority)
elif cmd == "wifi_remove_profile":
ssid = payload.get("ssid")
if not isinstance(ssid, str):
raise HTTPException(status_code=400, detail="ssid is required")
wifi_manager.remove_profile(ssid)
elif cmd == "wifi_reset":
wifi_manager.reset()
elif cmd == "spotify_set_tokens":
access_token = payload.get("access_token")
refresh_token = payload.get("refresh_token")
expires_at = payload.get("expires_at")
account_id = payload.get("account_id")
if not isinstance(access_token, str) or not isinstance(refresh_token, str):
raise HTTPException(status_code=400, detail="tokens are required")
if not isinstance(expires_at, int):
raise HTTPException(status_code=400, detail="expires_at is required")
if account_id is not None and not isinstance(account_id, str):
raise HTTPException(status_code=400, detail="account_id must be string")
spotify_manager.set_tokens(access_token, refresh_token, expires_at, account_id)
elif cmd == "spotify_clear":
spotify_manager.clear()
else:
raise HTTPException(status_code=400, detail="unknown command")
persist_resume()
return {"ok": True}
return app
def _server_status(storage: BoxStorage) -> str:
box_meta = storage.load_box_meta()
if box_meta.get("server_reachable") is False:
return "UNREACHABLE"
if box_meta.get("server_reachable") is True:
return "OK"
return "UNKNOWN"
box/api/__init__.py
View File
box/api/http.py
+127
View File
@@ -0,0 +1,127 @@
"""HTTP API (FastAPI) for the Box."""
from __future__ import annotations
from typing import Any, Callable
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from core.box_controller import BoxController
from core.lifecycle import NetworkLifecycle
from spotify.base import SpotifyBackend
from wifi.base import WifiBackend
class CommandRequest(BaseModel):
command: str
payload: dict[str, Any] = {}
def create_app(
controller: BoxController,
lifecycle: NetworkLifecycle,
wifi_backend: WifiBackend,
spotify_backend: SpotifyBackend,
persist_state: Callable[[], None],
) -> FastAPI:
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/status")
def status() -> dict[str, Any]:
snapshot = lifecycle.snapshot()
state = controller.state
return {
"box_id": state.box_id,
"wifi_state": snapshot.wifi_state.value,
"ip_address": snapshot.ip_address,
"connected_ssid": snapshot.connected_ssid,
"wifi_profiles_count": snapshot.wifi_profiles_count,
"spotify_status": spotify_backend.status,
"playback_state": {
"state": state.state.value,
"active_uid": state.active_uid,
"playback_root": state.playback_path,
"current_file": state.current_media,
"file_index": state.file_index,
"position": state.position,
"duration": state.current_duration,
"volume": state.volume,
"last_error": state.last_error,
},
}
@app.post("/command")
def command(request: CommandRequest) -> dict[str, Any]:
cmd = request.command
payload = request.payload or {}
if cmd == "nfc_on":
uid = payload.get("uid")
if not isinstance(uid, str):
raise HTTPException(status_code=400, detail="uid is required")
controller.handle_nfc_on(uid)
elif cmd == "nfc_off":
uid = payload.get("uid")
if not isinstance(uid, str):
raise HTTPException(status_code=400, detail="uid is required")
controller.handle_nfc_off(uid)
elif cmd == "play_pause":
controller.play_pause()
elif cmd == "next":
controller.next_track()
elif cmd == "prev":
controller.prev_track()
elif cmd in {"vol_up", "volume_up"}:
controller.volume_up()
elif cmd in {"vol_down", "volume_down"}:
controller.volume_down()
elif cmd == "stop":
if controller.state.active_uid:
controller.handle_nfc_off(controller.state.active_uid)
else:
controller.stop()
elif cmd == "trigger_error":
error_type = payload.get("type")
if not isinstance(error_type, str):
raise HTTPException(status_code=400, detail="type is required")
controller.trigger_error(error_type)
elif cmd == "wifi_add_profile":
ssid = payload.get("ssid")
password = payload.get("password")
priority = payload.get("priority", 0)
if not isinstance(ssid, str) or not isinstance(password, str):
raise HTTPException(status_code=400, detail="ssid and password required")
if not isinstance(priority, int):
priority = 0
wifi_backend.add_profile(ssid, password, priority)
elif cmd == "wifi_reset":
wifi_backend.reset()
elif cmd == "spotify_set_tokens":
access_token = payload.get("access_token")
refresh_token = payload.get("refresh_token")
expires_at = payload.get("expires_at")
account_id = payload.get("account_id")
if not isinstance(access_token, str) or not isinstance(refresh_token, str):
raise HTTPException(status_code=400, detail="tokens are required")
if not isinstance(expires_at, int):
raise HTTPException(status_code=400, detail="expires_at is required")
if account_id is not None and not isinstance(account_id, str):
raise HTTPException(status_code=400, detail="account_id must be string")
spotify_backend.set_tokens(access_token, refresh_token, expires_at, account_id)
elif cmd == "spotify_clear":
spotify_backend.clear()
else:
raise HTTPException(status_code=400, detail="unknown command")
persist_state()
return {"ok": True}
return app
box/audio/__init__.py
View File
box/audio/base.py
+16
View File
@@ -0,0 +1,16 @@
"""Audio backend interface (placeholder)."""
from __future__ import annotations
from typing import Protocol
class AudioBackend(Protocol):
def play(self, media_ref: str) -> None:
raise NotImplementedError
def pause(self) -> None:
raise NotImplementedError
def stop(self) -> None:
raise NotImplementedError
box/audio/mock.py
+19
View File
@@ -0,0 +1,19 @@
"""Mock audio backend (no real playback)."""
from __future__ import annotations
from audio.base import AudioBackend
class MockAudioBackend(AudioBackend):
def __init__(self) -> None:
self._current: str | None = None
def play(self, media_ref: str) -> None:
self._current = media_ref
def pause(self) -> None:
return None
def stop(self) -> None:
self._current = None
box/config/box_config.json
+9 -1
View File
@@ -24,12 +24,20 @@
"UID_5": {
"type": "folder",
"path": "../media/book_5"
},
"UID_6": {
"type": "folder",
"path": "../media/book_6"
}
},
"resume": {
"UID_1": {
"file_index": 0,
"position": 50
"position": 12
},
"UID_2": {
"file_index": 1,
"position": 0
}
}
}
box/core/__pycache__/box_controller.cpython-312.pyc
BIN
View File
Binary file not shown.
box/core/__pycache__/state.cpython-312.pyc
BIN
View File
Binary file not shown.
box/core/box_controller.py
+137 -14
View File
@@ -7,6 +7,7 @@ from typing import Any
from core.state import BoxState, RuntimeState
from player.base_player import BasePlayer
import hashlib
class BoxController:
@@ -31,10 +32,20 @@ class BoxController:
return None
def handle_nfc_on(self, uid: str) -> None:
if self._state.state == BoxState.ERROR:
return None
if self._state.active_uid == uid and self._state.state != BoxState.IDLE:
return None
if self._config.get("server_reachable") is False:
print("error: server not reachable")
self._trigger_error("server_unreachable")
return None
tag = self._resolve_tag(uid)
if tag is None:
print(f"error: uid '{uid}' not found in config tags")
self._trigger_error("uid_unmapped")
return None
self._state.last_error = None
playlist = self._build_playlist(tag)
if not playlist:
@@ -43,15 +54,24 @@ class BoxController:
print(
f"error: no playable files found for uid '{uid}' at path '{resolved}'"
)
self._trigger_error("no_playable_files")
return None
if self._state.active_uid == uid and uid in self._state._resume:
if uid in self._state._resume:
file_index, position = self._state._resume[uid]
else:
file_index, position = 0, 0
if file_index < 0 or file_index >= len(playlist):
file_index, position = 0, 0
print(
f"error: stored progress points outside playlist for uid '{uid}'"
)
self._trigger_error("invalid_resume")
return None
if self._state.active_uid and self._state.active_uid != uid:
self._store_progress(self._state.active_uid)
self.stop()
media_ref = playlist[file_index]
duration = self._duration_for(media_ref)
@@ -75,10 +95,11 @@ class BoxController:
def stop(self) -> None:
self._player.stop()
self._state.current_media = None
self._state.state = BoxState.IDLE
self._clear_playback_state()
def handle_nfc_off(self, uid: str) -> None:
if self._state.state == BoxState.ERROR:
return None
if self._state.active_uid != uid:
return None
self._store_progress(uid)
@@ -96,6 +117,8 @@ class BoxController:
return tag
def tick(self, seconds: int) -> None:
if self._state.state == BoxState.ERROR:
return None
if self._state.state != BoxState.PLAYING:
return None
if self._state.position is None or self._state.file_index is None:
@@ -124,12 +147,14 @@ class BoxController:
position = 0
if file_index >= len(self._state._playlist):
self._state.file_index = len(self._state._playlist) - 1
self._state.position = duration
self._state.current_media = None
self._state.current_file = None
self._state.current_duration = None
self._state.state = BoxState.IDLE
last_index = len(self._state._playlist) - 1
last_media = self._state._playlist[last_index]
last_duration = self._duration_for(last_media)
self._state.file_index = last_index
self._state.position = last_duration
if self._state.active_uid:
self._store_progress(self._state.active_uid)
self.stop()
return None
self._state.file_index = file_index
@@ -146,9 +171,17 @@ class BoxController:
path = self._resolve_path(path_value)
if not path.exists() or not path.is_dir():
return []
files = [p for p in path.iterdir() if p.is_file()]
files_sorted = sorted(files, key=lambda p: p.name)
return [str(p) for p in files_sorted]
return [str(p) for p in self._collect_files(path)]
def _collect_files(self, root: Path) -> list[Path]:
entries = list(root.iterdir())
dirs = sorted([p for p in entries if p.is_dir()], key=lambda p: p.name)
files = sorted([p for p in entries if p.is_file()], key=lambda p: p.name)
result: list[Path] = []
for directory in dirs:
result.extend(self._collect_files(directory))
result.extend(files)
return result
def _resolve_path(self, path_value: str) -> Path:
path = Path(path_value)
@@ -164,4 +197,94 @@ class BoxController:
def _duration_for(self, media_ref: str) -> int:
name = Path(media_ref).name
durations = {"01.mp3": 180, "02.mp3": 200, "03.mp3": 220}
return durations.get(name, 0)
if name in durations:
return durations[name]
if not name.lower().endswith(".mp3"):
return 0
digest = hashlib.md5(name.encode("utf-8")).hexdigest()
seed = int(digest[:8], 16)
return 120 + (seed % 241)
def play_pause(self) -> None:
if self._state.state in {BoxState.IDLE, BoxState.ERROR}:
return None
if self._state.state == BoxState.PLAYING:
self._state.state = BoxState.PAUSED
return None
if self._state.state == BoxState.PAUSED:
self._state.state = BoxState.PLAYING
return None
def next_track(self) -> None:
if self._state.state in {BoxState.IDLE, BoxState.ERROR}:
return None
if not self._state._playlist:
return None
if self._state.file_index is None:
return None
next_index = self._state.file_index + 1
if next_index >= len(self._state._playlist):
if self._state.active_uid:
self._store_progress(self._state.active_uid)
self.stop()
return None
self._jump_to_index(next_index, 0)
self._state.state = BoxState.PLAYING
def prev_track(self, threshold_seconds: int = 3) -> None:
if self._state.state in {BoxState.IDLE, BoxState.ERROR}:
return None
if self._state.file_index is None or self._state.position is None:
return None
if self._state.position > threshold_seconds:
self._state.position = 0
return None
prev_index = max(self._state.file_index - 1, 0)
self._jump_to_index(prev_index, 0)
self._state.state = BoxState.PLAYING
def volume_up(self, step: int = 5) -> None:
if self._state.state == BoxState.ERROR:
return None
self._state.volume = min(self._state.volume + step, self._state.max_volume)
def volume_down(self, step: int = 5) -> None:
if self._state.state == BoxState.ERROR:
return None
self._state.volume = max(self._state.volume - step, 0)
def _jump_to_index(self, file_index: int, position: int) -> None:
if not self._state._playlist:
return None
if file_index < 0 or file_index >= len(self._state._playlist):
return None
media_ref = self._state._playlist[file_index]
self._state.file_index = file_index
self._state.position = position
self._state.current_media = media_ref
self._state.current_file = Path(media_ref).name
self._state.current_duration = self._duration_for(media_ref)
self._player.play(media_ref)
def _trigger_error(self, code: str) -> None:
if self._state.state in {BoxState.PLAYING, BoxState.PAUSED}:
self._player.pause()
self._state.state = BoxState.ERROR
self._state.last_error = code
self._player.play(f"system:error:{code}")
self._clear_playback_state()
def _clear_playback_state(self) -> None:
self._state.state = BoxState.IDLE
self._state.active_uid = None
self._state.playback_type = None
self._state.playback_path = None
self._state.current_media = None
self._state.current_file = None
self._state.current_duration = None
self._state.file_index = None
self._state.position = None
self._state._playlist = []
def trigger_error(self, code: str) -> None:
self._trigger_error(code)
box/core/controller.py
+7
View File
@@ -0,0 +1,7 @@
"""Compatibility re-export for BoxController."""
from __future__ import annotations
from core.box_controller import BoxController
__all__ = ["BoxController"]
box/core/identity.py
+15
View File
@@ -0,0 +1,15 @@
"""Box identity generation and management."""
from __future__ import annotations
import secrets
import string
PREFIX = "klangkiste-"
ALPHABET = string.ascii_lowercase + string.digits
SUFFIX_LENGTH = 10
def generate_box_id() -> str:
suffix = "".join(secrets.choice(ALPHABET) for _ in range(SUFFIX_LENGTH))
return f"{PREFIX}{suffix}"
box/core/lifecycle.py
+52
View File
@@ -0,0 +1,52 @@
"""Network lifecycle state machine (logical)."""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from wifi.base import WifiBackend
class WifiState(str, Enum):
WIFI_UNCONFIGURED = "WIFI_UNCONFIGURED"
WIFI_CONNECTING = "WIFI_CONNECTING"
WIFI_ONLINE = "WIFI_ONLINE"
WIFI_OFFLINE = "WIFI_OFFLINE"
@dataclass
class NetworkSnapshot:
wifi_state: WifiState
ip_address: str
connected_ssid: str | None
wifi_profiles_count: int
class NetworkLifecycle:
def __init__(self, wifi_backend: WifiBackend) -> None:
self._wifi = wifi_backend
self._wifi_state = WifiState.WIFI_UNCONFIGURED
self._ip_address = "192.168.4.1"
self.refresh()
def refresh(self) -> None:
if self._wifi.profiles_count == 0:
self._wifi_state = WifiState.WIFI_UNCONFIGURED
self._ip_address = "192.168.4.1"
return
if self._wifi.connected_ssid:
self._wifi_state = WifiState.WIFI_ONLINE
self._ip_address = "127.0.0.1"
return
self._wifi_state = WifiState.WIFI_OFFLINE
self._ip_address = "127.0.0.1"
def snapshot(self) -> NetworkSnapshot:
self.refresh()
return NetworkSnapshot(
wifi_state=self._wifi_state,
ip_address=self._ip_address,
connected_ssid=self._wifi.connected_ssid,
wifi_profiles_count=self._wifi.profiles_count,
)
box/core/state.py
+3
View File
@@ -11,6 +11,7 @@ class BoxState(str, Enum):
IDLE = "IDLE"
PLAYING = "PLAYING"
PAUSED = "PAUSED"
ERROR = "ERROR"
@dataclass
@@ -26,5 +27,7 @@ class RuntimeState:
current_duration: Optional[int] = None
file_index: Optional[int] = None
position: Optional[int] = None
last_error: Optional[str] = None
_resume: dict[str, tuple[int, int]] = field(default_factory=dict)
_playlist: list[str] = field(default_factory=list)
max_volume: int = 100
box/data/.keep
View File
box/data/media/.keep
View File
box/data/media/book_1/01.mp3
View File
box/data/media/book_1/02.mp3
View File
box/data/media/book_1/03.mp3
View File
box/data/media/book_2/01.mp3
View File
box/data/media/book_2/02.mp3
View File
box/data/media/book_2/03.mp3
View File
box/data/media/book_3/01.mp3
View File
box/data/media/book_3/02.mp3
View File
box/data/media/book_3/03.mp3
View File
box/data/media/book_4/01.mp3
View File
box/data/media/book_4/02.mp3
View File
box/data/media/book_4/03.mp3
View File
box/data/media/book_5/01.mp3
View File
box/data/media/book_5/02.mp3
View File
box/data/media/book_5/03.mp3
View File
box/data/media/book_6/CD1/Titel A.mp3
View File
box/data/media/book_6/CD1/Titel B.mp3
View File
box/data/media/book_6/CD2/Titel A.mp3
View File
box/data/media/book_6/CD2/Titel B.mp3
View File
box/main.py
+214 -29
View File
@@ -3,12 +3,28 @@
from __future__ import annotations
import argparse
import asyncio
import contextlib
import os
import sys
from pathlib import Path
from typing import Any
import uvicorn
from api.http import create_app
from core.box_controller import BoxController
from core.state import RuntimeState
from core.identity import generate_box_id
from core.lifecycle import NetworkLifecycle
from core.state import BoxState, RuntimeState
from player.mock_player import MockPlayer
from setup.webui import create_setup_app
from spotify.mock import MockSpotifyBackend
from storage.box_store import BoxStore
from storage.state_store import StateStore
from storage.secret_store import SecretStore
from storage.json_storage import JsonStorage
from wifi.mock import MockWifiBackend
def build_parser() -> argparse.ArgumentParser:
@@ -19,7 +35,7 @@ def build_parser() -> argparse.ArgumentParser:
parser.add_argument(
"--config",
default="config/box_config.json",
help="Path to local box config JSON",
help=\"Path to local box config JSON (legacy seed)\",
)
subparsers = parser.add_subparsers(dest="command", required=True)
@@ -34,17 +50,24 @@ def build_parser() -> argparse.ArgumentParser:
tick_parser = subparsers.add_parser("tick", help="Simulate playback time")
tick_parser.add_argument("seconds", type=int, help="Seconds to advance")
subparsers.add_parser("run", help="Start interactive session")
subparsers.add_parser("run", help="Start interactive session + HTTP API")
return parser
def _state_from_config(config: dict[str, Any]) -> RuntimeState:
box_id = str(config.get("box_id", "unknown"))
settings = config.get("settings", {}) if isinstance(config.get("settings"), dict) else {}
def _state_from_data(
box_meta: dict[str, Any], state_data: dict[str, Any]
) -> RuntimeState:
box_id = str(box_meta.get("box_id", "unknown"))
settings = box_meta.get("settings", {}) if isinstance(box_meta.get("settings"), dict) else {}
default_volume = settings.get("default_volume", 0)
state = RuntimeState(box_id=box_id, volume=int(default_volume))
max_volume = settings.get("max_volume", 100)
state = RuntimeState(
box_id=box_id,
volume=int(default_volume),
max_volume=int(max_volume),
)
resume_raw = config.get("resume", {})
resume_raw = state_data.get("resume", {})
if isinstance(resume_raw, dict):
for uid, data in resume_raw.items():
if not isinstance(uid, str) or not isinstance(data, dict):
@@ -56,6 +79,19 @@ def _state_from_config(config: dict[str, Any]) -> RuntimeState:
return state
def _normalize_tag_paths(tags: dict[str, Any]) -> dict[str, Any]:
normalized = {}
for uid, tag in tags.items():
if not isinstance(tag, dict):
continue
path = tag.get("path")
if isinstance(path, str) and path.startswith("../media/"):
tag = dict(tag)
tag["path"] = path.replace("../media/", "media/", 1)
normalized[uid] = tag
return normalized
def _print_status(state: RuntimeState) -> None:
current = state.current_media if state.current_media is not None else "-"
current_file = state.current_file if state.current_file is not None else "-"
@@ -74,20 +110,48 @@ def _print_status(state: RuntimeState) -> None:
print(f"duration={current_duration}")
print(f"current_media={current}")
print(f"volume={state.volume}")
print(f"max_volume={state.max_volume}")
def main() -> int:
parser = build_parser()
args = parser.parse_args()
storage = JsonStorage(args.config)
config = storage.load_config()
state = _state_from_config(config)
config_path = Path(args.config)
if not config_path.is_absolute():
config_path = (Path(__file__).resolve().parent / config_path).resolve()
base_dir = Path(__file__).resolve().parent / "data"
base_dir.mkdir(parents=True, exist_ok=True)
(base_dir / "media").mkdir(parents=True, exist_ok=True)
box_store = BoxStore(base_dir)
box_meta = box_store.load()
if "box_id" not in box_meta:
box_meta["box_id"] = generate_box_id()
if "settings" not in box_meta:
box_meta["settings"] = {}
if "tags" not in box_meta:
legacy_storage = JsonStorage(str(config_path))
legacy_config = legacy_storage.load_config()
if isinstance(legacy_config.get("tags"), dict):
box_meta["tags"] = _normalize_tag_paths(legacy_config.get("tags", {}))
if isinstance(box_meta.get("tags"), dict):
box_meta["tags"] = _normalize_tag_paths(box_meta.get("tags", {}))
box_store.save(box_meta)
state_store = StateStore(base_dir)
state_data = state_store.load()
state = _state_from_data(box_meta, state_data)
secrets = SecretStore(base_dir, state.box_id)
wifi_backend = MockWifiBackend(state_store, secrets)
spotify_backend = MockSpotifyBackend(secrets)
lifecycle = NetworkLifecycle(wifi_backend)
controller = BoxController(
state=state,
player=MockPlayer(),
config=config,
config_path=args.config,
config=box_meta,
config_path=str(box_store.path),
)
if args.command == "status":
@@ -97,26 +161,31 @@ def main() -> int:
if args.command == "simulate-nfc":
controller.handle_nfc_on(args.uid)
_print_status(controller.state)
config["resume"] = _resume_to_json(controller.state)
storage.save_config(config)
_persist_resume(controller, state_store)
return 0
if args.command == "simulate-nfc-off":
controller.handle_nfc_off(args.uid)
_print_status(controller.state)
config["resume"] = _resume_to_json(controller.state)
storage.save_config(config)
_persist_resume(controller, state_store)
return 0
if args.command == "tick":
controller.tick(args.seconds)
_print_status(controller.state)
config["resume"] = _resume_to_json(controller.state)
storage.save_config(config)
_persist_resume(controller, state_store)
return 0
if args.command == "run":
_run_session(controller, storage, config)
asyncio.run(
_run_with_api(
controller,
state_store,
lifecycle,
wifi_backend,
spotify_backend,
)
)
return 0
return 0
@@ -129,12 +198,85 @@ def _resume_to_json(state: RuntimeState) -> dict[str, dict[str, int]]:
}
def _run_session(controller: BoxController, storage: JsonStorage, config: dict[str, Any]) -> None:
def _stdin_is_foreground() -> bool:
try:
if not sys.stdin.isatty():
return False
return os.getpgrp() == os.tcgetpgrp(sys.stdin.fileno())
except OSError:
return False
async def _run_with_api(
controller: BoxController,
storage: StateStore,
lifecycle: NetworkLifecycle,
wifi_backend: MockWifiBackend,
spotify_backend: MockSpotifyBackend,
) -> None:
app = create_app(
controller,
lifecycle,
wifi_backend,
spotify_backend,
lambda: _persist_resume(controller, storage),
)
api_server = uvicorn.Server(
uvicorn.Config(
app,
host="0.0.0.0",
port=8000,
log_level="warning",
access_log=False,
)
)
setup_app = create_setup_app(lifecycle, wifi_backend)
setup_server = uvicorn.Server(
uvicorn.Config(
setup_app,
host="127.0.0.1",
port=9000,
log_level="warning",
access_log=False,
)
)
stop_event = asyncio.Event()
tick_task = asyncio.create_task(
_auto_tick_loop(controller, storage, stop_event)
)
api_task = asyncio.create_task(api_server.serve())
setup_task = asyncio.create_task(setup_server.serve())
try:
if _stdin_is_foreground():
await _run_session(controller, storage)
api_server.should_exit = True
setup_server.should_exit = True
await api_task
except asyncio.CancelledError:
api_server.should_exit = True
setup_server.should_exit = True
finally:
stop_event.set()
tick_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await tick_task
setup_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await setup_task
async def _run_session(
controller: BoxController, storage: StateStore
) -> None:
print("Klangkiste Box - Interactive Mode")
print("Type 'help' for commands.")
while True:
try:
raw = input("> ").strip()
raw = await asyncio.to_thread(input, "> ")
raw = raw.strip()
except asyncio.CancelledError:
break
except EOFError:
print()
break
@@ -145,6 +287,9 @@ def _run_session(controller: BoxController, storage: JsonStorage, config: dict[s
args = parts[1:]
if command in {"exit", "quit"}:
if controller.state.active_uid is not None:
controller.handle_nfc_off(controller.state.active_uid)
_persist_resume(controller, storage)
break
if command == "help":
_print_help()
@@ -158,7 +303,7 @@ def _run_session(controller: BoxController, storage: JsonStorage, config: dict[s
continue
controller.handle_nfc_on(args[0])
_print_status(controller.state)
_persist_resume(controller, storage, config)
_persist_resume(controller, storage)
continue
if command == "nfc-off":
if len(args) != 1:
@@ -166,7 +311,7 @@ def _run_session(controller: BoxController, storage: JsonStorage, config: dict[s
continue
controller.handle_nfc_off(args[0])
_print_status(controller.state)
_persist_resume(controller, storage, config)
_persist_resume(controller, storage)
continue
if command == "tick":
if len(args) != 1:
@@ -179,17 +324,52 @@ def _run_session(controller: BoxController, storage: JsonStorage, config: dict[s
continue
controller.tick(seconds)
_print_status(controller.state)
_persist_resume(controller, storage, config)
_persist_resume(controller, storage)
continue
if command in {"play", "pause"}:
controller.play_pause()
_print_status(controller.state)
continue
if command in {"next", "vor"}:
controller.next_track()
_print_status(controller.state)
_persist_resume(controller, storage)
continue
if command in {"prev", "zurueck"}:
controller.prev_track()
_print_status(controller.state)
_persist_resume(controller, storage)
continue
if command in {"vol-up", "lauter"}:
controller.volume_up()
_print_status(controller.state)
continue
if command in {"vol-down", "leiser"}:
controller.volume_down()
_print_status(controller.state)
continue
print(f"error: unknown command '{command}'")
def _persist_resume(
controller: BoxController, storage: JsonStorage, config: dict[str, Any]
async def _auto_tick_loop(
controller: BoxController,
storage: StateStore,
stop_event: asyncio.Event,
) -> None:
config["resume"] = _resume_to_json(controller.state)
storage.save_config(config)
while not stop_event.is_set():
await asyncio.sleep(1)
if controller.state.state == BoxState.PLAYING:
controller.tick(1)
_persist_resume(controller, storage)
def _persist_resume(
controller: BoxController, storage: StateStore
) -> None:
data = storage.load()
data["resume"] = _resume_to_json(controller.state)
storage.save(data)
def _print_help() -> None:
@@ -197,6 +377,11 @@ def _print_help() -> None:
print(" nfc <UID> simulate NFC tag detection")
print(" nfc-off <UID> simulate NFC tag removal")
print(" tick <seconds> simulate time progression")
print(" play | pause toggle PLAYING/PAUSED")
print(" next | vor jump to next file")
print(" prev | zurueck jump to previous file")
print(" vol-up | lauter increase volume")
print(" vol-down | leiser decrease volume")
print(" status show current status")
print(" help show this help")
print(" exit | quit leave interactive mode")
box/media/book_6/CD1/Titel A.mp3
View File
box/media/book_6/CD1/Titel B.mp3
View File
box/media/book_6/CD2/Titel A.mp3
View File
box/media/book_6/CD2/Titel B.mp3
View File
box/requirements.txt
+4
View File
@@ -0,0 +1,4 @@
fastapi==0.115.6
uvicorn==0.30.6
pydantic==2.9.2
cryptography==43.0.3
box/schemas/box_config.schema.json
+3
View File
@@ -41,6 +41,9 @@
},
"additionalProperties": false
}
},
"server_reachable": {
"type": "boolean"
}
},
"additionalProperties": false
box/setup/__init__.py
View File
box/setup/webui.py
+71
View File
@@ -0,0 +1,71 @@
"""Setup Web UI for WiFi configuration (mock)."""
from __future__ import annotations
from fastapi import FastAPI, Form
from fastapi.responses import HTMLResponse
from core.lifecycle import NetworkLifecycle, WifiState
from wifi.base import WifiBackend
def create_setup_app(lifecycle: NetworkLifecycle, wifi_backend: WifiBackend) -> FastAPI:
app = FastAPI()
@app.get("/setup", response_class=HTMLResponse)
def setup_page() -> str:
snapshot = lifecycle.snapshot()
networks = wifi_backend.scan_available_networks()
options = "".join(f"<option value='{n}'>{n}</option>" for n in networks)
status = snapshot.wifi_state.value
connected = snapshot.connected_ssid or "-"
if snapshot.wifi_state != WifiState.WIFI_UNCONFIGURED:
return _render_readonly(status, connected)
return _render_form(status, connected, options)
@app.post("/setup", response_class=HTMLResponse)
def setup_submit(ssid: str = Form(...), password: str = Form("")) -> str:
wifi_backend.add_profile(ssid=ssid, password=password, priority=10)
snapshot = lifecycle.snapshot()
status = snapshot.wifi_state.value
connected = snapshot.connected_ssid or "-"
return _render_readonly(status, connected)
return app
def _render_form(status: str, connected: str, options: str) -> str:
return f"""
<!doctype html>
<html>
<head><meta charset="utf-8"><title>Klangkiste Setup</title></head>
<body>
<h1>WLAN Setup</h1>
<p>Status: {status}</p>
<p>Connected: {connected}</p>
<form method="post" action="/setup">
<label>SSID</label>
<select name="ssid">{options}</select>
<br />
<label>Password</label>
<input type="password" name="password" />
<br />
<button type="submit">Verbinden</button>
</form>
</body>
</html>
"""
def _render_readonly(status: str, connected: str) -> str:
return f"""
<!doctype html>
<html>
<head><meta charset="utf-8"><title>Klangkiste Setup</title></head>
<body>
<h1>WLAN Status</h1>
<p>Status: {status}</p>
<p>Connected: {connected}</p>
</body>
</html>
"""
box/spotify/__init__.py
View File
box/spotify/base.py
+23
View File
@@ -0,0 +1,23 @@
"""Spotify backend interface (logical)."""
from __future__ import annotations
from typing import Protocol
class SpotifyBackend(Protocol):
def set_tokens(
self,
access_token: str,
refresh_token: str,
expires_at: int,
account_id: str | None = None,
) -> None:
raise NotImplementedError
def clear(self) -> None:
raise NotImplementedError
@property
def status(self) -> str:
raise NotImplementedError
box/spotify/manager.py
+54
View File
@@ -0,0 +1,54 @@
"""Logical Spotify token manager (mock, no server dependency)."""
from __future__ import annotations
from typing import Any
from storage.secrets_store import SecretsStore
class SpotifyManager:
def __init__(self, secrets: SecretsStore) -> None:
self._secrets = secrets
self._tokens: dict[str, Any] = {}
self._load()
def set_tokens(
self,
access_token: str,
refresh_token: str,
expires_at: int,
account_id: str | None = None,
) -> None:
self._tokens = {
"access_token": access_token,
"refresh_token": refresh_token,
"expires_at": expires_at,
}
if account_id is not None:
self._tokens["account_id"] = account_id
self._save()
def clear(self) -> None:
self._tokens = {}
secrets = self._secrets.load_secrets()
if "spotify" in secrets:
secrets.pop("spotify")
self._secrets.save_secrets(secrets)
@property
def status(self) -> str:
if not self._tokens:
return "NOT_CONFIGURED"
return "READY"
def _load(self) -> None:
secrets = self._secrets.load_secrets()
tokens = secrets.get("spotify")
if isinstance(tokens, dict):
self._tokens = dict(tokens)
def _save(self) -> None:
secrets = self._secrets.load_secrets()
secrets["spotify"] = dict(self._tokens)
self._secrets.save_secrets(secrets)
box/spotify/mock.py
+55
View File
@@ -0,0 +1,55 @@
"""Mock Spotify backend (logical)."""
from __future__ import annotations
from typing import Any
from storage.secret_store import SecretStore
from spotify.base import SpotifyBackend
class MockSpotifyBackend(SpotifyBackend):
def __init__(self, secret_store: SecretStore) -> None:
self._secret_store = secret_store
self._tokens: dict[str, Any] = {}
self._load()
def set_tokens(
self,
access_token: str,
refresh_token: str,
expires_at: int,
account_id: str | None = None,
) -> None:
self._tokens = {
"access_token": access_token,
"refresh_token": refresh_token,
"expires_at": expires_at,
}
if account_id is not None:
self._tokens["account_id"] = account_id
self._save()
def clear(self) -> None:
self._tokens = {}
secrets = self._secret_store.load_secrets()
if "spotify" in secrets:
secrets.pop("spotify")
self._secret_store.save_secrets(secrets)
@property
def status(self) -> str:
if not self._tokens:
return "NOT_CONFIGURED"
return "READY"
def _load(self) -> None:
secrets = self._secret_store.load_secrets()
tokens = secrets.get("spotify")
if isinstance(tokens, dict):
self._tokens = dict(tokens)
def _save(self) -> None:
secrets = self._secret_store.load_secrets()
secrets["spotify"] = dict(self._tokens)
self._secret_store.save_secrets(secrets)
box/storage/box_storage.py
+52
View File
@@ -0,0 +1,52 @@
"""Persistent storage for Box metadata and runtime state."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
class BoxStorage:
def __init__(self, base_dir: str | None = None) -> None:
default_dir = Path(__file__).resolve().parents[1] / "data"
self._base_dir = Path(base_dir) if base_dir else default_dir
self._box_path = self._base_dir / "box.json"
self._state_path = self._base_dir / "state.json"
self._media_dir = self._base_dir / "media"
@property
def base_dir(self) -> Path:
return self._base_dir
@property
def media_dir(self) -> Path:
return self._media_dir
@property
def box_path(self) -> Path:
return self._box_path
@property
def state_path(self) -> Path:
return self._state_path
def ensure_layout(self) -> None:
self._base_dir.mkdir(parents=True, exist_ok=True)
self._media_dir.mkdir(parents=True, exist_ok=True)
def load_box_meta(self) -> dict[str, Any]:
if not self._box_path.exists():
return {}
return json.loads(self._box_path.read_text(encoding="utf-8"))
def save_box_meta(self, data: dict[str, Any]) -> None:
self._box_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
def load_state(self) -> dict[str, Any]:
if not self._state_path.exists():
return {}
return json.loads(self._state_path.read_text(encoding="utf-8"))
def save_state(self, data: dict[str, Any]) -> None:
self._state_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
box/storage/box_store.py
+34
View File
@@ -0,0 +1,34 @@
"""Box metadata storage (app-level, within box/data)."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from core.identity import generate_box_id
class BoxStore:
def __init__(self, base_dir: Path) -> None:
self._base_dir = base_dir
self._path = self._base_dir / "box.json"
@property
def path(self) -> Path:
return self._path
def load(self) -> dict[str, Any]:
if not self._path.exists():
return {}
return json.loads(self._path.read_text(encoding="utf-8"))
def save(self, data: dict[str, Any]) -> None:
self._path.write_text(json.dumps(data, indent=2), encoding="utf-8")
def ensure_box_id(self) -> str:
data = self.load()
if "box_id" not in data:
data["box_id"] = generate_box_id()
self.save(data)
return str(data["box_id"])
box/storage/secret_store.py
+57
View File
@@ -0,0 +1,57 @@
"""Encrypted secret storage for sensitive Box data (app-level)."""
from __future__ import annotations
import base64
import json
from hashlib import pbkdf2_hmac
from pathlib import Path
from typing import Any
from cryptography.fernet import Fernet, InvalidToken
STATIC_SALT = b"klangkiste-static-salt-v1"
KDF_ITERATIONS = 200_000
class SecretStore:
def __init__(self, base_dir: Path, box_id: str) -> None:
self._path = base_dir / "secrets.enc"
self._box_id = box_id
def load_secrets(self) -> dict[str, Any]:
if not self._path.exists():
return {}
token = self._path.read_bytes()
fernet = Fernet(self._derive_key())
try:
payload = fernet.decrypt(token)
except InvalidToken:
return {}
return json.loads(payload.decode("utf-8"))
def save_secrets(self, data: dict[str, Any]) -> None:
fernet = Fernet(self._derive_key())
payload = json.dumps(data, indent=2).encode("utf-8")
token = fernet.encrypt(payload)
self._path.write_bytes(token)
def clear_wifi_secrets(self) -> None:
data = self.load_secrets()
if "wifi_profiles" in data:
data.pop("wifi_profiles")
self.save_secrets(data)
def clear_all(self) -> None:
if self._path.exists():
self._path.unlink()
def _derive_key(self) -> bytes:
raw = pbkdf2_hmac(
"sha256",
self._box_id.encode("utf-8"),
STATIC_SALT,
KDF_ITERATIONS,
dklen=32,
)
return base64.urlsafe_b64encode(raw)
box/storage/secrets_store.py
+57
View File
@@ -0,0 +1,57 @@
"""Encrypted secret storage for sensitive Box data."""
from __future__ import annotations
import base64
import json
from hashlib import pbkdf2_hmac
from pathlib import Path
from typing import Any
from cryptography.fernet import Fernet, InvalidToken
STATIC_SALT = b"klangkiste-static-salt-v1"
KDF_ITERATIONS = 200_000
class SecretsStore:
def __init__(self, base_dir: Path, box_id: str) -> None:
self._path = base_dir / "secrets.enc"
self._box_id = box_id
def load_secrets(self) -> dict[str, Any]:
if not self._path.exists():
return {}
token = self._path.read_bytes()
fernet = Fernet(self._derive_key())
try:
payload = fernet.decrypt(token)
except InvalidToken:
return {}
return json.loads(payload.decode("utf-8"))
def save_secrets(self, data: dict[str, Any]) -> None:
fernet = Fernet(self._derive_key())
payload = json.dumps(data, indent=2).encode("utf-8")
token = fernet.encrypt(payload)
self._path.write_bytes(token)
def clear_wifi_secrets(self) -> None:
data = self.load_secrets()
if "wifi" in data:
data.pop("wifi")
self.save_secrets(data)
def clear_all(self) -> None:
if self._path.exists():
self._path.unlink()
def _derive_key(self) -> bytes:
raw = pbkdf2_hmac(
"sha256",
self._box_id.encode("utf-8"),
STATIC_SALT,
KDF_ITERATIONS,
dklen=32,
)
return base64.urlsafe_b64encode(raw)
box/storage/state_store.py
+24
View File
@@ -0,0 +1,24 @@
"""Runtime state storage (resume, wifi state)."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
class StateStore:
def __init__(self, base_dir: Path) -> None:
self._path = base_dir / "state.json"
@property
def path(self) -> Path:
return self._path
def load(self) -> dict[str, Any]:
if not self._path.exists():
return {}
return json.loads(self._path.read_text(encoding="utf-8"))
def save(self, data: dict[str, Any]) -> None:
self._path.write_text(json.dumps(data, indent=2), encoding="utf-8")
box/wifi/__init__.py
View File
box/wifi/base.py
+45
View File
@@ -0,0 +1,45 @@
"""WiFi backend interface (logical, no OS integration)."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol
@dataclass
class WifiProfile:
ssid: str
password: str
priority: int = 0
class WifiBackend(Protocol):
def scan_available_networks(self) -> list[str]:
raise NotImplementedError
def add_profile(self, ssid: str, password: str, priority: int = 0) -> None:
raise NotImplementedError
def remove_profile(self, ssid: str) -> None:
raise NotImplementedError
def reset(self) -> None:
raise NotImplementedError
def select_best_profile(self) -> WifiProfile | None:
raise NotImplementedError
def connect(self, profile: WifiProfile) -> None:
raise NotImplementedError
@property
def network_status(self) -> str:
raise NotImplementedError
@property
def connected_ssid(self) -> str | None:
raise NotImplementedError
@property
def profiles_count(self) -> int:
raise NotImplementedError
box/wifi/manager.py
+115
View File
@@ -0,0 +1,115 @@
"""Logical WiFi manager (mock, no OS integration)."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from storage.box_storage import BoxStorage
from storage.secrets_store import SecretsStore
@dataclass
class WifiProfile:
ssid: str
password: str
priority: int = 0
class WifiManager:
def __init__(self, storage: BoxStorage, secrets: SecretsStore) -> None:
self._storage = storage
self._secrets = secrets
self._profiles: list[WifiProfile] = []
self._connected_ssid: str | None = None
self._load_state()
def scan_available_networks(self) -> list[str]:
return []
def add_profile(self, ssid: str, password: str, priority: int = 0) -> None:
self._profiles = [p for p in self._profiles if p.ssid != ssid]
self._profiles.append(WifiProfile(ssid=ssid, password=password, priority=priority))
self._save_profiles()
self._auto_connect()
def remove_profile(self, ssid: str) -> None:
self._profiles = [p for p in self._profiles if p.ssid != ssid]
if self._connected_ssid == ssid:
self._connected_ssid = None
self._save_profiles()
self._save_state()
self._auto_connect()
def reset(self) -> None:
self._profiles = []
self._connected_ssid = None
self._secrets.clear_wifi_secrets()
self._save_state()
def select_best_profile(self) -> WifiProfile | None:
if not self._profiles:
return None
return sorted(self._profiles, key=lambda p: (-p.priority, p.ssid))[0]
def connect(self, profile: WifiProfile) -> None:
self._connected_ssid = profile.ssid
self._save_state()
@property
def network_status(self) -> str:
if not self._profiles:
return "NOT_CONFIGURED"
if self._connected_ssid:
return "CONNECTED"
return "DISCONNECTED"
@property
def connected_ssid(self) -> str | None:
return self._connected_ssid
@property
def profiles_count(self) -> int:
return len(self._profiles)
def _auto_connect(self) -> None:
profile = self.select_best_profile()
if profile is not None:
self.connect(profile)
def _load_state(self) -> None:
secrets = self._secrets.load_secrets()
profiles_raw = secrets.get("wifi_profiles", [])
if isinstance(profiles_raw, list):
for entry in profiles_raw:
if not isinstance(entry, dict):
continue
ssid = entry.get("ssid")
password = entry.get("password")
priority = entry.get("priority", 0)
if isinstance(ssid, str) and isinstance(password, str):
if not isinstance(priority, int):
priority = 0
self._profiles.append(
WifiProfile(ssid=ssid, password=password, priority=priority)
)
state = self._storage.load_state()
wifi_state = state.get("wifi", {}) if isinstance(state.get("wifi"), dict) else {}
connected = wifi_state.get("connected_ssid")
if isinstance(connected, str):
self._connected_ssid = connected
self._auto_connect()
def _save_profiles(self) -> None:
secrets = self._secrets.load_secrets()
secrets["wifi_profiles"] = [
{"ssid": p.ssid, "password": p.password, "priority": p.priority}
for p in self._profiles
]
self._secrets.save_secrets(secrets)
def _save_state(self) -> None:
state = self._storage.load_state()
state["wifi"] = {"connected_ssid": self._connected_ssid}
self._storage.save_state(state)
box/wifi/mock.py
+109
View File
@@ -0,0 +1,109 @@
"""Mock WiFi backend (logical simulation)."""
from __future__ import annotations
from typing import Any
from storage.state_store import StateStore
from storage.secret_store import SecretStore
from wifi.base import WifiBackend, WifiProfile
class MockWifiBackend(WifiBackend):
def __init__(self, state_store: StateStore, secret_store: SecretStore) -> None:
self._state_store = state_store
self._secret_store = secret_store
self._profiles: list[WifiProfile] = []
self._connected_ssid: str | None = None
self._available_networks = ["HomeWiFi", "Hotspot"]
self._load()
def scan_available_networks(self) -> list[str]:
return list(self._available_networks)
def add_profile(self, ssid: str, password: str, priority: int = 0) -> None:
self._profiles = [p for p in self._profiles if p.ssid != ssid]
self._profiles.append(WifiProfile(ssid=ssid, password=password, priority=priority))
self._save_profiles()
self._auto_connect()
def remove_profile(self, ssid: str) -> None:
self._profiles = [p for p in self._profiles if p.ssid != ssid]
if self._connected_ssid == ssid:
self._connected_ssid = None
self._save_profiles()
self._save_state()
self._auto_connect()
def reset(self) -> None:
self._profiles = []
self._connected_ssid = None
self._secret_store.clear_wifi_secrets()
self._save_state()
def select_best_profile(self) -> WifiProfile | None:
if not self._profiles:
return None
return sorted(self._profiles, key=lambda p: (-p.priority, p.ssid))[0]
def connect(self, profile: WifiProfile) -> None:
self._connected_ssid = profile.ssid
self._save_state()
@property
def network_status(self) -> str:
if not self._profiles:
return "WIFI_UNCONFIGURED"
if self._connected_ssid:
return "WIFI_ONLINE"
return "WIFI_OFFLINE"
@property
def connected_ssid(self) -> str | None:
return self._connected_ssid
@property
def profiles_count(self) -> int:
return len(self._profiles)
def _auto_connect(self) -> None:
profile = self.select_best_profile()
if profile is not None:
self.connect(profile)
def _load(self) -> None:
secrets = self._secret_store.load_secrets()
profiles_raw = secrets.get("wifi_profiles", [])
if isinstance(profiles_raw, list):
for entry in profiles_raw:
if not isinstance(entry, dict):
continue
ssid = entry.get("ssid")
password = entry.get("password")
priority = entry.get("priority", 0)
if isinstance(ssid, str) and isinstance(password, str):
if not isinstance(priority, int):
priority = 0
self._profiles.append(
WifiProfile(ssid=ssid, password=password, priority=priority)
)
state = self._state_store.load()
wifi_state = state.get("wifi", {}) if isinstance(state.get("wifi"), dict) else {}
connected = wifi_state.get("connected_ssid")
if isinstance(connected, str):
self._connected_ssid = connected
self._auto_connect()
def _save_profiles(self) -> None:
secrets = self._secret_store.load_secrets()
secrets["wifi_profiles"] = [
{"ssid": p.ssid, "password": p.password, "priority": p.priority}
for p in self._profiles
]
self._secret_store.save_secrets(secrets)
def _save_state(self) -> None:
state = self._state_store.load()
state["wifi"] = {"connected_ssid": self._connected_ssid}
self._state_store.save(state)