feat: config files, settings screen, hardware check
Build Android APK / build (push) Failing after 5m7s
Build Server .exe / build (push) Failing after 2m27s

- Add config.py with multi-source config (YAML, .env, env vars)
- Add .env.example and config.yaml.example
- Add settings API endpoints (GET/POST /api/v1/settings)
- Add status endpoint (GET /api/v1/status) with hardware subsystem check
- Add settings screen to web client (protected by API token)
- Add localization strings for settings UI
- Add tests for config module and new endpoints
- Update README with config documentation
This commit is contained in:
Imrayya
2026-07-01 13:32:08 +00:00
parent 8e2212f035
commit 871a3cc174
16 changed files with 1341 additions and 55 deletions
+13
View File
@@ -0,0 +1,13 @@
# PaperDash Configuration
# Copy this file to .env and fill in your values
# Dashboard PIN (4-digit code for the E-Ink client)
DASHBOARD_PIN=1234
# API Token (Bearer token for external scripts)
API_TOKEN=your-secret-token-here
# Server settings
HOST=0.0.0.0
PORT=8921
LOG_LEVEL=INFO
+176
View File
@@ -0,0 +1,176 @@
"""Configuration management for PaperDash.
Loads config from:
1. config.yaml (if exists)
2. .env file (if exists)
3. Environment variables (fallback)
"""
from __future__ import annotations
import logging
import os
import sys
from pathlib import Path
from typing import Any
logger = logging.getLogger("paperdash.config")
# Try to import yaml, fall back to simple parser
try:
import yaml
HAS_YAML = True
except ImportError:
HAS_YAML = False
class Config:
"""PaperDash configuration."""
def __init__(self) -> None:
self.dashboard_pin: str = ""
self.api_token: str = ""
self.host: str = "0.0.0.0"
self.port: int = 8921
self.log_level: str = "INFO"
def load(self) -> None:
"""Load configuration from all sources."""
# 1. Load config.yaml
config_file = self._find_config_file()
if config_file and config_file.exists():
self._load_yaml(config_file)
# 2. Load .env file
env_file = self._find_env_file()
if env_file and env_file.exists():
self._load_env(env_file)
# 3. Environment variables override everything
self._load_from_env()
# Validate
if not self.dashboard_pin:
logger.warning("DASHBOARD_PIN not set — WebSocket auth will be disabled")
if not self.api_token:
logger.warning("API_TOKEN not set — /api/v1/notify will reject all requests")
def _get_base_path(self) -> Path:
"""Get base path (exe directory or cwd)."""
if getattr(sys, 'frozen', False):
return Path(sys.executable).parent
return Path.cwd()
def _find_config_file(self) -> Path | None:
"""Find config.yaml relative to the executable or working directory."""
base = self._get_base_path()
for candidate in [base] + list(base.parents):
config_path = candidate / "config.yaml"
if config_path.exists():
return config_path
return None
def _find_env_file(self) -> Path | None:
"""Find .env file relative to the executable or working directory."""
base = self._get_base_path()
for candidate in [base] + list(base.parents):
env_path = candidate / ".env"
if env_path.exists():
return env_path
return None
def _load_yaml(self, path: Path) -> None:
"""Load configuration from YAML file."""
if not HAS_YAML:
logger.warning("PyYAML not installed — skipping config.yaml")
return
try:
with open(path) as f:
data = yaml.safe_load(f) or {} # type: ignore[possibly-undefined]
self.dashboard_pin = data.get("dashboard_pin", self.dashboard_pin)
self.api_token = data.get("api_token", self.api_token)
self.host = data.get("host", self.host)
self.port = data.get("port", self.port)
self.log_level = data.get("log_level", self.log_level).upper()
logger.info(f"Loaded config from {path}")
except Exception as e:
logger.error(f"Failed to load config from {path}: {e}")
def _load_env(self, path: Path) -> None:
"""Load configuration from .env file."""
try:
with open(path) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#") and "=" in line:
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip("'\"")
match key:
case "DASHBOARD_PIN":
self.dashboard_pin = value
case "API_TOKEN":
self.api_token = value
case "HOST":
self.host = value
case "PORT":
self.port = int(value)
case "LOG_LEVEL":
self.log_level = value.upper()
logger.info(f"Loaded config from {path}")
except Exception as e:
logger.error(f"Failed to load .env from {path}: {e}")
def _load_from_env(self) -> None:
"""Load configuration from environment variables (highest priority)."""
self.dashboard_pin = os.environ.get("DASHBOARD_PIN", self.dashboard_pin)
self.api_token = os.environ.get("API_TOKEN", self.api_token)
self.host = os.environ.get("HOST", self.host)
try:
self.port = int(os.environ.get("PORT", self.port))
except ValueError:
logger.warning("Invalid PORT environment variable")
self.log_level = os.environ.get("LOG_LEVEL", self.log_level).upper()
def save(self, path: Path | None = None) -> None:
"""Save configuration to YAML file."""
if path is None:
path = self._find_config_file() or Path.cwd() / "config.yaml"
if not HAS_YAML:
logger.error("PyYAML not installed — cannot save config")
return
data = {
"dashboard_pin": self.dashboard_pin,
"api_token": self.api_token,
"host": self.host,
"port": self.port,
"log_level": self.log_level,
}
try:
with open(path, "w") as f:
yaml.dump(data, f, default_flow_style=False) # type: ignore[possibly-undefined]
logger.info(f"Saved config to {path}")
except Exception as e:
logger.error(f"Failed to save config to {path}: {e}")
def to_dict(self) -> dict[str, Any]:
"""Return configuration as a dictionary (for API responses)."""
return {
"dashboard_pin": self.dashboard_pin,
"api_token": self.api_token,
"host": self.host,
"port": self.port,
"log_level": self.log_level,
}
# Global config instance
config = Config()
+166 -10
View File
@@ -17,6 +17,7 @@ from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from .audio_mixer import AudioMixerManager
from .config import config
from .media import MediaManager
from .notifications import NotificationManager, Notification
from .telemetry import TelemetryManager
@@ -24,9 +25,78 @@ from .telemetry import TelemetryManager
logger = logging.getLogger("paperdash")
# ---------------------------------------------------------------------------
# Hardware Subsystem Check
# ---------------------------------------------------------------------------
def check_subsystems() -> dict[str, bool]:
"""Check which hardware subsystems are available."""
results: dict[str, bool] = {}
# Check Voicemeeter
try:
import voicemeeter_api # noqa: F401 # type: ignore
results["voicemeeter"] = True
except ImportError:
results["voicemeeter"] = False
# Check Windows Audio (pycaw)
try:
import pycaw # noqa: F401 # type: ignore
results["windows_audio"] = True
except ImportError:
results["windows_audio"] = False
# Check Windows Media (winsdk)
try:
import winsdk # noqa: F401 # type: ignore
results["windows_media"] = True
except ImportError:
results["windows_media"] = False
# Check HWiNFO (pyhwinfo)
try:
import pyhwinfo # noqa: F401 # type: ignore
results["hwinfo"] = True
except ImportError:
results["hwinfo"] = False
return results
# ---------------------------------------------------------------------------
# Lifespan
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Start and stop background managers."""
# Load configuration
config.load()
# Configure logging
logging.basicConfig(
level=getattr(logging, config.log_level, logging.INFO),
format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
)
# Check hardware subsystems
subsystems = check_subsystems()
app.state.subsystems = subsystems
logger.info(f"Hardware subsystems: {subsystems}")
# Log configuration status
if config.dashboard_pin:
logger.info("Dashboard PIN is configured")
else:
logger.warning("Dashboard PIN is NOT configured — WebSocket auth disabled")
if config.api_token:
logger.info("API token is configured")
else:
logger.warning("API token is NOT configured — /api/v1/notify disabled")
# Start managers
app.state.media = MediaManager()
app.state.audio = AudioMixerManager()
@@ -59,7 +129,7 @@ app = FastAPI(
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Local network only — fine for this use case
allow_origins=["*"], # Local network only — fine for this use case # noqa: B008
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
@@ -78,7 +148,7 @@ class WebSocketManager:
self.connections: list[WebSocket] = []
async def connect(self, ws: WebSocket, pin: str) -> bool:
expected_pin = os.environ.get("DASHBOARD_PIN", "")
expected_pin = config.dashboard_pin
if not expected_pin or pin != expected_pin:
await ws.close(code=4001, reason="Invalid PIN")
return False
@@ -162,7 +232,7 @@ async def handle_client_command(data: dict[str, Any]) -> None:
def _verify_token(authorization: str | None) -> bool:
"""Verify Bearer token from Authorization header."""
expected = os.environ.get("API_TOKEN", "")
expected = config.api_token
if not expected:
return False
if not authorization or not authorization.startswith("Bearer "):
@@ -178,6 +248,7 @@ async def get_state():
"audio": app.state.audio.get_state(),
"telemetry": app.state.telemetry.get_state(),
"notifications": app.state.notifications.get_active(),
"subsystems": app.state.subsystems,
}
@@ -195,6 +266,83 @@ async def create_notification(
return {"id": created.id, "status": "created"}
# ---------------------------------------------------------------------------
# Settings (Protected by API Token)
# ---------------------------------------------------------------------------
class SettingsUpdate:
"""Settings update payload."""
def __init__(
self,
dashboard_pin: str | None = None,
api_token: str | None = None,
host: str | None = None,
port: int | None = None,
log_level: str | None = None,
):
self.dashboard_pin = dashboard_pin
self.api_token = api_token
self.host = host
self.port = port
self.log_level = log_level
@app.get("/api/v1/settings")
async def get_settings(
authorization: str | None = None,
):
"""Get current settings (requires API token)."""
if not _verify_token(authorization):
raise HTTPException(status_code=401, detail="Invalid or missing API token")
return config.to_dict()
@app.post("/api/v1/settings")
async def update_settings(
settings: SettingsUpdate,
authorization: str | None = None,
):
"""Update settings (requires API token)."""
if not _verify_token(authorization):
raise HTTPException(status_code=401, detail="Invalid or missing API token")
if settings.dashboard_pin is not None:
config.dashboard_pin = settings.dashboard_pin
if settings.api_token is not None:
config.api_token = settings.api_token
if settings.host is not None:
config.host = settings.host
if settings.port is not None:
config.port = settings.port
if settings.log_level is not None:
config.log_level = settings.log_level.upper()
# Save to config file
config.save()
logger.info("Settings updated")
return config.to_dict()
# ---------------------------------------------------------------------------
# Hardware Status
# ---------------------------------------------------------------------------
@app.get("/api/v1/status")
async def get_status():
"""Get server status including hardware subsystem availability."""
return {
"subsystems": app.state.subsystems,
"config": {
"pin_configured": bool(config.dashboard_pin),
"token_configured": bool(config.api_token),
},
}
# ---------------------------------------------------------------------------
# Health
# ---------------------------------------------------------------------------
@@ -206,16 +354,24 @@ async def health():
# ---------------------------------------------------------------------------
# Client HTML (served directly for Hermit WebView)
# Client HTML (served directly for WebView)
# ---------------------------------------------------------------------------
@app.get("/", response_class=HTMLResponse)
async def serve_client():
"""Serve the E-Ink dashboard client."""
client_dir = os.path.join(os.path.dirname(__file__), "..", "client")
index_path = os.path.join(client_dir, "index.html")
if os.path.exists(index_path):
with open(index_path, "r") as f:
return f.read()
return HTMLResponse("<h1>PaperDash</h1><p>Client files not found in client/</p>")
# Try multiple locations for the client files
possible_paths = [
os.path.join(os.path.dirname(__file__), "..", "client"),
os.path.join(os.path.dirname(__file__), "..", "..", "client"),
os.path.join(os.path.dirname(__file__), "client"),
]
for client_dir in possible_paths:
index_path = os.path.join(client_dir, "index.html")
if os.path.exists(index_path):
with open(index_path) as f:
return f.read()
return HTMLResponse("<h1>PaperDash</h1><p>Client files not found.</p>")
+1
View File
@@ -2,6 +2,7 @@
fastapi>=0.100.0
uvicorn[standard]>=0.23.0
pydantic>=2.0.0
pyyaml>=6.0
# Windows APIs
winsdk>=1.0.0
+134
View File
@@ -0,0 +1,134 @@
"""Tests for the configuration module."""
from __future__ import annotations
import os
import tempfile
from pathlib import Path
import pytest
from ..config import Config
@pytest.fixture
def config() -> Config:
return Config()
class TestConfigDefaults:
def test_defaults(self, config: Config) -> None:
assert config.dashboard_pin == ""
assert config.api_token == ""
assert config.host == "0.0.0.0"
assert config.port == 8921
assert config.log_level == "INFO"
class TestConfigLoad:
def test_load_from_env(self, config: Config) -> None:
os.environ["DASHBOARD_PIN"] = "5678"
os.environ["API_TOKEN"] = "test-token"
os.environ["HOST"] = "127.0.0.1"
os.environ["PORT"] = "9000"
os.environ["LOG_LEVEL"] = "DEBUG"
config.load()
assert config.dashboard_pin == "5678"
assert config.api_token == "test-token"
assert config.host == "127.0.0.1"
assert config.port == 9000
assert config.log_level == "DEBUG"
# Cleanup
del os.environ["DASHBOARD_PIN"]
del os.environ["API_TOKEN"]
del os.environ["HOST"]
del os.environ["PORT"]
del os.environ["LOG_LEVEL"]
def test_load_from_env_file(self, config: Config) -> None:
with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f:
f.write("DASHBOARD_PIN=1111\n")
f.write("API_TOKEN=env-token\n")
f.write("# Comment\n")
f.write("HOST=0.0.0.0\n")
f.write("PORT=8921\n")
env_file = f.name
try:
config._load_env(Path(env_file))
assert config.dashboard_pin == "1111"
assert config.api_token == "env-token"
finally:
os.unlink(env_file)
def test_load_from_yaml_file(self, config: Config) -> None:
"""Test YAML loading (requires PyYAML)."""
try:
import yaml # noqa: F401
except ImportError:
pytest.skip("PyYAML not installed")
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump( # type: ignore[possibly-undefined]
{
"dashboard_pin": "2222",
"api_token": "yaml-token",
"host": "0.0.0.0",
"port": 8921,
"log_level": "INFO",
},
f,
)
yaml_file = f.name
try:
config._load_yaml(Path(yaml_file))
assert config.dashboard_pin == "2222"
assert config.api_token == "yaml-token"
finally:
os.unlink(yaml_file)
class TestConfigSave:
def test_save_to_yaml(self, config: Config) -> None:
"""Test YAML saving (requires PyYAML)."""
try:
import yaml # noqa: F401
except ImportError:
pytest.skip("PyYAML not installed")
config.dashboard_pin = "3333"
config.api_token = "save-token"
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
save_path = f.name
try:
config.save(Path(save_path))
with open(save_path) as f:
data = yaml.safe_load(f) # type: ignore[possibly-undefined]
assert data["dashboard_pin"] == "3333"
assert data["api_token"] == "save-token"
finally:
os.unlink(save_path)
class TestConfigToDict:
def test_to_dict(self, config: Config) -> None:
config.dashboard_pin = "4444"
config.api_token = "dict-token"
config.host = "127.0.0.1"
config.port = 9000
config.log_level = "DEBUG"
d = config.to_dict()
assert d["dashboard_pin"] == "4444"
assert d["api_token"] == "dict-token"
assert d["host"] == "127.0.0.1"
assert d["port"] == 9000
assert d["log_level"] == "DEBUG"
+55
View File
@@ -92,3 +92,58 @@ class TestNotifyEndpoint:
assert response.status_code == 401
del os.environ["API_TOKEN"]
class TestSettingsEndpoint:
def test_get_settings_without_token_returns_401(self, client: TestClient) -> None:
response = client.get("/api/v1/settings")
assert response.status_code == 401
def test_get_settings_with_valid_token(self, client: TestClient) -> None:
os.environ["API_TOKEN"] = "settings-token"
response = client.get(
"/api/v1/settings",
headers={"Authorization": "Bearer settings-token"},
)
assert response.status_code == 200
data = response.json()
assert "dashboard_pin" in data
assert "api_token" in data
assert "host" in data
assert "port" in data
del os.environ["API_TOKEN"]
def test_update_settings(self, client: TestClient) -> None:
os.environ["API_TOKEN"] = "update-token"
response = client.post(
"/api/v1/settings",
json={
"dashboard_pin": "9999",
"api_token": "new-token",
},
headers={"Authorization": "Bearer update-token"},
)
assert response.status_code == 200
data = response.json()
assert data["dashboard_pin"] == "9999"
assert data["api_token"] == "new-token"
del os.environ["API_TOKEN"]
class TestStatusEndpoint:
def test_status_returns_subsystems(self, client: TestClient) -> None:
# Ensure subsystems are initialized
if not hasattr(app.state, "subsystems"):
app.state.subsystems = {}
response = client.get("/api/v1/status")
assert response.status_code == 200
data = response.json()
assert "subsystems" in data
assert "config" in data
assert "pin_configured" in data["config"]
assert "token_configured" in data["config"]