Centralize configuration into config.yaml

- All settings moved to config.yaml
- configs/load.py reads from config.yaml with env var overrides
- Environment variables still work for backward compatibility
- Added pyyaml to requirements
This commit is contained in:
2026-06-10 10:19:41 +08:00
parent 4485cbf702
commit c8fd057129
10 changed files with 123 additions and 40 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
from __future__ import annotations from __future__ import annotations
from app.diagnostics.crash_handler import enable_crash_handler from app.diagnostics.crash_handler import enable_crash_handler
from configs.default import LOG_DIR from configs.load import LOG_DIR
def startup() -> None: def startup() -> None:
+1 -1
View File
@@ -4,7 +4,7 @@ from pathlib import Path
from loguru import logger from loguru import logger
from configs.default import LOG_DIR, LOG_RETENTION, LOG_ROTATION from configs.load import LOG_DIR, LOG_RETENTION, LOG_ROTATION
def setup_logging() -> None: def setup_logging() -> None:
+5 -1
View File
@@ -13,7 +13,11 @@ from app.core.lifecycle import startup
from app.signaling.websocket_server import main as serve from app.signaling.websocket_server import main as serve
if __name__ == "__main__": def main():
startup() startup()
logger.info("Starting server...") logger.info("Starting server...")
asyncio.run(serve()) asyncio.run(serve())
if __name__ == "__main__":
main()
+1 -1
View File
@@ -7,7 +7,7 @@ import websockets
from loguru import logger from loguru import logger
from app.webrtc.peer_session import PeerSession from app.webrtc.peer_session import PeerSession
from configs.default import WS_HOST, WS_MAX_SIZE, WS_PORT from configs.load import WS_HOST, WS_MAX_SIZE, WS_PORT
async def handle_client(websocket): async def handle_client(websocket):
+1 -1
View File
@@ -10,7 +10,7 @@ from loguru import logger
from app.audio.rep_announcer import RepAnnouncer from app.audio.rep_announcer import RepAnnouncer
from app.exercises.dead_bug.detector import DeadBugDetector from app.exercises.dead_bug.detector import DeadBugDetector
from app.rendering.window_display import close_window, is_esc_pressed, show_frame from app.rendering.window_display import close_window, is_esc_pressed, show_frame
from configs.default import ( from configs.load import (
EXTENSION_CONFIRM_FRAMES, EXTENSION_CONFIRM_FRAMES,
MODEL_PATH, MODEL_PATH,
PREFER_GPU, PREFER_GPU,
+29
View File
@@ -0,0 +1,29 @@
# PoseFit Server Configuration
# ============================
server:
host: "0.0.0.0"
port: 8765
max_ws_size: 10485760 # 10 MB
video:
process_every_n_frames: 1
model:
path: "" # empty = auto-detect pose_models/pose_landmarker_full.task
prefer_gpu: true
dead_bug:
visibility_threshold: 0.45
extension_confirm_frames: 4
reset_confirm_frames: 3
audio:
rep_announcer_enabled: true
rep_announcer_rate: 185
rep_announcer_volume: 1.0
logging:
dir: logs
rotation: "20 MB"
retention: "14 days"
-32
View File
@@ -1,32 +0,0 @@
from __future__ import annotations
import os
from pathlib import Path
# ── Server ──────────────────────────────────────────────────────────────────
WS_HOST = os.getenv("POSEFIT_WS_HOST", "0.0.0.0")
WS_PORT = int(os.getenv("POSEFIT_WS_PORT", "8765"))
WS_MAX_SIZE = int(os.getenv("POSEFIT_WS_MAX_SIZE", str(10 * 1024 * 1024)))
# ── Video processing ────────────────────────────────────────────────────────
PROCESS_EVERY_N_FRAMES = max(1, int(os.getenv("POSEFIT_PROCESS_EVERY_N_FRAMES", "1")))
# ── Model ───────────────────────────────────────────────────────────────────
MODEL_DIR: Path = Path(__file__).resolve().parent.parent / "pose_models"
MODEL_PATH = os.getenv("POSEFIT_MODEL_PATH", str(MODEL_DIR / "pose_landmarker_full.task"))
PREFER_GPU = os.getenv("POSEFIT_PREFER_GPU", "1") not in ("0", "false", "False")
# ── Dead bug exercise ───────────────────────────────────────────────────────
VISIBILITY_THRESHOLD = float(os.getenv("POSEFIT_VISIBILITY_THRESHOLD", "0.45"))
EXTENSION_CONFIRM_FRAMES = int(os.getenv("POSEFIT_EXTENSION_CONFIRM_FRAMES", "4"))
RESET_CONFIRM_FRAMES = int(os.getenv("POSEFIT_RESET_CONFIRM_FRAMES", "3"))
# ── Audio ───────────────────────────────────────────────────────────────────
REP_ANNOUNCER_ENABLED = os.getenv("POSEFIT_REP_ANNOUNCER_ENABLED", "1") not in ("0", "false", "False")
REP_ANNOUNCER_RATE = int(os.getenv("POSEFIT_REP_ANNOUNCER_RATE", "185"))
REP_ANNOUNCER_VOLUME = float(os.getenv("POSEFIT_REP_ANNOUNCER_VOLUME", "1.0"))
# ── Logging ─────────────────────────────────────────────────────────────────
LOG_DIR: Path = Path(__file__).resolve().parent.parent / "logs"
LOG_ROTATION = os.getenv("POSEFIT_LOG_ROTATION", "20 MB")
LOG_RETENTION = os.getenv("POSEFIT_LOG_RETENTION", "14 days")
+83
View File
@@ -0,0 +1,83 @@
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
import yaml
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
_ENV_MAP = {
"POSEFIT_WS_HOST": ("server", "host"),
"POSEFIT_WS_PORT": ("server", "port", int),
"POSEFIT_WS_MAX_SIZE": ("server", "max_ws_size", int),
"POSEFIT_PROCESS_EVERY_N_FRAMES": ("video", "process_every_n_frames", int),
"POSEFIT_MODEL_PATH": ("model", "path"),
"POSEFIT_PREFER_GPU": ("model", "prefer_gpu", lambda v: v not in ("0", "false", "False")),
"POSEFIT_VISIBILITY_THRESHOLD": ("dead_bug", "visibility_threshold", float),
"POSEFIT_EXTENSION_CONFIRM_FRAMES": ("dead_bug", "extension_confirm_frames", int),
"POSEFIT_RESET_CONFIRM_FRAMES": ("dead_bug", "reset_confirm_frames", int),
"POSEFIT_REP_ANNOUNCER_ENABLED": ("audio", "rep_announcer_enabled", lambda v: v not in ("0", "false", "False")),
"POSEFIT_REP_ANNOUNCER_RATE": ("audio", "rep_announcer_rate", int),
"POSEFIT_REP_ANNOUNCER_VOLUME": ("audio", "rep_announcer_volume", float),
"POSEFIT_LOG_ROTATION": ("logging", "rotation"),
"POSEFIT_LOG_RETENTION": ("logging", "retention"),
"POSEFIT_LOG_DIR": ("logging", "dir"),
}
def _load_yaml() -> dict[str, Any]:
config_path = _PROJECT_ROOT / "config.yaml"
if config_path.exists():
with open(config_path, encoding="utf-8") as f:
return yaml.safe_load(f) or {}
return {}
def _apply_env_overrides(config: dict) -> None:
for env_var, (section, key, *rest) in _ENV_MAP.items():
value = os.getenv(env_var)
if value is None:
continue
if rest:
value = rest[0](value)
config.setdefault(section, {})[key] = value
_cfg = _load_yaml()
_apply_env_overrides(_cfg)
def _get(section: str, key: str, default: Any = None) -> Any:
return _cfg.get(section, {}).get(key, default)
# ── Server ──────────────────────────────────────────────────────────────────
WS_HOST = _get("server", "host", "0.0.0.0")
WS_PORT = _get("server", "port", 8765)
WS_MAX_SIZE = _get("server", "max_ws_size", 10485760)
# ── Video processing ────────────────────────────────────────────────────────
PROCESS_EVERY_N_FRAMES = max(1, _get("video", "process_every_n_frames", 1))
# ── Model ───────────────────────────────────────────────────────────────────
MODEL_DIR: Path = _PROJECT_ROOT / "pose_models"
_model_path = _get("model", "path", "")
MODEL_PATH = _model_path if _model_path else str(MODEL_DIR / "pose_landmarker_full.task")
PREFER_GPU = bool(_get("model", "prefer_gpu", True))
# ── Dead bug exercise ───────────────────────────────────────────────────────
VISIBILITY_THRESHOLD = float(_get("dead_bug", "visibility_threshold", 0.45))
EXTENSION_CONFIRM_FRAMES = int(_get("dead_bug", "extension_confirm_frames", 4))
RESET_CONFIRM_FRAMES = int(_get("dead_bug", "reset_confirm_frames", 3))
# ── Audio ───────────────────────────────────────────────────────────────────
REP_ANNOUNCER_ENABLED = bool(_get("audio", "rep_announcer_enabled", True))
REP_ANNOUNCER_RATE = int(_get("audio", "rep_announcer_rate", 185))
REP_ANNOUNCER_VOLUME = float(_get("audio", "rep_announcer_volume", 1.0))
# ── Logging ─────────────────────────────────────────────────────────────────
LOG_DIR: Path = _PROJECT_ROOT / _get("logging", "dir", "logs")
LOG_ROTATION = _get("logging", "rotation", "20 MB")
LOG_RETENTION = _get("logging", "retention", "14 days")
+1
View File
@@ -5,3 +5,4 @@ numpy>=1.26,<2
loguru>=0.7.0 loguru>=0.7.0
mediapipe==0.10.21 mediapipe==0.10.21
pyttsx3>=2.99 pyttsx3>=2.99
pyyaml>=6.0
+1 -3
View File
@@ -1,9 +1,7 @@
from __future__ import annotations from __future__ import annotations
from app.exercises.dead_bug.metrics import calculate_metrics
from app.exercises.dead_bug.rules import detect_diagonal_extension, has_required_visibility, is_ready_position from app.exercises.dead_bug.rules import detect_diagonal_extension, has_required_visibility, is_ready_position
from app.exercises.dead_bug.state_machine import DeadBugStateMachine from app.exercises.dead_bug.types import DeadBugMetrics, Point
from app.exercises.dead_bug.types import DeadBugMetrics, DeadBugPhase, Point
class TestDeadBugRules: class TestDeadBugRules: