forked from genewildish/Mainline
feat(presets): add TOML preset loader with validation
- Convert presets from YAML to TOML format (no external dep) - Add DEFAULT_PRESET fallback for graceful degradation - Add validate_preset() for preset validation - Add validate_signal_path() for circular dependency detection - Add generate_preset_toml() for skeleton generation - Use tomllib (Python 3.11+ stdlib)
This commit is contained in:
224
engine/pipeline/preset_loader.py
Normal file
224
engine/pipeline/preset_loader.py
Normal file
@@ -0,0 +1,224 @@
|
|||||||
|
"""
|
||||||
|
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_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)
|
||||||
106
presets.toml
Normal file
106
presets.toml
Normal file
@@ -0,0 +1,106 @@
|
|||||||
|
# Mainline Presets Configuration
|
||||||
|
# Human- and machine-readable preset definitions
|
||||||
|
#
|
||||||
|
# Format: TOML
|
||||||
|
# Usage: mainline --preset <name>
|
||||||
|
#
|
||||||
|
# Built-in presets can be overridden by user presets in:
|
||||||
|
# - ~/.config/mainline/presets.toml
|
||||||
|
# - ./presets.toml (local override)
|
||||||
|
|
||||||
|
[presets.demo]
|
||||||
|
description = "Demo mode with effect cycling and camera modes"
|
||||||
|
source = "headlines"
|
||||||
|
display = "pygame"
|
||||||
|
camera = "vertical"
|
||||||
|
effects = ["noise", "fade", "glitch", "firehose", "hud"]
|
||||||
|
viewport_width = 80
|
||||||
|
viewport_height = 24
|
||||||
|
camera_speed = 1.0
|
||||||
|
firehose_enabled = true
|
||||||
|
|
||||||
|
[presets.poetry]
|
||||||
|
description = "Poetry feed with subtle effects"
|
||||||
|
source = "poetry"
|
||||||
|
display = "pygame"
|
||||||
|
camera = "vertical"
|
||||||
|
effects = ["fade", "hud"]
|
||||||
|
viewport_width = 80
|
||||||
|
viewport_height = 24
|
||||||
|
camera_speed = 0.5
|
||||||
|
firehose_enabled = false
|
||||||
|
|
||||||
|
[presets.pipeline]
|
||||||
|
description = "Pipeline visualization mode"
|
||||||
|
source = "pipeline"
|
||||||
|
display = "terminal"
|
||||||
|
camera = "trace"
|
||||||
|
effects = ["hud"]
|
||||||
|
viewport_width = 80
|
||||||
|
viewport_height = 24
|
||||||
|
camera_speed = 1.0
|
||||||
|
firehose_enabled = false
|
||||||
|
|
||||||
|
[presets.websocket]
|
||||||
|
description = "WebSocket display mode"
|
||||||
|
source = "headlines"
|
||||||
|
display = "websocket"
|
||||||
|
camera = "vertical"
|
||||||
|
effects = ["noise", "fade", "glitch", "hud"]
|
||||||
|
viewport_width = 80
|
||||||
|
viewport_height = 24
|
||||||
|
camera_speed = 1.0
|
||||||
|
firehose_enabled = false
|
||||||
|
|
||||||
|
[presets.sixel]
|
||||||
|
description = "Sixel graphics display mode"
|
||||||
|
source = "headlines"
|
||||||
|
display = "sixel"
|
||||||
|
camera = "vertical"
|
||||||
|
effects = ["noise", "fade", "glitch", "hud"]
|
||||||
|
viewport_width = 80
|
||||||
|
viewport_height = 24
|
||||||
|
camera_speed = 1.0
|
||||||
|
firehose_enabled = false
|
||||||
|
|
||||||
|
[presets.firehose]
|
||||||
|
description = "High-speed firehose mode"
|
||||||
|
source = "headlines"
|
||||||
|
display = "pygame"
|
||||||
|
camera = "vertical"
|
||||||
|
effects = ["noise", "fade", "glitch", "firehose", "hud"]
|
||||||
|
viewport_width = 80
|
||||||
|
viewport_height = 24
|
||||||
|
camera_speed = 2.0
|
||||||
|
firehose_enabled = true
|
||||||
|
|
||||||
|
# Sensor configuration (for future use with param bindings)
|
||||||
|
[sensors.mic]
|
||||||
|
enabled = false
|
||||||
|
threshold_db = 50.0
|
||||||
|
|
||||||
|
[sensors.oscillator]
|
||||||
|
enabled = false
|
||||||
|
waveform = "sine"
|
||||||
|
frequency = 1.0
|
||||||
|
|
||||||
|
# Effect configurations
|
||||||
|
[effect_configs.noise]
|
||||||
|
enabled = true
|
||||||
|
intensity = 1.0
|
||||||
|
|
||||||
|
[effect_configs.fade]
|
||||||
|
enabled = true
|
||||||
|
intensity = 1.0
|
||||||
|
|
||||||
|
[effect_configs.glitch]
|
||||||
|
enabled = true
|
||||||
|
intensity = 0.5
|
||||||
|
|
||||||
|
[effect_configs.firehose]
|
||||||
|
enabled = true
|
||||||
|
intensity = 1.0
|
||||||
|
|
||||||
|
[effect_configs.hud]
|
||||||
|
enabled = true
|
||||||
|
intensity = 1.0
|
||||||
Reference in New Issue
Block a user