forked from genewildish/Mainline
David Gwilliam
9e4d54a82e
- Add EffectPlugin ABC with @abstractmethod decorators for interface enforcement - Add runtime interface checking in discover_plugins() with issubclass() - Add EffectContext factory with sensible defaults - Standardize Display __init__ (remove redundant init in TerminalDisplay) - Document effect behavior when ticker_height=0 - Evaluate legacy effects: document coexistence, no deprecation needed - Research plugin patterns (VST, Python entry points) - Fix pysixel dependency (removed broken dependency) Test coverage improvements: - Add DisplayRegistry tests - Add MultiDisplay tests - Add SixelDisplay tests - Add controller._get_display tests - Add effects controller command handling tests - Add benchmark regression tests (@pytest.mark.benchmark) - Add pytest marker for benchmark tests in pyproject.toml Documentation updates: - Update AGENTS.md with 56% coverage stats and effect plugin docs - Update README.md with Sixel display mode and benchmark commands - Add new modules to architecture section
73 lines
3.0 KiB
Python
73 lines
3.0 KiB
Python
import random
|
|
from datetime import datetime
|
|
|
|
from engine import config
|
|
from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
|
from engine.sources import FEEDS, POETRY_SOURCES
|
|
from engine.terminal import C_DIM, G_DIM, G_LO, RST, W_GHOST
|
|
|
|
|
|
class FirehoseEffect(EffectPlugin):
|
|
name = "firehose"
|
|
config = EffectConfig(enabled=True, intensity=1.0)
|
|
|
|
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
|
firehose_h = config.FIREHOSE_H if config.FIREHOSE else 0
|
|
if firehose_h <= 0 or not ctx.items:
|
|
return buf
|
|
|
|
result = list(buf)
|
|
intensity = self.config.intensity
|
|
h = ctx.terminal_height
|
|
|
|
for fr in range(firehose_h):
|
|
scr_row = h - firehose_h + fr + 1
|
|
fline = self._firehose_line(ctx.items, ctx.terminal_width, intensity)
|
|
result.append(f"\033[{scr_row};1H{fline}\033[K")
|
|
return result
|
|
|
|
def _firehose_line(self, items: list, w: int, intensity: float) -> str:
|
|
r = random.random()
|
|
if r < 0.35 * intensity:
|
|
title, src, ts = random.choice(items)
|
|
text = title[: w - 1]
|
|
color = random.choice([G_LO, G_DIM, W_GHOST, C_DIM])
|
|
return f"{color}{text}{RST}"
|
|
elif r < 0.55 * intensity:
|
|
d = random.choice([0.45, 0.55, 0.65, 0.75])
|
|
return "".join(
|
|
f"{random.choice([G_LO, G_DIM, C_DIM, W_GHOST])}"
|
|
f"{random.choice(config.GLITCH + config.KATA)}{RST}"
|
|
if random.random() < d
|
|
else " "
|
|
for _ in range(w)
|
|
)
|
|
elif r < 0.78 * intensity:
|
|
sources = FEEDS if config.MODE == "news" else POETRY_SOURCES
|
|
src = random.choice(list(sources.keys()))
|
|
msgs = [
|
|
f" SIGNAL :: {src} :: {datetime.now().strftime('%H:%M:%S.%f')[:-3]}",
|
|
f" ░░ FEED ACTIVE :: {src}",
|
|
f" >> DECODE 0x{random.randint(0x1000, 0xFFFF):04X} :: {src[:24]}",
|
|
f" ▒▒ ACQUIRE :: {random.choice(['TCP', 'UDP', 'RSS', 'ATOM', 'XML'])} :: {src}",
|
|
f" {''.join(random.choice(config.KATA) for _ in range(3))} STRM "
|
|
f"{random.randint(0, 255):02X}:{random.randint(0, 255):02X}",
|
|
]
|
|
text = random.choice(msgs)[: w - 1]
|
|
color = random.choice([G_LO, G_DIM, W_GHOST])
|
|
return f"{color}{text}{RST}"
|
|
else:
|
|
title, _, _ = random.choice(items)
|
|
start = random.randint(0, max(0, len(title) - 20))
|
|
frag = title[start : start + random.randint(10, 35)]
|
|
pad = random.randint(0, max(0, w - len(frag) - 8))
|
|
gp = "".join(
|
|
random.choice(config.GLITCH) for _ in range(random.randint(1, 3))
|
|
)
|
|
text = (" " * pad + gp + " " + frag)[: w - 1]
|
|
color = random.choice([G_LO, C_DIM, W_GHOST])
|
|
return f"{color}{text}{RST}"
|
|
|
|
def configure(self, config: EffectConfig) -> None:
|
|
self.config = config
|