forked from genewildish/Mainline
David Gwilliam
3e9c1be6d2
- Add --demo flag that runs effect showcase with pygame display - Add HUD effect plugin (effects_plugins/hud.py) that displays: - FPS and frame time - Current effect name with intensity bar - Pipeline order - Demo mode cycles through noise, fade, glitch, firehose effects - Ramps intensity 0→1→0 over 5 seconds per effect
574 lines
17 KiB
Python
574 lines
17 KiB
Python
"""
|
|
Application orchestrator — boot sequence, signal handling, main loop wiring.
|
|
"""
|
|
|
|
import atexit
|
|
import os
|
|
import signal
|
|
import sys
|
|
import termios
|
|
import time
|
|
import tty
|
|
|
|
from engine import config, render
|
|
from engine.controller import StreamController
|
|
from engine.fetch import fetch_all, fetch_poetry, load_cache, save_cache
|
|
from engine.terminal import (
|
|
CLR,
|
|
CURSOR_OFF,
|
|
CURSOR_ON,
|
|
G_DIM,
|
|
G_HI,
|
|
G_MID,
|
|
RST,
|
|
W_DIM,
|
|
W_GHOST,
|
|
boot_ln,
|
|
slow_print,
|
|
tw,
|
|
)
|
|
|
|
TITLE = [
|
|
" ███╗ ███╗ █████╗ ██╗███╗ ██╗██╗ ██╗███╗ ██╗███████╗",
|
|
" ████╗ ████║██╔══██╗██║████╗ ██║██║ ██║████╗ ██║██╔════╝",
|
|
" ██╔████╔██║███████║██║██╔██╗ ██║██║ ██║██╔██╗ ██║█████╗ ",
|
|
" ██║╚██╔╝██║██╔══██║██║██║╚██╗██║██║ ██║██║╚██╗██║██╔══╝ ",
|
|
" ██║ ╚═╝ ██║██║ ██║██║██║ ╚████║███████╗██║██║ ╚████║███████╗",
|
|
" ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚═╝ ╚═══╝╚══════╝╚═╝╚═╝ ╚═══╝╚══════╝",
|
|
]
|
|
|
|
|
|
def _read_picker_key():
|
|
ch = sys.stdin.read(1)
|
|
if ch == "\x03":
|
|
return "interrupt"
|
|
if ch in ("\r", "\n"):
|
|
return "enter"
|
|
if ch == "\x1b":
|
|
c1 = sys.stdin.read(1)
|
|
if c1 != "[":
|
|
return None
|
|
c2 = sys.stdin.read(1)
|
|
if c2 == "A":
|
|
return "up"
|
|
if c2 == "B":
|
|
return "down"
|
|
return None
|
|
if ch in ("k", "K"):
|
|
return "up"
|
|
if ch in ("j", "J"):
|
|
return "down"
|
|
if ch in ("q", "Q"):
|
|
return "enter"
|
|
return None
|
|
|
|
|
|
def _normalize_preview_rows(rows):
|
|
"""Trim shared left padding and trailing spaces for stable on-screen previews."""
|
|
non_empty = [r for r in rows if r.strip()]
|
|
if not non_empty:
|
|
return [""]
|
|
left_pad = min(len(r) - len(r.lstrip(" ")) for r in non_empty)
|
|
out = []
|
|
for row in rows:
|
|
if left_pad < len(row):
|
|
out.append(row[left_pad:].rstrip())
|
|
else:
|
|
out.append(row.rstrip())
|
|
return out
|
|
|
|
|
|
def _draw_font_picker(faces, selected):
|
|
w = tw()
|
|
h = 24
|
|
try:
|
|
h = os.get_terminal_size().lines
|
|
except Exception:
|
|
pass
|
|
|
|
max_preview_w = max(24, w - 8)
|
|
header_h = 6
|
|
footer_h = 3
|
|
preview_h = max(4, min(config.RENDER_H + 2, max(4, h // 2)))
|
|
visible = max(1, h - header_h - preview_h - footer_h)
|
|
top = max(0, selected - (visible // 2))
|
|
bottom = min(len(faces), top + visible)
|
|
top = max(0, bottom - visible)
|
|
|
|
print(CLR, end="")
|
|
print(CURSOR_OFF, end="")
|
|
print()
|
|
print(f" {G_HI}FONT PICKER{RST}")
|
|
print(f" {W_GHOST}{'─' * (w - 4)}{RST}")
|
|
print(f" {W_DIM}{config.FONT_DIR[:max_preview_w]}{RST}")
|
|
print(f" {W_GHOST}↑/↓ move · Enter select · q accept current{RST}")
|
|
print()
|
|
|
|
for pos in range(top, bottom):
|
|
face = faces[pos]
|
|
active = pos == selected
|
|
pointer = "▶" if active else " "
|
|
color = G_HI if active else W_DIM
|
|
print(
|
|
f" {color}{pointer} {face['name']}{RST}{W_GHOST} · {face['file_name']}{RST}"
|
|
)
|
|
|
|
if top > 0:
|
|
print(f" {W_GHOST}… {top} above{RST}")
|
|
if bottom < len(faces):
|
|
print(f" {W_GHOST}… {len(faces) - bottom} below{RST}")
|
|
|
|
print()
|
|
print(f" {W_GHOST}{'─' * (w - 4)}{RST}")
|
|
print(
|
|
f" {W_DIM}Preview: {faces[selected]['name']} · {faces[selected]['file_name']}{RST}"
|
|
)
|
|
preview_rows = faces[selected]["preview_rows"][:preview_h]
|
|
for row in preview_rows:
|
|
shown = row[:max_preview_w]
|
|
print(f" {shown}")
|
|
|
|
|
|
def pick_font_face():
|
|
"""Interactive startup picker for selecting a face from repo OTF files."""
|
|
if not config.FONT_PICKER:
|
|
return
|
|
|
|
font_files = config.list_repo_font_files()
|
|
if not font_files:
|
|
print(CLR, end="")
|
|
print(CURSOR_OFF, end="")
|
|
print()
|
|
print(f" {G_HI}FONT PICKER{RST}")
|
|
print(f" {W_GHOST}{'─' * (tw() - 4)}{RST}")
|
|
print(f" {G_DIM}> no .otf/.ttf/.ttc files found in: {config.FONT_DIR}{RST}")
|
|
print(f" {W_GHOST}> add font files to the fonts folder, then rerun{RST}")
|
|
time.sleep(1.8)
|
|
sys.exit(1)
|
|
|
|
prepared = []
|
|
for font_path in font_files:
|
|
try:
|
|
faces = render.list_font_faces(font_path, max_faces=64)
|
|
except Exception:
|
|
fallback = os.path.splitext(os.path.basename(font_path))[0]
|
|
faces = [{"index": 0, "name": fallback}]
|
|
for face in faces:
|
|
idx = face["index"]
|
|
name = face["name"]
|
|
file_name = os.path.basename(font_path)
|
|
try:
|
|
fnt = render.load_font_face(font_path, idx)
|
|
rows = _normalize_preview_rows(render.render_line(name, fnt))
|
|
except Exception:
|
|
rows = ["(preview unavailable)"]
|
|
prepared.append(
|
|
{
|
|
"font_path": font_path,
|
|
"font_index": idx,
|
|
"name": name,
|
|
"file_name": file_name,
|
|
"preview_rows": rows,
|
|
}
|
|
)
|
|
|
|
if not prepared:
|
|
print(CLR, end="")
|
|
print(CURSOR_OFF, end="")
|
|
print()
|
|
print(f" {G_HI}FONT PICKER{RST}")
|
|
print(f" {W_GHOST}{'─' * (tw() - 4)}{RST}")
|
|
print(f" {G_DIM}> no readable font faces found in: {config.FONT_DIR}{RST}")
|
|
time.sleep(1.8)
|
|
sys.exit(1)
|
|
|
|
def _same_path(a, b):
|
|
try:
|
|
return os.path.samefile(a, b)
|
|
except Exception:
|
|
return os.path.abspath(a) == os.path.abspath(b)
|
|
|
|
selected = next(
|
|
(
|
|
i
|
|
for i, f in enumerate(prepared)
|
|
if _same_path(f["font_path"], config.FONT_PATH)
|
|
and f["font_index"] == config.FONT_INDEX
|
|
),
|
|
0,
|
|
)
|
|
|
|
if not sys.stdin.isatty():
|
|
selected_font = prepared[selected]
|
|
config.set_font_selection(
|
|
font_path=selected_font["font_path"],
|
|
font_index=selected_font["font_index"],
|
|
)
|
|
render.clear_font_cache()
|
|
print(
|
|
f" {G_DIM}> using {selected_font['name']} ({selected_font['file_name']}){RST}"
|
|
)
|
|
time.sleep(0.8)
|
|
print(CLR, end="")
|
|
print(CURSOR_OFF, end="")
|
|
print()
|
|
return
|
|
|
|
fd = sys.stdin.fileno()
|
|
old_settings = termios.tcgetattr(fd)
|
|
try:
|
|
tty.setcbreak(fd)
|
|
while True:
|
|
_draw_font_picker(prepared, selected)
|
|
key = _read_picker_key()
|
|
if key == "up":
|
|
selected = max(0, selected - 1)
|
|
elif key == "down":
|
|
selected = min(len(prepared) - 1, selected + 1)
|
|
elif key == "enter":
|
|
break
|
|
elif key == "interrupt":
|
|
raise KeyboardInterrupt
|
|
finally:
|
|
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
|
|
|
selected_font = prepared[selected]
|
|
config.set_font_selection(
|
|
font_path=selected_font["font_path"],
|
|
font_index=selected_font["font_index"],
|
|
)
|
|
render.clear_font_cache()
|
|
print(
|
|
f" {G_DIM}> using {selected_font['name']} ({selected_font['file_name']}){RST}"
|
|
)
|
|
time.sleep(0.8)
|
|
print(CLR, end="")
|
|
print(CURSOR_OFF, end="")
|
|
print()
|
|
|
|
|
|
def pick_effects_config():
|
|
"""Interactive picker for configuring effects pipeline."""
|
|
import effects_plugins
|
|
from engine.effects import get_effect_chain, get_registry
|
|
|
|
effects_plugins.discover_plugins()
|
|
|
|
registry = get_registry()
|
|
chain = get_effect_chain()
|
|
chain.set_order(["noise", "fade", "glitch", "firehose"])
|
|
|
|
effects = list(registry.list_all().values())
|
|
if not effects:
|
|
return
|
|
|
|
selected = 0
|
|
editing_intensity = False
|
|
intensity_value = 1.0
|
|
|
|
def _draw_effects_picker():
|
|
w = tw()
|
|
print(CLR, end="")
|
|
print("\033[1;1H", end="")
|
|
print(" \033[1;38;5;231mEFFECTS CONFIG\033[0m")
|
|
print(f" \033[2;38;5;37m{'─' * (w - 4)}\033[0m")
|
|
print()
|
|
|
|
for i, effect in enumerate(effects):
|
|
prefix = " > " if i == selected else " "
|
|
marker = "[*]" if effect.config.enabled else "[ ]"
|
|
if editing_intensity and i == selected:
|
|
print(
|
|
f"{prefix}{marker} \033[1;38;5;82m{effect.name}\033[0m: intensity={intensity_value:.2f} (use +/- to adjust, Enter to confirm)"
|
|
)
|
|
else:
|
|
print(
|
|
f"{prefix}{marker} {effect.name}: intensity={effect.config.intensity:.2f}"
|
|
)
|
|
|
|
print()
|
|
print(f" \033[2;38;5;37m{'─' * (w - 4)}\033[0m")
|
|
print(
|
|
" \033[38;5;245mControls: space=toggle on/off | +/-=adjust intensity | arrows=move | Enter=next effect | q=done\033[0m"
|
|
)
|
|
|
|
def _read_effects_key():
|
|
ch = sys.stdin.read(1)
|
|
if ch == "\x03":
|
|
return "interrupt"
|
|
if ch in ("\r", "\n"):
|
|
return "enter"
|
|
if ch == " ":
|
|
return "toggle"
|
|
if ch == "q":
|
|
return "quit"
|
|
if ch == "+" or ch == "=":
|
|
return "up"
|
|
if ch == "-" or ch == "_":
|
|
return "down"
|
|
if ch == "\x1b":
|
|
c1 = sys.stdin.read(1)
|
|
if c1 != "[":
|
|
return None
|
|
c2 = sys.stdin.read(1)
|
|
if c2 == "A":
|
|
return "up"
|
|
if c2 == "B":
|
|
return "down"
|
|
return None
|
|
return None
|
|
|
|
if not sys.stdin.isatty():
|
|
return
|
|
|
|
fd = sys.stdin.fileno()
|
|
old_settings = termios.tcgetattr(fd)
|
|
try:
|
|
tty.setcbreak(fd)
|
|
while True:
|
|
_draw_effects_picker()
|
|
key = _read_effects_key()
|
|
|
|
if key == "quit" or key == "enter":
|
|
break
|
|
elif key == "up" and editing_intensity:
|
|
intensity_value = min(1.0, intensity_value + 0.1)
|
|
effects[selected].config.intensity = intensity_value
|
|
elif key == "down" and editing_intensity:
|
|
intensity_value = max(0.0, intensity_value - 0.1)
|
|
effects[selected].config.intensity = intensity_value
|
|
elif key == "up":
|
|
selected = max(0, selected - 1)
|
|
intensity_value = effects[selected].config.intensity
|
|
elif key == "down":
|
|
selected = min(len(effects) - 1, selected + 1)
|
|
intensity_value = effects[selected].config.intensity
|
|
elif key == "toggle":
|
|
effects[selected].config.enabled = not effects[selected].config.enabled
|
|
elif key == "interrupt":
|
|
raise KeyboardInterrupt
|
|
finally:
|
|
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
|
|
|
|
|
def run_demo_mode():
|
|
"""Run demo mode - showcases effects with pygame display."""
|
|
import random
|
|
|
|
from engine import config
|
|
from engine.display import DisplayRegistry
|
|
from engine.effects import (
|
|
EffectContext,
|
|
PerformanceMonitor,
|
|
get_effect_chain,
|
|
get_registry,
|
|
set_monitor,
|
|
)
|
|
|
|
print(" \033[1;38;5;46mMAINLINE DEMO MODE\033[0m")
|
|
print(" \033[38;5;245mInitializing pygame display...\033[0m")
|
|
|
|
import effects_plugins
|
|
|
|
effects_plugins.discover_plugins()
|
|
|
|
registry = get_registry()
|
|
chain = get_effect_chain()
|
|
chain.set_order(["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)
|
|
|
|
display.init(80, 24)
|
|
display.clear()
|
|
|
|
effects_to_demo = ["noise", "fade", "glitch", "firehose"]
|
|
w, h = 80, 24
|
|
|
|
base_buffer = []
|
|
for row in range(h):
|
|
line = ""
|
|
for col in range(w):
|
|
char = random.choice("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789 ")
|
|
line += char
|
|
base_buffer.append(line)
|
|
|
|
print(" \033[38;5;82mStarting effect demo...\033[0m")
|
|
print(" \033[38;5;245mPress Ctrl+C to exit\033[0m\n")
|
|
|
|
effect_idx = 0
|
|
effect_name = effects_to_demo[effect_idx]
|
|
effect_start_time = time.time()
|
|
current_intensity = 0.0
|
|
ramping_up = True
|
|
|
|
frame_count = 0
|
|
|
|
try:
|
|
while True:
|
|
elapsed = time.time() - effect_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
|
|
|
|
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_name):
|
|
effect.config.enabled = False
|
|
|
|
hud_effect = registry.get("hud")
|
|
if hud_effect:
|
|
hud_effect.config.params["display_effect"] = effect_name
|
|
hud_effect.config.params["display_intensity"] = current_intensity
|
|
|
|
ctx = EffectContext(
|
|
terminal_width=w,
|
|
terminal_height=h,
|
|
scroll_cam=0,
|
|
ticker_height=h,
|
|
mic_excess=0.0,
|
|
grad_offset=time.time() % 1.0,
|
|
frame_number=frame_count,
|
|
has_message=False,
|
|
items=[],
|
|
)
|
|
|
|
result = chain.process(base_buffer, ctx)
|
|
display.show(result)
|
|
|
|
frame_count += 1
|
|
time.sleep(1 / 60)
|
|
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
display.cleanup()
|
|
print("\n \033[38;5;245mDemo ended\033[0m")
|
|
|
|
|
|
def main():
|
|
if config.DEMO:
|
|
run_demo_mode()
|
|
return
|
|
|
|
atexit.register(lambda: print(CURSOR_ON, end="", flush=True))
|
|
|
|
def handle_sigint(*_):
|
|
print(f"\n\n {G_DIM}> SIGNAL LOST{RST}")
|
|
print(f" {W_GHOST}> connection terminated{RST}\n")
|
|
sys.exit(0)
|
|
|
|
signal.signal(signal.SIGINT, handle_sigint)
|
|
|
|
StreamController.warmup_topics()
|
|
|
|
w = tw()
|
|
print(CLR, end="")
|
|
print(CURSOR_OFF, end="")
|
|
pick_font_face()
|
|
pick_effects_config()
|
|
w = tw()
|
|
print()
|
|
time.sleep(0.4)
|
|
|
|
for ln in TITLE:
|
|
print(f"{G_HI}{ln}{RST}")
|
|
time.sleep(0.07)
|
|
|
|
print()
|
|
_subtitle = (
|
|
"literary consciousness stream"
|
|
if config.MODE == "poetry"
|
|
else "digital consciousness stream"
|
|
)
|
|
print(f" {W_DIM}v0.1 · {_subtitle}{RST}")
|
|
print(f" {W_GHOST}{'─' * (w - 4)}{RST}")
|
|
print()
|
|
time.sleep(0.4)
|
|
|
|
cached = load_cache() if "--refresh" not in sys.argv else None
|
|
if cached:
|
|
items = cached
|
|
boot_ln("Cache", f"LOADED [{len(items)} SIGNALS]", True)
|
|
elif config.MODE == "poetry":
|
|
slow_print(" > INITIALIZING LITERARY CORPUS...\n")
|
|
time.sleep(0.2)
|
|
print()
|
|
items, linked, failed = fetch_poetry()
|
|
print()
|
|
print(
|
|
f" {G_DIM}>{RST} {G_MID}{linked} TEXTS LOADED{RST} {W_GHOST}· {failed} DARK{RST}"
|
|
)
|
|
print(f" {G_DIM}>{RST} {G_MID}{len(items)} STANZAS ACQUIRED{RST}")
|
|
save_cache(items)
|
|
else:
|
|
slow_print(" > INITIALIZING FEED ARRAY...\n")
|
|
time.sleep(0.2)
|
|
print()
|
|
items, linked, failed = fetch_all()
|
|
print()
|
|
print(
|
|
f" {G_DIM}>{RST} {G_MID}{linked} SOURCES LINKED{RST} {W_GHOST}· {failed} DARK{RST}"
|
|
)
|
|
print(f" {G_DIM}>{RST} {G_MID}{len(items)} SIGNALS ACQUIRED{RST}")
|
|
save_cache(items)
|
|
|
|
if not items:
|
|
print(f"\n {W_DIM}> NO SIGNAL — check network{RST}")
|
|
sys.exit(1)
|
|
|
|
print()
|
|
controller = StreamController()
|
|
mic_ok, ntfy_ok = controller.initialize_sources()
|
|
|
|
if controller.mic and controller.mic.available:
|
|
boot_ln(
|
|
"Microphone",
|
|
"ACTIVE"
|
|
if mic_ok
|
|
else "OFFLINE · check System Settings → Privacy → Microphone",
|
|
bool(mic_ok),
|
|
)
|
|
|
|
boot_ln("ntfy", "LISTENING" if ntfy_ok else "OFFLINE", ntfy_ok)
|
|
|
|
if config.FIREHOSE:
|
|
boot_ln("Firehose", "ENGAGED", True)
|
|
|
|
time.sleep(0.4)
|
|
slow_print(" > STREAMING...\n")
|
|
time.sleep(0.2)
|
|
print(f" {W_GHOST}{'─' * (w - 4)}{RST}")
|
|
print()
|
|
time.sleep(0.4)
|
|
|
|
controller.run(items)
|
|
|
|
print()
|
|
print(f" {W_GHOST}{'─' * (tw() - 4)}{RST}")
|
|
print(f" {G_DIM}> {config.HEADLINE_LIMIT} SIGNALS PROCESSED{RST}")
|
|
print(f" {W_GHOST}> end of stream{RST}")
|
|
print()
|