35 Commits

Author SHA1 Message Date
9415e18679 docs: use david's most recent README from feature/vector_display branch 2026-03-18 23:29:02 -07:00
0819f8d160 Merge pull request 'refactor: Make EffectPlugin an abstract base class, update effects to inherit from it, and improve plugin discovery.' (#32) from chore/plugin-migration into main
Reviewed-on: #32
2026-03-19 06:15:41 +00:00
edd1416407 refactor: Make EffectPlugin an abstract base class, update effects to inherit from it, and improve plugin discovery. 2026-03-18 23:06:04 -07:00
ac9b47f668 Merge pull request 'fix: use theme-aware msg_gradient for ntfy messages' (#31) from feat/color-pick into main
Reviewed-on: #31
2026-03-16 10:29:46 +00:00
b149825bcb Merge branch 'main' into feat/color-pick 2026-03-16 10:29:28 +00:00
1b29e91f9d fix: use theme-aware msg_gradient for ntfy messages
Replace lr_gradient_opposite() with msg_gradient() in render_message_overlay().
Messages now render in complementary colors matching the active theme:
- Green theme → magenta messages
- Orange theme → blue messages
- Purple theme → yellow messages
2026-03-16 03:25:52 -07:00
001158214c Merge pull request 'feat/color-pick' (#30) from feat/color-pick into main
Reviewed-on: #30
2026-03-16 10:13:47 +00:00
31f5d9f171 Merge origin/main into feat/color-pick
Resolved conflicts in tests/test_config.py by keeping TestActiveTheme tests.
Main branch has new architecture components (controller, events, layers, effects).
Color scheme feature preserved and compatible.
2026-03-16 03:13:13 -07:00
bc20a35ea9 refactor: Fix import ordering and code formatting with ruff
Organize imports in render.py and scroll.py to meet ruff style requirements.
Add blank lines for code formatting compliance.
2026-03-16 03:01:22 -07:00
d4d0344a12 feat: add pick_color_theme() UI and integration 2026-03-16 02:58:18 -07:00
84cb16d463 feat: update scroll.py to use theme message gradient
Add msg_gradient() helper function to render.py that applies message
(ntfy) gradient using the active theme's message_gradient property.
This replaces hardcoded magenta gradient in scroll.py with a theme-aware
approach that uses complementary colors from the active theme.

- Add msg_gradient(rows, offset) helper in render.py with fallback to
  default magenta gradient when no theme is active
- Update scroll.py imports to use msg_gradient instead of
  lr_gradient_opposite
- Replace lr_gradient_opposite() call with msg_gradient() in message
  overlay rendering
- Add 6 comprehensive tests for msg_gradient covering theme usage,
  fallback behavior, and edge cases

All tests pass (121 passed), no regressions detected.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-03-16 02:55:17 -07:00
d67423fe4c feat: update lr_gradient to use config.ACTIVE_THEME
- Remove hardcoded GRAD_COLS and MSG_GRAD_COLS module constants
- Add _default_green_gradient() and _default_magenta_gradient() fallback functions
- Add _color_codes_to_ansi() to convert integer color codes from themes to ANSI escape strings
- Update lr_gradient() signature: cols parameter (was grad_cols)
- lr_gradient() now pulls colors from config.ACTIVE_THEME when available
- Falls back to default green gradient when no theme is active
- Existing calls with explicit cols parameter continue to work
- Add comprehensive tests for new functionality

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-03-16 02:53:22 -07:00
ebe7b04ba5 feat: add ACTIVE_THEME global and set_active_theme() to config 2026-03-16 02:50:36 -07:00
abc4483859 feat: create Theme class and registry with finalized color gradients 2026-03-16 02:49:15 -07:00
d9422b1fec docs: add color scheme implementation plan
Comprehensive plan with 6 chunks, each containing bite-sized TDD tasks:
- Chunk 1: Theme class and registry
- Chunk 2: Config integration
- Chunk 3: Render pipeline
- Chunk 4: Message gradient integration
- Chunk 5: Color picker UI
- Chunk 6: Integration and validation

Each step includes exact code, test commands, and expected output.
2026-03-16 02:47:25 -07:00
6daea90b0a docs: add color scheme feature documentation to README
- Update opening description to mention selectable color gradients
- Add new 'Color Schemes' section with picker usage instructions
- Document three available themes (Green, Orange, Purple)
- Clarify that boot UI uses hardcoded green, not theme colors
2026-03-16 02:44:59 -07:00
9d9172ef0d docs: add terminal resize handling clarification 2026-03-16 02:42:59 -07:00
667bef2685 docs: revise color scheme design spec to address review feedback
- Clarify boot messages use hardcoded green, not theme gradients
- Finalize all gradient ANSI color codes (no TBD)
- Add initialization guarantee for ACTIVE_THEME
- Resolve circular import risk with data-only themes.py
- Update theme ID naming to match menu labels
- Expand test strategy and mock approach
2026-03-16 02:42:19 -07:00
f085042dee docs: add color scheme switcher design spec 2026-03-16 02:40:32 -07:00
8b696c96ce Merge pull request 'feat/code-scroll' (#29) from feat/code-scroll into main
Reviewed-on: #29
2026-03-16 09:16:53 +00:00
72d21459ca Merge branch 'main' into feat/code-scroll 2026-03-16 09:16:36 +00:00
58dbbbdba7 Merge pull request 'plugin-based effects architecture, daemon mode with command-and-control (C&C), and display abstraction' (#25) from david/Mainline:effects_plugins into main
Reviewed-on: #25
2026-03-16 09:13:22 +00:00
7ff78c66ed Merge pull request 'refactor to improve testability and modularization of the mainline terminal application' (#24) from david/Mainline:testability_modularization into main
Reviewed-on: #24
2026-03-16 09:11:13 +00:00
2229ccdea4 feat: introduce a 'code' mode to display source code lines, add new font assets, and include dedicated tests for code fetching. 2026-03-16 02:09:56 -07:00
f13e89f823 docs: add code-scroll mode design spec
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-16 01:48:18 -07:00
4228400c43 feat(daemon): add display abstraction and daemon mode with C&C 2026-03-15 19:17:23 -07:00
05cc475858 feat(cmdline): add command-line interface for mainline control 2026-03-15 19:17:10 -07:00
cfd7e8931e feat(effects): add plugin architecture with performance monitoring 2026-03-15 19:17:10 -07:00
15de46722a refactor: phase 4 - event-driven architecture foundation
- Add EventBus class with pub/sub messaging (thread-safe)
- Add emitter Protocol classes (EventEmitter, Startable, Stoppable)
- Add event emission to NtfyPoller (NtfyMessageEvent)
- Add event emission to MicMonitor (MicLevelEvent)
- Update StreamController to publish stream start/end events
- Add comprehensive tests for eventbus and emitters modules
2026-03-15 19:15:08 -07:00
35e5c8d38b refactor: phase 3 - API efficiency improvements
Add typed dataclasses for tuple returns:
- types.py: HeadlineItem, FetchResult, Block dataclasses with legacy tuple converters
- fetch.py: Add type hints and HeadlineTuple type alias

Add pyright for static type checking:
- Add pyright to dependencies
- Verify type coverage with pyright (0 errors in core modules)

This enables:
- Named types instead of raw tuples (better IDE support, self-documenting)
- Type-safe APIs across modules
- Backward compatibility via to_tuple/from_tuple methods

Note: Lazy imports skipped for render.py - startup impact is minimal.
2026-03-15 19:13:32 -07:00
cdc8094de2 refactor: phase 2 - modularization of scroll engine
Split monolithic scroll.py into focused modules:
- viewport.py: terminal size (tw/th), ANSI positioning helpers
- frame.py: FrameTimer class, scroll step calculation
- layers.py: message overlay, ticker zone, firehose rendering
- scroll.py: simplified orchestrator, imports from new modules

Add stream controller and event types for future event-driven architecture:
- controller.py: StreamController for source initialization and stream lifecycle
- events.py: EventType enum and event dataclasses (HeadlineEvent, FrameTickEvent, etc.)

Added tests for new modules:
- test_viewport.py: 8 tests for viewport utilities
- test_frame.py: 10 tests for frame timing
- test_layers.py: 13 tests for layer compositing
- test_events.py: 11 tests for event types
- test_controller.py: 6 tests for stream controller

This enables:
- Testable chunks with clear responsibilities
- Reusable viewport utilities across modules
- Better separation of concerns in render pipeline
- Foundation for future event-driven architecture

Also includes Phase 1 documentation updates in code comments.
2026-03-15 19:13:32 -07:00
f170143939 refactor: phase 1 - testability improvements
- Add Config dataclass with get_config()/set_config() for injection
- Add Config.from_args() for CLI argument parsing (testable)
- Add platform font path detection (Darwin/Linux)
- Bound translate cache with @lru_cache(maxsize=500)
- Add fixtures for external dependencies (network, feeds, config)
- Add 15 tests for Config class, from_args, and platform detection

This enables testability by:
- Allowing config injection instead of global mutable state
- Supporting custom argv in from_args() for testing
- Providing reusable fixtures for mocking network/filesystem
- Preventing unbounded memory growth in translation cache

Fixes: _arg_value/_arg_int not accepting custom argv
2026-03-15 19:13:32 -07:00
19fb4bc4fe Merge pull request 'docs/update-readme' (#23) from docs/update-readme into main
Reviewed-on: #23
2026-03-16 00:09:10 +00:00
ae10fd78ca refactor: Restructure README, add uv and mise commands, and detail component extension and development workflows. 2026-03-15 17:08:32 -07:00
4afab642f7 docs: add README update design spec
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-15 16:56:58 -07:00
35 changed files with 4367 additions and 156 deletions

296
README.md
View File

@@ -6,25 +6,45 @@ A full-screen terminal news ticker that renders live global headlines in large O
--- ---
## Run ## Using
### Run
```bash ```bash
python3 mainline.py # news stream python3 mainline.py # news stream
python3 mainline.py --poetry # literary consciousness mode python3 mainline.py --poetry # literary consciousness mode
python3 mainline.py -p # same python3 mainline.py -p # same
python3 mainline.py --firehose # dense rapid-fire headline mode python3 mainline.py --firehose # dense rapid-fire headline mode
python3 mainline.py --refresh # force re-fetch (bypass cache) python3 mainline.py --display websocket # web browser display only
python3 mainline.py --display both # terminal + web browser
python3 mainline.py --no-font-picker # skip interactive font picker 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-file path.otf # use a specific font file
python3 mainline.py --font-dir ~/fonts # scan a different font folder python3 mainline.py --font-dir ~/fonts # scan a different font folder
python3 mainline.py --font-index 1 # select face index within a collection python3 mainline.py --font-index 1 # select face index within a collection
``` ```
First run bootstraps a local `.mainline_venv/` and installs deps (`feedparser`, `Pillow`, `sounddevice`, `numpy`). Subsequent runs start immediately, loading from cache. Or with uv:
--- ```bash
uv run mainline.py
```
## Config 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
All constants live in `engine/config.py`: All constants live in `engine/config.py`:
@@ -33,90 +53,50 @@ All constants live in `engine/config.py`:
| `HEADLINE_LIMIT` | `1000` | Total headlines per session | | `HEADLINE_LIMIT` | `1000` | Total headlines per session |
| `FEED_TIMEOUT` | `10` | Per-feed HTTP timeout (seconds) | | `FEED_TIMEOUT` | `10` | Per-feed HTTP timeout (seconds) |
| `MIC_THRESHOLD_DB` | `50` | dB floor above which glitches spike | | `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_DIR` | `fonts/` | Folder scanned for `.otf`, `.ttf`, `.ttc` files |
| `FONT_PATH` | first file in `FONT_DIR` | Active display font (overridden by picker or `--font-file`) | | `FONT_PATH` | first file in `FONT_DIR` | Active display font |
| `FONT_INDEX` | `0` | Face index within a font collection file | | `FONT_PICKER` | `True` | Show interactive font picker at boot |
| `FONT_PICKER` | `True` | Show interactive font picker at boot (`--no-font-picker` to skip) |
| `FONT_SZ` | `60` | Font render size (affects block density) | | `FONT_SZ` | `60` | Font render size (affects block density) |
| `RENDER_H` | `8` | Terminal rows per headline line | | `RENDER_H` | `8` | Terminal rows per headline line |
| `SSAA` | `4` | Super-sampling factor (render at 4× then downsample) | | `SSAA` | `4` | Super-sampling factor |
| `SCROLL_DUR` | `5.625` | Seconds per headline | | `SCROLL_DUR` | `5.625` | Seconds per headline |
| `FRAME_DT` | `0.05` | Frame interval in seconds (20 FPS) | | `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) | | `FIREHOSE_H` | `12` | Firehose zone height (terminal rows) |
| `NTFY_TOPIC` | klubhaus URL | ntfy.sh JSON endpoint to poll | | `GRAD_SPEED` | `0.08` | Gradient sweep speed |
| `NTFY_POLL_INTERVAL` | `15` | Seconds between ntfy polls |
| `MESSAGE_DISPLAY_SECS` | `30` | How long an ntfy message holds the screen |
--- ### Display Modes
## Fonts Mainline supports multiple display backends:
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. - **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
Navigation: `↑`/`↓` or `j`/`k` to move, `Enter` or `q` to select. The selected face persists for that session. WebSocket mode serves a web client at http://localhost:8766 with ANSI color support and fullscreen mode.
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. ### Feeds
---
## 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`. ~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`. **Poetry mode** pulls from Project Gutenberg: Whitman, Dickinson, Thoreau, Emerson. Sources are in `engine/sources.py``POETRY_SOURCES`.
--- ### Fonts
## ntfy.sh Integration A `fonts/` directory is bundled with demo faces. On startup, an interactive picker lists all discovered faces with a live half-block preview.
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. 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.
To push a message: To push a message:
@@ -124,44 +104,160 @@ To push a message:
curl -d "Body text" -H "Title: Alert title" https://ntfy.sh/your_topic 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: ---
```python ## Internals
from engine.ntfy import NtfyPoller
poller = NtfyPoller("https://ntfy.sh/my_topic/json?since=20s&poll=1") ### How it works
poller.start()
# in render loop: - On launch, the font picker scans `fonts/` and presents a live-rendered TUI for face selection
msg = poller.get_active_message() # returns (title, body, timestamp) or None - 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
``` ```
--- ---
## Ideas / Future ## Development
### Performance ### Setup
- **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 Requires Python 3.10+ and [uv](https://docs.astral.sh/uv/).
- **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 ```bash
- **Keyword watch list** — highlight or strobe any headline matching tracked terms (names, topics, tickers) uv sync # minimal (no mic)
- **Breaking interrupt** — full-screen flash + synthesized blip when a high-priority keyword hits uv sync --all-extras # with mic support
- **Live data overlay** — secondary ticker strip at screen edge: BTC price, ISS position, geomagnetic index uv sync --all-extras --group dev # full dev environment
- **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 ### Tasks
- **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 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`.
--- ---
*macOS only (script/system font paths for translation are hardcoded). Primary display font is user-selectable via the bundled `fonts/` picker. Python 3.9+.* ## Roadmap
# test
### Performance
- Concurrent feed fetching with ThreadPoolExecutor
- Background feed refresh daemon
- Translation pre-fetch during boot
### Graphics
- Matrix rain katakana underlay
- CRT scanline simulation
- Sixel/iTerm2 inline images
- Parallax secondary column
### 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)
---
*Python 3.10+. Primary display font is user-selectable via bundled `fonts/` picker.*

250
cmdline.py Normal file
View File

@@ -0,0 +1,250 @@
#!/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()

View File

@@ -0,0 +1,894 @@
# 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

View File

@@ -0,0 +1,145 @@
# 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

View File

@@ -0,0 +1,154 @@
# 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

View File

@@ -0,0 +1,299 @@
# 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

View File

@@ -0,0 +1,36 @@
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

58
effects_plugins/fade.py Normal file
View File

@@ -0,0 +1,58 @@
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

View File

@@ -0,0 +1,72 @@
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

37
effects_plugins/glitch.py Normal file
View File

@@ -0,0 +1,37 @@
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

36
effects_plugins/noise.py Normal file
View File

@@ -0,0 +1,36 @@
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

View File

@@ -10,7 +10,7 @@ import termios
import time import time
import tty import tty
from engine import config, render from engine import config, render, themes
from engine.fetch import fetch_all, fetch_poetry, load_cache, save_cache from engine.fetch import fetch_all, fetch_poetry, load_cache, save_cache
from engine.mic import MicMonitor from engine.mic import MicMonitor
from engine.ntfy import NtfyPoller from engine.ntfy import NtfyPoller
@@ -65,6 +65,30 @@ def _read_picker_key():
return None 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): def _normalize_preview_rows(rows):
"""Trim shared left padding and trailing spaces for stable on-screen previews.""" """Trim shared left padding and trailing spaces for stable on-screen previews."""
non_empty = [r for r in rows if r.strip()] non_empty = [r for r in rows if r.strip()]
@@ -131,6 +155,50 @@ def _draw_font_picker(faces, selected):
print(f" {shown}") 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(): def pick_font_face():
"""Interactive startup picker for selecting a face from repo OTF files.""" """Interactive startup picker for selecting a face from repo OTF files."""
if not config.FONT_PICKER: if not config.FONT_PICKER:
@@ -262,6 +330,7 @@ def main():
w = tw() w = tw()
print(CLR, end="") print(CLR, end="")
print(CURSOR_OFF, end="") print(CURSOR_OFF, end="")
pick_color_theme()
pick_font_face() pick_font_face()
w = tw() w = tw()
print() print()
@@ -272,11 +341,10 @@ def main():
time.sleep(0.07) time.sleep(0.07)
print() print()
_subtitle = ( _subtitle = {
"literary consciousness stream" "poetry": "literary consciousness stream",
if config.MODE == "poetry" "code": "source consciousness stream",
else "digital consciousness stream" }.get(config.MODE, "digital consciousness stream")
)
print(f" {W_DIM}v0.1 · {_subtitle}{RST}") print(f" {W_DIM}v0.1 · {_subtitle}{RST}")
print(f" {W_GHOST}{'' * (w - 4)}{RST}") print(f" {W_GHOST}{'' * (w - 4)}{RST}")
print() print()
@@ -297,6 +365,15 @@ def main():
) )
print(f" {G_DIM}>{RST} {G_MID}{len(items)} STANZAS ACQUIRED{RST}") print(f" {G_DIM}>{RST} {G_MID}{len(items)} STANZAS ACQUIRED{RST}")
save_cache(items) 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: else:
slow_print(" > INITIALIZING FEED ARRAY...\n") slow_print(" > INITIALIZING FEED ARRAY...\n")
time.sleep(0.2) time.sleep(0.2)

View File

@@ -188,7 +188,13 @@ def set_config(config: Config) -> None:
HEADLINE_LIMIT = 1000 HEADLINE_LIMIT = 1000
FEED_TIMEOUT = 10 FEED_TIMEOUT = 10
MIC_THRESHOLD_DB = 50 # dB above which glitches intensify MIC_THRESHOLD_DB = 50 # dB above which glitches intensify
MODE = "poetry" if "--poetry" in sys.argv or "-p" in sys.argv else "news" MODE = (
"poetry"
if "--poetry" in sys.argv or "-p" in sys.argv
else "code"
if "--code" in sys.argv
else "news"
)
FIREHOSE = "--firehose" in sys.argv FIREHOSE = "--firehose" in sys.argv
# ─── NTFY MESSAGE QUEUE ────────────────────────────────── # ─── NTFY MESSAGE QUEUE ──────────────────────────────────
@@ -231,3 +237,26 @@ def set_font_selection(font_path=None, font_index=None):
FONT_PATH = _resolve_font_path(font_path) FONT_PATH = _resolve_font_path(font_path)
if font_index is not None: if font_index is not None:
FONT_INDEX = max(0, int(font_index)) 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)

102
engine/display.py Normal file
View File

@@ -0,0 +1,102 @@
"""
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

View File

@@ -0,0 +1,42 @@
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",
]

71
engine/effects/chain.py Normal file
View File

@@ -0,0 +1,71 @@
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

View File

@@ -0,0 +1,144 @@
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)

View File

@@ -0,0 +1,103 @@
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

View File

@@ -0,0 +1,59 @@
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

68
engine/effects/types.py Normal file
View File

@@ -0,0 +1,68 @@
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)

67
engine/fetch_code.py Normal file
View File

@@ -0,0 +1,67 @@
"""
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

View File

@@ -10,13 +10,15 @@ from datetime import datetime
from engine import config from engine import config
from engine.effects import ( from engine.effects import (
EffectChain,
EffectContext,
fade_line, fade_line,
firehose_line, firehose_line,
glitch_bar, glitch_bar,
noise, noise,
vis_trunc, vis_trunc,
) )
from engine.render import big_wrap, lr_gradient, lr_gradient_opposite from engine.render import big_wrap, lr_gradient, msg_gradient
from engine.terminal import RST, W_COOL from engine.terminal import RST, W_COOL
MSG_META = "\033[38;5;245m" MSG_META = "\033[38;5;245m"
@@ -55,7 +57,7 @@ def render_message_overlay(
else: else:
msg_rows = msg_cache[1] msg_rows = msg_cache[1]
msg_rows = lr_gradient_opposite( msg_rows = msg_gradient(
msg_rows, (time.monotonic() * config.GRAD_SPEED) % 1.0 msg_rows, (time.monotonic() * config.GRAD_SPEED) % 1.0
) )
@@ -199,3 +201,60 @@ def render_firehose(items: list, w: int, fh: int, h: int) -> list[str]:
fline = firehose_line(items, w) fline = firehose_line(items, w)
buf.append(f"\033[{scr_row};1H{fline}\033[K") buf.append(f"\033[{scr_row};1H{fline}\033[K")
return buf 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

View File

@@ -15,38 +15,72 @@ from engine.sources import NO_UPPER, SCRIPT_FONTS, SOURCE_LANGS
from engine.terminal import RST from engine.terminal import RST
from engine.translate import detect_location_language, translate_headline from engine.translate import detect_location_language, translate_headline
# ─── GRADIENT ─────────────────────────────────────────────
# 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
"\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
]
# Complementary sweep for queue messages (opposite hue family from ticker greens) # ─── GRADIENT ─────────────────────────────────────────────
MSG_GRAD_COLS = [ def _color_codes_to_ansi(color_codes):
"\033[1;38;5;231m", # white """Convert a list of 256-color codes to ANSI escape code strings.
"\033[1;38;5;225m", # pale pink-white
"\033[38;5;219m", # bright pink Pattern: first 2 are bold, middle 8 are normal, last 2 are dim.
"\033[38;5;213m", # hot pink
"\033[38;5;207m", # magenta Args:
"\033[38;5;201m", # bright magenta color_codes: List of 12 integers (256-color palette codes)
"\033[38;5;165m", # orchid-red
"\033[38;5;161m", # ruby-magenta Returns:
"\033[38;5;125m", # dark magenta List of ANSI escape code strings
"\033[38;5;89m", # deep maroon-magenta """
"\033[2;38;5;89m", # dim deep maroon-magenta if not color_codes or len(color_codes) != 12:
"\033[2;38;5;235m", # near black # 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 [
"\033[1;38;5;231m", # white
"\033[1;38;5;195m", # pale cyan-white
"\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():
"""Return the default 12-color magenta gradient for fallback when no theme is active."""
return [
"\033[1;38;5;231m", # white
"\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
]
# ─── FONT LOADING ───────────────────────────────────────── # ─── FONT LOADING ─────────────────────────────────────────
_FONT_OBJ = None _FONT_OBJ = None
@@ -189,9 +223,15 @@ def big_wrap(text, max_w, fnt=None):
return out return out
def lr_gradient(rows, offset=0.0, grad_cols=None): def lr_gradient(rows, offset=0.0, cols=None):
"""Color each non-space block character with a shifting left-to-right gradient.""" """Color each non-space block character with a shifting left-to-right gradient."""
cols = grad_cols or GRAD_COLS 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()
n = len(cols) n = len(cols)
max_x = max((len(r.rstrip()) for r in rows if r.strip()), default=1) max_x = max((len(r.rstrip()) for r in rows if r.strip()), default=1)
out = [] out = []
@@ -213,7 +253,30 @@ def lr_gradient(rows, offset=0.0, grad_cols=None):
def lr_gradient_opposite(rows, offset=0.0): def lr_gradient_opposite(rows, offset=0.0):
"""Complementary (opposite wheel) gradient used for queue message panels.""" """Complementary (opposite wheel) gradient used for queue message panels."""
return lr_gradient(rows, offset, MSG_GRAD_COLS) 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)
# ─── HEADLINE BLOCK ASSEMBLY ───────────────────────────── # ─── HEADLINE BLOCK ASSEMBLY ─────────────────────────────

View File

@@ -4,33 +4,42 @@ Orchestrates viewport, frame timing, and layers.
""" """
import random import random
import sys
import time import time
from engine import config 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.frame import calculate_scroll_step
from engine.layers import ( from engine.layers import (
apply_glitch, apply_glitch,
process_effects,
render_firehose, render_firehose,
render_message_overlay, render_message_overlay,
render_ticker_zone, render_ticker_zone,
) )
from engine.terminal import CLR
from engine.viewport import th, tw from engine.viewport import th, tw
USE_EFFECT_CHAIN = True
def stream(items, ntfy_poller, mic_monitor):
def stream(items, ntfy_poller, mic_monitor, display: Display | None = None):
"""Main render loop with four layers: message, ticker, scroll motion, firehose.""" """Main render loop with four layers: message, ticker, scroll motion, firehose."""
if display is None:
display = TerminalDisplay()
random.shuffle(items) random.shuffle(items)
pool = list(items) pool = list(items)
seen = set() seen = set()
queued = 0 queued = 0
time.sleep(0.5) time.sleep(0.5)
sys.stdout.write(CLR)
sys.stdout.flush()
w, h = tw(), th() w, h = tw(), th()
display.init(w, h)
display.clear()
fh = config.FIREHOSE_H if config.FIREHOSE else 0 fh = config.FIREHOSE_H if config.FIREHOSE else 0
ticker_view_h = h - fh ticker_view_h = h - fh
GAP = 3 GAP = 3
@@ -42,6 +51,7 @@ def stream(items, ntfy_poller, mic_monitor):
noise_cache = {} noise_cache = {}
scroll_motion_accum = 0.0 scroll_motion_accum = 0.0
msg_cache = (None, None) msg_cache = (None, None)
frame_number = 0
while True: while True:
if queued >= config.HEADLINE_LIMIT and not active: if queued >= config.HEADLINE_LIMIT and not active:
@@ -93,19 +103,39 @@ def stream(items, ntfy_poller, mic_monitor):
buf.extend(ticker_buf) buf.extend(ticker_buf)
mic_excess = mic_monitor.excess mic_excess = mic_monitor.excess
buf = apply_glitch(buf, ticker_buf_start, mic_excess, w) render_start = time.perf_counter()
firehose_buf = render_firehose(items, w, fh, h) if USE_EFFECT_CHAIN:
buf.extend(firehose_buf) 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: if msg_overlay:
buf.extend(msg_overlay) buf.extend(msg_overlay)
sys.stdout.buffer.write("".join(buf).encode()) render_elapsed = (time.perf_counter() - render_start) * 1000
sys.stdout.flush() 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)
elapsed = time.monotonic() - t0 elapsed = time.monotonic() - t0
time.sleep(max(0, config.FRAME_DT - elapsed)) time.sleep(max(0, config.FRAME_DT - elapsed))
frame_number += 1
sys.stdout.write(CLR) display.cleanup()
sys.stdout.flush()

60
engine/themes.py Normal file
View File

@@ -0,0 +1,60 @@
"""
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]

BIN
fonts/Kapiler.otf Normal file

Binary file not shown.

BIN
fonts/Kapiler.ttf Normal file

Binary file not shown.

View File

@@ -83,3 +83,35 @@ class TestStreamControllerCleanup:
controller.cleanup() controller.cleanup()
mock_mic_instance.stop.assert_called_once() 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()

79
tests/test_display.py Normal file
View File

@@ -0,0 +1,79 @@
"""
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()

427
tests/test_effects.py Normal file
View File

@@ -0,0 +1,427 @@
"""
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

View File

@@ -0,0 +1,117 @@
"""
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

35
tests/test_fetch_code.py Normal file
View File

@@ -0,0 +1,35 @@
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}"

301
tests/test_render.py Normal file
View File

@@ -0,0 +1,301 @@
"""
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

169
tests/test_themes.py Normal file
View File

@@ -0,0 +1,169 @@
"""
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}"