forked from genewildish/Mainline
380 lines
11 KiB
Python
380 lines
11 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.fetch import fetch_all, fetch_poetry, load_cache, save_cache
|
|
from engine.mic import MicMonitor
|
|
from engine.ntfy import NtfyPoller
|
|
from engine.scroll import stream
|
|
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,
|
|
)
|
|
from engine.websocket_display import WebSocketDisplay
|
|
|
|
|
|
def _get_display():
|
|
"""Get the appropriate display(s) based on config."""
|
|
from engine.display import MultiDisplay, TerminalDisplay
|
|
|
|
displays = []
|
|
|
|
if config.DISPLAY in ("terminal", "both"):
|
|
displays.append(TerminalDisplay())
|
|
|
|
if config.DISPLAY in ("websocket", "both") or config.WEBSOCKET:
|
|
ws = WebSocketDisplay(host="0.0.0.0", port=config.WEBSOCKET_PORT)
|
|
ws.start_server()
|
|
ws.start_http_server()
|
|
displays.append(ws)
|
|
|
|
if not displays:
|
|
return None
|
|
if len(displays) == 1:
|
|
return displays[0]
|
|
return MultiDisplay(displays)
|
|
|
|
|
|
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 main():
|
|
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)
|
|
|
|
w = tw()
|
|
print(CLR, end="")
|
|
print(CURSOR_OFF, end="")
|
|
pick_font_face()
|
|
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()
|
|
mic = MicMonitor(threshold_db=config.MIC_THRESHOLD_DB)
|
|
mic_ok = mic.start()
|
|
if mic.available:
|
|
boot_ln(
|
|
"Microphone",
|
|
"ACTIVE"
|
|
if mic_ok
|
|
else "OFFLINE · check System Settings → Privacy → Microphone",
|
|
bool(mic_ok),
|
|
)
|
|
|
|
ntfy = NtfyPoller(
|
|
config.NTFY_TOPIC,
|
|
reconnect_delay=config.NTFY_RECONNECT_DELAY,
|
|
display_secs=config.MESSAGE_DISPLAY_SECS,
|
|
)
|
|
ntfy_ok = ntfy.start()
|
|
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)
|
|
|
|
display = _get_display()
|
|
stream(items, ntfy, mic, display)
|
|
if display:
|
|
display.cleanup()
|
|
|
|
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()
|