forked from genewildish/Mainline
David Gwilliam
affafe810c
Two critical fixes: 1. ListDataSource Cache Bug - Previously, ListDataSource.__init__ cached raw tuples directly - get_items() would return cached raw tuples without converting to SourceItem - This caused SourceItemsToBufferStage to receive tuples and stringify them - Results: ugly tuple representations in terminal/pygame instead of formatted text - Fix: Store raw items in _raw_items, let fetch() convert to SourceItem - Cache now contains proper SourceItem objects 2. Camera Dependency Resolution - CameraStage declared dependency on 'source.items' exactly - DataSourceStage provides 'source.headlines' (or 'source.poetry', etc.) - Capability matching didn't trigger prefix match for exact dependency - Fix: Change CameraStage dependency to 'source' for prefix matching 3. Added app.py Camera Stage Support - Pipeline now adds camera stage from preset.camera config - Supports vertical, horizontal, omni, floating, bounce modes - Tests now passing with proper data flow through all stages Tests: All 502 tests passing, 16 skipped
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
from engine.effects.types import (
|
|
EffectConfig,
|
|
EffectContext,
|
|
EffectPlugin,
|
|
PartialUpdate,
|
|
)
|
|
|
|
|
|
class HudEffect(EffectPlugin):
|
|
name = "hud"
|
|
config = EffectConfig(enabled=True, intensity=1.0)
|
|
supports_partial_updates = True # Enable partial update optimization
|
|
|
|
# Cache last HUD content to detect changes
|
|
_last_hud_content: tuple | None = None
|
|
|
|
def process_partial(
|
|
self, buf: list[str], ctx: EffectContext, partial: PartialUpdate
|
|
) -> list[str]:
|
|
# If full buffer requested, process normally
|
|
if partial.full_buffer:
|
|
return self.process(buf, ctx)
|
|
|
|
# If HUD rows (0, 1, 2) aren't dirty, skip processing
|
|
if partial.dirty:
|
|
hud_rows = {0, 1, 2}
|
|
dirty_hud_rows = partial.dirty & hud_rows
|
|
if not dirty_hud_rows:
|
|
return buf # Nothing for HUD to do
|
|
|
|
# Proceed with full processing
|
|
return self.process(buf, ctx)
|
|
|
|
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
|
result = list(buf)
|
|
|
|
# Read metrics from pipeline context (first-class citizen)
|
|
# Falls back to global monitor for backwards compatibility
|
|
metrics = ctx.get_state("metrics")
|
|
if not metrics:
|
|
# Fallback to global monitor for backwards compatibility
|
|
from engine.effects.performance import get_monitor
|
|
|
|
monitor = get_monitor()
|
|
if monitor:
|
|
stats = monitor.get_stats()
|
|
if stats and "pipeline" in stats:
|
|
metrics = stats
|
|
|
|
fps = 0.0
|
|
frame_time = 0.0
|
|
if metrics:
|
|
if "error" in metrics:
|
|
pass # No metrics available yet
|
|
elif "pipeline" in metrics:
|
|
frame_time = metrics["pipeline"].get("avg_ms", 0.0)
|
|
frame_count = metrics.get("frame_count", 0)
|
|
if frame_count > 0 and frame_time > 0:
|
|
fps = 1000.0 / frame_time
|
|
elif "avg_ms" in metrics:
|
|
# Direct metrics format
|
|
frame_time = metrics.get("avg_ms", 0.0)
|
|
frame_count = metrics.get("frame_count", 0)
|
|
if frame_count > 0 and frame_time > 0:
|
|
fps = 1000.0 / frame_time
|
|
|
|
w = ctx.terminal_width
|
|
h = ctx.terminal_height
|
|
|
|
effect_name = self.config.params.get("display_effect", "none")
|
|
effect_intensity = self.config.params.get("display_intensity", 0.0)
|
|
|
|
hud_lines = []
|
|
hud_lines.append(
|
|
f"\033[1;1H\033[38;5;46mMAINLINE DEMO\033[0m \033[38;5;245m|\033[0m \033[38;5;39mFPS: {fps:.1f}\033[0m \033[38;5;245m|\033[0m \033[38;5;208m{frame_time:.1f}ms\033[0m"
|
|
)
|
|
|
|
bar_width = 20
|
|
filled = int(bar_width * effect_intensity)
|
|
bar = (
|
|
"\033[38;5;82m"
|
|
+ "█" * filled
|
|
+ "\033[38;5;240m"
|
|
+ "░" * (bar_width - filled)
|
|
+ "\033[0m"
|
|
)
|
|
hud_lines.append(
|
|
f"\033[2;1H\033[38;5;45mEFFECT:\033[0m \033[1;38;5;227m{effect_name:12s}\033[0m \033[38;5;245m|\033[0m {bar} \033[38;5;245m|\033[0m \033[38;5;219m{effect_intensity * 100:.0f}%\033[0m"
|
|
)
|
|
|
|
# Get pipeline order from context
|
|
pipeline_order = ctx.get_state("pipeline_order")
|
|
pipeline_str = ",".join(pipeline_order) if pipeline_order else "(none)"
|
|
hud_lines.append(f"\033[3;1H\033[38;5;44mPIPELINE:\033[0m {pipeline_str}")
|
|
|
|
for i, line in enumerate(hud_lines):
|
|
if i < len(result):
|
|
result[i] = line + result[i][len(line) :]
|
|
else:
|
|
result.append(line)
|
|
|
|
return result
|
|
|
|
def configure(self, config: EffectConfig) -> None:
|
|
self.config = config
|