Compare commits
4 Commits
docs/updat
...
testabilit
| Author | SHA1 | Date | |
|---|---|---|---|
| 3551cc249f | |||
| fba183526a | |||
| 7193e7487b | |||
| b5d6eeedc0 |
292
README.md
292
README.md
@@ -6,45 +6,25 @@ A full-screen terminal news ticker that renders live global headlines in large O
|
||||
|
||||
---
|
||||
|
||||
## Using
|
||||
|
||||
### Run
|
||||
## Run
|
||||
|
||||
```bash
|
||||
python3 mainline.py # news stream
|
||||
python3 mainline.py --poetry # literary consciousness mode
|
||||
python3 mainline.py -p # same
|
||||
python3 mainline.py --firehose # dense rapid-fire headline mode
|
||||
python3 mainline.py --display websocket # web browser display only
|
||||
python3 mainline.py --display both # terminal + web browser
|
||||
python3 mainline.py --refresh # force re-fetch (bypass cache)
|
||||
python3 mainline.py --no-font-picker # skip interactive font picker
|
||||
python3 mainline.py --font-file path.otf # use a specific font file
|
||||
python3 mainline.py --font-dir ~/fonts # scan a different font folder
|
||||
python3 mainline.py --font-index 1 # select face index within a collection
|
||||
```
|
||||
|
||||
Or with uv:
|
||||
First run bootstraps a local `.mainline_venv/` and installs deps (`feedparser`, `Pillow`, `sounddevice`, `numpy`). Subsequent runs start immediately, loading from cache.
|
||||
|
||||
```bash
|
||||
uv run mainline.py
|
||||
```
|
||||
---
|
||||
|
||||
First run bootstraps dependencies. Use `uv sync --all-extras` for mic support.
|
||||
|
||||
### Command & Control (C&C)
|
||||
|
||||
Control mainline remotely using `cmdline.py`:
|
||||
|
||||
```bash
|
||||
uv run cmdline.py # Interactive TUI
|
||||
uv run cmdline.py /effects list # List all effects
|
||||
uv run cmdline.py /effects stats # Show performance stats
|
||||
uv run cmdline.py -w /effects stats # Watch mode (auto-refresh)
|
||||
```
|
||||
|
||||
Commands are sent via ntfy.sh topics - useful for controlling a daemonized mainline instance.
|
||||
|
||||
### Config
|
||||
## Config
|
||||
|
||||
All constants live in `engine/config.py`:
|
||||
|
||||
@@ -53,50 +33,90 @@ All constants live in `engine/config.py`:
|
||||
| `HEADLINE_LIMIT` | `1000` | Total headlines per session |
|
||||
| `FEED_TIMEOUT` | `10` | Per-feed HTTP timeout (seconds) |
|
||||
| `MIC_THRESHOLD_DB` | `50` | dB floor above which glitches spike |
|
||||
| `NTFY_TOPIC` | klubhaus URL | ntfy.sh JSON stream for messages |
|
||||
| `NTFY_CC_CMD_TOPIC` | klubhaus URL | ntfy.sh topic for C&C commands |
|
||||
| `NTFY_CC_RESP_TOPIC` | klubhaus URL | ntfy.sh topic for C&C responses |
|
||||
| `NTFY_RECONNECT_DELAY` | `5` | Seconds before reconnecting after dropped SSE |
|
||||
| `MESSAGE_DISPLAY_SECS` | `30` | How long an ntfy message holds the screen |
|
||||
| `FONT_DIR` | `fonts/` | Folder scanned for `.otf`, `.ttf`, `.ttc` files |
|
||||
| `FONT_PATH` | first file in `FONT_DIR` | Active display font |
|
||||
| `FONT_PICKER` | `True` | Show interactive font picker at boot |
|
||||
| `FONT_PATH` | first file in `FONT_DIR` | Active display font (overridden by picker or `--font-file`) |
|
||||
| `FONT_INDEX` | `0` | Face index within a font collection file |
|
||||
| `FONT_PICKER` | `True` | Show interactive font picker at boot (`--no-font-picker` to skip) |
|
||||
| `FONT_SZ` | `60` | Font render size (affects block density) |
|
||||
| `RENDER_H` | `8` | Terminal rows per headline line |
|
||||
| `SSAA` | `4` | Super-sampling factor |
|
||||
| `SSAA` | `4` | Super-sampling factor (render at 4× then downsample) |
|
||||
| `SCROLL_DUR` | `5.625` | Seconds per headline |
|
||||
| `FRAME_DT` | `0.05` | Frame interval in seconds (20 FPS) |
|
||||
| `GRAD_SPEED` | `0.08` | Gradient sweep speed (cycles/sec, ~12s full sweep) |
|
||||
| `FIREHOSE_H` | `12` | Firehose zone height (terminal rows) |
|
||||
| `GRAD_SPEED` | `0.08` | Gradient sweep speed |
|
||||
| `NTFY_TOPIC` | klubhaus URL | ntfy.sh JSON endpoint to poll |
|
||||
| `NTFY_POLL_INTERVAL` | `15` | Seconds between ntfy polls |
|
||||
| `MESSAGE_DISPLAY_SECS` | `30` | How long an ntfy message holds the screen |
|
||||
|
||||
### Display Modes
|
||||
---
|
||||
|
||||
Mainline supports multiple display backends:
|
||||
## Fonts
|
||||
|
||||
- **Terminal** (`--display terminal`): ANSI terminal output (default)
|
||||
- **WebSocket** (`--display websocket`): Stream to web browser clients
|
||||
- **Sixel** (`--display sixel`): Sixel graphics in supported terminals (iTerm2, mintty)
|
||||
- **Both** (`--display both`): Terminal + WebSocket simultaneously
|
||||
A `fonts/` directory is bundled with demo faces (AlphatronDemo, CSBishopDrawn, CyberformDemo, KATA, Microbots, Neoform, Pixel Sparta, Robocops, Xeonic, and others). On startup, an interactive picker lists all discovered faces with a live half-block preview rendered at your configured size.
|
||||
|
||||
WebSocket mode serves a web client at http://localhost:8766 with ANSI color support and fullscreen mode.
|
||||
Navigation: `↑`/`↓` or `j`/`k` to move, `Enter` or `q` to select. The selected face persists for that session.
|
||||
|
||||
### Feeds
|
||||
To add your own fonts, drop `.otf`, `.ttf`, or `.ttc` files into `fonts/` (or point `--font-dir` at any other folder). Font collections (`.ttc`, multi-face `.otf`) are enumerated face-by-face.
|
||||
|
||||
---
|
||||
|
||||
## How it works
|
||||
|
||||
- On launch, the font picker scans `fonts/` and presents a live-rendered TUI for face selection; `--no-font-picker` skips directly to stream
|
||||
- Feeds are fetched and filtered on startup (sports and vapid content stripped); results are cached to `.mainline_cache_news.json` / `.mainline_cache_poetry.json` for fast restarts
|
||||
- Headlines are rasterized via Pillow with 4× SSAA into half-block characters (`▀▄█ `) at the configured font size
|
||||
- The ticker uses a sweeping white-hot → deep green gradient; ntfy messages use a complementary white-hot → magenta/maroon gradient to distinguish them visually
|
||||
- Subject-region detection runs a regex pass on each headline; matches trigger a Google Translate call and font swap to the appropriate script (CJK, Arabic, Devanagari, etc.) using macOS system fonts
|
||||
- The mic stream runs in a background thread, feeding RMS dB into the glitch probability calculation each frame
|
||||
- The viewport scrolls through a virtual canvas of pre-rendered blocks; fade zones at top and bottom dissolve characters probabilistically
|
||||
- An ntfy.sh poller runs in a background thread; incoming messages interrupt the scroll and render full-screen until dismissed or expired
|
||||
|
||||
---
|
||||
|
||||
## Architecture
|
||||
|
||||
`mainline.py` is a thin entrypoint (venv bootstrap → `engine.app.main()`). All logic lives in the `engine/` package:
|
||||
|
||||
```
|
||||
engine/
|
||||
__init__.py package marker
|
||||
app.py main(), font picker TUI, boot sequence, signal handler
|
||||
config.py constants, CLI flags, glyph tables
|
||||
sources.py FEEDS, POETRY_SOURCES, language/script maps
|
||||
terminal.py ANSI codes, tw/th, type_out, boot_ln
|
||||
filter.py HTML stripping, content filter
|
||||
translate.py Google Translate wrapper + region detection
|
||||
render.py OTF → half-block pipeline (SSAA, gradient)
|
||||
effects.py noise, glitch_bar, fade, firehose
|
||||
fetch.py RSS/Gutenberg fetching + cache load/save
|
||||
ntfy.py NtfyPoller — standalone, zero internal deps
|
||||
mic.py MicMonitor — standalone, graceful fallback
|
||||
scroll.py stream() frame loop + message rendering
|
||||
viewport.py terminal dimension tracking (tw/th)
|
||||
frame.py scroll step calculation, timing
|
||||
layers.py ticker zone, firehose, message overlay rendering
|
||||
eventbus.py thread-safe event publishing for decoupled communication
|
||||
events.py event types and definitions
|
||||
controller.py coordinates ntfy/mic monitoring and event publishing
|
||||
emitters.py background emitters for ntfy and mic
|
||||
types.py type definitions and dataclasses
|
||||
```
|
||||
|
||||
`ntfy.py` and `mic.py` have zero internal dependencies and can be imported by any other visualizer.
|
||||
|
||||
---
|
||||
|
||||
## Feeds
|
||||
|
||||
~25 sources across four categories: **Science & Technology**, **Economics & Business**, **World & Politics**, **Culture & Ideas**. Add or swap feeds in `engine/sources.py` → `FEEDS`.
|
||||
|
||||
**Poetry mode** pulls from Project Gutenberg: Whitman, Dickinson, Thoreau, Emerson. Sources are in `engine/sources.py` → `POETRY_SOURCES`.
|
||||
|
||||
### Fonts
|
||||
---
|
||||
|
||||
A `fonts/` directory is bundled with demo faces. On startup, an interactive picker lists all discovered faces with a live half-block preview.
|
||||
## ntfy.sh Integration
|
||||
|
||||
Navigation: `↑`/`↓` or `j`/`k` to move, `Enter` or `q` to select.
|
||||
|
||||
To add your own fonts, drop `.otf`, `.ttf`, or `.ttc` files into `fonts/`.
|
||||
|
||||
### ntfy.sh
|
||||
|
||||
Mainline polls a configurable ntfy.sh topic in the background. When a message arrives, the scroll pauses and the message renders full-screen.
|
||||
Mainline polls a configurable ntfy.sh topic in the background. When a message arrives, the scroll pauses and the message renders full-screen for `MESSAGE_DISPLAY_SECS` seconds, then the stream resumes.
|
||||
|
||||
To push a message:
|
||||
|
||||
@@ -104,160 +124,44 @@ To push a message:
|
||||
curl -d "Body text" -H "Title: Alert title" https://ntfy.sh/your_topic
|
||||
```
|
||||
|
||||
---
|
||||
Update `NTFY_TOPIC` in `engine/config.py` to point at your own topic. The `NtfyPoller` class is fully standalone and can be reused by other visualizers:
|
||||
|
||||
## Internals
|
||||
|
||||
### How it works
|
||||
|
||||
- On launch, the font picker scans `fonts/` and presents a live-rendered TUI for face selection
|
||||
- Feeds are fetched and filtered on startup; results are cached for fast restarts
|
||||
- Headlines are rasterized via Pillow with 4× SSAA into half-block characters
|
||||
- The ticker uses a sweeping white-hot → deep green gradient
|
||||
- Subject-region detection triggers Google Translate and font swap for non-Latin scripts
|
||||
- The mic stream runs in a background thread, feeding RMS dB into glitch probability
|
||||
- The viewport scrolls through pre-rendered blocks with fade zones
|
||||
- An ntfy.sh SSE stream runs in a background thread for messages and C&C commands
|
||||
|
||||
### Architecture
|
||||
|
||||
```
|
||||
engine/
|
||||
__init__.py package marker
|
||||
app.py main(), font picker TUI, boot sequence, C&C poller
|
||||
config.py constants, CLI flags, glyph tables
|
||||
sources.py FEEDS, POETRY_SOURCES, language/script maps
|
||||
terminal.py ANSI codes, tw/th, type_out, boot_ln
|
||||
filter.py HTML stripping, content filter
|
||||
translate.py Google Translate wrapper + region detection
|
||||
render.py OTF → half-block pipeline (SSAA, gradient)
|
||||
effects/ plugin architecture for visual effects
|
||||
types.py EffectPlugin ABC, EffectConfig, EffectContext
|
||||
registry.py effect registration and lookup
|
||||
chain.py effect pipeline chaining
|
||||
controller.py handles /effects commands
|
||||
performance.py performance monitoring
|
||||
legacy.py legacy functional effects
|
||||
effects_plugins/ effect plugin implementations
|
||||
noise.py noise effect
|
||||
fade.py fade effect
|
||||
glitch.py glitch effect
|
||||
firehose.py firehose effect
|
||||
fetch.py RSS/Gutenberg fetching + cache
|
||||
ntfy.py NtfyPoller — standalone, zero internal deps
|
||||
mic.py MicMonitor — standalone, graceful fallback
|
||||
scroll.py stream() frame loop + message rendering
|
||||
viewport.py terminal dimension tracking
|
||||
frame.py scroll step calculation, timing
|
||||
layers.py ticker zone, firehose, message overlay
|
||||
eventbus.py thread-safe event publishing
|
||||
events.py event types and definitions
|
||||
controller.py coordinates ntfy/mic monitoring
|
||||
emitters.py background emitters
|
||||
types.py type definitions
|
||||
display/ Display backend system
|
||||
__init__.py DisplayRegistry, get_monitor
|
||||
backends/
|
||||
terminal.py ANSI terminal display
|
||||
websocket.py WebSocket server for browser clients
|
||||
sixel.py Sixel graphics (pure Python)
|
||||
null.py headless display for testing
|
||||
multi.py forwards to multiple displays
|
||||
benchmark.py performance benchmarking tool
|
||||
```python
|
||||
from engine.ntfy import NtfyPoller
|
||||
poller = NtfyPoller("https://ntfy.sh/my_topic/json?since=20s&poll=1")
|
||||
poller.start()
|
||||
# in render loop:
|
||||
msg = poller.get_active_message() # returns (title, body, timestamp) or None
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Development
|
||||
|
||||
### Setup
|
||||
|
||||
Requires Python 3.10+ and [uv](https://docs.astral.sh/uv/).
|
||||
|
||||
```bash
|
||||
uv sync # minimal (no mic)
|
||||
uv sync --all-extras # with mic support
|
||||
uv sync --all-extras --group dev # full dev environment
|
||||
```
|
||||
|
||||
### Tasks
|
||||
|
||||
With [mise](https://mise.jdx.dev/):
|
||||
|
||||
```bash
|
||||
mise run test # run test suite
|
||||
mise run test-cov # run with coverage report
|
||||
|
||||
mise run lint # ruff check
|
||||
mise run lint-fix # ruff check --fix
|
||||
mise run format # ruff format
|
||||
|
||||
mise run run # terminal display
|
||||
mise run run-websocket # web display only
|
||||
mise run run-sixel # sixel graphics
|
||||
mise run run-both # terminal + web
|
||||
mise run run-client # both + open browser
|
||||
|
||||
mise run cmd # C&C command interface
|
||||
mise run cmd-stats # watch effects stats
|
||||
|
||||
mise run benchmark # run performance benchmarks
|
||||
mise run benchmark-json # save as JSON
|
||||
|
||||
mise run topics-init # initialize ntfy topics
|
||||
```
|
||||
|
||||
### Testing
|
||||
|
||||
```bash
|
||||
uv run pytest
|
||||
uv run pytest --cov=engine --cov-report=term-missing
|
||||
|
||||
# Run with mise
|
||||
mise run test
|
||||
mise run test-cov
|
||||
|
||||
# Run performance benchmarks
|
||||
mise run benchmark
|
||||
mise run benchmark-json
|
||||
|
||||
# Run benchmark hook mode (for CI)
|
||||
uv run python -m engine.benchmark --hook
|
||||
```
|
||||
|
||||
Performance regression tests are in `tests/test_benchmark.py` marked with `@pytest.mark.benchmark`.
|
||||
|
||||
### Linting
|
||||
|
||||
```bash
|
||||
uv run ruff check engine/ mainline.py
|
||||
uv run ruff format engine/ mainline.py
|
||||
```
|
||||
|
||||
Pre-commit hooks run lint automatically via `hk`.
|
||||
|
||||
---
|
||||
|
||||
## Roadmap
|
||||
## Ideas / Future
|
||||
|
||||
### Performance
|
||||
- Concurrent feed fetching with ThreadPoolExecutor
|
||||
- Background feed refresh daemon
|
||||
- Translation pre-fetch during boot
|
||||
- **Concurrent feed fetching** — startup currently blocks sequentially on ~25 HTTP requests; `concurrent.futures.ThreadPoolExecutor` would cut load time to the slowest single feed
|
||||
- **Background refresh** — re-fetch feeds in a daemon thread so a long session stays current without restart
|
||||
- **Translation pre-fetch** — run translate calls concurrently during the boot sequence rather than on first render
|
||||
|
||||
### Graphics
|
||||
- Matrix rain katakana underlay
|
||||
- CRT scanline simulation
|
||||
- Sixel/iTerm2 inline images
|
||||
- Parallax secondary column
|
||||
- **Matrix rain underlay** — katakana column rain rendered at low opacity beneath the scrolling blocks as a background layer
|
||||
- **CRT simulation** — subtle dim scanlines every N rows, occasional brightness ripple across the full screen
|
||||
- **Sixel / iTerm2 inline images** — bypass half-blocks entirely and stream actual bitmap frames for true resolution; would require a capable terminal
|
||||
- **Parallax secondary column** — a second, dimmer, faster-scrolling stream of ambient text at reduced opacity on one side
|
||||
|
||||
### Cyberpunk Vibes
|
||||
- Keyword watch list with strobe effects
|
||||
- Breaking interrupt with synthesized audio
|
||||
- Live data overlay (BTC, ISS position)
|
||||
- Theme switcher (amber, ice, red)
|
||||
- Persona modes (surveillance, oracle, underground)
|
||||
- **Keyword watch list** — highlight or strobe any headline matching tracked terms (names, topics, tickers)
|
||||
- **Breaking interrupt** — full-screen flash + synthesized blip when a high-priority keyword hits
|
||||
- **Live data overlay** — secondary ticker strip at screen edge: BTC price, ISS position, geomagnetic index
|
||||
- **Theme switcher** — `--amber` (phosphor), `--ice` (electric cyan), `--red` (alert state) palette modes via CLI flag
|
||||
- **Persona modes** — `--surveillance`, `--oracle`, `--underground` as feed presets with matching color themes and boot copy
|
||||
- **Synthesized audio** — short static bursts tied to glitch events, independent of mic input
|
||||
|
||||
### Extensibility
|
||||
- **serve.py** — HTTP server that imports `engine.render` and `engine.fetch` directly to stream 1-bit bitmaps to an ESP32 display
|
||||
- **Rust port** — `ntfy.py` and `render.py` are the natural first targets; clear module boundaries make incremental porting viable
|
||||
|
||||
---
|
||||
|
||||
*Python 3.10+. Primary display font is user-selectable via bundled `fonts/` picker.*
|
||||
*macOS only (script/system font paths for translation are hardcoded). Primary display font is user-selectable via the bundled `fonts/` picker. Python 3.9+.*
|
||||
# test
|
||||
|
||||
250
cmdline.py
250
cmdline.py
@@ -1,250 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Command-line utility for interacting with mainline via ntfy.
|
||||
|
||||
Usage:
|
||||
python cmdline.py # Interactive TUI mode
|
||||
python cmdline.py --help # Show help
|
||||
python cmdline.py /effects list # Send single command via ntfy
|
||||
python cmdline.py /effects stats # Get performance stats via ntfy
|
||||
python cmdline.py -w /effects stats # Watch mode (polls for stats)
|
||||
|
||||
The TUI mode provides:
|
||||
- Arrow keys to navigate command history
|
||||
- Tab completion for commands
|
||||
- Auto-refresh for performance stats
|
||||
|
||||
C&C works like a serial port:
|
||||
1. Send command to ntfy_cc_topic
|
||||
2. Mainline receives, processes, responds to same topic
|
||||
3. Cmdline polls for response
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
import threading
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
from engine import config
|
||||
from engine.terminal import CLR, CURSOR_OFF, CURSOR_ON, G_DIM, G_HI, RST, W_GHOST
|
||||
|
||||
try:
|
||||
CC_CMD_TOPIC = config.NTFY_CC_CMD_TOPIC
|
||||
CC_RESP_TOPIC = config.NTFY_CC_RESP_TOPIC
|
||||
except AttributeError:
|
||||
CC_CMD_TOPIC = "https://ntfy.sh/klubhaus_terminal_mainline_cc_cmd/json"
|
||||
CC_RESP_TOPIC = "https://ntfy.sh/klubhaus_terminal_mainline_cc_resp/json"
|
||||
|
||||
|
||||
class NtfyResponsePoller:
|
||||
"""Polls ntfy for command responses."""
|
||||
|
||||
def __init__(self, cmd_topic: str, resp_topic: str, timeout: float = 10.0):
|
||||
self.cmd_topic = cmd_topic
|
||||
self.resp_topic = resp_topic
|
||||
self.timeout = timeout
|
||||
self._last_id = None
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def _build_url(self) -> str:
|
||||
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
|
||||
|
||||
parsed = urlparse(self.resp_topic)
|
||||
params = parse_qs(parsed.query, keep_blank_values=True)
|
||||
params["since"] = [self._last_id if self._last_id else "20s"]
|
||||
new_query = urlencode({k: v[0] for k, v in params.items()})
|
||||
return urlunparse(parsed._replace(query=new_query))
|
||||
|
||||
def send_and_wait(self, cmd: str) -> str:
|
||||
"""Send command and wait for response."""
|
||||
url = self.cmd_topic.replace("/json", "")
|
||||
data = cmd.encode("utf-8")
|
||||
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=data,
|
||||
headers={
|
||||
"User-Agent": "mainline-cmdline/0.1",
|
||||
"Content-Type": "text/plain",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
|
||||
try:
|
||||
urllib.request.urlopen(req, timeout=5)
|
||||
except Exception as e:
|
||||
return f"Error sending command: {e}"
|
||||
|
||||
return self._wait_for_response(cmd)
|
||||
|
||||
def _wait_for_response(self, expected_cmd: str = "") -> str:
|
||||
"""Poll for response message."""
|
||||
start = time.time()
|
||||
while time.time() - start < self.timeout:
|
||||
try:
|
||||
url = self._build_url()
|
||||
req = urllib.request.Request(
|
||||
url, headers={"User-Agent": "mainline-cmdline/0.1"}
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
for line in resp:
|
||||
try:
|
||||
data = json.loads(line.decode("utf-8", errors="replace"))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if data.get("event") == "message":
|
||||
self._last_id = data.get("id")
|
||||
msg = data.get("message", "")
|
||||
if msg:
|
||||
return msg
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(0.5)
|
||||
return "Timeout waiting for response"
|
||||
|
||||
|
||||
AVAILABLE_COMMANDS = """Available commands:
|
||||
/effects list - List all effects and status
|
||||
/effects <name> on - Enable an effect
|
||||
/effects <name> off - Disable an effect
|
||||
/effects <name> intensity <0.0-1.0> - Set effect intensity
|
||||
/effects reorder <name1>,<name2>,... - Reorder pipeline
|
||||
/effects stats - Show performance statistics
|
||||
/help - Show this help
|
||||
/quit - Exit
|
||||
"""
|
||||
|
||||
|
||||
def print_header():
|
||||
w = 60
|
||||
print(CLR, end="")
|
||||
print(CURSOR_OFF, end="")
|
||||
print(f"\033[1;1H", end="")
|
||||
print(f" \033[1;38;5;231m╔{'═' * (w - 6)}╗\033[0m")
|
||||
print(
|
||||
f" \033[1;38;5;231m║\033[0m \033[1;38;5;82mMAINLINE\033[0m \033[3;38;5;245mCommand Center\033[0m \033[1;38;5;231m ║\033[0m"
|
||||
)
|
||||
print(f" \033[1;38;5;231m╚{'═' * (w - 6)}╝\033[0m")
|
||||
print(f" \033[2;38;5;37mCMD: {CC_CMD_TOPIC.split('/')[-2]}\033[0m")
|
||||
print(f" \033[2;38;5;37mRESP: {CC_RESP_TOPIC.split('/')[-2]}\033[0m")
|
||||
print()
|
||||
|
||||
|
||||
def print_response(response: str, is_error: bool = False) -> None:
|
||||
"""Print response with nice formatting."""
|
||||
print()
|
||||
if is_error:
|
||||
print(f" \033[1;38;5;196m✗ Error\033[0m")
|
||||
print(f" \033[38;5;196m{'─' * 40}\033[0m")
|
||||
else:
|
||||
print(f" \033[1;38;5;82m✓ Response\033[0m")
|
||||
print(f" \033[38;5;37m{'─' * 40}\033[0m")
|
||||
|
||||
for line in response.split("\n"):
|
||||
print(f" {line}")
|
||||
print()
|
||||
|
||||
|
||||
def interactive_mode():
|
||||
"""Interactive TUI for sending commands."""
|
||||
import readline
|
||||
|
||||
print_header()
|
||||
poller = NtfyResponsePoller(CC_CMD_TOPIC, CC_RESP_TOPIC)
|
||||
|
||||
print(f" \033[38;5;245mType /help for commands, /quit to exit\033[0m")
|
||||
print()
|
||||
|
||||
while True:
|
||||
try:
|
||||
cmd = input(f" \033[1;38;5;82m❯\033[0m {G_HI}").strip()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print()
|
||||
break
|
||||
|
||||
if not cmd:
|
||||
continue
|
||||
|
||||
if cmd.startswith("/"):
|
||||
if cmd == "/quit" or cmd == "/exit":
|
||||
print(f"\n \033[1;38;5;245mGoodbye!{RST}\n")
|
||||
break
|
||||
|
||||
if cmd == "/help":
|
||||
print(f"\n{AVAILABLE_COMMANDS}\n")
|
||||
continue
|
||||
|
||||
print(f" \033[38;5;245m⟳ Sending to mainline...{RST}")
|
||||
result = poller.send_and_wait(cmd)
|
||||
print_response(result, is_error=result.startswith("Error"))
|
||||
else:
|
||||
print(f"\n \033[1;38;5;196m⚠ Commands must start with /{RST}\n")
|
||||
|
||||
print(CURSOR_ON, end="")
|
||||
return 0
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Mainline command-line interface",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog=AVAILABLE_COMMANDS,
|
||||
)
|
||||
parser.add_argument(
|
||||
"command",
|
||||
nargs="?",
|
||||
default=None,
|
||||
help="Command to send (e.g., /effects list)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--watch",
|
||||
"-w",
|
||||
action="store_true",
|
||||
help="Watch mode: continuously poll for stats (Ctrl+C to exit)",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.command is None:
|
||||
return interactive_mode()
|
||||
|
||||
poller = NtfyResponsePoller(CC_CMD_TOPIC, CC_RESP_TOPIC)
|
||||
|
||||
if args.watch and "/effects stats" in args.command:
|
||||
import signal
|
||||
|
||||
def handle_sigterm(*_):
|
||||
print(f"\n \033[1;38;5;245mStopped watching{RST}")
|
||||
print(CURSOR_ON, end="")
|
||||
sys.exit(0)
|
||||
|
||||
signal.signal(signal.SIGTERM, handle_sigterm)
|
||||
|
||||
print_header()
|
||||
print(f" \033[38;5;245mWatching /effects stats (Ctrl+C to exit)...{RST}\n")
|
||||
try:
|
||||
while True:
|
||||
result = poller.send_and_wait(args.command)
|
||||
print(f"\033[2J\033[1;1H", end="")
|
||||
print(
|
||||
f" \033[1;38;5;82m❯\033[0m Performance Stats - \033[1;38;5;245m{time.strftime('%H:%M:%S')}{RST}"
|
||||
)
|
||||
print(f" \033[38;5;37m{'─' * 44}{RST}")
|
||||
for line in result.split("\n"):
|
||||
print(f" {line}")
|
||||
time.sleep(2)
|
||||
except KeyboardInterrupt:
|
||||
print(f"\n \033[1;38;5;245mStopped watching{RST}")
|
||||
return 0
|
||||
return 0
|
||||
|
||||
result = poller.send_and_wait(args.command)
|
||||
print(result)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,894 +0,0 @@
|
||||
# Color Scheme Switcher Implementation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** Implement interactive color theme picker at startup that lets users choose between green, orange, or purple gradients with complementary message queue colors.
|
||||
|
||||
**Architecture:** New `themes.py` data module defines Theme class and THEME_REGISTRY. Config adds `ACTIVE_THEME` global set by picker. Render functions read from active theme instead of hardcoded constants. App adds picker UI that mirrors font picker pattern.
|
||||
|
||||
**Tech Stack:** Python 3.10+, ANSI 256-color codes, existing terminal I/O utilities
|
||||
|
||||
---
|
||||
|
||||
## File Structure
|
||||
|
||||
| File | Purpose | Change Type |
|
||||
|------|---------|------------|
|
||||
| `engine/themes.py` | Theme class, THEME_REGISTRY, color codes | Create |
|
||||
| `engine/config.py` | ACTIVE_THEME global, set_active_theme() | Modify |
|
||||
| `engine/render.py` | Replace GRAD_COLS/MSG_GRAD_COLS with config lookup | Modify |
|
||||
| `engine/scroll.py` | Update message gradient call | Modify |
|
||||
| `engine/app.py` | pick_color_theme(), call in main() | Modify |
|
||||
| `tests/test_themes.py` | Theme class and registry unit tests | Create |
|
||||
|
||||
---
|
||||
|
||||
## Chunk 1: Theme Data Module
|
||||
|
||||
### Task 1: Create themes.py with Theme class and registry
|
||||
|
||||
**Files:**
|
||||
- Create: `engine/themes.py`
|
||||
- Test: `tests/test_themes.py`
|
||||
|
||||
- [ ] **Step 1: Write failing test for Theme class**
|
||||
|
||||
Create `tests/test_themes.py`:
|
||||
|
||||
```python
|
||||
"""Test color themes and registry."""
|
||||
from engine.themes import Theme, THEME_REGISTRY, get_theme
|
||||
|
||||
|
||||
def test_theme_construction():
|
||||
"""Theme stores name and gradient lists."""
|
||||
main = ["\033[1;38;5;231m"] * 12
|
||||
msg = ["\033[1;38;5;225m"] * 12
|
||||
theme = Theme(name="Test Green", main_gradient=main, message_gradient=msg)
|
||||
|
||||
assert theme.name == "Test Green"
|
||||
assert theme.main_gradient == main
|
||||
assert theme.message_gradient == msg
|
||||
|
||||
|
||||
def test_gradient_length():
|
||||
"""Each gradient must have exactly 12 ANSI codes."""
|
||||
for theme_id, theme in THEME_REGISTRY.items():
|
||||
assert len(theme.main_gradient) == 12, f"{theme_id} main gradient wrong length"
|
||||
assert len(theme.message_gradient) == 12, f"{theme_id} message gradient wrong length"
|
||||
|
||||
|
||||
def test_theme_registry_has_three_themes():
|
||||
"""Registry contains green, orange, purple."""
|
||||
assert len(THEME_REGISTRY) == 3
|
||||
assert "green" in THEME_REGISTRY
|
||||
assert "orange" in THEME_REGISTRY
|
||||
assert "purple" in THEME_REGISTRY
|
||||
|
||||
|
||||
def test_get_theme_valid():
|
||||
"""get_theme returns Theme object for valid ID."""
|
||||
theme = get_theme("green")
|
||||
assert isinstance(theme, Theme)
|
||||
assert theme.name == "Verdant Green"
|
||||
|
||||
|
||||
def test_get_theme_invalid():
|
||||
"""get_theme raises KeyError for invalid ID."""
|
||||
with pytest.raises(KeyError):
|
||||
get_theme("invalid_theme")
|
||||
|
||||
|
||||
def test_green_theme_unchanged():
|
||||
"""Green theme uses original green → magenta colors."""
|
||||
green_theme = get_theme("green")
|
||||
# First color should be white (bold)
|
||||
assert green_theme.main_gradient[0] == "\033[1;38;5;231m"
|
||||
# Last deep green
|
||||
assert green_theme.main_gradient[9] == "\033[38;5;22m"
|
||||
# Message gradient is magenta
|
||||
assert green_theme.message_gradient[9] == "\033[38;5;89m"
|
||||
```
|
||||
|
||||
Run: `pytest tests/test_themes.py -v`
|
||||
Expected: FAIL (module doesn't exist)
|
||||
|
||||
- [ ] **Step 2: Create themes.py with Theme class and finalized gradients**
|
||||
|
||||
Create `engine/themes.py`:
|
||||
|
||||
```python
|
||||
"""Color theme definitions and registry."""
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class Theme:
|
||||
"""Encapsulates a color scheme: name, main gradient, message gradient."""
|
||||
|
||||
def __init__(self, name: str, main_gradient: list[str], message_gradient: list[str]):
|
||||
"""Initialize theme with display name and gradient lists.
|
||||
|
||||
Args:
|
||||
name: Display name (e.g., "Verdant Green")
|
||||
main_gradient: List of 12 ANSI 256-color codes (white → primary color)
|
||||
message_gradient: List of 12 ANSI codes (white → complementary color)
|
||||
"""
|
||||
self.name = name
|
||||
self.main_gradient = main_gradient
|
||||
self.message_gradient = message_gradient
|
||||
|
||||
|
||||
# ─── FINALIZED GRADIENTS ──────────────────────────────────────────────────
|
||||
# Each gradient: white → primary/complementary, 12 steps total
|
||||
# Format: "\033[<brightness>;<color>m" where color is 38;5;<colorcode>
|
||||
|
||||
_GREEN_MAIN = [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;195m", # pale white-tint
|
||||
"\033[38;5;123m", # bright cyan
|
||||
"\033[38;5;118m", # bright lime
|
||||
"\033[38;5;82m", # lime
|
||||
"\033[38;5;46m", # bright green
|
||||
"\033[38;5;40m", # green
|
||||
"\033[38;5;34m", # medium green
|
||||
"\033[38;5;28m", # dark green
|
||||
"\033[38;5;22m", # deep green
|
||||
"\033[2;38;5;22m", # dim deep green
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
_GREEN_MESSAGE = [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;225m", # pale pink-white
|
||||
"\033[38;5;219m", # bright pink
|
||||
"\033[38;5;213m", # hot pink
|
||||
"\033[38;5;207m", # magenta
|
||||
"\033[38;5;201m", # bright magenta
|
||||
"\033[38;5;165m", # orchid-red
|
||||
"\033[38;5;161m", # ruby-magenta
|
||||
"\033[38;5;125m", # dark magenta
|
||||
"\033[38;5;89m", # deep maroon-magenta
|
||||
"\033[2;38;5;89m", # dim deep maroon-magenta
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
_ORANGE_MAIN = [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;215m", # pale orange-white
|
||||
"\033[38;5;209m", # bright orange
|
||||
"\033[38;5;208m", # vibrant orange
|
||||
"\033[38;5;202m", # orange
|
||||
"\033[38;5;166m", # dark orange
|
||||
"\033[38;5;130m", # burnt orange
|
||||
"\033[38;5;94m", # rust
|
||||
"\033[38;5;58m", # dark rust
|
||||
"\033[38;5;94m", # rust (hold)
|
||||
"\033[2;38;5;94m", # dim rust
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
_ORANGE_MESSAGE = [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;195m", # pale cyan-white
|
||||
"\033[38;5;33m", # bright blue
|
||||
"\033[38;5;27m", # blue
|
||||
"\033[38;5;21m", # deep blue
|
||||
"\033[38;5;21m", # deep blue (hold)
|
||||
"\033[38;5;21m", # deep blue (hold)
|
||||
"\033[38;5;18m", # navy
|
||||
"\033[38;5;18m", # navy (hold)
|
||||
"\033[38;5;18m", # navy (hold)
|
||||
"\033[2;38;5;18m", # dim navy
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
_PURPLE_MAIN = [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;225m", # pale purple-white
|
||||
"\033[38;5;177m", # bright purple
|
||||
"\033[38;5;171m", # vibrant purple
|
||||
"\033[38;5;165m", # purple
|
||||
"\033[38;5;135m", # medium purple
|
||||
"\033[38;5;129m", # purple
|
||||
"\033[38;5;93m", # dark purple
|
||||
"\033[38;5;57m", # deep purple
|
||||
"\033[38;5;57m", # deep purple (hold)
|
||||
"\033[2;38;5;57m", # dim deep purple
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
_PURPLE_MESSAGE = [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;226m", # pale yellow-white
|
||||
"\033[38;5;226m", # bright yellow
|
||||
"\033[38;5;220m", # yellow
|
||||
"\033[38;5;220m", # yellow (hold)
|
||||
"\033[38;5;184m", # dark yellow
|
||||
"\033[38;5;184m", # dark yellow (hold)
|
||||
"\033[38;5;178m", # olive-yellow
|
||||
"\033[38;5;178m", # olive-yellow (hold)
|
||||
"\033[38;5;172m", # golden
|
||||
"\033[2;38;5;172m", # dim golden
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
# ─── THEME REGISTRY ───────────────────────────────────────────────────────
|
||||
|
||||
THEME_REGISTRY = {
|
||||
"green": Theme(
|
||||
name="Verdant Green",
|
||||
main_gradient=_GREEN_MAIN,
|
||||
message_gradient=_GREEN_MESSAGE,
|
||||
),
|
||||
"orange": Theme(
|
||||
name="Molten Orange",
|
||||
main_gradient=_ORANGE_MAIN,
|
||||
message_gradient=_ORANGE_MESSAGE,
|
||||
),
|
||||
"purple": Theme(
|
||||
name="Violet Purple",
|
||||
main_gradient=_PURPLE_MAIN,
|
||||
message_gradient=_PURPLE_MESSAGE,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def get_theme(theme_id: str) -> Theme:
|
||||
"""Retrieve a theme by ID.
|
||||
|
||||
Args:
|
||||
theme_id: One of "green", "orange", "purple"
|
||||
|
||||
Returns:
|
||||
Theme object
|
||||
|
||||
Raises:
|
||||
KeyError: If theme_id not found in registry
|
||||
"""
|
||||
if theme_id not in THEME_REGISTRY:
|
||||
raise KeyError(f"Unknown theme: {theme_id}. Available: {list(THEME_REGISTRY.keys())}")
|
||||
return THEME_REGISTRY[theme_id]
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Run tests to verify they pass**
|
||||
|
||||
Run: `pytest tests/test_themes.py -v`
|
||||
Expected: PASS (all 6 tests)
|
||||
|
||||
- [ ] **Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add engine/themes.py tests/test_themes.py
|
||||
git commit -m "feat: create Theme class and registry with finalized color gradients
|
||||
|
||||
- Define Theme class to encapsulate name and main/message gradients
|
||||
- Create THEME_REGISTRY with green, orange, purple themes
|
||||
- Each gradient has 12 ANSI 256-color codes finalized
|
||||
- Complementary color pairs: green/magenta, orange/blue, purple/yellow
|
||||
- Add get_theme() lookup with error handling
|
||||
- Add comprehensive unit tests"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Chunk 2: Config Integration
|
||||
|
||||
### Task 2: Add ACTIVE_THEME global and set_active_theme() to config.py
|
||||
|
||||
**Files:**
|
||||
- Modify: `engine/config.py:1-30`
|
||||
- Test: `tests/test_config.py` (expand existing)
|
||||
|
||||
- [ ] **Step 1: Write failing tests for config changes**
|
||||
|
||||
Add to `tests/test_config.py`:
|
||||
|
||||
```python
|
||||
def test_active_theme_initially_none():
|
||||
"""ACTIVE_THEME is None before initialization."""
|
||||
# This test may fail if config is already initialized
|
||||
# We'll set it to None first for testing
|
||||
import engine.config
|
||||
engine.config.ACTIVE_THEME = None
|
||||
assert engine.config.ACTIVE_THEME is None
|
||||
|
||||
|
||||
def test_set_active_theme_green():
|
||||
"""set_active_theme('green') sets ACTIVE_THEME to green theme."""
|
||||
from engine.config import set_active_theme
|
||||
from engine.themes import get_theme
|
||||
|
||||
set_active_theme("green")
|
||||
|
||||
assert config.ACTIVE_THEME is not None
|
||||
assert config.ACTIVE_THEME.name == "Verdant Green"
|
||||
assert config.ACTIVE_THEME == get_theme("green")
|
||||
|
||||
|
||||
def test_set_active_theme_default():
|
||||
"""set_active_theme() with no args defaults to green."""
|
||||
from engine.config import set_active_theme
|
||||
|
||||
set_active_theme()
|
||||
|
||||
assert config.ACTIVE_THEME.name == "Verdant Green"
|
||||
|
||||
|
||||
def test_set_active_theme_invalid():
|
||||
"""set_active_theme() with invalid ID raises KeyError."""
|
||||
from engine.config import set_active_theme
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
set_active_theme("invalid")
|
||||
```
|
||||
|
||||
Run: `pytest tests/test_config.py -v`
|
||||
Expected: FAIL (functions don't exist yet)
|
||||
|
||||
- [ ] **Step 2: Add ACTIVE_THEME global and set_active_theme() to config.py**
|
||||
|
||||
Edit `engine/config.py`, add after line 30 (after `_resolve_font_path` function):
|
||||
|
||||
```python
|
||||
# ─── COLOR THEME ──────────────────────────────────────────────────────────
|
||||
ACTIVE_THEME = None # set by set_active_theme() after picker
|
||||
|
||||
|
||||
def set_active_theme(theme_id: str = "green"):
|
||||
"""Set the active color theme. Defaults to 'green' if not specified.
|
||||
|
||||
Args:
|
||||
theme_id: One of "green", "orange", "purple"
|
||||
|
||||
Raises:
|
||||
KeyError: If theme_id is invalid
|
||||
"""
|
||||
global ACTIVE_THEME
|
||||
from engine import themes
|
||||
ACTIVE_THEME = themes.get_theme(theme_id)
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Remove hardcoded GRAD_COLS and MSG_GRAD_COLS from render.py**
|
||||
|
||||
Edit `engine/render.py`, find and delete lines 20-49 (the hardcoded gradient arrays):
|
||||
|
||||
```python
|
||||
# DELETED:
|
||||
# GRAD_COLS = [...]
|
||||
# MSG_GRAD_COLS = [...]
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Run tests to verify they pass**
|
||||
|
||||
Run: `pytest tests/test_config.py::test_active_theme_initially_none -v`
|
||||
Run: `pytest tests/test_config.py::test_set_active_theme_green -v`
|
||||
Run: `pytest tests/test_config.py::test_set_active_theme_default -v`
|
||||
Run: `pytest tests/test_config.py::test_set_active_theme_invalid -v`
|
||||
|
||||
Expected: PASS (all 4 new tests)
|
||||
|
||||
- [ ] **Step 5: Verify existing config tests still pass**
|
||||
|
||||
Run: `pytest tests/test_config.py -v`
|
||||
|
||||
Expected: PASS (all existing + new tests)
|
||||
|
||||
- [ ] **Step 6: Commit**
|
||||
|
||||
```bash
|
||||
git add engine/config.py tests/test_config.py
|
||||
git commit -m "feat: add ACTIVE_THEME global and set_active_theme() to config
|
||||
|
||||
- Add ACTIVE_THEME global (initialized to None)
|
||||
- Add set_active_theme(theme_id) function with green default
|
||||
- Remove hardcoded GRAD_COLS and MSG_GRAD_COLS (move to themes.py)
|
||||
- Add comprehensive tests for theme setting"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Chunk 3: Render Pipeline Integration
|
||||
|
||||
### Task 3: Update render.py to use config.ACTIVE_THEME
|
||||
|
||||
**Files:**
|
||||
- Modify: `engine/render.py:15-220`
|
||||
- Test: `tests/test_render.py` (expand existing)
|
||||
|
||||
- [ ] **Step 1: Write failing test for lr_gradient with theme**
|
||||
|
||||
Add to `tests/test_render.py`:
|
||||
|
||||
```python
|
||||
def test_lr_gradient_uses_active_theme(monkeypatch):
|
||||
"""lr_gradient uses config.ACTIVE_THEME when cols=None."""
|
||||
from engine import config, render
|
||||
from engine.themes import get_theme
|
||||
|
||||
# Set orange theme
|
||||
config.set_active_theme("orange")
|
||||
|
||||
# Create simple rows
|
||||
rows = ["test row"]
|
||||
result = render.lr_gradient(rows, offset=0, cols=None)
|
||||
|
||||
# Result should start with first color from orange main gradient
|
||||
assert result[0].startswith("\033[1;38;5;231m") # white (same for all)
|
||||
|
||||
|
||||
def test_lr_gradient_fallback_when_no_theme(monkeypatch):
|
||||
"""lr_gradient uses fallback when ACTIVE_THEME is None."""
|
||||
from engine import config, render
|
||||
|
||||
# Clear active theme
|
||||
config.ACTIVE_THEME = None
|
||||
|
||||
rows = ["test row"]
|
||||
result = render.lr_gradient(rows, offset=0, cols=None)
|
||||
|
||||
# Should not crash and should return something
|
||||
assert result is not None
|
||||
assert len(result) > 0
|
||||
|
||||
|
||||
def test_default_green_gradient_length():
|
||||
"""_default_green_gradient returns 12 colors."""
|
||||
from engine import render
|
||||
|
||||
colors = render._default_green_gradient()
|
||||
assert len(colors) == 12
|
||||
```
|
||||
|
||||
Run: `pytest tests/test_render.py::test_lr_gradient_uses_active_theme -v`
|
||||
Expected: FAIL (function signature doesn't match)
|
||||
|
||||
- [ ] **Step 2: Update lr_gradient() to use config.ACTIVE_THEME**
|
||||
|
||||
Edit `engine/render.py`, find the `lr_gradient()` function (around line 194) and update it:
|
||||
|
||||
```python
|
||||
def lr_gradient(rows, offset, cols=None):
|
||||
"""
|
||||
Render rows through a left-to-right color sweep.
|
||||
|
||||
Args:
|
||||
rows: List of text rows to colorize
|
||||
offset: Gradient position offset (for animation)
|
||||
cols: Optional list of color codes. If None, uses active theme.
|
||||
|
||||
Returns:
|
||||
List of colorized rows
|
||||
"""
|
||||
if cols is None:
|
||||
from engine import config
|
||||
cols = (
|
||||
config.ACTIVE_THEME.main_gradient
|
||||
if config.ACTIVE_THEME
|
||||
else _default_green_gradient()
|
||||
)
|
||||
|
||||
# ... rest of function unchanged ...
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Add _default_green_gradient() fallback function**
|
||||
|
||||
Add to `engine/render.py` before `lr_gradient()`:
|
||||
|
||||
```python
|
||||
def _default_green_gradient():
|
||||
"""Fallback green gradient (original colors) for initialization."""
|
||||
return [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;195m", # pale white-tint
|
||||
"\033[38;5;123m", # bright cyan
|
||||
"\033[38;5;118m", # bright lime
|
||||
"\033[38;5;82m", # lime
|
||||
"\033[38;5;46m", # bright green
|
||||
"\033[38;5;40m", # green
|
||||
"\033[38;5;34m", # medium green
|
||||
"\033[38;5;28m", # dark green
|
||||
"\033[38;5;22m", # deep green
|
||||
"\033[2;38;5;22m", # dim deep green
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
|
||||
def _default_magenta_gradient():
|
||||
"""Fallback magenta gradient (original message colors) for initialization."""
|
||||
return [
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;225m", # pale pink-white
|
||||
"\033[38;5;219m", # bright pink
|
||||
"\033[38;5;213m", # hot pink
|
||||
"\033[38;5;207m", # magenta
|
||||
"\033[38;5;201m", # bright magenta
|
||||
"\033[38;5;165m", # orchid-red
|
||||
"\033[38;5;161m", # ruby-magenta
|
||||
"\033[38;5;125m", # dark magenta
|
||||
"\033[38;5;89m", # deep maroon-magenta
|
||||
"\033[2;38;5;89m", # dim deep maroon-magenta
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Run tests to verify they pass**
|
||||
|
||||
Run: `pytest tests/test_render.py::test_lr_gradient_uses_active_theme -v`
|
||||
Run: `pytest tests/test_render.py::test_lr_gradient_fallback_when_no_theme -v`
|
||||
Run: `pytest tests/test_render.py::test_default_green_gradient_length -v`
|
||||
|
||||
Expected: PASS (all 3 new tests)
|
||||
|
||||
- [ ] **Step 5: Run full render test suite**
|
||||
|
||||
Run: `pytest tests/test_render.py -v`
|
||||
|
||||
Expected: PASS (existing tests may need adjustment for mocking)
|
||||
|
||||
- [ ] **Step 6: Commit**
|
||||
|
||||
```bash
|
||||
git add engine/render.py tests/test_render.py
|
||||
git commit -m "feat: update lr_gradient to use config.ACTIVE_THEME
|
||||
|
||||
- Update lr_gradient(cols=None) to check config.ACTIVE_THEME
|
||||
- Add _default_green_gradient() and _default_magenta_gradient() fallbacks
|
||||
- Fallback used when ACTIVE_THEME is None (non-interactive init)
|
||||
- Add tests for theme-aware and fallback gradient rendering"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Chunk 4: Message Gradient Integration
|
||||
|
||||
### Task 4: Update scroll.py to use message gradient from config
|
||||
|
||||
**Files:**
|
||||
- Modify: `engine/scroll.py:85-95`
|
||||
- Test: existing `tests/test_scroll.py`
|
||||
|
||||
- [ ] **Step 1: Locate message gradient calls in scroll.py**
|
||||
|
||||
Run: `grep -n "MSG_GRAD_COLS\|lr_gradient_opposite" /Users/genejohnson/Dev/mainline/engine/scroll.py`
|
||||
|
||||
Expected: Should find line(s) where `MSG_GRAD_COLS` or similar is used
|
||||
|
||||
- [ ] **Step 2: Update scroll.py to use theme message gradient**
|
||||
|
||||
Edit `engine/scroll.py`, find the line that uses message gradients (around line 89 based on spec) and update:
|
||||
|
||||
Old code:
|
||||
```python
|
||||
# Some variation of:
|
||||
rows = lr_gradient(rows, offset, MSG_GRAD_COLS)
|
||||
```
|
||||
|
||||
New code:
|
||||
```python
|
||||
from engine import config
|
||||
msg_cols = (
|
||||
config.ACTIVE_THEME.message_gradient
|
||||
if config.ACTIVE_THEME
|
||||
else render._default_magenta_gradient()
|
||||
)
|
||||
rows = lr_gradient(rows, offset, msg_cols)
|
||||
```
|
||||
|
||||
Or use the helper approach (create `msg_gradient()` in render.py):
|
||||
|
||||
```python
|
||||
def msg_gradient(rows, offset):
|
||||
"""Apply message (ntfy) gradient using theme complementary colors."""
|
||||
from engine import config
|
||||
cols = (
|
||||
config.ACTIVE_THEME.message_gradient
|
||||
if config.ACTIVE_THEME
|
||||
else _default_magenta_gradient()
|
||||
)
|
||||
return lr_gradient(rows, offset, cols)
|
||||
```
|
||||
|
||||
Then in scroll.py:
|
||||
```python
|
||||
rows = render.msg_gradient(rows, offset)
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Run existing scroll tests**
|
||||
|
||||
Run: `pytest tests/test_scroll.py -v`
|
||||
|
||||
Expected: PASS (existing functionality unchanged)
|
||||
|
||||
- [ ] **Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add engine/scroll.py engine/render.py
|
||||
git commit -m "feat: update scroll.py to use theme message gradient
|
||||
|
||||
- Replace MSG_GRAD_COLS reference with config.ACTIVE_THEME.message_gradient
|
||||
- Use fallback magenta gradient when theme not initialized
|
||||
- Ensure ntfy messages render in complementary color from selected theme"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Chunk 5: Color Picker UI
|
||||
|
||||
### Task 5: Create pick_color_theme() function in app.py
|
||||
|
||||
**Files:**
|
||||
- Modify: `engine/app.py:1-300`
|
||||
- Test: manual/integration (interactive)
|
||||
|
||||
- [ ] **Step 1: Write helper functions for color picker UI**
|
||||
|
||||
Edit `engine/app.py`, add before `pick_font_face()` function:
|
||||
|
||||
```python
|
||||
def _draw_color_picker(themes_list, selected):
|
||||
"""Draw the color theme picker menu."""
|
||||
import sys
|
||||
from engine.terminal import CLR, W_GHOST, G_HI, G_DIM, tw
|
||||
|
||||
print(CLR, end="")
|
||||
print()
|
||||
print(f" {G_HI}▼ COLOR THEME{W_GHOST} ─ ↑/↓ or j/k to move, Enter/q to select{G_DIM}")
|
||||
print(f" {W_GHOST}{'─' * (tw() - 4)}\n")
|
||||
|
||||
for i, (theme_id, theme) in enumerate(themes_list):
|
||||
prefix = " ▶ " if i == selected else " "
|
||||
color = G_HI if i == selected else ""
|
||||
reset = "" if i == selected else W_GHOST
|
||||
print(f"{prefix}{color}{theme.name}{reset}")
|
||||
|
||||
print()
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Create pick_color_theme() function**
|
||||
|
||||
Edit `engine/app.py`, add after helper function:
|
||||
|
||||
```python
|
||||
def pick_color_theme():
|
||||
"""Interactive color theme picker. Defaults to 'green' if not TTY."""
|
||||
import sys
|
||||
import termios
|
||||
import tty
|
||||
from engine import config, themes
|
||||
|
||||
# Non-interactive fallback: use green
|
||||
if not sys.stdin.isatty():
|
||||
config.set_active_theme("green")
|
||||
return
|
||||
|
||||
themes_list = list(themes.THEME_REGISTRY.items())
|
||||
selected = 0
|
||||
|
||||
fd = sys.stdin.fileno()
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
try:
|
||||
tty.setcbreak(fd)
|
||||
while True:
|
||||
_draw_color_picker(themes_list, selected)
|
||||
key = _read_picker_key()
|
||||
if key == "up":
|
||||
selected = max(0, selected - 1)
|
||||
elif key == "down":
|
||||
selected = min(len(themes_list) - 1, selected + 1)
|
||||
elif key == "enter":
|
||||
break
|
||||
elif key == "interrupt":
|
||||
raise KeyboardInterrupt
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
|
||||
selected_theme_id = themes_list[selected][0]
|
||||
config.set_active_theme(selected_theme_id)
|
||||
|
||||
theme_name = themes_list[selected][1].name
|
||||
print(f" {G_DIM}> using {theme_name}{RST}")
|
||||
time.sleep(0.8)
|
||||
print(CLR, end="")
|
||||
print(CURSOR_OFF, end="")
|
||||
print()
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Update main() to call pick_color_theme() before pick_font_face()**
|
||||
|
||||
Edit `engine/app.py`, find the `main()` function and locate where `pick_font_face()` is called (around line 265). Add before it:
|
||||
|
||||
```python
|
||||
def main():
|
||||
# ... existing signal handler setup ...
|
||||
|
||||
pick_color_theme() # NEW LINE - before font picker
|
||||
pick_font_face()
|
||||
|
||||
# ... rest of main unchanged ...
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Manual test - run in interactive terminal**
|
||||
|
||||
Run: `python3 mainline.py`
|
||||
|
||||
Expected:
|
||||
- See color theme picker menu before font picker
|
||||
- Can navigate with ↑/↓ or j/k
|
||||
- Can select with Enter or q
|
||||
- Selected theme applies to scrolling headlines
|
||||
- Can select different themes and see colors change
|
||||
|
||||
- [ ] **Step 5: Manual test - run in non-interactive environment**
|
||||
|
||||
Run: `echo "" | python3 mainline.py`
|
||||
|
||||
Expected:
|
||||
- No color picker menu shown
|
||||
- Defaults to green theme
|
||||
- App runs without error
|
||||
|
||||
- [ ] **Step 6: Commit**
|
||||
|
||||
```bash
|
||||
git add engine/app.py
|
||||
git commit -m "feat: add pick_color_theme() UI and integration
|
||||
|
||||
- Create _draw_color_picker() to render menu
|
||||
- Create pick_color_theme() function mirroring font picker pattern
|
||||
- Integrate into main() before font picker
|
||||
- Fallback to green theme in non-interactive environments
|
||||
- Support arrow keys and j/k navigation"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Chunk 6: Integration & Validation
|
||||
|
||||
### Task 6: End-to-end testing and cleanup
|
||||
|
||||
**Files:**
|
||||
- Test: All modified files
|
||||
- Verify: App functionality
|
||||
|
||||
- [ ] **Step 1: Run full test suite**
|
||||
|
||||
Run: `pytest tests/ -v`
|
||||
|
||||
Expected: PASS (all tests, including new ones)
|
||||
|
||||
- [ ] **Step 2: Run linter**
|
||||
|
||||
Run: `ruff check engine/ mainline.py`
|
||||
|
||||
Expected: No errors (fix any style issues)
|
||||
|
||||
- [ ] **Step 3: Manual integration test - green theme**
|
||||
|
||||
Run: `python3 mainline.py`
|
||||
|
||||
Then select "Verdant Green" from picker.
|
||||
|
||||
Expected:
|
||||
- Headlines render in green → deep green
|
||||
- ntfy messages render in magenta gradient
|
||||
- Both work correctly during streaming
|
||||
|
||||
- [ ] **Step 4: Manual integration test - orange theme**
|
||||
|
||||
Run: `python3 mainline.py`
|
||||
|
||||
Then select "Molten Orange" from picker.
|
||||
|
||||
Expected:
|
||||
- Headlines render in orange → deep orange
|
||||
- ntfy messages render in blue gradient
|
||||
- Colors are visually distinct from green
|
||||
|
||||
- [ ] **Step 5: Manual integration test - purple theme**
|
||||
|
||||
Run: `python3 mainline.py`
|
||||
|
||||
Then select "Violet Purple" from picker.
|
||||
|
||||
Expected:
|
||||
- Headlines render in purple → deep purple
|
||||
- ntfy messages render in yellow gradient
|
||||
- Colors are visually distinct from green and orange
|
||||
|
||||
- [ ] **Step 6: Test poetry mode with color picker**
|
||||
|
||||
Run: `python3 mainline.py --poetry`
|
||||
|
||||
Then select "orange" from picker.
|
||||
|
||||
Expected:
|
||||
- Poetry mode works with color picker
|
||||
- Colors apply to poetry rendering
|
||||
|
||||
- [ ] **Step 7: Test code mode with color picker**
|
||||
|
||||
Run: `python3 mainline.py --code`
|
||||
|
||||
Then select "purple" from picker.
|
||||
|
||||
Expected:
|
||||
- Code mode works with color picker
|
||||
- Colors apply to code rendering
|
||||
|
||||
- [ ] **Step 8: Verify acceptance criteria**
|
||||
|
||||
✓ Color picker displays 3 theme options at startup
|
||||
✓ Selection applies to all headline and message gradients
|
||||
✓ Boot UI (title, status) uses hardcoded green (not theme)
|
||||
✓ Scrolling headlines and ntfy messages use theme gradients
|
||||
✓ No persistence between runs (each run picks fresh)
|
||||
✓ Non-TTY environments default to green without error
|
||||
✓ Architecture supports future random/animation modes
|
||||
✓ All gradient color codes finalized with no TBD values
|
||||
|
||||
- [ ] **Step 9: Final commit**
|
||||
|
||||
```bash
|
||||
git add -A
|
||||
git commit -m "feat: color scheme switcher implementation complete
|
||||
|
||||
Closes color-pick feature with:
|
||||
- Three selectable color themes (green, orange, purple)
|
||||
- Interactive menu at startup (mirrors font picker UI)
|
||||
- Complementary colors for ntfy message queue
|
||||
- Fallback to green in non-interactive environments
|
||||
- All tests passing, manual validation complete"
|
||||
```
|
||||
|
||||
- [ ] **Step 10: Create feature branch PR summary**
|
||||
|
||||
```
|
||||
## Color Scheme Switcher
|
||||
|
||||
Implements interactive color theme selection for Mainline news ticker.
|
||||
|
||||
### What's New
|
||||
- 3 color themes: Verdant Green, Molten Orange, Violet Purple
|
||||
- Interactive picker at startup (↑/↓ or j/k, Enter to select)
|
||||
- Complementary gradients for ntfy messages (magenta, blue, yellow)
|
||||
- Fresh theme selection each run (no persistence)
|
||||
|
||||
### Files Changed
|
||||
- `engine/themes.py` (new)
|
||||
- `engine/config.py` (ACTIVE_THEME, set_active_theme)
|
||||
- `engine/render.py` (theme-aware gradients)
|
||||
- `engine/scroll.py` (message gradient integration)
|
||||
- `engine/app.py` (pick_color_theme UI)
|
||||
- `tests/test_themes.py` (new theme tests)
|
||||
- `README.md` (documentation)
|
||||
|
||||
### Acceptance Criteria
|
||||
All met. App fully tested and ready for merge.
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Testing Checklist
|
||||
|
||||
- [ ] Unit tests: `pytest tests/test_themes.py -v`
|
||||
- [ ] Unit tests: `pytest tests/test_config.py -v`
|
||||
- [ ] Unit tests: `pytest tests/test_render.py -v`
|
||||
- [ ] Full suite: `pytest tests/ -v`
|
||||
- [ ] Linting: `ruff check engine/ mainline.py`
|
||||
- [ ] Manual: Green theme selection
|
||||
- [ ] Manual: Orange theme selection
|
||||
- [ ] Manual: Purple theme selection
|
||||
- [ ] Manual: Poetry mode with colors
|
||||
- [ ] Manual: Code mode with colors
|
||||
- [ ] Manual: Non-TTY fallback
|
||||
|
||||
---
|
||||
|
||||
## Notes
|
||||
|
||||
- `themes.py` is data-only; never import config or render to prevent cycles
|
||||
- `ACTIVE_THEME` initialized to None; guaranteed non-None before stream() via pick_color_theme()
|
||||
- Font picker UI remains hardcoded green; title/subtitle use G_HI/G_DIM constants (not theme)
|
||||
- Message gradients use complementary colors; lookup in scroll.py
|
||||
- Each gradient has 12 colors; verify length in tests
|
||||
- No persistence; fresh picker each run
|
||||
@@ -1,145 +0,0 @@
|
||||
# README Update Design — 2026-03-15
|
||||
|
||||
## Goal
|
||||
|
||||
Restructure and expand `README.md` to:
|
||||
1. Align with the current codebase (Python 3.10+, uv/mise/pytest/ruff toolchain, 6 new fonts)
|
||||
2. Add extensibility-focused content (`Extending` section)
|
||||
3. Add developer workflow coverage (`Development` section)
|
||||
4. Improve navigability via top-level grouping (Approach C)
|
||||
|
||||
---
|
||||
|
||||
## Proposed Structure
|
||||
|
||||
```
|
||||
# MAINLINE
|
||||
> tagline + description
|
||||
|
||||
## Using
|
||||
### Run
|
||||
### Config
|
||||
### Feeds
|
||||
### Fonts
|
||||
### ntfy.sh
|
||||
|
||||
## Internals
|
||||
### How it works
|
||||
### Architecture
|
||||
|
||||
## Extending
|
||||
### NtfyPoller
|
||||
### MicMonitor
|
||||
### Render pipeline
|
||||
|
||||
## Development
|
||||
### Setup
|
||||
### Tasks
|
||||
### Testing
|
||||
### Linting
|
||||
|
||||
## Roadmap
|
||||
|
||||
---
|
||||
*footer*
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Section-by-section design
|
||||
|
||||
### Using
|
||||
|
||||
All existing content preserved verbatim. Two changes:
|
||||
- **Run**: add `uv run mainline.py` as an alternative invocation; expand bootstrap note to mention `uv sync` / `uv sync --all-extras`
|
||||
- **ntfy.sh**: remove `NtfyPoller` reuse code example (moves to Extending); keep push instructions and topic config
|
||||
|
||||
Subsections moved into Using (currently standalone):
|
||||
- `Feeds` — it's configuration, not a concept
|
||||
- `ntfy.sh` (usage half)
|
||||
|
||||
### Internals
|
||||
|
||||
All existing content preserved verbatim. One change:
|
||||
- **Architecture**: append `tests/` directory listing to the module tree
|
||||
|
||||
### Extending
|
||||
|
||||
Entirely new section. Three subsections:
|
||||
|
||||
**NtfyPoller**
|
||||
- Minimal working import + usage example
|
||||
- Note: stdlib only dependencies
|
||||
|
||||
```python
|
||||
from engine.ntfy import NtfyPoller
|
||||
|
||||
poller = NtfyPoller("https://ntfy.sh/my_topic/json?since=20s&poll=1")
|
||||
poller.start()
|
||||
|
||||
# in your render loop:
|
||||
msg = poller.get_active_message() # → (title, body, timestamp) or None
|
||||
if msg:
|
||||
title, body, ts = msg
|
||||
render_my_message(title, body) # visualizer-specific
|
||||
```
|
||||
|
||||
**MicMonitor**
|
||||
- Minimal working import + usage example
|
||||
- Note: sounddevice/numpy optional, degrades gracefully
|
||||
|
||||
```python
|
||||
from engine.mic import MicMonitor
|
||||
|
||||
mic = MicMonitor(threshold_db=50)
|
||||
if mic.start(): # returns False if sounddevice unavailable
|
||||
excess = mic.excess # dB above threshold, clamped to 0
|
||||
db = mic.db # raw RMS dB level
|
||||
```
|
||||
|
||||
**Render pipeline**
|
||||
- Brief prose about `engine.render` as importable pipeline
|
||||
- Minimal sketch of serve.py / ESP32 usage pattern
|
||||
- Reference to `Mainline Renderer + ntfy Message Queue for ESP32.md`
|
||||
|
||||
### Development
|
||||
|
||||
Entirely new section. Four subsections:
|
||||
|
||||
**Setup**
|
||||
- Hard requirements: Python 3.10+, uv
|
||||
- `uv sync` / `uv sync --all-extras` / `uv sync --group dev`
|
||||
|
||||
**Tasks** (via mise)
|
||||
- `mise run test`, `test-cov`, `lint`, `lint-fix`, `format`, `run`, `run-poetry`, `run-firehose`
|
||||
|
||||
**Testing**
|
||||
- Tests in `tests/` covering config, filter, mic, ntfy, sources, terminal
|
||||
- `uv run pytest` and `uv run pytest --cov=engine --cov-report=term-missing`
|
||||
|
||||
**Linting**
|
||||
- `uv run ruff check` and `uv run ruff format`
|
||||
- Note: pre-commit hooks run lint via `hk`
|
||||
|
||||
### Roadmap
|
||||
|
||||
Existing `## Ideas / Future` content preserved verbatim. Only change: rename heading to `## Roadmap`.
|
||||
|
||||
### Footer
|
||||
|
||||
Update `Python 3.9+` → `Python 3.10+`.
|
||||
|
||||
---
|
||||
|
||||
## Files changed
|
||||
|
||||
- `README.md` — restructured and expanded as above
|
||||
- No other files
|
||||
|
||||
---
|
||||
|
||||
## What is not changing
|
||||
|
||||
- All existing prose, examples, and config table values — preserved verbatim where retained
|
||||
- The Ideas/Future content — kept intact under the new Roadmap heading
|
||||
- The cyberpunk voice and terse style of the existing README
|
||||
@@ -1,154 +0,0 @@
|
||||
# Code Scroll Mode — Design Spec
|
||||
|
||||
**Date:** 2026-03-16
|
||||
**Branch:** feat/code-scroll
|
||||
**Status:** Approved
|
||||
|
||||
---
|
||||
|
||||
## Overview
|
||||
|
||||
Add a `--code` CLI flag that puts MAINLINE into "source consciousness" mode. Instead of RSS headlines or poetry stanzas, the program's own source code scrolls upward as large OTF half-block characters with the standard white-hot → deep green gradient. Each scroll item is one non-blank, non-comment line from `engine/*.py`, attributed to its enclosing function/class scope and dotted module path.
|
||||
|
||||
---
|
||||
|
||||
## Goals
|
||||
|
||||
- Mirror the existing `--poetry` mode pattern as closely as possible
|
||||
- Zero new runtime dependencies (stdlib `ast` and `pathlib` only)
|
||||
- No changes to `scroll.py` or the render pipeline
|
||||
- The item tuple shape `(text, src, ts)` is unchanged
|
||||
|
||||
---
|
||||
|
||||
## New Files
|
||||
|
||||
### `engine/fetch_code.py`
|
||||
|
||||
Single public function `fetch_code()` that returns `(items, line_count, 0)`.
|
||||
|
||||
**Algorithm:**
|
||||
|
||||
1. Glob `engine/*.py` in sorted order
|
||||
2. For each file:
|
||||
a. Read source text
|
||||
b. `ast.parse(source)` → build a `{line_number: scope_label}` map by walking all `FunctionDef`, `AsyncFunctionDef`, and `ClassDef` nodes. Each node covers its full line range. Inner scopes override outer ones.
|
||||
c. Iterate source lines (1-indexed). Skip if:
|
||||
- The stripped line is empty
|
||||
- The stripped line starts with `#`
|
||||
d. For each kept line emit:
|
||||
- `text` = `line.rstrip()` (preserve indentation for readability in the big render)
|
||||
- `src` = scope label from the AST map, e.g. `stream()` for functions, `MicMonitor` for classes, `<module>` for top-level lines
|
||||
- `ts` = dotted module path derived from filename, e.g. `engine/scroll.py` → `engine.scroll`
|
||||
3. Return `(items, len(items), 0)`
|
||||
|
||||
**Scope label rules:**
|
||||
- `FunctionDef` / `AsyncFunctionDef` → `name()`
|
||||
- `ClassDef` → `name` (no parens)
|
||||
- No enclosing node → `<module>`
|
||||
|
||||
**Dependencies:** `ast`, `pathlib` — stdlib only.
|
||||
|
||||
---
|
||||
|
||||
## Modified Files
|
||||
|
||||
### `engine/config.py`
|
||||
|
||||
Extend `MODE` detection to recognise `--code`:
|
||||
|
||||
```python
|
||||
MODE = (
|
||||
"poetry" if "--poetry" in sys.argv or "-p" in sys.argv
|
||||
else "code" if "--code" in sys.argv
|
||||
else "news"
|
||||
)
|
||||
```
|
||||
|
||||
### `engine/app.py`
|
||||
|
||||
**Subtitle line** — extend the subtitle dict:
|
||||
|
||||
```python
|
||||
_subtitle = {
|
||||
"poetry": "literary consciousness stream",
|
||||
"code": "source consciousness stream",
|
||||
}.get(config.MODE, "digital consciousness stream")
|
||||
```
|
||||
|
||||
**Boot sequence** — add `elif config.MODE == "code":` branch after the poetry branch:
|
||||
|
||||
```python
|
||||
elif config.MODE == "code":
|
||||
from engine.fetch_code import fetch_code
|
||||
slow_print(" > INITIALIZING SOURCE ARRAY...\n")
|
||||
time.sleep(0.2)
|
||||
print()
|
||||
items, line_count, _ = fetch_code()
|
||||
print()
|
||||
print(f" {G_DIM}>{RST} {G_MID}{line_count} LINES ACQUIRED{RST}")
|
||||
```
|
||||
|
||||
No cache save/load — local source files are read instantly and change only on disk writes.
|
||||
|
||||
---
|
||||
|
||||
## Data Flow
|
||||
|
||||
```
|
||||
engine/*.py (sorted)
|
||||
│
|
||||
▼
|
||||
fetch_code()
|
||||
│ ast.parse → scope map
|
||||
│ filter blank + comment lines
|
||||
│ emit (line, scope(), engine.module)
|
||||
▼
|
||||
items: List[Tuple[str, str, str]]
|
||||
│
|
||||
▼
|
||||
stream(items, ntfy, mic) ← unchanged
|
||||
│
|
||||
▼
|
||||
next_headline() shuffles + recycles automatically
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Error Handling
|
||||
|
||||
- If a file fails to `ast.parse` (malformed source), fall back to `<module>` scope for all lines in that file — do not crash.
|
||||
- If `engine/` contains no `.py` files (shouldn't happen in practice), `fetch_code()` returns an empty list; `app.py`'s existing `if not items:` guard handles this.
|
||||
|
||||
---
|
||||
|
||||
## Testing
|
||||
|
||||
New file: `tests/test_fetch_code.py`
|
||||
|
||||
| Test | Assertion |
|
||||
|------|-----------|
|
||||
| `test_items_are_tuples` | Every item from `fetch_code()` is a 3-tuple of strings |
|
||||
| `test_blank_and_comment_lines_excluded` | No item text is empty; no item text (stripped) starts with `#` |
|
||||
| `test_module_path_format` | Every `ts` field matches pattern `engine\.\w+` |
|
||||
|
||||
No mocking — tests read the real engine source files, keeping them honest against actual content.
|
||||
|
||||
---
|
||||
|
||||
## CLI
|
||||
|
||||
```bash
|
||||
python3 mainline.py --code # source consciousness mode
|
||||
uv run mainline.py --code
|
||||
```
|
||||
|
||||
Compatible with all existing flags (`--no-font-picker`, `--font-file`, `--firehose`, etc.).
|
||||
|
||||
---
|
||||
|
||||
## Out of Scope
|
||||
|
||||
- Syntax highlighting / token-aware coloring (can be added later)
|
||||
- `--code-dir` flag for pointing at arbitrary directories (YAGNI)
|
||||
- Caching code items to disk
|
||||
@@ -1,299 +0,0 @@
|
||||
# Color Scheme Switcher Design
|
||||
|
||||
**Date:** 2026-03-16
|
||||
**Status:** Revised after review
|
||||
**Scope:** Interactive color theme selection for Mainline news ticker
|
||||
|
||||
---
|
||||
|
||||
## Overview
|
||||
|
||||
Mainline currently renders news headlines with a fixed white-hot → deep green gradient. This feature adds an interactive theme picker at startup that lets users choose between three precise color schemes (green, orange, purple), each with complementary message queue colors.
|
||||
|
||||
The implementation uses a dedicated `Theme` class to encapsulate gradients and metadata, enabling future extensions like random rotation, animation, or additional themes without architectural changes.
|
||||
|
||||
---
|
||||
|
||||
## Requirements
|
||||
|
||||
**Functional:**
|
||||
1. User selects a color theme from an interactive menu at startup (green, orange, or purple)
|
||||
2. Main headline gradient uses the selected primary color (white → color)
|
||||
3. Message queue (ntfy) gradient uses the precise complementary color (white → opposite)
|
||||
4. Selection is fresh each run (no persistence)
|
||||
5. Design supports future "random rotation" mode without refactoring
|
||||
|
||||
**Complementary colors (precise opposites):**
|
||||
- Green (38;5;22) → Magenta (38;5;89) *(current, unchanged)*
|
||||
- Orange (38;5;208) → Blue (38;5;21)
|
||||
- Purple (38;5;129) → Yellow (38;5;226)
|
||||
|
||||
**Non-functional:**
|
||||
- Reuse the existing font picker pattern for UI consistency
|
||||
- Zero runtime overhead during streaming (theme lookup happens once at startup)
|
||||
- **Boot UI (title, subtitle, status lines) use hardcoded green color constants (G_HI, G_DIM, G_MID); only scrolling headlines and ntfy messages use theme gradients**
|
||||
- Font picker UI remains hardcoded green for visual continuity
|
||||
|
||||
---
|
||||
|
||||
## Architecture
|
||||
|
||||
### New Module: `engine/themes.py`
|
||||
|
||||
**Data-only module:** Contains Theme class, THEME_REGISTRY, and get_theme() function. **Imports only typing; does NOT import config or render** to prevent circular dependencies.
|
||||
|
||||
```python
|
||||
class Theme:
|
||||
"""Encapsulates a color scheme: name, main gradient, message gradient."""
|
||||
|
||||
def __init__(self, name: str, main_gradient: list[str], message_gradient: list[str]):
|
||||
self.name = name
|
||||
self.main_gradient = main_gradient # white → primary color
|
||||
self.message_gradient = message_gradient # white → complementary
|
||||
```
|
||||
|
||||
**Theme Registry:**
|
||||
Three instances registered by ID: `"green"`, `"orange"`, `"purple"` (IDs match menu labels for clarity).
|
||||
|
||||
Each gradient is a list of 12 ANSI 256-color codes matching the current green gradient:
|
||||
```
|
||||
[
|
||||
"\033[1;38;5;231m", # white (bold)
|
||||
"\033[1;38;5;195m", # pale white-tint
|
||||
"\033[38;5;123m", # bright cyan
|
||||
"\033[38;5;118m", # bright lime
|
||||
"\033[38;5;82m", # lime
|
||||
"\033[38;5;46m", # bright color
|
||||
"\033[38;5;40m", # color
|
||||
"\033[38;5;34m", # medium color
|
||||
"\033[38;5;28m", # dark color
|
||||
"\033[38;5;22m", # deep color
|
||||
"\033[2;38;5;22m", # dim deep color
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
```
|
||||
|
||||
**Finalized color codes:**
|
||||
|
||||
**Green (primary: 22, complementary: 89)** — unchanged from current
|
||||
- Main: `[231, 195, 123, 118, 82, 46, 40, 34, 28, 22, 22(dim), 235]`
|
||||
- Messages: `[231, 225, 219, 213, 207, 201, 165, 161, 125, 89, 89(dim), 235]`
|
||||
|
||||
**Orange (primary: 208, complementary: 21)**
|
||||
- Main: `[231, 215, 209, 208, 202, 166, 130, 94, 58, 94, 94(dim), 235]`
|
||||
- Messages: `[231, 195, 33, 27, 21, 21, 21, 18, 18, 18, 18(dim), 235]`
|
||||
|
||||
**Purple (primary: 129, complementary: 226)**
|
||||
- Main: `[231, 225, 177, 171, 165, 135, 129, 93, 57, 57, 57(dim), 235]`
|
||||
- Messages: `[231, 226, 226, 220, 220, 184, 184, 178, 178, 172, 172(dim), 235]`
|
||||
|
||||
**Public API:**
|
||||
- `get_theme(theme_id: str) -> Theme` — lookup by ID, raises KeyError if not found
|
||||
- `THEME_REGISTRY` — dict of all available themes (for picker)
|
||||
|
||||
---
|
||||
|
||||
### Modified: `engine/config.py`
|
||||
|
||||
**New globals:**
|
||||
```python
|
||||
ACTIVE_THEME = None # set by set_active_theme() after picker; guaranteed non-None during stream()
|
||||
```
|
||||
|
||||
**New function:**
|
||||
```python
|
||||
def set_active_theme(theme_id: str = "green"):
|
||||
"""Set the active theme. Defaults to 'green' if not specified."""
|
||||
global ACTIVE_THEME
|
||||
from engine import themes
|
||||
ACTIVE_THEME = themes.get_theme(theme_id)
|
||||
```
|
||||
|
||||
**Behavior:**
|
||||
- Called by `app.pick_color_theme()` with user selection
|
||||
- Has default fallback to "green" for non-interactive environments (CI, testing, piped stdin)
|
||||
- Guarantees `ACTIVE_THEME` is set before any render functions are called
|
||||
|
||||
**Removal:**
|
||||
- Delete hardcoded `GRAD_COLS` and `MSG_GRAD_COLS` constants
|
||||
|
||||
---
|
||||
|
||||
### Modified: `engine/render.py`
|
||||
|
||||
**Updated gradient access in existing functions:**
|
||||
|
||||
Current pattern (will be removed):
|
||||
```python
|
||||
GRAD_COLS = [...] # hardcoded green
|
||||
MSG_GRAD_COLS = [...] # hardcoded magenta
|
||||
```
|
||||
|
||||
New pattern — update `lr_gradient()` function:
|
||||
```python
|
||||
def lr_gradient(rows, offset, cols=None):
|
||||
if cols is None:
|
||||
from engine import config
|
||||
cols = (config.ACTIVE_THEME.main_gradient
|
||||
if config.ACTIVE_THEME
|
||||
else _default_green_gradient())
|
||||
# ... rest of function unchanged
|
||||
```
|
||||
|
||||
**Define fallback:**
|
||||
```python
|
||||
def _default_green_gradient():
|
||||
"""Fallback green gradient (current colors)."""
|
||||
return [
|
||||
"\033[1;38;5;231m", "\033[1;38;5;195m", "\033[38;5;123m",
|
||||
"\033[38;5;118m", "\033[38;5;82m", "\033[38;5;46m",
|
||||
"\033[38;5;40m", "\033[38;5;34m", "\033[38;5;28m",
|
||||
"\033[38;5;22m", "\033[2;38;5;22m", "\033[2;38;5;235m",
|
||||
]
|
||||
```
|
||||
|
||||
**Message gradient handling:**
|
||||
|
||||
The existing code (scroll.py line 89) calls `lr_gradient()` with `MSG_GRAD_COLS`. Change this call to:
|
||||
```python
|
||||
# Instead of: lr_gradient(rows, offset, MSG_GRAD_COLS)
|
||||
# Use:
|
||||
from engine import config
|
||||
cols = (config.ACTIVE_THEME.message_gradient
|
||||
if config.ACTIVE_THEME
|
||||
else _default_magenta_gradient())
|
||||
lr_gradient(rows, offset, cols)
|
||||
```
|
||||
|
||||
or define a helper:
|
||||
```python
|
||||
def msg_gradient(rows, offset):
|
||||
"""Apply message (ntfy) gradient using theme complementary colors."""
|
||||
from engine import config
|
||||
cols = (config.ACTIVE_THEME.message_gradient
|
||||
if config.ACTIVE_THEME
|
||||
else _default_magenta_gradient())
|
||||
return lr_gradient(rows, offset, cols)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Modified: `engine/app.py`
|
||||
|
||||
**New function: `pick_color_theme()`**
|
||||
|
||||
Mirrors `pick_font_face()` pattern:
|
||||
|
||||
```python
|
||||
def pick_color_theme():
|
||||
"""Interactive color theme picker. Defaults to 'green' if not TTY."""
|
||||
import sys
|
||||
from engine import config, themes
|
||||
|
||||
# Non-interactive fallback: use default
|
||||
if not sys.stdin.isatty():
|
||||
config.set_active_theme("green")
|
||||
return
|
||||
|
||||
# Interactive picker (similar to font picker)
|
||||
themes_list = list(themes.THEME_REGISTRY.items())
|
||||
selected = 0
|
||||
|
||||
# ... render menu, handle arrow keys j/k, ↑/↓ ...
|
||||
# ... on Enter, call config.set_active_theme(themes_list[selected][0]) ...
|
||||
```
|
||||
|
||||
**Placement in `main()`:**
|
||||
```python
|
||||
def main():
|
||||
# ... signal handler setup ...
|
||||
pick_color_theme() # NEW — before title/subtitle
|
||||
pick_font_face()
|
||||
# ... rest of boot sequence, title/subtitle use hardcoded G_HI/G_DIM ...
|
||||
```
|
||||
|
||||
**Important:** The title and subtitle render with hardcoded `G_HI`/`G_DIM` constants, not theme gradients. This is intentional for visual consistency with the font picker menu.
|
||||
|
||||
---
|
||||
|
||||
## Data Flow
|
||||
|
||||
```
|
||||
User starts: mainline.py
|
||||
↓
|
||||
main() called
|
||||
↓
|
||||
pick_color_theme()
|
||||
→ If TTY: display menu, read input, call config.set_active_theme(user_choice)
|
||||
→ If not TTY: silently call config.set_active_theme("green")
|
||||
↓
|
||||
pick_font_face() — renders in hardcoded green UI colors
|
||||
↓
|
||||
Boot messages (title, status) — all use hardcoded G_HI/G_DIM (not theme gradients)
|
||||
↓
|
||||
stream() — headlines + ntfy messages use config.ACTIVE_THEME gradients
|
||||
↓
|
||||
On exit: no persistence
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
### Initialization Guarantee
|
||||
`config.ACTIVE_THEME` is guaranteed to be non-None before `stream()` is called because:
|
||||
1. `pick_color_theme()` always sets it (either interactively or via fallback)
|
||||
2. It's called before any rendering happens
|
||||
3. Default fallback ensures non-TTY environments don't crash
|
||||
|
||||
### Module Independence
|
||||
`themes.py` is a pure data module with no imports of `config` or `render`. This prevents circular dependencies and allows it to be imported by multiple consumers without side effects.
|
||||
|
||||
### Color Code Finalization
|
||||
All three gradient sequences (green, orange, purple main + complementary) are now finalized with specific ANSI codes. No TBD placeholders remain.
|
||||
|
||||
### Theme ID Naming
|
||||
IDs are `"green"`, `"orange"`, `"purple"` — matching the menu labels exactly for clarity.
|
||||
|
||||
### Terminal Resize Handling
|
||||
The `pick_color_theme()` function mirrors `pick_font_face()`, which does not handle terminal resizing during the picker display. If the terminal is resized while the picker menu is shown, the menu redraw may be incomplete; pressing any key (arrow, j/k, q) continues normally. This is acceptable because:
|
||||
1. The picker completes quickly (< 5 seconds typical interaction)
|
||||
2. Once a theme is selected, the menu closes and rendering begins
|
||||
3. The streaming phase (`stream()`) is resilient to terminal resizing and auto-reflows to new dimensions
|
||||
|
||||
No special resize handling is needed for the color picker beyond what exists for the font picker.
|
||||
|
||||
### Testing Strategy
|
||||
1. **Unit tests** (`tests/test_themes.py`):
|
||||
- Verify Theme class construction
|
||||
- Test THEME_REGISTRY lookup (valid and invalid IDs)
|
||||
- Confirm gradient lists have correct length (12)
|
||||
|
||||
2. **Integration tests** (`tests/test_render.py`):
|
||||
- Mock `config.ACTIVE_THEME` to each theme
|
||||
- Verify `lr_gradient()` uses correct colors
|
||||
- Verify fallback works when `ACTIVE_THEME` is None
|
||||
|
||||
3. **Existing tests:**
|
||||
- Render tests that check gradient output will need to mock `config.ACTIVE_THEME`
|
||||
- Use pytest fixtures to set theme per test case
|
||||
|
||||
---
|
||||
|
||||
## Files Changed
|
||||
- `engine/themes.py` (new)
|
||||
- `engine/config.py` (add `ACTIVE_THEME`, `set_active_theme()`)
|
||||
- `engine/render.py` (replace GRAD_COLS/MSG_GRAD_COLS references with config lookups)
|
||||
- `engine/app.py` (add `pick_color_theme()`, call in main)
|
||||
- `tests/test_themes.py` (new unit tests)
|
||||
- `tests/test_render.py` (update mocking strategy)
|
||||
|
||||
## Acceptance Criteria
|
||||
1. ✓ Color picker displays 3 theme options at startup
|
||||
2. ✓ Selection applies to all headline and message gradients
|
||||
3. ✓ Boot UI (title, status) uses hardcoded green (not theme)
|
||||
4. ✓ Scrolling headlines and ntfy messages use theme gradients
|
||||
5. ✓ No persistence between runs
|
||||
6. ✓ Non-TTY environments default to green without error
|
||||
7. ✓ Architecture supports future random/animation modes
|
||||
8. ✓ All gradient color codes finalized with no TBD values
|
||||
@@ -1,36 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
PLUGIN_DIR = Path(__file__).parent
|
||||
|
||||
|
||||
def discover_plugins():
|
||||
from engine.effects.registry import get_registry
|
||||
from engine.effects.types import EffectPlugin
|
||||
|
||||
registry = get_registry()
|
||||
imported = {}
|
||||
|
||||
for file_path in PLUGIN_DIR.glob("*.py"):
|
||||
if file_path.name.startswith("_"):
|
||||
continue
|
||||
module_name = file_path.stem
|
||||
if module_name in ("base", "types"):
|
||||
continue
|
||||
|
||||
try:
|
||||
module = __import__(f"effects_plugins.{module_name}", fromlist=[""])
|
||||
for attr_name in dir(module):
|
||||
attr = getattr(module, attr_name)
|
||||
if (
|
||||
isinstance(attr, type)
|
||||
and issubclass(attr, EffectPlugin)
|
||||
and attr is not EffectPlugin
|
||||
and attr_name.endswith("Effect")
|
||||
):
|
||||
plugin = attr()
|
||||
registry.register(plugin)
|
||||
imported[plugin.name] = plugin
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return imported
|
||||
@@ -1,58 +0,0 @@
|
||||
import random
|
||||
|
||||
from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
||||
|
||||
|
||||
class FadeEffect(EffectPlugin):
|
||||
name = "fade"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
||||
if not ctx.ticker_height:
|
||||
return buf
|
||||
result = list(buf)
|
||||
intensity = self.config.intensity
|
||||
|
||||
top_zone = max(1, int(ctx.ticker_height * 0.25))
|
||||
bot_zone = max(1, int(ctx.ticker_height * 0.10))
|
||||
|
||||
for r in range(len(result)):
|
||||
if r >= ctx.ticker_height:
|
||||
continue
|
||||
top_f = min(1.0, r / top_zone) if top_zone > 0 else 1.0
|
||||
bot_f = (
|
||||
min(1.0, (ctx.ticker_height - 1 - r) / bot_zone)
|
||||
if bot_zone > 0
|
||||
else 1.0
|
||||
)
|
||||
row_fade = min(top_f, bot_f) * intensity
|
||||
|
||||
if row_fade < 1.0 and result[r].strip():
|
||||
result[r] = self._fade_line(result[r], row_fade)
|
||||
|
||||
return result
|
||||
|
||||
def _fade_line(self, s: str, fade: float) -> str:
|
||||
if fade >= 1.0:
|
||||
return s
|
||||
if fade <= 0.0:
|
||||
return ""
|
||||
result = []
|
||||
i = 0
|
||||
while i < len(s):
|
||||
if s[i] == "\033" and i + 1 < len(s) and s[i + 1] == "[":
|
||||
j = i + 2
|
||||
while j < len(s) and not s[j].isalpha():
|
||||
j += 1
|
||||
result.append(s[i : j + 1])
|
||||
i = j + 1
|
||||
elif s[i] == " ":
|
||||
result.append(" ")
|
||||
i += 1
|
||||
else:
|
||||
result.append(s[i] if random.random() < fade else " ")
|
||||
i += 1
|
||||
return "".join(result)
|
||||
|
||||
def configure(self, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
@@ -1,72 +0,0 @@
|
||||
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, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
@@ -1,37 +0,0 @@
|
||||
import random
|
||||
|
||||
from engine import config
|
||||
from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
||||
from engine.terminal import C_DIM, DIM, G_DIM, G_LO, RST
|
||||
|
||||
|
||||
class GlitchEffect(EffectPlugin):
|
||||
name = "glitch"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
||||
if not buf:
|
||||
return buf
|
||||
result = list(buf)
|
||||
intensity = self.config.intensity
|
||||
|
||||
glitch_prob = 0.32 + min(0.9, ctx.mic_excess * 0.16)
|
||||
glitch_prob = glitch_prob * intensity
|
||||
n_hits = 4 + int(ctx.mic_excess / 2)
|
||||
n_hits = int(n_hits * intensity)
|
||||
|
||||
if random.random() < glitch_prob:
|
||||
for _ in range(min(n_hits, len(result))):
|
||||
gi = random.randint(0, len(result) - 1)
|
||||
scr_row = gi + 1
|
||||
result[gi] = f"\033[{scr_row};1H{self._glitch_bar(ctx.terminal_width)}"
|
||||
return result
|
||||
|
||||
def _glitch_bar(self, w: int) -> str:
|
||||
c = random.choice(["░", "▒", "─", "\xc2"])
|
||||
n = random.randint(3, w // 2)
|
||||
o = random.randint(0, w - n)
|
||||
return " " * o + f"{G_LO}{DIM}" + c * n + RST
|
||||
|
||||
def configure(self, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
@@ -1,36 +0,0 @@
|
||||
import random
|
||||
|
||||
from engine import config
|
||||
from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
||||
from engine.terminal import C_DIM, G_DIM, G_LO, RST, W_GHOST
|
||||
|
||||
|
||||
class NoiseEffect(EffectPlugin):
|
||||
name = "noise"
|
||||
config = EffectConfig(enabled=True, intensity=0.15)
|
||||
|
||||
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
||||
if not ctx.ticker_height:
|
||||
return buf
|
||||
result = list(buf)
|
||||
intensity = self.config.intensity
|
||||
probability = intensity * 0.15
|
||||
|
||||
for r in range(len(result)):
|
||||
cy = ctx.scroll_cam + r
|
||||
if random.random() < probability:
|
||||
result[r] = self._generate_noise(ctx.terminal_width, cy)
|
||||
return result
|
||||
|
||||
def _generate_noise(self, w: int, cy: int) -> str:
|
||||
d = random.choice([0.15, 0.25, 0.35, 0.12])
|
||||
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)
|
||||
)
|
||||
|
||||
def configure(self, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
@@ -10,7 +10,7 @@ import termios
|
||||
import time
|
||||
import tty
|
||||
|
||||
from engine import config, render, themes
|
||||
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
|
||||
@@ -65,30 +65,6 @@ def _read_picker_key():
|
||||
return None
|
||||
|
||||
|
||||
def _draw_color_picker(themes_list, selected):
|
||||
"""Draw the color theme picker menu.
|
||||
|
||||
Args:
|
||||
themes_list: List of (theme_id, Theme) tuples from THEME_REGISTRY.items()
|
||||
selected: Index of currently selected theme (0-2)
|
||||
"""
|
||||
print(CLR, end="")
|
||||
print()
|
||||
|
||||
print(
|
||||
f" {G_HI}▼ COLOR THEME{RST} {W_GHOST}─ ↑/↓ or j/k to move, Enter/q to select{RST}"
|
||||
)
|
||||
print(f" {W_GHOST}{'─' * (tw() - 4)}{RST}\n")
|
||||
|
||||
for i, (theme_id, theme) in enumerate(themes_list):
|
||||
prefix = " ▶ " if i == selected else " "
|
||||
color = G_HI if i == selected else ""
|
||||
reset = "" if i == selected else W_GHOST
|
||||
print(f"{prefix}{color}{theme.name}{reset}")
|
||||
|
||||
print()
|
||||
|
||||
|
||||
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()]
|
||||
@@ -155,50 +131,6 @@ def _draw_font_picker(faces, selected):
|
||||
print(f" {shown}")
|
||||
|
||||
|
||||
def pick_color_theme():
|
||||
"""Interactive color theme picker. Defaults to 'green' if not TTY.
|
||||
|
||||
Displays a menu of available themes and lets user select with arrow keys.
|
||||
Non-interactive environments (piped stdin, CI) silently default to green.
|
||||
"""
|
||||
# Non-interactive fallback
|
||||
if not sys.stdin.isatty():
|
||||
config.set_active_theme("green")
|
||||
return
|
||||
|
||||
# Interactive picker
|
||||
themes_list = list(themes.THEME_REGISTRY.items())
|
||||
selected = 0
|
||||
|
||||
fd = sys.stdin.fileno()
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
try:
|
||||
tty.setcbreak(fd)
|
||||
while True:
|
||||
_draw_color_picker(themes_list, selected)
|
||||
key = _read_picker_key()
|
||||
if key == "up":
|
||||
selected = max(0, selected - 1)
|
||||
elif key == "down":
|
||||
selected = min(len(themes_list) - 1, selected + 1)
|
||||
elif key == "enter":
|
||||
break
|
||||
elif key == "interrupt":
|
||||
raise KeyboardInterrupt
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
|
||||
selected_theme_id = themes_list[selected][0]
|
||||
config.set_active_theme(selected_theme_id)
|
||||
|
||||
theme_name = themes_list[selected][1].name
|
||||
print(f" {G_DIM}> using {theme_name}{RST}")
|
||||
time.sleep(0.8)
|
||||
print(CLR, end="")
|
||||
print(CURSOR_OFF, end="")
|
||||
print()
|
||||
|
||||
|
||||
def pick_font_face():
|
||||
"""Interactive startup picker for selecting a face from repo OTF files."""
|
||||
if not config.FONT_PICKER:
|
||||
@@ -330,7 +262,6 @@ def main():
|
||||
w = tw()
|
||||
print(CLR, end="")
|
||||
print(CURSOR_OFF, end="")
|
||||
pick_color_theme()
|
||||
pick_font_face()
|
||||
w = tw()
|
||||
print()
|
||||
@@ -341,10 +272,11 @@ def main():
|
||||
time.sleep(0.07)
|
||||
|
||||
print()
|
||||
_subtitle = {
|
||||
"poetry": "literary consciousness stream",
|
||||
"code": "source consciousness stream",
|
||||
}.get(config.MODE, "digital consciousness stream")
|
||||
_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()
|
||||
@@ -365,15 +297,6 @@ def main():
|
||||
)
|
||||
print(f" {G_DIM}>{RST} {G_MID}{len(items)} STANZAS ACQUIRED{RST}")
|
||||
save_cache(items)
|
||||
elif config.MODE == "code":
|
||||
from engine.fetch_code import fetch_code
|
||||
|
||||
slow_print(" > INITIALIZING SOURCE ARRAY...\n")
|
||||
time.sleep(0.2)
|
||||
print()
|
||||
items, line_count, _ = fetch_code()
|
||||
print()
|
||||
print(f" {G_DIM}>{RST} {G_MID}{line_count} LINES ACQUIRED{RST}")
|
||||
else:
|
||||
slow_print(" > INITIALIZING FEED ARRAY...\n")
|
||||
time.sleep(0.2)
|
||||
|
||||
@@ -188,13 +188,7 @@ def set_config(config: Config) -> None:
|
||||
HEADLINE_LIMIT = 1000
|
||||
FEED_TIMEOUT = 10
|
||||
MIC_THRESHOLD_DB = 50 # dB above which glitches intensify
|
||||
MODE = (
|
||||
"poetry"
|
||||
if "--poetry" in sys.argv or "-p" in sys.argv
|
||||
else "code"
|
||||
if "--code" in sys.argv
|
||||
else "news"
|
||||
)
|
||||
MODE = "poetry" if "--poetry" in sys.argv or "-p" in sys.argv else "news"
|
||||
FIREHOSE = "--firehose" in sys.argv
|
||||
|
||||
# ─── NTFY MESSAGE QUEUE ──────────────────────────────────
|
||||
@@ -237,26 +231,3 @@ def set_font_selection(font_path=None, font_index=None):
|
||||
FONT_PATH = _resolve_font_path(font_path)
|
||||
if font_index is not None:
|
||||
FONT_INDEX = max(0, int(font_index))
|
||||
|
||||
|
||||
# ─── THEME MANAGEMENT ─────────────────────────────────────────
|
||||
ACTIVE_THEME = None
|
||||
|
||||
|
||||
def set_active_theme(theme_id: str = "green"):
|
||||
"""Set the active theme by ID.
|
||||
|
||||
Args:
|
||||
theme_id: Theme identifier ("green", "orange", or "purple")
|
||||
Defaults to "green"
|
||||
|
||||
Raises:
|
||||
KeyError: If theme_id is not in the theme registry
|
||||
|
||||
Side Effects:
|
||||
Sets the ACTIVE_THEME global variable
|
||||
"""
|
||||
global ACTIVE_THEME
|
||||
from engine import themes
|
||||
|
||||
ACTIVE_THEME = themes.get_theme(theme_id)
|
||||
|
||||
@@ -1,102 +0,0 @@
|
||||
"""
|
||||
Display output abstraction - allows swapping output backends.
|
||||
|
||||
Protocol:
|
||||
- init(width, height): Initialize display with terminal dimensions
|
||||
- show(buffer): Render buffer (list of strings) to display
|
||||
- clear(): Clear the display
|
||||
- cleanup(): Shutdown display
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Protocol
|
||||
|
||||
|
||||
class Display(Protocol):
|
||||
"""Protocol for display backends."""
|
||||
|
||||
def init(self, width: int, height: int) -> None:
|
||||
"""Initialize display with dimensions."""
|
||||
...
|
||||
|
||||
def show(self, buffer: list[str]) -> None:
|
||||
"""Show buffer on display."""
|
||||
...
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Clear display."""
|
||||
...
|
||||
|
||||
def cleanup(self) -> None:
|
||||
"""Shutdown display."""
|
||||
...
|
||||
|
||||
|
||||
def get_monitor():
|
||||
"""Get the performance monitor."""
|
||||
try:
|
||||
from engine.effects.performance import get_monitor as _get_monitor
|
||||
|
||||
return _get_monitor()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
class TerminalDisplay:
|
||||
"""ANSI terminal display backend."""
|
||||
|
||||
def __init__(self):
|
||||
self.width = 80
|
||||
self.height = 24
|
||||
|
||||
def init(self, width: int, height: int) -> None:
|
||||
from engine.terminal import CURSOR_OFF
|
||||
|
||||
self.width = width
|
||||
self.height = height
|
||||
print(CURSOR_OFF, end="", flush=True)
|
||||
|
||||
def show(self, buffer: list[str]) -> None:
|
||||
import sys
|
||||
|
||||
t0 = time.perf_counter()
|
||||
sys.stdout.buffer.write("".join(buffer).encode())
|
||||
sys.stdout.flush()
|
||||
elapsed_ms = (time.perf_counter() - t0) * 1000
|
||||
|
||||
monitor = get_monitor()
|
||||
if monitor:
|
||||
chars_in = sum(len(line) for line in buffer)
|
||||
monitor.record_effect("terminal_display", elapsed_ms, chars_in, chars_in)
|
||||
|
||||
def clear(self) -> None:
|
||||
from engine.terminal import CLR
|
||||
|
||||
print(CLR, end="", flush=True)
|
||||
|
||||
def cleanup(self) -> None:
|
||||
from engine.terminal import CURSOR_ON
|
||||
|
||||
print(CURSOR_ON, end="", flush=True)
|
||||
|
||||
|
||||
class NullDisplay:
|
||||
"""Headless/null display - discards all output."""
|
||||
|
||||
def init(self, width: int, height: int) -> None:
|
||||
self.width = width
|
||||
self.height = height
|
||||
|
||||
def show(self, buffer: list[str]) -> None:
|
||||
monitor = get_monitor()
|
||||
if monitor:
|
||||
t0 = time.perf_counter()
|
||||
chars_in = sum(len(line) for line in buffer)
|
||||
elapsed_ms = (time.perf_counter() - t0) * 1000
|
||||
monitor.record_effect("null_display", elapsed_ms, chars_in, chars_in)
|
||||
|
||||
def clear(self) -> None:
|
||||
pass
|
||||
|
||||
def cleanup(self) -> None:
|
||||
pass
|
||||
@@ -1,42 +0,0 @@
|
||||
from engine.effects.chain import EffectChain
|
||||
from engine.effects.controller import handle_effects_command, show_effects_menu
|
||||
from engine.effects.legacy import (
|
||||
fade_line,
|
||||
firehose_line,
|
||||
glitch_bar,
|
||||
next_headline,
|
||||
noise,
|
||||
vis_trunc,
|
||||
)
|
||||
from engine.effects.performance import PerformanceMonitor, get_monitor, set_monitor
|
||||
from engine.effects.registry import EffectRegistry, get_registry, set_registry
|
||||
from engine.effects.types import EffectConfig, EffectContext, PipelineConfig
|
||||
|
||||
|
||||
def get_effect_chain():
|
||||
from engine.layers import get_effect_chain as _chain
|
||||
|
||||
return _chain()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"EffectChain",
|
||||
"EffectRegistry",
|
||||
"EffectConfig",
|
||||
"EffectContext",
|
||||
"PipelineConfig",
|
||||
"get_registry",
|
||||
"set_registry",
|
||||
"get_effect_chain",
|
||||
"get_monitor",
|
||||
"set_monitor",
|
||||
"PerformanceMonitor",
|
||||
"handle_effects_command",
|
||||
"show_effects_menu",
|
||||
"fade_line",
|
||||
"firehose_line",
|
||||
"glitch_bar",
|
||||
"noise",
|
||||
"next_headline",
|
||||
"vis_trunc",
|
||||
]
|
||||
@@ -1,71 +0,0 @@
|
||||
import time
|
||||
|
||||
from engine.effects.performance import PerformanceMonitor, get_monitor
|
||||
from engine.effects.registry import EffectRegistry
|
||||
from engine.effects.types import EffectContext
|
||||
|
||||
|
||||
class EffectChain:
|
||||
def __init__(
|
||||
self, registry: EffectRegistry, monitor: PerformanceMonitor | None = None
|
||||
):
|
||||
self._registry = registry
|
||||
self._order: list[str] = []
|
||||
self._monitor = monitor
|
||||
|
||||
def _get_monitor(self) -> PerformanceMonitor:
|
||||
if self._monitor is not None:
|
||||
return self._monitor
|
||||
return get_monitor()
|
||||
|
||||
def set_order(self, names: list[str]) -> None:
|
||||
self._order = list(names)
|
||||
|
||||
def get_order(self) -> list[str]:
|
||||
return self._order.copy()
|
||||
|
||||
def add_effect(self, name: str, position: int | None = None) -> bool:
|
||||
if name not in self._registry.list_all():
|
||||
return False
|
||||
if position is None:
|
||||
self._order.append(name)
|
||||
else:
|
||||
self._order.insert(position, name)
|
||||
return True
|
||||
|
||||
def remove_effect(self, name: str) -> bool:
|
||||
if name in self._order:
|
||||
self._order.remove(name)
|
||||
return True
|
||||
return False
|
||||
|
||||
def reorder(self, new_order: list[str]) -> bool:
|
||||
all_plugins = set(self._registry.list_all().keys())
|
||||
if not all(name in all_plugins for name in new_order):
|
||||
return False
|
||||
self._order = list(new_order)
|
||||
return True
|
||||
|
||||
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
||||
monitor = self._get_monitor()
|
||||
frame_number = ctx.frame_number
|
||||
monitor.start_frame(frame_number)
|
||||
|
||||
frame_start = time.perf_counter()
|
||||
result = list(buf)
|
||||
for name in self._order:
|
||||
plugin = self._registry.get(name)
|
||||
if plugin and plugin.config.enabled:
|
||||
chars_in = sum(len(line) for line in result)
|
||||
effect_start = time.perf_counter()
|
||||
try:
|
||||
result = plugin.process(result, ctx)
|
||||
except Exception:
|
||||
plugin.config.enabled = False
|
||||
elapsed = time.perf_counter() - effect_start
|
||||
chars_out = sum(len(line) for line in result)
|
||||
monitor.record_effect(name, elapsed * 1000, chars_in, chars_out)
|
||||
|
||||
total_elapsed = time.perf_counter() - frame_start
|
||||
monitor.end_frame(frame_number, total_elapsed * 1000)
|
||||
return result
|
||||
@@ -1,144 +0,0 @@
|
||||
from engine.effects.performance import get_monitor
|
||||
from engine.effects.registry import get_registry
|
||||
|
||||
_effect_chain_ref = None
|
||||
|
||||
|
||||
def _get_effect_chain():
|
||||
global _effect_chain_ref
|
||||
if _effect_chain_ref is not None:
|
||||
return _effect_chain_ref
|
||||
try:
|
||||
from engine.layers import get_effect_chain as _chain
|
||||
|
||||
return _chain()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def set_effect_chain_ref(chain) -> None:
|
||||
global _effect_chain_ref
|
||||
_effect_chain_ref = chain
|
||||
|
||||
|
||||
def handle_effects_command(cmd: str) -> str:
|
||||
"""Handle /effects command from NTFY message.
|
||||
|
||||
Commands:
|
||||
/effects list - list all effects and their status
|
||||
/effects <name> on - enable an effect
|
||||
/effects <name> off - disable an effect
|
||||
/effects <name> intensity <0.0-1.0> - set intensity
|
||||
/effects reorder <name1>,<name2>,... - reorder pipeline
|
||||
/effects stats - show performance statistics
|
||||
"""
|
||||
parts = cmd.strip().split()
|
||||
if not parts or parts[0] != "/effects":
|
||||
return "Unknown command"
|
||||
|
||||
registry = get_registry()
|
||||
chain = _get_effect_chain()
|
||||
|
||||
if len(parts) == 1 or parts[1] == "list":
|
||||
result = ["Effects:"]
|
||||
for name, plugin in registry.list_all().items():
|
||||
status = "ON" if plugin.config.enabled else "OFF"
|
||||
intensity = plugin.config.intensity
|
||||
result.append(f" {name}: {status} (intensity={intensity})")
|
||||
if chain:
|
||||
result.append(f"Order: {chain.get_order()}")
|
||||
return "\n".join(result)
|
||||
|
||||
if parts[1] == "stats":
|
||||
return _format_stats()
|
||||
|
||||
if parts[1] == "reorder" and len(parts) >= 3:
|
||||
new_order = parts[2].split(",")
|
||||
if chain and chain.reorder(new_order):
|
||||
return f"Reordered pipeline: {new_order}"
|
||||
return "Failed to reorder pipeline"
|
||||
|
||||
if len(parts) < 3:
|
||||
return "Usage: /effects <name> on|off|intensity <value>"
|
||||
|
||||
effect_name = parts[1]
|
||||
action = parts[2]
|
||||
|
||||
if effect_name not in registry.list_all():
|
||||
return f"Unknown effect: {effect_name}"
|
||||
|
||||
if action == "on":
|
||||
registry.enable(effect_name)
|
||||
return f"Enabled: {effect_name}"
|
||||
|
||||
if action == "off":
|
||||
registry.disable(effect_name)
|
||||
return f"Disabled: {effect_name}"
|
||||
|
||||
if action == "intensity" and len(parts) >= 4:
|
||||
try:
|
||||
value = float(parts[3])
|
||||
if not 0.0 <= value <= 1.0:
|
||||
return "Intensity must be between 0.0 and 1.0"
|
||||
plugin = registry.get(effect_name)
|
||||
if plugin:
|
||||
plugin.config.intensity = value
|
||||
return f"Set {effect_name} intensity to {value}"
|
||||
except ValueError:
|
||||
return "Invalid intensity value"
|
||||
|
||||
return f"Unknown action: {action}"
|
||||
|
||||
|
||||
def _format_stats() -> str:
|
||||
monitor = get_monitor()
|
||||
stats = monitor.get_stats()
|
||||
|
||||
if "error" in stats:
|
||||
return stats["error"]
|
||||
|
||||
lines = ["Performance Stats:"]
|
||||
|
||||
pipeline = stats["pipeline"]
|
||||
lines.append(
|
||||
f" Pipeline: avg={pipeline['avg_ms']:.2f}ms min={pipeline['min_ms']:.2f}ms max={pipeline['max_ms']:.2f}ms (over {stats['frame_count']} frames)"
|
||||
)
|
||||
|
||||
if stats["effects"]:
|
||||
lines.append(" Per-effect (avg ms):")
|
||||
for name, effect_stats in stats["effects"].items():
|
||||
lines.append(
|
||||
f" {name}: avg={effect_stats['avg_ms']:.2f}ms min={effect_stats['min_ms']:.2f}ms max={effect_stats['max_ms']:.2f}ms"
|
||||
)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def show_effects_menu() -> str:
|
||||
"""Generate effects menu text for display."""
|
||||
registry = get_registry()
|
||||
chain = _get_effect_chain()
|
||||
|
||||
lines = [
|
||||
"\033[1;38;5;231m=== EFFECTS MENU ===\033[0m",
|
||||
"",
|
||||
"Effects:",
|
||||
]
|
||||
|
||||
for name, plugin in registry.list_all().items():
|
||||
status = "ON" if plugin.config.enabled else "OFF"
|
||||
intensity = plugin.config.intensity
|
||||
lines.append(f" [{status:3}] {name}: intensity={intensity:.2f}")
|
||||
|
||||
if chain:
|
||||
lines.append("")
|
||||
lines.append(f"Pipeline order: {' -> '.join(chain.get_order())}")
|
||||
|
||||
lines.append("")
|
||||
lines.append("Controls:")
|
||||
lines.append(" /effects <name> on|off")
|
||||
lines.append(" /effects <name> intensity <0.0-1.0>")
|
||||
lines.append(" /effects reorder name1,name2,...")
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
@@ -1,103 +0,0 @@
|
||||
from collections import deque
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class EffectTiming:
|
||||
name: str
|
||||
duration_ms: float
|
||||
buffer_chars_in: int
|
||||
buffer_chars_out: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class FrameTiming:
|
||||
frame_number: int
|
||||
total_ms: float
|
||||
effects: list[EffectTiming]
|
||||
|
||||
|
||||
class PerformanceMonitor:
|
||||
"""Collects and stores performance metrics for effect pipeline."""
|
||||
|
||||
def __init__(self, max_frames: int = 60):
|
||||
self._max_frames = max_frames
|
||||
self._frames: deque[FrameTiming] = deque(maxlen=max_frames)
|
||||
self._current_frame: list[EffectTiming] = []
|
||||
|
||||
def start_frame(self, frame_number: int) -> None:
|
||||
self._current_frame = []
|
||||
|
||||
def record_effect(
|
||||
self, name: str, duration_ms: float, chars_in: int, chars_out: int
|
||||
) -> None:
|
||||
self._current_frame.append(
|
||||
EffectTiming(
|
||||
name=name,
|
||||
duration_ms=duration_ms,
|
||||
buffer_chars_in=chars_in,
|
||||
buffer_chars_out=chars_out,
|
||||
)
|
||||
)
|
||||
|
||||
def end_frame(self, frame_number: int, total_ms: float) -> None:
|
||||
self._frames.append(
|
||||
FrameTiming(
|
||||
frame_number=frame_number,
|
||||
total_ms=total_ms,
|
||||
effects=self._current_frame,
|
||||
)
|
||||
)
|
||||
|
||||
def get_stats(self) -> dict:
|
||||
if not self._frames:
|
||||
return {"error": "No timing data available"}
|
||||
|
||||
total_times = [f.total_ms for f in self._frames]
|
||||
avg_total = sum(total_times) / len(total_times)
|
||||
min_total = min(total_times)
|
||||
max_total = max(total_times)
|
||||
|
||||
effect_stats: dict[str, dict] = {}
|
||||
for frame in self._frames:
|
||||
for effect in frame.effects:
|
||||
if effect.name not in effect_stats:
|
||||
effect_stats[effect.name] = {"times": [], "total_chars": 0}
|
||||
effect_stats[effect.name]["times"].append(effect.duration_ms)
|
||||
effect_stats[effect.name]["total_chars"] += effect.buffer_chars_out
|
||||
|
||||
for name, stats in effect_stats.items():
|
||||
times = stats["times"]
|
||||
stats["avg_ms"] = sum(times) / len(times)
|
||||
stats["min_ms"] = min(times)
|
||||
stats["max_ms"] = max(times)
|
||||
del stats["times"]
|
||||
|
||||
return {
|
||||
"frame_count": len(self._frames),
|
||||
"pipeline": {
|
||||
"avg_ms": avg_total,
|
||||
"min_ms": min_total,
|
||||
"max_ms": max_total,
|
||||
},
|
||||
"effects": effect_stats,
|
||||
}
|
||||
|
||||
def reset(self) -> None:
|
||||
self._frames.clear()
|
||||
self._current_frame = []
|
||||
|
||||
|
||||
_monitor: PerformanceMonitor | None = None
|
||||
|
||||
|
||||
def get_monitor() -> PerformanceMonitor:
|
||||
global _monitor
|
||||
if _monitor is None:
|
||||
_monitor = PerformanceMonitor()
|
||||
return _monitor
|
||||
|
||||
|
||||
def set_monitor(monitor: PerformanceMonitor) -> None:
|
||||
global _monitor
|
||||
_monitor = monitor
|
||||
@@ -1,59 +0,0 @@
|
||||
from engine.effects.types import EffectConfig, EffectPlugin
|
||||
|
||||
|
||||
class EffectRegistry:
|
||||
def __init__(self):
|
||||
self._plugins: dict[str, EffectPlugin] = {}
|
||||
self._discovered: bool = False
|
||||
|
||||
def register(self, plugin: EffectPlugin) -> None:
|
||||
self._plugins[plugin.name] = plugin
|
||||
|
||||
def get(self, name: str) -> EffectPlugin | None:
|
||||
return self._plugins.get(name)
|
||||
|
||||
def list_all(self) -> dict[str, EffectPlugin]:
|
||||
return self._plugins.copy()
|
||||
|
||||
def list_enabled(self) -> list[EffectPlugin]:
|
||||
return [p for p in self._plugins.values() if p.config.enabled]
|
||||
|
||||
def enable(self, name: str) -> bool:
|
||||
plugin = self._plugins.get(name)
|
||||
if plugin:
|
||||
plugin.config.enabled = True
|
||||
return True
|
||||
return False
|
||||
|
||||
def disable(self, name: str) -> bool:
|
||||
plugin = self._plugins.get(name)
|
||||
if plugin:
|
||||
plugin.config.enabled = False
|
||||
return True
|
||||
return False
|
||||
|
||||
def configure(self, name: str, config: EffectConfig) -> bool:
|
||||
plugin = self._plugins.get(name)
|
||||
if plugin:
|
||||
plugin.configure(config)
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_enabled(self, name: str) -> bool:
|
||||
plugin = self._plugins.get(name)
|
||||
return plugin.config.enabled if plugin else False
|
||||
|
||||
|
||||
_registry: EffectRegistry | None = None
|
||||
|
||||
|
||||
def get_registry() -> EffectRegistry:
|
||||
global _registry
|
||||
if _registry is None:
|
||||
_registry = EffectRegistry()
|
||||
return _registry
|
||||
|
||||
|
||||
def set_registry(registry: EffectRegistry) -> None:
|
||||
global _registry
|
||||
_registry = registry
|
||||
@@ -1,68 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class EffectContext:
|
||||
terminal_width: int
|
||||
terminal_height: int
|
||||
scroll_cam: int
|
||||
ticker_height: int
|
||||
camera_x: int = 0
|
||||
mic_excess: float = 0.0
|
||||
grad_offset: float = 0.0
|
||||
frame_number: int = 0
|
||||
has_message: bool = False
|
||||
items: list = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EffectConfig:
|
||||
enabled: bool = True
|
||||
intensity: float = 1.0
|
||||
params: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
class EffectPlugin(ABC):
|
||||
name: str
|
||||
config: EffectConfig
|
||||
|
||||
@abstractmethod
|
||||
def process(self, buf: list[str], ctx: EffectContext) -> list[str]: ...
|
||||
|
||||
@abstractmethod
|
||||
def configure(self, config: EffectConfig) -> None: ...
|
||||
|
||||
|
||||
def create_effect_context(
|
||||
terminal_width: int = 80,
|
||||
terminal_height: int = 24,
|
||||
scroll_cam: int = 0,
|
||||
ticker_height: int = 0,
|
||||
camera_x: int = 0,
|
||||
mic_excess: float = 0.0,
|
||||
grad_offset: float = 0.0,
|
||||
frame_number: int = 0,
|
||||
has_message: bool = False,
|
||||
items: list | None = None,
|
||||
) -> EffectContext:
|
||||
"""Factory function to create EffectContext with sensible defaults."""
|
||||
return EffectContext(
|
||||
terminal_width=terminal_width,
|
||||
terminal_height=terminal_height,
|
||||
scroll_cam=scroll_cam,
|
||||
ticker_height=ticker_height,
|
||||
camera_x=camera_x,
|
||||
mic_excess=mic_excess,
|
||||
grad_offset=grad_offset,
|
||||
frame_number=frame_number,
|
||||
has_message=has_message,
|
||||
items=items or [],
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PipelineConfig:
|
||||
order: list[str] = field(default_factory=list)
|
||||
effects: dict[str, EffectConfig] = field(default_factory=dict)
|
||||
@@ -1,67 +0,0 @@
|
||||
"""
|
||||
Source code feed — reads engine/*.py and emits non-blank, non-comment lines
|
||||
as scroll items. Used by --code mode.
|
||||
Depends on: nothing (stdlib only).
|
||||
"""
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
_ENGINE_DIR = Path(__file__).resolve().parent
|
||||
|
||||
|
||||
def _scope_map(source: str) -> dict[int, str]:
|
||||
"""Return {line_number: scope_label} for every line in source.
|
||||
|
||||
Nodes are sorted by range size descending so inner scopes overwrite
|
||||
outer ones, guaranteeing the narrowest enclosing scope wins.
|
||||
"""
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
return {}
|
||||
|
||||
nodes = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||||
end = getattr(node, "end_lineno", node.lineno)
|
||||
span = end - node.lineno
|
||||
nodes.append((span, node))
|
||||
|
||||
# Largest range first → inner scopes overwrite on second pass
|
||||
nodes.sort(key=lambda x: x[0], reverse=True)
|
||||
|
||||
scope = {}
|
||||
for _, node in nodes:
|
||||
end = getattr(node, "end_lineno", node.lineno)
|
||||
if isinstance(node, ast.ClassDef):
|
||||
label = node.name
|
||||
else:
|
||||
label = f"{node.name}()"
|
||||
for ln in range(node.lineno, end + 1):
|
||||
scope[ln] = label
|
||||
|
||||
return scope
|
||||
|
||||
|
||||
def fetch_code():
|
||||
"""Read engine/*.py and return (items, line_count, 0).
|
||||
|
||||
Each item is (text, src, ts) where:
|
||||
text = the code line (rstripped, indentation preserved)
|
||||
src = enclosing function/class name, e.g. 'stream()' or '<module>'
|
||||
ts = dotted module path, e.g. 'engine.scroll'
|
||||
"""
|
||||
items = []
|
||||
for path in sorted(_ENGINE_DIR.glob("*.py")):
|
||||
module = f"engine.{path.stem}"
|
||||
source = path.read_text(encoding="utf-8")
|
||||
scope = _scope_map(source)
|
||||
for lineno, raw in enumerate(source.splitlines(), start=1):
|
||||
stripped = raw.strip()
|
||||
if not stripped or stripped.startswith("#"):
|
||||
continue
|
||||
label = scope.get(lineno, "<module>")
|
||||
items.append((raw.rstrip(), label, module))
|
||||
|
||||
return items, len(items), 0
|
||||
@@ -10,15 +10,13 @@ from datetime import datetime
|
||||
|
||||
from engine import config
|
||||
from engine.effects import (
|
||||
EffectChain,
|
||||
EffectContext,
|
||||
fade_line,
|
||||
firehose_line,
|
||||
glitch_bar,
|
||||
noise,
|
||||
vis_trunc,
|
||||
)
|
||||
from engine.render import big_wrap, lr_gradient, msg_gradient
|
||||
from engine.render import big_wrap, lr_gradient, lr_gradient_opposite
|
||||
from engine.terminal import RST, W_COOL
|
||||
|
||||
MSG_META = "\033[38;5;245m"
|
||||
@@ -57,7 +55,7 @@ def render_message_overlay(
|
||||
else:
|
||||
msg_rows = msg_cache[1]
|
||||
|
||||
msg_rows = msg_gradient(
|
||||
msg_rows = lr_gradient_opposite(
|
||||
msg_rows, (time.monotonic() * config.GRAD_SPEED) % 1.0
|
||||
)
|
||||
|
||||
@@ -201,60 +199,3 @@ def render_firehose(items: list, w: int, fh: int, h: int) -> list[str]:
|
||||
fline = firehose_line(items, w)
|
||||
buf.append(f"\033[{scr_row};1H{fline}\033[K")
|
||||
return buf
|
||||
|
||||
|
||||
_effect_chain = None
|
||||
|
||||
|
||||
def init_effects() -> None:
|
||||
"""Initialize effect plugins and chain."""
|
||||
global _effect_chain
|
||||
from engine.effects import EffectChain, get_registry
|
||||
|
||||
registry = get_registry()
|
||||
|
||||
import effects_plugins
|
||||
|
||||
effects_plugins.discover_plugins()
|
||||
|
||||
chain = EffectChain(registry)
|
||||
chain.set_order(["noise", "fade", "glitch", "firehose"])
|
||||
_effect_chain = chain
|
||||
|
||||
|
||||
def process_effects(
|
||||
buf: list[str],
|
||||
w: int,
|
||||
h: int,
|
||||
scroll_cam: int,
|
||||
ticker_h: int,
|
||||
mic_excess: float,
|
||||
grad_offset: float,
|
||||
frame_number: int,
|
||||
has_message: bool,
|
||||
items: list,
|
||||
) -> list[str]:
|
||||
"""Process buffer through effect chain."""
|
||||
if _effect_chain is None:
|
||||
init_effects()
|
||||
|
||||
ctx = EffectContext(
|
||||
terminal_width=w,
|
||||
terminal_height=h,
|
||||
scroll_cam=scroll_cam,
|
||||
ticker_height=ticker_h,
|
||||
mic_excess=mic_excess,
|
||||
grad_offset=grad_offset,
|
||||
frame_number=frame_number,
|
||||
has_message=has_message,
|
||||
items=items,
|
||||
)
|
||||
return _effect_chain.process(buf, ctx)
|
||||
|
||||
|
||||
def get_effect_chain() -> EffectChain | None:
|
||||
"""Get the effect chain instance."""
|
||||
global _effect_chain
|
||||
if _effect_chain is None:
|
||||
init_effects()
|
||||
return _effect_chain
|
||||
|
||||
@@ -15,40 +15,9 @@ from engine.sources import NO_UPPER, SCRIPT_FONTS, SOURCE_LANGS
|
||||
from engine.terminal import RST
|
||||
from engine.translate import detect_location_language, translate_headline
|
||||
|
||||
|
||||
# ─── GRADIENT ─────────────────────────────────────────────
|
||||
def _color_codes_to_ansi(color_codes):
|
||||
"""Convert a list of 256-color codes to ANSI escape code strings.
|
||||
|
||||
Pattern: first 2 are bold, middle 8 are normal, last 2 are dim.
|
||||
|
||||
Args:
|
||||
color_codes: List of 12 integers (256-color palette codes)
|
||||
|
||||
Returns:
|
||||
List of ANSI escape code strings
|
||||
"""
|
||||
if not color_codes or len(color_codes) != 12:
|
||||
# Fallback to default green if invalid
|
||||
return _default_green_gradient()
|
||||
|
||||
result = []
|
||||
for i, code in enumerate(color_codes):
|
||||
if i < 2:
|
||||
# Bold for first 2 (bright leading edge)
|
||||
result.append(f"\033[1;38;5;{code}m")
|
||||
elif i < 10:
|
||||
# Normal for middle 8
|
||||
result.append(f"\033[38;5;{code}m")
|
||||
else:
|
||||
# Dim for last 2 (dark trailing edge)
|
||||
result.append(f"\033[2;38;5;{code}m")
|
||||
return result
|
||||
|
||||
|
||||
def _default_green_gradient():
|
||||
"""Return the default 12-color green gradient for fallback when no theme is active."""
|
||||
return [
|
||||
# Left → right: white-hot leading edge fades to near-black
|
||||
GRAD_COLS = [
|
||||
"\033[1;38;5;231m", # white
|
||||
"\033[1;38;5;195m", # pale cyan-white
|
||||
"\033[38;5;123m", # bright cyan
|
||||
@@ -63,10 +32,8 @@ def _default_green_gradient():
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
|
||||
def _default_magenta_gradient():
|
||||
"""Return the default 12-color magenta gradient for fallback when no theme is active."""
|
||||
return [
|
||||
# Complementary sweep for queue messages (opposite hue family from ticker greens)
|
||||
MSG_GRAD_COLS = [
|
||||
"\033[1;38;5;231m", # white
|
||||
"\033[1;38;5;225m", # pale pink-white
|
||||
"\033[38;5;219m", # bright pink
|
||||
@@ -81,7 +48,6 @@ def _default_magenta_gradient():
|
||||
"\033[2;38;5;235m", # near black
|
||||
]
|
||||
|
||||
|
||||
# ─── FONT LOADING ─────────────────────────────────────────
|
||||
_FONT_OBJ = None
|
||||
_FONT_OBJ_KEY = None
|
||||
@@ -223,15 +189,9 @@ def big_wrap(text, max_w, fnt=None):
|
||||
return out
|
||||
|
||||
|
||||
def lr_gradient(rows, offset=0.0, cols=None):
|
||||
def lr_gradient(rows, offset=0.0, grad_cols=None):
|
||||
"""Color each non-space block character with a shifting left-to-right gradient."""
|
||||
if cols is None:
|
||||
from engine import config
|
||||
|
||||
if config.ACTIVE_THEME:
|
||||
cols = _color_codes_to_ansi(config.ACTIVE_THEME.main_gradient)
|
||||
else:
|
||||
cols = _default_green_gradient()
|
||||
cols = grad_cols or GRAD_COLS
|
||||
n = len(cols)
|
||||
max_x = max((len(r.rstrip()) for r in rows if r.strip()), default=1)
|
||||
out = []
|
||||
@@ -253,30 +213,7 @@ def lr_gradient(rows, offset=0.0, cols=None):
|
||||
|
||||
def lr_gradient_opposite(rows, offset=0.0):
|
||||
"""Complementary (opposite wheel) gradient used for queue message panels."""
|
||||
return lr_gradient(rows, offset, _default_magenta_gradient())
|
||||
|
||||
|
||||
def msg_gradient(rows, offset):
|
||||
"""Apply message (ntfy) gradient using theme complementary colors.
|
||||
|
||||
Returns colored rows using ACTIVE_THEME.message_gradient if available,
|
||||
falling back to default magenta if no theme is set.
|
||||
|
||||
Args:
|
||||
rows: List of text strings to colorize
|
||||
offset: Gradient offset (0.0-1.0) for animation
|
||||
|
||||
Returns:
|
||||
List of rows with ANSI color codes applied
|
||||
"""
|
||||
from engine import config
|
||||
|
||||
cols = (
|
||||
_color_codes_to_ansi(config.ACTIVE_THEME.message_gradient)
|
||||
if config.ACTIVE_THEME
|
||||
else _default_magenta_gradient()
|
||||
)
|
||||
return lr_gradient(rows, offset, cols)
|
||||
return lr_gradient(rows, offset, MSG_GRAD_COLS)
|
||||
|
||||
|
||||
# ─── HEADLINE BLOCK ASSEMBLY ─────────────────────────────
|
||||
|
||||
@@ -4,42 +4,33 @@ Orchestrates viewport, frame timing, and layers.
|
||||
"""
|
||||
|
||||
import random
|
||||
import sys
|
||||
import time
|
||||
|
||||
from engine import config
|
||||
from engine.display import (
|
||||
Display,
|
||||
TerminalDisplay,
|
||||
)
|
||||
from engine.display import (
|
||||
get_monitor as _get_display_monitor,
|
||||
)
|
||||
from engine.frame import calculate_scroll_step
|
||||
from engine.layers import (
|
||||
apply_glitch,
|
||||
process_effects,
|
||||
render_firehose,
|
||||
render_message_overlay,
|
||||
render_ticker_zone,
|
||||
)
|
||||
from engine.terminal import CLR
|
||||
from engine.viewport import th, tw
|
||||
|
||||
USE_EFFECT_CHAIN = True
|
||||
|
||||
|
||||
def stream(items, ntfy_poller, mic_monitor, display: Display | None = None):
|
||||
def stream(items, ntfy_poller, mic_monitor):
|
||||
"""Main render loop with four layers: message, ticker, scroll motion, firehose."""
|
||||
if display is None:
|
||||
display = TerminalDisplay()
|
||||
random.shuffle(items)
|
||||
pool = list(items)
|
||||
seen = set()
|
||||
queued = 0
|
||||
|
||||
time.sleep(0.5)
|
||||
sys.stdout.write(CLR)
|
||||
sys.stdout.flush()
|
||||
|
||||
w, h = tw(), th()
|
||||
display.init(w, h)
|
||||
display.clear()
|
||||
fh = config.FIREHOSE_H if config.FIREHOSE else 0
|
||||
ticker_view_h = h - fh
|
||||
GAP = 3
|
||||
@@ -51,7 +42,6 @@ def stream(items, ntfy_poller, mic_monitor, display: Display | None = None):
|
||||
noise_cache = {}
|
||||
scroll_motion_accum = 0.0
|
||||
msg_cache = (None, None)
|
||||
frame_number = 0
|
||||
|
||||
while True:
|
||||
if queued >= config.HEADLINE_LIMIT and not active:
|
||||
@@ -103,39 +93,19 @@ def stream(items, ntfy_poller, mic_monitor, display: Display | None = None):
|
||||
buf.extend(ticker_buf)
|
||||
|
||||
mic_excess = mic_monitor.excess
|
||||
render_start = time.perf_counter()
|
||||
|
||||
if USE_EFFECT_CHAIN:
|
||||
buf = process_effects(
|
||||
buf,
|
||||
w,
|
||||
h,
|
||||
scroll_cam,
|
||||
ticker_h,
|
||||
mic_excess,
|
||||
grad_offset,
|
||||
frame_number,
|
||||
msg is not None,
|
||||
items,
|
||||
)
|
||||
else:
|
||||
buf = apply_glitch(buf, ticker_buf_start, mic_excess, w)
|
||||
|
||||
firehose_buf = render_firehose(items, w, fh, h)
|
||||
buf.extend(firehose_buf)
|
||||
|
||||
if msg_overlay:
|
||||
buf.extend(msg_overlay)
|
||||
|
||||
render_elapsed = (time.perf_counter() - render_start) * 1000
|
||||
monitor = _get_display_monitor()
|
||||
if monitor:
|
||||
chars = sum(len(line) for line in buf)
|
||||
monitor.record_effect("render", render_elapsed, chars, chars)
|
||||
|
||||
display.show(buf)
|
||||
sys.stdout.buffer.write("".join(buf).encode())
|
||||
sys.stdout.flush()
|
||||
|
||||
elapsed = time.monotonic() - t0
|
||||
time.sleep(max(0, config.FRAME_DT - elapsed))
|
||||
frame_number += 1
|
||||
|
||||
display.cleanup()
|
||||
sys.stdout.write(CLR)
|
||||
sys.stdout.flush()
|
||||
|
||||
@@ -1,60 +0,0 @@
|
||||
"""
|
||||
Theme definitions with color gradients for terminal rendering.
|
||||
|
||||
This module is data-only and does not import config or render
|
||||
to prevent circular dependencies.
|
||||
"""
|
||||
|
||||
|
||||
class Theme:
|
||||
"""Represents a color theme with two gradients."""
|
||||
|
||||
def __init__(self, name, main_gradient, message_gradient):
|
||||
"""Initialize a theme with name and color gradients.
|
||||
|
||||
Args:
|
||||
name: Theme identifier string
|
||||
main_gradient: List of 12 ANSI 256-color codes for main gradient
|
||||
message_gradient: List of 12 ANSI 256-color codes for message gradient
|
||||
"""
|
||||
self.name = name
|
||||
self.main_gradient = main_gradient
|
||||
self.message_gradient = message_gradient
|
||||
|
||||
|
||||
# ─── GRADIENT DEFINITIONS ─────────────────────────────────────────────────
|
||||
# Each gradient is 12 ANSI 256-color codes in sequence
|
||||
# Format: [light...] → [medium...] → [dark...] → [black]
|
||||
|
||||
_GREEN_MAIN = [231, 195, 123, 118, 82, 46, 40, 34, 28, 22, 22, 235]
|
||||
_GREEN_MSG = [231, 225, 219, 213, 207, 201, 165, 161, 125, 89, 89, 235]
|
||||
|
||||
_ORANGE_MAIN = [231, 215, 209, 208, 202, 166, 130, 94, 58, 94, 94, 235]
|
||||
_ORANGE_MSG = [231, 195, 33, 27, 21, 21, 21, 18, 18, 18, 18, 235]
|
||||
|
||||
_PURPLE_MAIN = [231, 225, 177, 171, 165, 135, 129, 93, 57, 57, 57, 235]
|
||||
_PURPLE_MSG = [231, 226, 226, 220, 220, 184, 184, 178, 178, 172, 172, 235]
|
||||
|
||||
|
||||
# ─── THEME REGISTRY ───────────────────────────────────────────────────────
|
||||
|
||||
THEME_REGISTRY = {
|
||||
"green": Theme("green", _GREEN_MAIN, _GREEN_MSG),
|
||||
"orange": Theme("orange", _ORANGE_MAIN, _ORANGE_MSG),
|
||||
"purple": Theme("purple", _PURPLE_MAIN, _PURPLE_MSG),
|
||||
}
|
||||
|
||||
|
||||
def get_theme(theme_id):
|
||||
"""Retrieve a theme by ID.
|
||||
|
||||
Args:
|
||||
theme_id: Theme identifier string
|
||||
|
||||
Returns:
|
||||
Theme object matching the ID
|
||||
|
||||
Raises:
|
||||
KeyError: If theme_id is not in registry
|
||||
"""
|
||||
return THEME_REGISTRY[theme_id]
|
||||
Binary file not shown.
Binary file not shown.
@@ -83,35 +83,3 @@ class TestStreamControllerCleanup:
|
||||
controller.cleanup()
|
||||
|
||||
mock_mic_instance.stop.assert_called_once()
|
||||
|
||||
|
||||
class TestStreamControllerWarmup:
|
||||
"""Tests for StreamController topic warmup."""
|
||||
|
||||
def test_warmup_topics_idempotent(self):
|
||||
"""warmup_topics can be called multiple times."""
|
||||
StreamController._topics_warmed = False
|
||||
|
||||
with patch("urllib.request.urlopen") as mock_urlopen:
|
||||
StreamController.warmup_topics()
|
||||
StreamController.warmup_topics()
|
||||
|
||||
assert mock_urlopen.call_count >= 3
|
||||
|
||||
def test_warmup_topics_sets_flag(self):
|
||||
"""warmup_topics sets the warmed flag."""
|
||||
StreamController._topics_warmed = False
|
||||
|
||||
with patch("urllib.request.urlopen"):
|
||||
StreamController.warmup_topics()
|
||||
|
||||
assert StreamController._topics_warmed is True
|
||||
|
||||
def test_warmup_topics_skips_after_first(self):
|
||||
"""warmup_topics skips after first call."""
|
||||
StreamController._topics_warmed = True
|
||||
|
||||
with patch("urllib.request.urlopen") as mock_urlopen:
|
||||
StreamController.warmup_topics()
|
||||
|
||||
mock_urlopen.assert_not_called()
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
"""
|
||||
Tests for engine.display module.
|
||||
"""
|
||||
|
||||
from engine.display import NullDisplay, TerminalDisplay
|
||||
|
||||
|
||||
class TestDisplayProtocol:
|
||||
"""Test that display backends satisfy the Display protocol."""
|
||||
|
||||
def test_terminal_display_is_display(self):
|
||||
"""TerminalDisplay satisfies Display protocol."""
|
||||
display = TerminalDisplay()
|
||||
assert hasattr(display, "init")
|
||||
assert hasattr(display, "show")
|
||||
assert hasattr(display, "clear")
|
||||
assert hasattr(display, "cleanup")
|
||||
|
||||
def test_null_display_is_display(self):
|
||||
"""NullDisplay satisfies Display protocol."""
|
||||
display = NullDisplay()
|
||||
assert hasattr(display, "init")
|
||||
assert hasattr(display, "show")
|
||||
assert hasattr(display, "clear")
|
||||
assert hasattr(display, "cleanup")
|
||||
|
||||
|
||||
class TestTerminalDisplay:
|
||||
"""Tests for TerminalDisplay class."""
|
||||
|
||||
def test_init_sets_dimensions(self):
|
||||
"""init stores terminal dimensions."""
|
||||
display = TerminalDisplay()
|
||||
display.init(80, 24)
|
||||
assert display.width == 80
|
||||
assert display.height == 24
|
||||
|
||||
def test_show_returns_none(self):
|
||||
"""show returns None after writing to stdout."""
|
||||
display = TerminalDisplay()
|
||||
display.width = 80
|
||||
display.height = 24
|
||||
display.show(["line1", "line2"])
|
||||
|
||||
def test_clear_does_not_error(self):
|
||||
"""clear works without error."""
|
||||
display = TerminalDisplay()
|
||||
display.clear()
|
||||
|
||||
def test_cleanup_does_not_error(self):
|
||||
"""cleanup works without error."""
|
||||
display = TerminalDisplay()
|
||||
display.cleanup()
|
||||
|
||||
|
||||
class TestNullDisplay:
|
||||
"""Tests for NullDisplay class."""
|
||||
|
||||
def test_init_stores_dimensions(self):
|
||||
"""init stores dimensions."""
|
||||
display = NullDisplay()
|
||||
display.init(100, 50)
|
||||
assert display.width == 100
|
||||
assert display.height == 50
|
||||
|
||||
def test_show_does_nothing(self):
|
||||
"""show discards buffer without error."""
|
||||
display = NullDisplay()
|
||||
display.show(["line1", "line2", "line3"])
|
||||
|
||||
def test_clear_does_nothing(self):
|
||||
"""clear does nothing."""
|
||||
display = NullDisplay()
|
||||
display.clear()
|
||||
|
||||
def test_cleanup_does_nothing(self):
|
||||
"""cleanup does nothing."""
|
||||
display = NullDisplay()
|
||||
display.cleanup()
|
||||
@@ -1,427 +0,0 @@
|
||||
"""
|
||||
Tests for engine.effects module.
|
||||
"""
|
||||
|
||||
from engine.effects import EffectChain, EffectConfig, EffectContext, EffectRegistry
|
||||
|
||||
|
||||
class MockEffect:
|
||||
name = "mock"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
def __init__(self):
|
||||
self.processed = False
|
||||
self.last_ctx = None
|
||||
|
||||
def process(self, buf, ctx):
|
||||
self.processed = True
|
||||
self.last_ctx = ctx
|
||||
return buf + ["processed"]
|
||||
|
||||
def configure(self, config):
|
||||
self.config = config
|
||||
|
||||
|
||||
class TestEffectConfig:
|
||||
def test_defaults(self):
|
||||
cfg = EffectConfig()
|
||||
assert cfg.enabled is True
|
||||
assert cfg.intensity == 1.0
|
||||
assert cfg.params == {}
|
||||
|
||||
def test_custom_values(self):
|
||||
cfg = EffectConfig(enabled=False, intensity=0.5, params={"key": "value"})
|
||||
assert cfg.enabled is False
|
||||
assert cfg.intensity == 0.5
|
||||
assert cfg.params == {"key": "value"}
|
||||
|
||||
|
||||
class TestEffectContext:
|
||||
def test_defaults(self):
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=0,
|
||||
has_message=False,
|
||||
)
|
||||
assert ctx.terminal_width == 80
|
||||
assert ctx.terminal_height == 24
|
||||
assert ctx.ticker_height == 20
|
||||
assert ctx.items == []
|
||||
|
||||
def test_with_items(self):
|
||||
items = [("Title", "Source", "12:00")]
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=0,
|
||||
has_message=False,
|
||||
items=items,
|
||||
)
|
||||
assert ctx.items == items
|
||||
|
||||
|
||||
class TestEffectRegistry:
|
||||
def test_init_empty(self):
|
||||
registry = EffectRegistry()
|
||||
assert len(registry.list_all()) == 0
|
||||
|
||||
def test_register(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
registry.register(effect)
|
||||
assert "mock" in registry.list_all()
|
||||
|
||||
def test_get(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
registry.register(effect)
|
||||
retrieved = registry.get("mock")
|
||||
assert retrieved is effect
|
||||
|
||||
def test_get_nonexistent(self):
|
||||
registry = EffectRegistry()
|
||||
assert registry.get("nonexistent") is None
|
||||
|
||||
def test_enable(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.config.enabled = False
|
||||
registry.register(effect)
|
||||
registry.enable("mock")
|
||||
assert effect.config.enabled is True
|
||||
|
||||
def test_disable(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.config.enabled = True
|
||||
registry.register(effect)
|
||||
registry.disable("mock")
|
||||
assert effect.config.enabled is False
|
||||
|
||||
def test_list_enabled(self):
|
||||
registry = EffectRegistry()
|
||||
|
||||
class EnabledEffect:
|
||||
name = "enabled_effect"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
class DisabledEffect:
|
||||
name = "disabled_effect"
|
||||
config = EffectConfig(enabled=False, intensity=1.0)
|
||||
|
||||
registry.register(EnabledEffect())
|
||||
registry.register(DisabledEffect())
|
||||
enabled = registry.list_enabled()
|
||||
assert len(enabled) == 1
|
||||
assert enabled[0].name == "enabled_effect"
|
||||
|
||||
def test_configure(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
registry.register(effect)
|
||||
new_config = EffectConfig(enabled=False, intensity=0.3)
|
||||
registry.configure("mock", new_config)
|
||||
assert effect.config.enabled is False
|
||||
assert effect.config.intensity == 0.3
|
||||
|
||||
def test_is_enabled(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.config.enabled = True
|
||||
registry.register(effect)
|
||||
assert registry.is_enabled("mock") is True
|
||||
assert registry.is_enabled("nonexistent") is False
|
||||
|
||||
|
||||
class TestEffectChain:
|
||||
def test_init(self):
|
||||
registry = EffectRegistry()
|
||||
chain = EffectChain(registry)
|
||||
assert chain.get_order() == []
|
||||
|
||||
def test_set_order(self):
|
||||
registry = EffectRegistry()
|
||||
effect1 = MockEffect()
|
||||
effect1.name = "effect1"
|
||||
effect2 = MockEffect()
|
||||
effect2.name = "effect2"
|
||||
registry.register(effect1)
|
||||
registry.register(effect2)
|
||||
chain = EffectChain(registry)
|
||||
chain.set_order(["effect1", "effect2"])
|
||||
assert chain.get_order() == ["effect1", "effect2"]
|
||||
|
||||
def test_add_effect(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.name = "test_effect"
|
||||
registry.register(effect)
|
||||
chain = EffectChain(registry)
|
||||
chain.add_effect("test_effect")
|
||||
assert "test_effect" in chain.get_order()
|
||||
|
||||
def test_add_effect_invalid(self):
|
||||
registry = EffectRegistry()
|
||||
chain = EffectChain(registry)
|
||||
result = chain.add_effect("nonexistent")
|
||||
assert result is False
|
||||
|
||||
def test_remove_effect(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.name = "test_effect"
|
||||
registry.register(effect)
|
||||
chain = EffectChain(registry)
|
||||
chain.set_order(["test_effect"])
|
||||
chain.remove_effect("test_effect")
|
||||
assert "test_effect" not in chain.get_order()
|
||||
|
||||
def test_reorder(self):
|
||||
registry = EffectRegistry()
|
||||
effect1 = MockEffect()
|
||||
effect1.name = "effect1"
|
||||
effect2 = MockEffect()
|
||||
effect2.name = "effect2"
|
||||
effect3 = MockEffect()
|
||||
effect3.name = "effect3"
|
||||
registry.register(effect1)
|
||||
registry.register(effect2)
|
||||
registry.register(effect3)
|
||||
chain = EffectChain(registry)
|
||||
chain.set_order(["effect1", "effect2", "effect3"])
|
||||
result = chain.reorder(["effect3", "effect1", "effect2"])
|
||||
assert result is True
|
||||
assert chain.get_order() == ["effect3", "effect1", "effect2"]
|
||||
|
||||
def test_reorder_invalid(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.name = "effect1"
|
||||
registry.register(effect)
|
||||
chain = EffectChain(registry)
|
||||
result = chain.reorder(["effect1", "nonexistent"])
|
||||
assert result is False
|
||||
|
||||
def test_process_empty_chain(self):
|
||||
registry = EffectRegistry()
|
||||
chain = EffectChain(registry)
|
||||
buf = ["line1", "line2"]
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=0,
|
||||
has_message=False,
|
||||
)
|
||||
result = chain.process(buf, ctx)
|
||||
assert result == buf
|
||||
|
||||
def test_process_with_effects(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.name = "test_effect"
|
||||
registry.register(effect)
|
||||
chain = EffectChain(registry)
|
||||
chain.set_order(["test_effect"])
|
||||
buf = ["line1", "line2"]
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=0,
|
||||
has_message=False,
|
||||
)
|
||||
result = chain.process(buf, ctx)
|
||||
assert result == ["line1", "line2", "processed"]
|
||||
assert effect.processed is True
|
||||
assert effect.last_ctx is ctx
|
||||
|
||||
def test_process_disabled_effect(self):
|
||||
registry = EffectRegistry()
|
||||
effect = MockEffect()
|
||||
effect.name = "test_effect"
|
||||
effect.config.enabled = False
|
||||
registry.register(effect)
|
||||
chain = EffectChain(registry)
|
||||
chain.set_order(["test_effect"])
|
||||
buf = ["line1"]
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=0,
|
||||
has_message=False,
|
||||
)
|
||||
result = chain.process(buf, ctx)
|
||||
assert result == ["line1"]
|
||||
assert effect.processed is False
|
||||
|
||||
|
||||
class TestEffectsExports:
|
||||
def test_all_exports_are_importable(self):
|
||||
"""Verify all exports in __all__ can actually be imported."""
|
||||
import engine.effects as effects_module
|
||||
|
||||
for name in effects_module.__all__:
|
||||
getattr(effects_module, name)
|
||||
|
||||
|
||||
class TestPerformanceMonitor:
|
||||
def test_empty_stats(self):
|
||||
from engine.effects.performance import PerformanceMonitor
|
||||
|
||||
monitor = PerformanceMonitor()
|
||||
stats = monitor.get_stats()
|
||||
assert "error" in stats
|
||||
|
||||
def test_record_and_retrieve(self):
|
||||
from engine.effects.performance import PerformanceMonitor
|
||||
|
||||
monitor = PerformanceMonitor()
|
||||
monitor.start_frame(1)
|
||||
monitor.record_effect("test_effect", 1.5, 100, 150)
|
||||
monitor.end_frame(1, 2.0)
|
||||
|
||||
stats = monitor.get_stats()
|
||||
assert "error" not in stats
|
||||
assert stats["frame_count"] == 1
|
||||
assert "test_effect" in stats["effects"]
|
||||
|
||||
def test_multiple_frames(self):
|
||||
from engine.effects.performance import PerformanceMonitor
|
||||
|
||||
monitor = PerformanceMonitor(max_frames=3)
|
||||
for i in range(5):
|
||||
monitor.start_frame(i)
|
||||
monitor.record_effect("effect1", 1.0, 100, 100)
|
||||
monitor.record_effect("effect2", 0.5, 100, 100)
|
||||
monitor.end_frame(i, 1.5)
|
||||
|
||||
stats = monitor.get_stats()
|
||||
assert stats["frame_count"] == 3
|
||||
assert "effect1" in stats["effects"]
|
||||
assert "effect2" in stats["effects"]
|
||||
|
||||
def test_reset(self):
|
||||
from engine.effects.performance import PerformanceMonitor
|
||||
|
||||
monitor = PerformanceMonitor()
|
||||
monitor.start_frame(1)
|
||||
monitor.record_effect("test", 1.0, 100, 100)
|
||||
monitor.end_frame(1, 1.0)
|
||||
|
||||
monitor.reset()
|
||||
stats = monitor.get_stats()
|
||||
assert "error" in stats
|
||||
|
||||
|
||||
class TestEffectPipelinePerformance:
|
||||
def test_pipeline_stays_within_frame_budget(self):
|
||||
"""Verify effect pipeline completes within frame budget (33ms for 30fps)."""
|
||||
from engine.effects import (
|
||||
EffectChain,
|
||||
EffectConfig,
|
||||
EffectContext,
|
||||
EffectRegistry,
|
||||
)
|
||||
|
||||
class DummyEffect:
|
||||
name = "dummy"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
def process(self, buf, ctx):
|
||||
return [line * 2 for line in buf]
|
||||
|
||||
registry = EffectRegistry()
|
||||
registry.register(DummyEffect())
|
||||
|
||||
from engine.effects.performance import PerformanceMonitor
|
||||
|
||||
monitor = PerformanceMonitor(max_frames=10)
|
||||
chain = EffectChain(registry, monitor)
|
||||
chain.set_order(["dummy"])
|
||||
|
||||
buf = ["x" * 80] * 20
|
||||
|
||||
for i in range(10):
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=i,
|
||||
has_message=False,
|
||||
)
|
||||
chain.process(buf, ctx)
|
||||
|
||||
stats = monitor.get_stats()
|
||||
assert "error" not in stats
|
||||
assert stats["pipeline"]["max_ms"] < 33.0
|
||||
|
||||
def test_individual_effects_performance(self):
|
||||
"""Verify individual effects don't exceed 10ms per frame."""
|
||||
from engine.effects import (
|
||||
EffectChain,
|
||||
EffectConfig,
|
||||
EffectContext,
|
||||
EffectRegistry,
|
||||
)
|
||||
|
||||
class SlowEffect:
|
||||
name = "slow"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
def process(self, buf, ctx):
|
||||
result = []
|
||||
for line in buf:
|
||||
result.append(line)
|
||||
result.append(line + line)
|
||||
return result
|
||||
|
||||
registry = EffectRegistry()
|
||||
registry.register(SlowEffect())
|
||||
|
||||
from engine.effects.performance import PerformanceMonitor
|
||||
|
||||
monitor = PerformanceMonitor(max_frames=5)
|
||||
chain = EffectChain(registry, monitor)
|
||||
chain.set_order(["slow"])
|
||||
|
||||
buf = ["x" * 80] * 10
|
||||
|
||||
for i in range(5):
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=i,
|
||||
has_message=False,
|
||||
)
|
||||
chain.process(buf, ctx)
|
||||
|
||||
stats = monitor.get_stats()
|
||||
assert "error" not in stats
|
||||
assert stats["effects"]["slow"]["max_ms"] < 10.0
|
||||
@@ -1,117 +0,0 @@
|
||||
"""
|
||||
Tests for engine.effects.controller module.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from engine.effects.controller import (
|
||||
handle_effects_command,
|
||||
set_effect_chain_ref,
|
||||
)
|
||||
|
||||
|
||||
class TestHandleEffectsCommand:
|
||||
"""Tests for handle_effects_command function."""
|
||||
|
||||
def test_list_effects(self):
|
||||
"""list command returns formatted effects list."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_plugin.config.enabled = True
|
||||
mock_plugin.config.intensity = 0.5
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
with patch("engine.effects.controller._get_effect_chain") as mock_chain:
|
||||
mock_chain.return_value.get_order.return_value = ["noise"]
|
||||
|
||||
result = handle_effects_command("/effects list")
|
||||
|
||||
assert "noise: ON" in result
|
||||
assert "intensity=0.5" in result
|
||||
|
||||
def test_enable_effect(self):
|
||||
"""enable command calls registry.enable."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_registry.return_value.get.return_value = mock_plugin
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
result = handle_effects_command("/effects noise on")
|
||||
|
||||
assert "Enabled: noise" in result
|
||||
mock_registry.return_value.enable.assert_called_once_with("noise")
|
||||
|
||||
def test_disable_effect(self):
|
||||
"""disable command calls registry.disable."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_registry.return_value.get.return_value = mock_plugin
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
result = handle_effects_command("/effects noise off")
|
||||
|
||||
assert "Disabled: noise" in result
|
||||
mock_registry.return_value.disable.assert_called_once_with("noise")
|
||||
|
||||
def test_set_intensity(self):
|
||||
"""intensity command sets plugin intensity."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_plugin.config.intensity = 0.5
|
||||
mock_registry.return_value.get.return_value = mock_plugin
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
result = handle_effects_command("/effects noise intensity 0.8")
|
||||
|
||||
assert "intensity to 0.8" in result
|
||||
assert mock_plugin.config.intensity == 0.8
|
||||
|
||||
def test_invalid_intensity_range(self):
|
||||
"""intensity outside 0.0-1.0 returns error."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_registry.return_value.get.return_value = mock_plugin
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
result = handle_effects_command("/effects noise intensity 1.5")
|
||||
|
||||
assert "between 0.0 and 1.0" in result
|
||||
|
||||
def test_reorder_pipeline(self):
|
||||
"""reorder command calls chain.reorder."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_registry.return_value.list_all.return_value = {}
|
||||
|
||||
with patch("engine.effects.controller._get_effect_chain") as mock_chain:
|
||||
mock_chain_instance = MagicMock()
|
||||
mock_chain_instance.reorder.return_value = True
|
||||
mock_chain.return_value = mock_chain_instance
|
||||
|
||||
result = handle_effects_command("/effects reorder noise,fade")
|
||||
|
||||
assert "Reordered pipeline" in result
|
||||
mock_chain_instance.reorder.assert_called_once_with(["noise", "fade"])
|
||||
|
||||
def test_unknown_command(self):
|
||||
"""unknown command returns error."""
|
||||
result = handle_effects_command("/unknown")
|
||||
assert "Unknown command" in result
|
||||
|
||||
def test_non_effects_command(self):
|
||||
"""non-effects command returns error."""
|
||||
result = handle_effects_command("not a command")
|
||||
assert "Unknown command" in result
|
||||
|
||||
|
||||
class TestSetEffectChainRef:
|
||||
"""Tests for set_effect_chain_ref function."""
|
||||
|
||||
def test_sets_global_ref(self):
|
||||
"""set_effect_chain_ref updates global reference."""
|
||||
mock_chain = MagicMock()
|
||||
set_effect_chain_ref(mock_chain)
|
||||
|
||||
from engine.effects.controller import _get_effect_chain
|
||||
|
||||
result = _get_effect_chain()
|
||||
assert result == mock_chain
|
||||
@@ -1,35 +0,0 @@
|
||||
import re
|
||||
|
||||
from engine.fetch_code import fetch_code
|
||||
|
||||
|
||||
def test_return_shape():
|
||||
items, line_count, ignored = fetch_code()
|
||||
assert isinstance(items, list)
|
||||
assert line_count == len(items)
|
||||
assert ignored == 0
|
||||
|
||||
|
||||
def test_items_are_tuples():
|
||||
items, _, _ = fetch_code()
|
||||
assert items, "expected at least one code line"
|
||||
for item in items:
|
||||
assert isinstance(item, tuple) and len(item) == 3
|
||||
text, src, ts = item
|
||||
assert isinstance(text, str)
|
||||
assert isinstance(src, str)
|
||||
assert isinstance(ts, str)
|
||||
|
||||
|
||||
def test_blank_and_comment_lines_excluded():
|
||||
items, _, _ = fetch_code()
|
||||
for text, _, _ in items:
|
||||
assert text.strip(), "blank line should have been filtered"
|
||||
assert not text.strip().startswith("#"), "comment line should have been filtered"
|
||||
|
||||
|
||||
def test_module_path_format():
|
||||
items, _, _ = fetch_code()
|
||||
pattern = re.compile(r"^engine\.\w+$")
|
||||
for _, _, ts in items:
|
||||
assert pattern.match(ts), f"unexpected module path: {ts!r}"
|
||||
@@ -1,301 +0,0 @@
|
||||
"""
|
||||
Tests for engine.render module.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from engine import config, render
|
||||
|
||||
|
||||
class TestDefaultGradients:
|
||||
"""Tests for default gradient fallback functions."""
|
||||
|
||||
def test_default_green_gradient_length(self):
|
||||
"""_default_green_gradient returns 12 colors."""
|
||||
gradient = render._default_green_gradient()
|
||||
assert len(gradient) == 12
|
||||
|
||||
def test_default_green_gradient_is_list(self):
|
||||
"""_default_green_gradient returns a list."""
|
||||
gradient = render._default_green_gradient()
|
||||
assert isinstance(gradient, list)
|
||||
|
||||
def test_default_green_gradient_all_strings(self):
|
||||
"""_default_green_gradient returns list of ANSI code strings."""
|
||||
gradient = render._default_green_gradient()
|
||||
assert all(isinstance(code, str) for code in gradient)
|
||||
|
||||
def test_default_magenta_gradient_length(self):
|
||||
"""_default_magenta_gradient returns 12 colors."""
|
||||
gradient = render._default_magenta_gradient()
|
||||
assert len(gradient) == 12
|
||||
|
||||
def test_default_magenta_gradient_is_list(self):
|
||||
"""_default_magenta_gradient returns a list."""
|
||||
gradient = render._default_magenta_gradient()
|
||||
assert isinstance(gradient, list)
|
||||
|
||||
def test_default_magenta_gradient_all_strings(self):
|
||||
"""_default_magenta_gradient returns list of ANSI code strings."""
|
||||
gradient = render._default_magenta_gradient()
|
||||
assert all(isinstance(code, str) for code in gradient)
|
||||
|
||||
|
||||
class TestLrGradientUsesActiveTheme:
|
||||
"""Tests for lr_gradient using active theme."""
|
||||
|
||||
def test_lr_gradient_uses_active_theme_when_cols_none(self):
|
||||
"""lr_gradient uses ACTIVE_THEME.main_gradient when cols=None."""
|
||||
# Save original state
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
# Set a theme
|
||||
config.set_active_theme("green")
|
||||
|
||||
# Create simple test data
|
||||
rows = ["text"]
|
||||
|
||||
# Call without cols parameter (cols=None)
|
||||
result = render.lr_gradient(rows, offset=0.0)
|
||||
|
||||
# Should not raise and should return colored output
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
# Should have ANSI codes (no plain "text")
|
||||
assert result[0] != "text"
|
||||
finally:
|
||||
# Restore original state
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_lr_gradient_fallback_when_no_theme(self):
|
||||
"""lr_gradient uses fallback green when ACTIVE_THEME is None."""
|
||||
# Save original state
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
# Clear the theme
|
||||
config.ACTIVE_THEME = None
|
||||
|
||||
# Create simple test data
|
||||
rows = ["text"]
|
||||
|
||||
# Call without cols parameter (should use fallback)
|
||||
result = render.lr_gradient(rows, offset=0.0)
|
||||
|
||||
# Should not raise and should return colored output
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
# Should have ANSI codes (no plain "text")
|
||||
assert result[0] != "text"
|
||||
finally:
|
||||
# Restore original state
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_lr_gradient_explicit_cols_parameter_still_works(self):
|
||||
"""lr_gradient with explicit cols parameter overrides theme."""
|
||||
# Custom gradient
|
||||
custom_cols = ["\033[38;5;1m", "\033[38;5;2m"] * 6
|
||||
|
||||
rows = ["xy"]
|
||||
result = render.lr_gradient(rows, offset=0.0, cols=custom_cols)
|
||||
|
||||
# Should use the provided cols
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
|
||||
def test_lr_gradient_respects_cols_parameter_name(self):
|
||||
"""lr_gradient accepts cols as keyword argument."""
|
||||
custom_cols = ["\033[38;5;1m", "\033[38;5;2m"] * 6
|
||||
|
||||
rows = ["xy"]
|
||||
# Call with cols as keyword
|
||||
result = render.lr_gradient(rows, offset=0.0, cols=custom_cols)
|
||||
|
||||
assert isinstance(result, list)
|
||||
|
||||
|
||||
class TestLrGradientBasicFunctionality:
|
||||
"""Tests to ensure lr_gradient basic functionality still works."""
|
||||
|
||||
def test_lr_gradient_colors_non_space_chars(self):
|
||||
"""lr_gradient colors non-space characters."""
|
||||
rows = ["hello"]
|
||||
|
||||
# Set a theme for the test
|
||||
original_theme = config.ACTIVE_THEME
|
||||
try:
|
||||
config.set_active_theme("green")
|
||||
result = render.lr_gradient(rows, offset=0.0)
|
||||
|
||||
# Result should have ANSI codes
|
||||
assert any("\033[" in r for r in result), "Expected ANSI codes in result"
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_lr_gradient_preserves_spaces(self):
|
||||
"""lr_gradient preserves spaces in output."""
|
||||
rows = ["a b c"]
|
||||
|
||||
original_theme = config.ACTIVE_THEME
|
||||
try:
|
||||
config.set_active_theme("green")
|
||||
result = render.lr_gradient(rows, offset=0.0)
|
||||
|
||||
# Spaces should be preserved (not colored)
|
||||
assert " " in result[0]
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_lr_gradient_empty_rows(self):
|
||||
"""lr_gradient handles empty rows correctly."""
|
||||
rows = [""]
|
||||
|
||||
original_theme = config.ACTIVE_THEME
|
||||
try:
|
||||
config.set_active_theme("green")
|
||||
result = render.lr_gradient(rows, offset=0.0)
|
||||
|
||||
assert result == [""]
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_lr_gradient_multiple_rows(self):
|
||||
"""lr_gradient handles multiple rows."""
|
||||
rows = ["row1", "row2", "row3"]
|
||||
|
||||
original_theme = config.ACTIVE_THEME
|
||||
try:
|
||||
config.set_active_theme("green")
|
||||
result = render.lr_gradient(rows, offset=0.0)
|
||||
|
||||
assert len(result) == 3
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
|
||||
class TestMsgGradient:
|
||||
"""Tests for msg_gradient function (message/ntfy overlay coloring)."""
|
||||
|
||||
def test_msg_gradient_uses_active_theme(self):
|
||||
"""msg_gradient uses ACTIVE_THEME.message_gradient when theme is set."""
|
||||
# Save original state
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
# Set a theme
|
||||
config.set_active_theme("green")
|
||||
|
||||
# Create simple test data
|
||||
rows = ["MESSAGE"]
|
||||
|
||||
# Call msg_gradient
|
||||
result = render.msg_gradient(rows, offset=0.0)
|
||||
|
||||
# Should return colored output using theme's message_gradient
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
# Should have ANSI codes from the message gradient
|
||||
assert result[0] != "MESSAGE"
|
||||
assert "\033[" in result[0]
|
||||
finally:
|
||||
# Restore original state
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_msg_gradient_fallback_when_no_theme(self):
|
||||
"""msg_gradient uses fallback magenta when ACTIVE_THEME is None."""
|
||||
# Save original state
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
# Clear the theme
|
||||
config.ACTIVE_THEME = None
|
||||
|
||||
# Create simple test data
|
||||
rows = ["MESSAGE"]
|
||||
|
||||
# Call msg_gradient
|
||||
result = render.msg_gradient(rows, offset=0.0)
|
||||
|
||||
# Should return colored output using default magenta
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
# Should have ANSI codes
|
||||
assert result[0] != "MESSAGE"
|
||||
assert "\033[" in result[0]
|
||||
finally:
|
||||
# Restore original state
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_msg_gradient_returns_colored_rows(self):
|
||||
"""msg_gradient returns properly colored rows with animation offset."""
|
||||
# Save original state
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
# Set a theme
|
||||
config.set_active_theme("orange")
|
||||
|
||||
rows = ["NTFY", "ALERT"]
|
||||
|
||||
# Call with offset
|
||||
result = render.msg_gradient(rows, offset=0.5)
|
||||
|
||||
# Should return same number of rows
|
||||
assert len(result) == 2
|
||||
# Both should be colored
|
||||
assert all("\033[" in r for r in result)
|
||||
# Should not be the original text
|
||||
assert result != rows
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_msg_gradient_different_themes_produce_different_results(self):
|
||||
"""msg_gradient produces different colors for different themes."""
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
rows = ["TEST"]
|
||||
|
||||
# Get result with green theme
|
||||
config.set_active_theme("green")
|
||||
result_green = render.msg_gradient(rows, offset=0.0)
|
||||
|
||||
# Get result with orange theme
|
||||
config.set_active_theme("orange")
|
||||
result_orange = render.msg_gradient(rows, offset=0.0)
|
||||
|
||||
# Results should be different (different message gradients)
|
||||
assert result_green != result_orange
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_msg_gradient_preserves_spacing(self):
|
||||
"""msg_gradient preserves spaces in rows."""
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
config.set_active_theme("purple")
|
||||
rows = ["M E S S A G E"]
|
||||
|
||||
result = render.msg_gradient(rows, offset=0.0)
|
||||
|
||||
# Spaces should be preserved
|
||||
assert " " in result[0]
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
|
||||
def test_msg_gradient_empty_rows(self):
|
||||
"""msg_gradient handles empty rows correctly."""
|
||||
original_theme = config.ACTIVE_THEME
|
||||
|
||||
try:
|
||||
config.set_active_theme("green")
|
||||
rows = [""]
|
||||
|
||||
result = render.msg_gradient(rows, offset=0.0)
|
||||
|
||||
# Empty row should stay empty
|
||||
assert result == [""]
|
||||
finally:
|
||||
config.ACTIVE_THEME = original_theme
|
||||
@@ -1,169 +0,0 @@
|
||||
"""
|
||||
Tests for engine.themes module.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from engine import themes
|
||||
|
||||
|
||||
class TestThemeConstruction:
|
||||
"""Tests for Theme class initialization."""
|
||||
|
||||
def test_theme_construction(self):
|
||||
"""Theme stores name and gradients correctly."""
|
||||
main_grad = ["color1", "color2", "color3"]
|
||||
msg_grad = ["msg1", "msg2", "msg3"]
|
||||
theme = themes.Theme("test_theme", main_grad, msg_grad)
|
||||
|
||||
assert theme.name == "test_theme"
|
||||
assert theme.main_gradient == main_grad
|
||||
assert theme.message_gradient == msg_grad
|
||||
|
||||
|
||||
class TestGradientLength:
|
||||
"""Tests for gradient length validation."""
|
||||
|
||||
def test_gradient_length_green(self):
|
||||
"""Green theme has exactly 12 colors in each gradient."""
|
||||
green = themes.THEME_REGISTRY["green"]
|
||||
assert len(green.main_gradient) == 12
|
||||
assert len(green.message_gradient) == 12
|
||||
|
||||
def test_gradient_length_orange(self):
|
||||
"""Orange theme has exactly 12 colors in each gradient."""
|
||||
orange = themes.THEME_REGISTRY["orange"]
|
||||
assert len(orange.main_gradient) == 12
|
||||
assert len(orange.message_gradient) == 12
|
||||
|
||||
def test_gradient_length_purple(self):
|
||||
"""Purple theme has exactly 12 colors in each gradient."""
|
||||
purple = themes.THEME_REGISTRY["purple"]
|
||||
assert len(purple.main_gradient) == 12
|
||||
assert len(purple.message_gradient) == 12
|
||||
|
||||
|
||||
class TestThemeRegistry:
|
||||
"""Tests for THEME_REGISTRY dictionary."""
|
||||
|
||||
def test_theme_registry_has_three_themes(self):
|
||||
"""Registry contains exactly three themes: green, orange, purple."""
|
||||
assert len(themes.THEME_REGISTRY) == 3
|
||||
assert set(themes.THEME_REGISTRY.keys()) == {"green", "orange", "purple"}
|
||||
|
||||
def test_registry_values_are_themes(self):
|
||||
"""All registry values are Theme instances."""
|
||||
for theme_id, theme in themes.THEME_REGISTRY.items():
|
||||
assert isinstance(theme, themes.Theme)
|
||||
assert theme.name == theme_id
|
||||
|
||||
|
||||
class TestGetTheme:
|
||||
"""Tests for get_theme function."""
|
||||
|
||||
def test_get_theme_valid_green(self):
|
||||
"""get_theme('green') returns correct green Theme."""
|
||||
green = themes.get_theme("green")
|
||||
assert isinstance(green, themes.Theme)
|
||||
assert green.name == "green"
|
||||
|
||||
def test_get_theme_valid_orange(self):
|
||||
"""get_theme('orange') returns correct orange Theme."""
|
||||
orange = themes.get_theme("orange")
|
||||
assert isinstance(orange, themes.Theme)
|
||||
assert orange.name == "orange"
|
||||
|
||||
def test_get_theme_valid_purple(self):
|
||||
"""get_theme('purple') returns correct purple Theme."""
|
||||
purple = themes.get_theme("purple")
|
||||
assert isinstance(purple, themes.Theme)
|
||||
assert purple.name == "purple"
|
||||
|
||||
def test_get_theme_invalid(self):
|
||||
"""get_theme with invalid ID raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
themes.get_theme("invalid_theme")
|
||||
|
||||
def test_get_theme_invalid_none(self):
|
||||
"""get_theme with None raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
themes.get_theme(None)
|
||||
|
||||
|
||||
class TestGreenTheme:
|
||||
"""Tests for green theme specific values."""
|
||||
|
||||
def test_green_theme_unchanged(self):
|
||||
"""Green theme maintains original color sequence."""
|
||||
green = themes.get_theme("green")
|
||||
|
||||
# Expected main gradient: 231→195→123→118→82→46→40→34→28→22→22(dim)→235
|
||||
expected_main = [231, 195, 123, 118, 82, 46, 40, 34, 28, 22, 22, 235]
|
||||
# Expected msg gradient: 231→225→219→213→207→201→165→161→125→89→89(dim)→235
|
||||
expected_msg = [231, 225, 219, 213, 207, 201, 165, 161, 125, 89, 89, 235]
|
||||
|
||||
assert green.main_gradient == expected_main
|
||||
assert green.message_gradient == expected_msg
|
||||
|
||||
def test_green_theme_name(self):
|
||||
"""Green theme has correct name."""
|
||||
green = themes.get_theme("green")
|
||||
assert green.name == "green"
|
||||
|
||||
|
||||
class TestOrangeTheme:
|
||||
"""Tests for orange theme specific values."""
|
||||
|
||||
def test_orange_theme_unchanged(self):
|
||||
"""Orange theme maintains original color sequence."""
|
||||
orange = themes.get_theme("orange")
|
||||
|
||||
# Expected main gradient: 231→215→209→208→202→166→130→94→58→94→94(dim)→235
|
||||
expected_main = [231, 215, 209, 208, 202, 166, 130, 94, 58, 94, 94, 235]
|
||||
# Expected msg gradient: 231→195→33→27→21→21→21→18→18→18→18(dim)→235
|
||||
expected_msg = [231, 195, 33, 27, 21, 21, 21, 18, 18, 18, 18, 235]
|
||||
|
||||
assert orange.main_gradient == expected_main
|
||||
assert orange.message_gradient == expected_msg
|
||||
|
||||
def test_orange_theme_name(self):
|
||||
"""Orange theme has correct name."""
|
||||
orange = themes.get_theme("orange")
|
||||
assert orange.name == "orange"
|
||||
|
||||
|
||||
class TestPurpleTheme:
|
||||
"""Tests for purple theme specific values."""
|
||||
|
||||
def test_purple_theme_unchanged(self):
|
||||
"""Purple theme maintains original color sequence."""
|
||||
purple = themes.get_theme("purple")
|
||||
|
||||
# Expected main gradient: 231→225→177→171→165→135→129→93→57→57→57(dim)→235
|
||||
expected_main = [231, 225, 177, 171, 165, 135, 129, 93, 57, 57, 57, 235]
|
||||
# Expected msg gradient: 231→226→226→220→220→184→184→178→178→172→172(dim)→235
|
||||
expected_msg = [231, 226, 226, 220, 220, 184, 184, 178, 178, 172, 172, 235]
|
||||
|
||||
assert purple.main_gradient == expected_main
|
||||
assert purple.message_gradient == expected_msg
|
||||
|
||||
def test_purple_theme_name(self):
|
||||
"""Purple theme has correct name."""
|
||||
purple = themes.get_theme("purple")
|
||||
assert purple.name == "purple"
|
||||
|
||||
|
||||
class TestThemeDataOnly:
|
||||
"""Tests to ensure themes module has no problematic imports."""
|
||||
|
||||
def test_themes_module_imports(self):
|
||||
"""themes module should be data-only without config/render imports."""
|
||||
import inspect
|
||||
source = inspect.getsource(themes)
|
||||
# Verify no imports of config or render (look for actual import statements)
|
||||
lines = source.split('\n')
|
||||
import_lines = [line for line in lines if line.strip().startswith('import ') or line.strip().startswith('from ')]
|
||||
# Filter out empty and comment lines
|
||||
import_lines = [line for line in import_lines if line.strip() and not line.strip().startswith('#')]
|
||||
# Should have no import lines
|
||||
assert len(import_lines) == 0, f"Found unexpected imports: {import_lines}"
|
||||
Reference in New Issue
Block a user