forked from genewildish/Mainline
feat(demo): add pipeline visualization demo mode
- Add --pipeline-demo flag for ASCII pipeline animation - Create engine/pipeline_viz.py with animated pipeline graphics - Shows data flow, camera modes, FPS counter - Run with: python mainline.py --pipeline-demo --display pygame
This commit is contained in:
145
engine/app.py
145
engine/app.py
@@ -558,6 +558,147 @@ def run_demo_mode():
|
||||
print("\n \033[38;5;245mDemo ended\033[0m")
|
||||
|
||||
|
||||
def run_pipeline_demo():
|
||||
"""Run pipeline visualization demo mode - shows ASCII pipeline animation."""
|
||||
import time
|
||||
|
||||
from engine import config
|
||||
from engine.camera import Camera, CameraMode
|
||||
from engine.display import DisplayRegistry
|
||||
from engine.effects import (
|
||||
EffectContext,
|
||||
PerformanceMonitor,
|
||||
get_effect_chain,
|
||||
get_registry,
|
||||
set_monitor,
|
||||
)
|
||||
from engine.pipeline_viz import generate_animated_pipeline
|
||||
|
||||
print(" \033[1;38;5;46mMAINLINE PIPELINE DEMO\033[0m")
|
||||
print(" \033[38;5;245mInitializing...\033[0m")
|
||||
|
||||
import effects_plugins
|
||||
|
||||
effects_plugins.discover_plugins()
|
||||
|
||||
registry = get_registry()
|
||||
chain = get_effect_chain()
|
||||
chain.set_order(["noise", "fade", "glitch", "firehose", "hud"])
|
||||
|
||||
monitor = PerformanceMonitor()
|
||||
set_monitor(monitor)
|
||||
chain._monitor = monitor
|
||||
|
||||
display = DisplayRegistry.create("pygame")
|
||||
if not display:
|
||||
print(" \033[38;5;196mFailed to create pygame display\033[0m")
|
||||
sys.exit(1)
|
||||
|
||||
w, h = 80, 24
|
||||
display.init(w, h)
|
||||
display.clear()
|
||||
|
||||
camera = Camera.vertical(speed=1.0)
|
||||
|
||||
effects_to_demo = ["noise", "fade", "glitch", "firehose"]
|
||||
effect_idx = 0
|
||||
effect_name = effects_to_demo[effect_idx]
|
||||
effect_start_time = time.time()
|
||||
current_intensity = 0.0
|
||||
ramping_up = True
|
||||
|
||||
camera_modes = [
|
||||
(CameraMode.VERTICAL, "vertical"),
|
||||
(CameraMode.HORIZONTAL, "horizontal"),
|
||||
(CameraMode.OMNI, "omni"),
|
||||
(CameraMode.FLOATING, "floating"),
|
||||
]
|
||||
camera_mode_idx = 0
|
||||
camera_start_time = time.time()
|
||||
|
||||
frame_number = 0
|
||||
|
||||
print(" \033[38;5;82mStarting pipeline visualization...\033[0m")
|
||||
print(" \033[38;5;245mPress Ctrl+C to exit\033[0m\n")
|
||||
|
||||
try:
|
||||
while True:
|
||||
elapsed = time.time() - effect_start_time
|
||||
camera_elapsed = time.time() - camera_start_time
|
||||
duration = config.DEMO_EFFECT_DURATION
|
||||
|
||||
if elapsed >= duration:
|
||||
effect_idx = (effect_idx + 1) % len(effects_to_demo)
|
||||
effect_name = effects_to_demo[effect_idx]
|
||||
effect_start_time = time.time()
|
||||
elapsed = 0
|
||||
current_intensity = 0.0
|
||||
ramping_up = True
|
||||
|
||||
if camera_elapsed >= duration * 2:
|
||||
camera_mode_idx = (camera_mode_idx + 1) % len(camera_modes)
|
||||
mode, mode_name = camera_modes[camera_mode_idx]
|
||||
camera = Camera(mode=mode, speed=1.0)
|
||||
camera_start_time = time.time()
|
||||
camera_elapsed = 0
|
||||
|
||||
progress = elapsed / duration
|
||||
if ramping_up:
|
||||
current_intensity = progress
|
||||
if progress >= 1.0:
|
||||
ramping_up = False
|
||||
else:
|
||||
current_intensity = 1.0 - progress
|
||||
|
||||
for effect in registry.list_all().values():
|
||||
if effect.name == effect_name:
|
||||
effect.config.enabled = True
|
||||
effect.config.intensity = current_intensity
|
||||
elif effect.name not in ("hud",):
|
||||
effect.config.enabled = False
|
||||
|
||||
hud_effect = registry.get("hud")
|
||||
if hud_effect:
|
||||
mode_name = camera_modes[camera_mode_idx][1]
|
||||
hud_effect.config.params["display_effect"] = (
|
||||
f"{effect_name} / {mode_name}"
|
||||
)
|
||||
hud_effect.config.params["display_intensity"] = current_intensity
|
||||
|
||||
camera.update(config.FRAME_DT)
|
||||
|
||||
buf = generate_animated_pipeline(w, frame_number)
|
||||
|
||||
ctx = EffectContext(
|
||||
terminal_width=w,
|
||||
terminal_height=h,
|
||||
scroll_cam=camera.y,
|
||||
ticker_height=h,
|
||||
camera_x=camera.x,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=frame_number,
|
||||
has_message=False,
|
||||
items=[],
|
||||
)
|
||||
|
||||
result = chain.process(buf, ctx)
|
||||
display.show(result)
|
||||
|
||||
new_w, new_h = display.get_dimensions()
|
||||
if new_w != w or new_h != h:
|
||||
w, h = new_w, new_h
|
||||
|
||||
frame_number += 1
|
||||
time.sleep(1 / 60)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
display.cleanup()
|
||||
print("\n \033[38;5;245mPipeline demo ended\033[0m")
|
||||
|
||||
|
||||
def main():
|
||||
from engine import config
|
||||
from engine.pipeline import generate_pipeline_diagram
|
||||
@@ -566,6 +707,10 @@ def main():
|
||||
print(generate_pipeline_diagram())
|
||||
return
|
||||
|
||||
if config.PIPELINE_DEMO:
|
||||
run_pipeline_demo()
|
||||
return
|
||||
|
||||
if config.DEMO:
|
||||
run_demo_mode()
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user