David Gwilliam
f43920e2f0
- Remove ~700 lines of legacy code from app.py (run_demo_mode, run_pipeline_demo, run_preset_mode, font picker, effects picker) - HUD now reads metrics from pipeline context (first-class citizen) with fallback to global monitor for backwards compatibility - Add validate_signal_flow() for PureData-style type validation in presets - Update MicSensor documentation (self-contained, doesn't use MicMonitor) - Delete test_app.py (was testing removed legacy code) - Update AGENTS.md with pipeline architecture documentation
283 lines
8.4 KiB
Python
283 lines
8.4 KiB
Python
"""
|
|
Preset loader - Loads presets from TOML files.
|
|
|
|
Supports:
|
|
- Built-in presets.toml in the package
|
|
- User overrides in ~/.config/mainline/presets.toml
|
|
- Local override in ./presets.toml
|
|
- Fallback DEFAULT_PRESET when loading fails
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import tomllib
|
|
|
|
DEFAULT_PRESET: dict[str, Any] = {
|
|
"description": "Default fallback preset",
|
|
"source": "headlines",
|
|
"display": "terminal",
|
|
"camera": "vertical",
|
|
"effects": ["hud"],
|
|
"viewport": {"width": 80, "height": 24},
|
|
"camera_speed": 1.0,
|
|
"firehose_enabled": False,
|
|
}
|
|
|
|
|
|
def get_preset_paths() -> list[Path]:
|
|
"""Get list of preset file paths in load order (later overrides earlier)."""
|
|
paths = []
|
|
|
|
builtin = Path(__file__).parent.parent / "presets.toml"
|
|
if builtin.exists():
|
|
paths.append(builtin)
|
|
|
|
user_config = Path(os.path.expanduser("~/.config/mainline/presets.toml"))
|
|
if user_config.exists():
|
|
paths.append(user_config)
|
|
|
|
local = Path("presets.toml")
|
|
if local.exists():
|
|
paths.append(local)
|
|
|
|
return paths
|
|
|
|
|
|
def load_presets() -> dict[str, Any]:
|
|
"""Load all presets, merging from multiple sources."""
|
|
merged: dict[str, Any] = {"presets": {}, "sensors": {}, "effect_configs": {}}
|
|
|
|
for path in get_preset_paths():
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = tomllib.load(f)
|
|
|
|
if "presets" in data:
|
|
merged["presets"].update(data["presets"])
|
|
|
|
if "sensors" in data:
|
|
merged["sensors"].update(data["sensors"])
|
|
|
|
if "effect_configs" in data:
|
|
merged["effect_configs"].update(data["effect_configs"])
|
|
|
|
except Exception as e:
|
|
print(f"Warning: Failed to load presets from {path}: {e}")
|
|
|
|
return merged
|
|
|
|
|
|
def get_preset(name: str) -> dict[str, Any] | None:
|
|
"""Get a preset by name."""
|
|
presets = load_presets()
|
|
return presets.get("presets", {}).get(name)
|
|
|
|
|
|
def list_preset_names() -> list[str]:
|
|
"""List all available preset names."""
|
|
presets = load_presets()
|
|
return list(presets.get("presets", {}).keys())
|
|
|
|
|
|
def get_sensor_config(name: str) -> dict[str, Any] | None:
|
|
"""Get sensor configuration by name."""
|
|
sensors = load_presets()
|
|
return sensors.get("sensors", {}).get(name)
|
|
|
|
|
|
def get_effect_config(name: str) -> dict[str, Any] | None:
|
|
"""Get effect configuration by name."""
|
|
configs = load_presets()
|
|
return configs.get("effect_configs", {}).get(name)
|
|
|
|
|
|
def get_all_effect_configs() -> dict[str, Any]:
|
|
"""Get all effect configurations."""
|
|
configs = load_presets()
|
|
return configs.get("effect_configs", {})
|
|
|
|
|
|
def get_preset_or_default(name: str) -> dict[str, Any]:
|
|
"""Get a preset by name, or return DEFAULT_PRESET if not found."""
|
|
preset = get_preset(name)
|
|
if preset is not None:
|
|
return preset
|
|
return DEFAULT_PRESET.copy()
|
|
|
|
|
|
def ensure_preset_available(name: str | None) -> dict[str, Any]:
|
|
"""Ensure a preset is available, falling back to DEFAULT_PRESET."""
|
|
if name is None:
|
|
return DEFAULT_PRESET.copy()
|
|
return get_preset_or_default(name)
|
|
|
|
|
|
class PresetValidationError(Exception):
|
|
"""Raised when preset validation fails."""
|
|
|
|
pass
|
|
|
|
|
|
def validate_preset(preset: dict[str, Any]) -> list[str]:
|
|
"""Validate a preset and return list of errors (empty if valid)."""
|
|
errors: list[str] = []
|
|
|
|
required_fields = ["source", "display", "effects"]
|
|
for field in required_fields:
|
|
if field not in preset:
|
|
errors.append(f"Missing required field: {field}")
|
|
|
|
if "effects" in preset:
|
|
if not isinstance(preset["effects"], list):
|
|
errors.append("'effects' must be a list")
|
|
else:
|
|
for effect in preset["effects"]:
|
|
if not isinstance(effect, str):
|
|
errors.append(
|
|
f"Effect must be string, got {type(effect)}: {effect}"
|
|
)
|
|
|
|
if "viewport" in preset:
|
|
viewport = preset["viewport"]
|
|
if not isinstance(viewport, dict):
|
|
errors.append("'viewport' must be a dict")
|
|
else:
|
|
if "width" in viewport and not isinstance(viewport["width"], int):
|
|
errors.append("'viewport.width' must be an int")
|
|
if "height" in viewport and not isinstance(viewport["height"], int):
|
|
errors.append("'viewport.height' must be an int")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_signal_flow(stages: list[dict]) -> list[str]:
|
|
"""Validate signal flow based on inlet/outlet types.
|
|
|
|
This validates that the preset's stage configuration produces valid
|
|
data flow using the PureData-style type system.
|
|
|
|
Args:
|
|
stages: List of stage configs with 'name', 'category', 'inlet_types', 'outlet_types'
|
|
|
|
Returns:
|
|
List of errors (empty if valid)
|
|
"""
|
|
errors: list[str] = []
|
|
|
|
if not stages:
|
|
errors.append("Signal flow is empty")
|
|
return errors
|
|
|
|
# Define expected types for each category
|
|
type_map = {
|
|
"source": {"inlet": "NONE", "outlet": "SOURCE_ITEMS"},
|
|
"data": {"inlet": "ANY", "outlet": "SOURCE_ITEMS"},
|
|
"transform": {"inlet": "SOURCE_ITEMS", "outlet": "TEXT_BUFFER"},
|
|
"effect": {"inlet": "TEXT_BUFFER", "outlet": "TEXT_BUFFER"},
|
|
"overlay": {"inlet": "TEXT_BUFFER", "outlet": "TEXT_BUFFER"},
|
|
"camera": {"inlet": "TEXT_BUFFER", "outlet": "TEXT_BUFFER"},
|
|
"display": {"inlet": "TEXT_BUFFER", "outlet": "NONE"},
|
|
"render": {"inlet": "SOURCE_ITEMS", "outlet": "TEXT_BUFFER"},
|
|
}
|
|
|
|
# Check stage order and type compatibility
|
|
for i, stage in enumerate(stages):
|
|
category = stage.get("category", "unknown")
|
|
name = stage.get("name", f"stage_{i}")
|
|
|
|
if category not in type_map:
|
|
continue # Skip unknown categories
|
|
|
|
expected = type_map[category]
|
|
|
|
# Check against previous stage
|
|
if i > 0:
|
|
prev = stages[i - 1]
|
|
prev_category = prev.get("category", "unknown")
|
|
if prev_category in type_map:
|
|
prev_outlet = type_map[prev_category]["outlet"]
|
|
inlet = expected["inlet"]
|
|
|
|
# Validate type compatibility
|
|
if inlet != "ANY" and prev_outlet != "ANY" and inlet != prev_outlet:
|
|
errors.append(
|
|
f"Type mismatch at '{name}': "
|
|
f"expects {inlet} but previous stage outputs {prev_outlet}"
|
|
)
|
|
|
|
return errors
|
|
|
|
|
|
def validate_signal_path(stages: list[str]) -> list[str]:
|
|
"""Validate signal path for circular dependencies and connectivity.
|
|
|
|
Args:
|
|
stages: List of stage names in execution order
|
|
|
|
Returns:
|
|
List of errors (empty if valid)
|
|
"""
|
|
errors: list[str] = []
|
|
|
|
if not stages:
|
|
errors.append("Signal path is empty")
|
|
return errors
|
|
|
|
seen: set[str] = set()
|
|
for i, stage in enumerate(stages):
|
|
if stage in seen:
|
|
errors.append(
|
|
f"Circular dependency: '{stage}' appears multiple times at index {i}"
|
|
)
|
|
seen.add(stage)
|
|
|
|
return errors
|
|
|
|
|
|
def generate_preset_toml(
|
|
name: str,
|
|
source: str = "headlines",
|
|
display: str = "terminal",
|
|
effects: list[str] | None = None,
|
|
viewport_width: int = 80,
|
|
viewport_height: int = 24,
|
|
camera: str = "vertical",
|
|
camera_speed: float = 1.0,
|
|
firehose_enabled: bool = False,
|
|
) -> str:
|
|
"""Generate a TOML preset skeleton with default values.
|
|
|
|
Args:
|
|
name: Preset name
|
|
source: Data source name
|
|
display: Display backend
|
|
effects: List of effect names
|
|
viewport_width: Viewport width in columns
|
|
viewport_height: Viewport height in rows
|
|
camera: Camera mode
|
|
camera_speed: Camera scroll speed
|
|
firehose_enabled: Enable firehose mode
|
|
|
|
Returns:
|
|
TOML string for the preset
|
|
"""
|
|
|
|
if effects is None:
|
|
effects = ["fade", "hud"]
|
|
|
|
output = []
|
|
output.append(f"[presets.{name}]")
|
|
output.append(f'description = "Auto-generated preset: {name}"')
|
|
output.append(f'source = "{source}"')
|
|
output.append(f'display = "{display}"')
|
|
output.append(f'camera = "{camera}"')
|
|
output.append(f"effects = {effects}")
|
|
output.append(f"viewport_width = {viewport_width}")
|
|
output.append(f"viewport_height = {viewport_height}")
|
|
output.append(f"camera_speed = {camera_speed}")
|
|
output.append(f"firehose_enabled = {str(firehose_enabled).lower()}")
|
|
|
|
return "\n".join(output)
|