klubhaus/sideline
2
0
Fork 0
forked from genewildish/Mainline
Code Issues 5 Pull Requests Projects 1 Releases Wiki Activity

fix: resolve terminal display wobble and effect dimension stability

Browse Source 
- Fix TerminalDisplay: add screen clear each frame (cursor home + erase down)
- Fix CameraStage: use set_canvas_size instead of read-only viewport properties
- Fix Glitch effect: preserve visible line lengths, remove cursor positioning
- Fix Fade effect: return original line when fade=0 instead of empty string
- Fix Noise effect: use input line length instead of terminal_width
- Remove HUD effect from all presets (redundant with border FPS display)
- Add regression tests for effect dimension stability
- Add docs/ARCHITECTURE.md with Mermaid diagrams
- Add mise tasks: diagram-ascii, diagram-validate, diagram-check
- Move markdown docs to docs/ (ARCHITECTURE, Refactor, hardware specs)
- Remove redundant requirements files (use pyproject.toml)
- Add *.dot and *.png to .gitignore

Closes #25
This commit is contained in:
David Gwilliam
2026-03-18 03:34:36 -07:00
parent a65fb50464
commit b926b346ad
30 changed files with 1472 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -10,3 +10,5 @@ htmlcov/
.pytest_cache/
*.egg-info/
coverage.xml
*.dot
*.png

78
.opencode/skills/mainline-architecture/SKILL.md Normal file
View File

@@ -0,0 +1,78 @@
---
name: mainline-architecture
description: Pipeline stages, capability resolution, and core architecture patterns
compatibility: opencode
metadata:
audience: developers
source_type: codebase
---
## What This Skill Covers
This skill covers Mainline's pipeline architecture - the Stage-based system for dependency resolution, data flow, and component composition.
## Key Concepts
### Stage Class (engine/pipeline/core.py)
The `Stage` ABC is the foundation. All pipeline components inherit from it:
```python
class Stage(ABC):
name: str
category: str # "source", "effect", "overlay", "display", "camera"
optional: bool = False
@property
def capabilities(self) -> set[str]:
"""What this stage provides (e.g., 'source.headlines')"""
return set()
@property
def dependencies(self) -> list[str]:
"""What this stage needs (e.g., ['source'])"""
return []
```
### Capability-Based Dependencies
The Pipeline resolves dependencies using **prefix matching**:
- `"source"` matches `"source.headlines"`, `"source.poetry"`, etc.
- This allows flexible composition without hardcoding specific stage names
### DataType Enum
PureData-style data types for inlet/outlet validation:
- `SOURCE_ITEMS`: List[SourceItem] - raw items from sources
- `ITEM_TUPLES`: List[tuple] - (title, source, timestamp) tuples
- `TEXT_BUFFER`: List[str] - rendered ANSI buffer
- `RAW_TEXT`: str - raw text strings
- `PIL_IMAGE`: PIL Image object
### Pipeline Execution
The Pipeline (engine/pipeline/controller.py):
1. Collects all stages from StageRegistry
2. Resolves dependencies using prefix matching
3. Executes stages in dependency order
4. Handles errors for non-optional stages
### Canvas & Camera
- **Canvas** (`engine/canvas.py`): 2D rendering surface with dirty region tracking
- **Camera** (`engine/camera.py`): Viewport controller for scrolling content
Canvas tracks dirty regions automatically when content is written via `put_region`, `put_text`, `fill`, enabling partial buffer updates.
## Adding New Stages
1. Create a class inheriting from `Stage`
2. Define `capabilities` and `dependencies` properties
3. Implement required abstract methods
4. Register in StageRegistry or use as adapter
## Common Patterns
- Use adapters (engine/pipeline/adapters.py) to wrap existing components as stages
- Set `optional=True` for stages that can fail gracefully
- Use `stage_type` and `render_order` for execution ordering

86
.opencode/skills/mainline-display/SKILL.md Normal file
View File

@@ -0,0 +1,86 @@
---
name: mainline-display
description: Display backend implementation and the Display protocol
compatibility: opencode
metadata:
audience: developers
source_type: codebase
---
## What This Skill Covers
This skill covers Mainline's display backend system - how to implement new display backends and how the Display protocol works.
## Key Concepts
### Display Protocol
All backends implement a common Display protocol (in `engine/display/__init__.py`):
```python
class Display(Protocol):
def show(self, buf: list[str]) -> None:
"""Display the buffer"""
...
def clear(self) -> None:
"""Clear the display"""
...
def size(self) -> tuple[int, int]:
"""Return (width, height)"""
...
```
### DisplayRegistry
Discovers and manages backends:
```python
from engine.display import get_monitor
display = get_monitor("terminal") # or "websocket", "sixel", "null", "multi"
```
### Available Backends
| Backend | File | Description |
|---------|------|-------------|
| terminal | backends/terminal.py | ANSI terminal output |
| websocket | backends/websocket.py | Web browser via WebSocket |
| sixel | backends/sixel.py | Sixel graphics (pure Python) |
| null | backends/null.py | Headless for testing |
| multi | backends/multi.py | Forwards to multiple displays |
### WebSocket Backend
- WebSocket server: port 8765
- HTTP server: port 8766 (serves client/index.html)
- Client has ANSI color parsing and fullscreen support
### Multi Backend
Forwards to multiple displays simultaneously - useful for `terminal + websocket`.
## Adding a New Backend
1. Create `engine/display/backends/my_backend.py`
2. Implement the Display protocol methods
3. Register in `engine/display/__init__.py`'s `DisplayRegistry`
Required methods:
- `show(buf: list[str])` - Display buffer
- `clear()` - Clear screen
- `size() -> tuple[int, int]` - Terminal dimensions
Optional methods:
- `title(text: str)` - Set window title
- `cursor(show: bool)` - Control cursor
## Usage
```bash
python mainline.py --display terminal # default
python mainline.py --display websocket
python mainline.py --display sixel
python mainline.py --display both # terminal + websocket
```

113
.opencode/skills/mainline-effects/SKILL.md Normal file
View File

@@ -0,0 +1,113 @@
---
name: mainline-effects
description: How to add new effect plugins to Mainline's effect system
compatibility: opencode
metadata:
audience: developers
source_type: codebase
---
## What This Skill Covers
This skill covers Mainline's effect plugin system - how to create, configure, and integrate visual effects into the pipeline.
## Key Concepts
### EffectPlugin ABC (engine/effects/types.py)
All effects must inherit from `EffectPlugin` and implement:
```python
class EffectPlugin(ABC):
name: str
config: EffectConfig
param_bindings: dict[str, dict[str, str | float]] = {}
supports_partial_updates: bool = False
@abstractmethod
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
"""Process buffer with effect applied"""
...
@abstractmethod
def configure(self, config: EffectConfig) -> None:
"""Configure the effect"""
...
```
### EffectContext
Passed to every effect's process method:
```python
@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)
_state: dict[str, Any] = field(default_factory=dict)
```
Access sensor values via `ctx.get_sensor_value("sensor_name")`.
### EffectConfig
Configuration dataclass:
```python
@dataclass
class EffectConfig:
enabled: bool = True
intensity: float = 1.0
params: dict[str, Any] = field(default_factory=dict)
```
### Partial Updates
For performance optimization, set `supports_partial_updates = True` and implement `process_partial`:
```python
class MyEffect(EffectPlugin):
supports_partial_updates = True
def process_partial(self, buf, ctx, partial: PartialUpdate) -> list[str]:
# Only process changed regions
...
```
## Adding a New Effect
1. Create file in `effects_plugins/my_effect.py`
2. Inherit from `EffectPlugin`
3. Implement `process()` and `configure()`
4. Add to `effects_plugins/__init__.py` (runtime discovery via issubclass checks)
## Param Bindings
Declarative sensor-to-param mappings:
```python
param_bindings = {
"intensity": {"sensor": "mic", "transform": "linear"},
"rate": {"sensor": "oscillator", "transform": "exponential"},
}
```
Transforms: `linear`, `exponential`, `threshold`
## Effect Chain
Effects are chained via `engine/effects/chain.py` - processes each effect in order, passing output to next.
## Existing Effects
See `effects_plugins/`:
- noise.py, fade.py, glitch.py, firehose.py
- border.py, crop.py, tint.py, hud.py

103
.opencode/skills/mainline-presets/SKILL.md Normal file
View File

@@ -0,0 +1,103 @@
---
name: mainline-presets
description: Creating pipeline presets in TOML format for Mainline
compatibility: opencode
metadata:
audience: developers
source_type: codebase
---
## What This Skill Covers
This skill covers how to create pipeline presets in TOML format for Mainline's rendering pipeline.
## Key Concepts
### Preset Loading Order
Presets are loaded from multiple locations (later overrides earlier):
1. Built-in: `engine/presets.toml`
2. User config: `~/.config/mainline/presets.toml`
3. Local override: `./presets.toml`
### PipelinePreset Dataclass
```python
@dataclass
class PipelinePreset:
name: str
description: str = ""
source: str = "headlines" # Data source
display: str = "terminal" # Display backend
camera: str = "scroll" # Camera mode
effects: list[str] = field(default_factory=list)
border: bool = False
```
### TOML Format
```toml
[presets.my-preset]
description = "My custom pipeline"
source = "headlines"
display = "terminal"
camera = "scroll"
effects = ["noise", "fade"]
border = true
```
## Creating a Preset
### Option 1: User Config
Create/edit `~/.config/mainline/presets.toml`:
```toml
[presets.my-cool-preset]
description = "Noise and glitch effects"
source = "headlines"
display = "terminal"
effects = ["noise", "glitch"]
```
### Option 2: Local Override
Create `./presets.toml` in project root:
```toml
[presets.dev-inspect]
description = "Pipeline introspection for development"
source = "headlines"
display = "terminal"
effects = ["hud"]
```
### Option 3: Built-in
Edit `engine/presets.toml` (requires PR to repository).
## Available Sources
- `headlines` - RSS news feeds
- `poetry` - Literature mode
- `pipeline-inspect` - Live DAG visualization
## Available Displays
- `terminal` - ANSI terminal
- `websocket` - Web browser
- `sixel` - Sixel graphics
- `null` - Headless
## Available Effects
See `effects_plugins/`:
- noise, fade, glitch, firehose
- border, crop, tint, hud
## Validation Functions
Use these from `engine/pipeline/presets.py`:
- `validate_preset()` - Validate preset structure
- `validate_signal_path()` - Detect circular dependencies
- `generate_preset_toml()` - Generate skeleton preset

136
.opencode/skills/mainline-sensors/SKILL.md Normal file
View File

@@ -0,0 +1,136 @@
---
name: mainline-sensors
description: Sensor framework for real-time input in Mainline
compatibility: opencode
metadata:
audience: developers
source_type: codebase
---
## What This Skill Covers
This skill covers Mainline's sensor framework - how to use, create, and integrate sensors for real-time input.
## Key Concepts
### Sensor Base Class (engine/sensors/__init__.py)
```python
class Sensor(ABC):
name: str
unit: str = ""
@property
def available(self) -> bool:
"""Whether sensor is currently available"""
return True
@abstractmethod
def read(self) -> SensorValue | None:
"""Read current sensor value"""
...
def start(self) -> None:
"""Initialize sensor (optional)"""
pass
def stop(self) -> None:
"""Clean up sensor (optional)"""
pass
```
### SensorValue Dataclass
```python
@dataclass
class SensorValue:
sensor_name: str
value: float
timestamp: float
unit: str = ""
```
### SensorRegistry
Discovers and manages sensors globally:
```python
from engine.sensors import SensorRegistry
registry = SensorRegistry()
sensor = registry.get("mic")
```
### SensorStage
Pipeline adapter that provides sensor values to effects:
```python
from engine.pipeline.adapters import SensorStage
stage = SensorStage(sensor_name="mic")
```
## Built-in Sensors
| Sensor | File | Description |
|--------|------|-------------|
| MicSensor | sensors/mic.py | Microphone input (RMS dB) |
| OscillatorSensor | sensors/oscillator.py | Test sine wave generator |
| PipelineMetricsSensor | sensors/pipeline_metrics.py | FPS, frame time, etc. |
## Param Bindings
Effects declare sensor-to-param mappings:
```python
class GlitchEffect(EffectPlugin):
param_bindings = {
"intensity": {"sensor": "mic", "transform": "linear"},
}
```
### Transform Functions
- `linear` - Direct mapping to param range
- `exponential` - Exponential scaling
- `threshold` - Binary on/off
## Adding a New Sensor
1. Create `engine/sensors/my_sensor.py`
2. Inherit from `Sensor` ABC
3. Implement required methods
4. Register in `SensorRegistry`
Example:
```python
class MySensor(Sensor):
name = "my-sensor"
unit = "units"
def read(self) -> SensorValue | None:
return SensorValue(
sensor_name=self.name,
value=self._read_hardware(),
timestamp=time.time(),
unit=self.unit
)
```
## Using Sensors in Effects
Access sensor values via EffectContext:
```python
def process(self, buf, ctx):
mic_level = ctx.get_sensor_value("mic")
if mic_level and mic_level > 0.5:
# Apply intense effect
...
```
Or via param_bindings (automatic):
```python
# If intensity is bound to "mic", it's automatically
# available in self.config.intensity
```

87
.opencode/skills/mainline-sources/SKILL.md Normal file
View File

@@ -0,0 +1,87 @@
---
name: mainline-sources
description: Adding new RSS feeds and data sources to Mainline
compatibility: opencode
metadata:
audience: developers
source_type: codebase
---
## What This Skill Covers
This skill covers how to add new data sources (RSS feeds, poetry) to Mainline.
## Key Concepts
### Feeds Dictionary (engine/sources.py)
All feeds are defined in a simple dictionary:
```python
FEEDS = {
"Feed Name": "https://example.com/feed.xml",
# Category comments help organize:
# Science & Technology
# Economics & Business
# World & Politics
# Culture & Ideas
}
```
### Poetry Sources
Project Gutenberg URLs for public domain literature:
```python
POETRY_SOURCES = {
"Author Name": "https://www.gutenberg.org/cache/epub/1234/pg1234.txt",
}
```
### Language & Script Mapping
The sources.py also contains language/script detection mappings used for auto-translation and font selection.
## Adding a New RSS Feed
1. Edit `engine/sources.py`
2. Add entry to `FEEDS` dict under appropriate category:
```python
"My Feed": "https://example.com/feed.xml",
```
3. The feed will be automatically discovered on next run
### Feed Requirements
- Must be valid RSS or Atom XML
- Should have `<title>` elements for items
- Must be HTTP/HTTPS accessible
## Adding Poetry Sources
1. Edit `engine/sources.py`
2. Add to `POETRY_SOURCES` dict:
```python
"Author": "https://www.gutenberg.org/cache/epub/XXXX/pgXXXX.txt",
```
### Poetry Requirements
- Plain text (UTF-8)
- Project Gutenberg format preferred
- No DRM-protected sources
## Data Flow
Feeds are fetched via `engine/fetch.py`:
- `fetch_feed(url)` - Fetches and parses RSS/Atom
- Results cached for fast restarts
- Filtered via `engine/filter.py` for content cleaning
## Categories
Organize new feeds by category using comments:
- Science & Technology
- Economics & Business
- World & Politics
- Culture & Ideas

156
docs/ARCHITECTURE.md Normal file
View File

@@ -0,0 +1,156 @@
# Mainline Architecture Diagrams
> These diagrams use Mermaid. Render with: `npx @mermaid-js/mermaid-cli -i ARCHITECTURE.md` or view in GitHub/GitLab/Notion.
## Class Hierarchy (Mermaid)
```mermaid
classDiagram
class Stage {
<<abstract>>
+str name
+set[str] capabilities
+set[str] dependencies
+process(data, ctx) Any
}
Stage <|-- DataSourceStage
Stage <|-- CameraStage
Stage <|-- FontStage
Stage <|-- ViewportFilterStage
Stage <|-- EffectPluginStage
Stage <|-- DisplayStage
Stage <|-- SourceItemsToBufferStage
Stage <|-- PassthroughStage
Stage <|-- ImageToTextStage
Stage <|-- CanvasStage
class EffectPlugin {
<<abstract>>
+str name
+EffectConfig config
+process(buf, ctx) list[str]
+configure(config) None
}
EffectPlugin <|-- NoiseEffect
EffectPlugin <|-- FadeEffect
EffectPlugin <|-- GlitchEffect
EffectPlugin <|-- FirehoseEffect
EffectPlugin <|-- CropEffect
EffectPlugin <|-- TintEffect
class Display {
<<protocol>>
+int width
+int height
+init(width, height, reuse)
+show(buffer, border)
+clear() None
+cleanup() None
}
Display <|.. TerminalDisplay
Display <|.. NullDisplay
Display <|.. PygameDisplay
Display <|.. WebSocketDisplay
Display <|.. SixelDisplay
class Camera {
+int viewport_width
+int viewport_height
+CameraMode mode
+apply(buffer, width, height) list[str]
}
class Pipeline {
+dict[str, Stage] stages
+PipelineContext context
+execute(data) StageResult
}
Pipeline --> Stage
Stage --> Display
```
## Data Flow (Mermaid)
```mermaid
flowchart LR
DataSource[Data Source] --> DataSourceStage
DataSourceStage --> FontStage
FontStage --> CameraStage
CameraStage --> EffectStages
EffectStages --> DisplayStage
DisplayStage --> TerminalDisplay
DisplayStage --> BrowserWebSocket
DisplayStage --> SixelDisplay
DisplayStage --> NullDisplay
```
## Effect Chain (Mermaid)
```mermaid
flowchart LR
InputBuffer --> NoiseEffect
NoiseEffect --> FadeEffect
FadeEffect --> GlitchEffect
GlitchEffect --> FirehoseEffect
FirehoseEffect --> Output
```
> **Note:** Each effect must preserve buffer dimensions (line count and visible width).
## Stage Capabilities
```mermaid
flowchart TB
subgraph "Capability Resolution"
D[DataSource<br/>provides: source.*]
C[Camera<br/>provides: render.output]
E[Effects<br/>provides: render.effect]
DIS[Display<br/>provides: display.output]
end
```
---
## Legacy ASCII Diagrams
### Stage Inheritance
```
Stage(ABC)
├── DataSourceStage
├── CameraStage
├── FontStage
├── ViewportFilterStage
├── EffectPluginStage
├── DisplayStage
├── SourceItemsToBufferStage
├── PassthroughStage
├── ImageToTextStage
└── CanvasStage
```
### Display Backends
```
Display(Protocol)
├── TerminalDisplay
├── NullDisplay
├── PygameDisplay
├── WebSocketDisplay
├── SixelDisplay
├── KittyDisplay
└── MultiDisplay
```
### Camera Modes
```
Camera
├── FEED # Static view
├── SCROLL # Horizontal scroll
├── VERTICAL # Vertical scroll
├── HORIZONTAL # Same as scroll
├── OMNI # Omnidirectional
├── FLOATING # Floating particles
└── BOUNCE # Bouncing camera

0
Mainline Renderer + ntfy Message Queue for ESP32.md → docs/Mainline Renderer + ntfy Message Queue for ESP32.md
View File

79
Refactor mainline.md → docs/Refactor mainline.md
View File

@@ -1,11 +1,18 @@
# Refactor mainline\.py into modular package
#
Refactor mainline\.py into modular package
## Problem
`mainline.py` is a single 1085\-line file with ~10 interleaved concerns\. This prevents:
* Reusing the ntfy doorbell interrupt in other visualizers
* Importing the render pipeline from `serve.py` \(future ESP32 HTTP server\)
* Testing any concern in isolation
* Porting individual layers to Rust independently
## Target structure
```warp-runnable-command
mainline.py # thin entrypoint: venv bootstrap → engine.app.main()
engine/
@@ -23,8 +30,11 @@ engine/
scroll.py # stream() frame loop + message rendering
app.py # main(), TITLE art, boot sequence, signal handler
```
The package is named `engine/` to avoid a naming conflict with the `mainline.py` entrypoint\.
## Module dependency graph
```warp-runnable-command
config ← (nothing)
sources ← (nothing)
@@ -39,64 +49,92 @@ mic ← (nothing — sounddevice only)
scroll ← config, terminal, render, effects, ntfy, mic
app ← everything above
```
Critical property: **ntfy\.py and mic\.py have zero internal dependencies**, making ntfy reusable by any visualizer\.
## Module details
### mainline\.py \(entrypoint — slimmed down\)
Keeps only the venv bootstrap \(lines 10\-38\) which must run before any third\-party imports\. After bootstrap, delegates to `engine.app.main()`\.
### engine/config\.py
From current mainline\.py:
* `HEADLINE_LIMIT`, `FEED_TIMEOUT`, `MIC_THRESHOLD_DB` \(lines 55\-57\)
* `MODE`, `FIREHOSE` CLI flag parsing \(lines 58\-59\)
* `NTFY_TOPIC`, `NTFY_POLL_INTERVAL`, `MESSAGE_DISPLAY_SECS` \(lines 62\-64\)
* `_FONT_PATH`, `_FONT_SZ`, `_RENDER_H` \(lines 147\-150\)
* `_SCROLL_DUR`, `_FRAME_DT`, `FIREHOSE_H` \(lines 505\-507\)
* `GLITCH`, `KATA` glyph tables \(lines 143\-144\)
### engine/sources\.py
Pure data, no logic:
* `FEEDS` dict \(lines 102\-140\)
* `POETRY_SOURCES` dict \(lines 67\-80\)
* `SOURCE_LANGS` dict \(lines 258\-266\)
* `_LOCATION_LANGS` dict \(lines 269\-289\)
* `_SCRIPT_FONTS` dict \(lines 153\-165\)
* `_NO_UPPER` set \(line 167\)
### engine/terminal\.py
ANSI primitives and terminal I/O:
* All ANSI constants: `RST`, `BOLD`, `DIM`, `G_HI`, `G_MID`, `G_LO`, `G_DIM`, `W_COOL`, `W_DIM`, `W_GHOST`, `C_DIM`, `CLR`, `CURSOR_OFF`, `CURSOR_ON` \(lines 83\-99\)
* `tw()`, `th()` \(lines 223\-234\)
* `type_out()`, `slow_print()`, `boot_ln()` \(lines 355\-386\)
### engine/filter\.py
* `_Strip` HTML parser class \(lines 205\-214\)
* `strip_tags()` \(lines 217\-220\)
* `_SKIP_RE` compiled regex \(lines 322\-346\)
* `_skip()` predicate \(lines 349\-351\)
### engine/translate\.py
* `_TRANSLATE_CACHE` \(line 291\)
* `_detect_location_language()` \(lines 294\-300\) — imports `_LOCATION_LANGS` from sources
* `_translate_headline()` \(lines 303\-319\)
### engine/render\.py
The OTF→terminal pipeline\. This is exactly what `serve.py` will import to produce 1\-bit bitmaps for the ESP32\.
* `_GRAD_COLS` gradient table \(lines 169\-182\)
* `_font()`, `_font_for_lang()` with lazy\-load \+ cache \(lines 185\-202\)
* `_render_line()` — OTF text → half\-block terminal rows \(lines 567\-605\)
* `_big_wrap()` — word\-wrap \+ render \(lines 608\-636\)
* `_lr_gradient()` — apply left→right color gradient \(lines 639\-656\)
* `_make_block()` — composite: translate → render → colorize a headline \(lines 718\-756\)\. Imports from translate, sources\.
### engine/effects\.py
Visual effects applied during the frame loop:
* `noise()` \(lines 237\-245\)
* `glitch_bar()` \(lines 248\-252\)
* `_fade_line()` — probabilistic character dissolve \(lines 659\-680\)
* `_vis_trunc()` — ANSI\-aware width truncation \(lines 683\-701\)
* `_firehose_line()` \(lines 759\-801\) — imports config\.MODE, sources\.FEEDS/POETRY\_SOURCES
* `_next_headline()` — pool management \(lines 704\-715\)
### engine/fetch\.py
* `fetch_feed()` \(lines 390\-396\)
* `fetch_all()` \(lines 399\-426\) — imports filter\.\_skip, filter\.strip\_tags, terminal\.boot\_ln
* `_fetch_gutenberg()` \(lines 429\-456\)
* `fetch_poetry()` \(lines 459\-472\)
* `_cache_path()`, `_load_cache()`, `_save_cache()` \(lines 476\-501\)
### engine/ntfy\.py — standalone, reusable
Refactored from the current globals \+ thread \(lines 531\-564\) and the message rendering section of `stream()` \(lines 845\-909\) into a class:
```python
class NtfyPoller:
def __init__(self, topic_url, poll_interval=15, display_secs=30):
@@ -108,8 +146,10 @@ class NtfyPoller:
def dismiss(self):
"""Manually dismiss current message."""
```
Dependencies: `urllib.request`, `json`, `threading`, `time` — all stdlib\. No internal imports\.
Other visualizers use it like:
```python
from engine.ntfy import NtfyPoller
poller = NtfyPoller("https://ntfy.sh/my_topic/json?since=20s&poll=1")
@@ -120,8 +160,11 @@ if msg:
title, body, ts = msg
render_my_message(title, body) # visualizer-specific
```
### engine/mic\.py — standalone
Refactored from the current globals \(lines 508\-528\) into a class:
```python
class MicMonitor:
def __init__(self, threshold_db=50):
@@ -137,41 +180,75 @@ class MicMonitor:
def excess(self) -> float:
"""dB above threshold (clamped to 0)."""
```
Dependencies: `sounddevice`, `numpy` \(both optional — graceful fallback\)\.
### engine/scroll\.py
The `stream()` function \(lines 804\-990\)\. Receives its dependencies via arguments or imports:
* `stream(items, ntfy_poller, mic_monitor, config)` or similar
* Message rendering \(lines 855\-909\) stays here since it's terminal\-display\-specific — a different visualizer would render messages differently
### engine/app\.py
The orchestrator:
* `TITLE` ASCII art \(lines 994\-1001\)
* `main()` \(lines 1004\-1084\): CLI handling, signal setup, boot animation, fetch, wire up ntfy/mic/scroll
## Execution order
### Step 1: Create engine/ package skeleton
Create `engine/__init__.py` and all empty module files\.
### Step 2: Extract pure data modules \(zero\-dep\)
Move constants and data dicts into `config.py`, `sources.py`\. These have no logic dependencies\.
### Step 3: Extract terminal\.py
Move ANSI codes and terminal I/O helpers\. No internal deps\.
### Step 4: Extract filter\.py and translate\.py
Both are small, self\-contained\. translate imports from sources\.
### Step 5: Extract render\.py
Font loading \+ the OTF→half\-block pipeline\. Imports from config, terminal, sources\. This is the module `serve.py` will later import\.
### Step 6: Extract effects\.py
Visual effects\. Imports from config, terminal, sources\.
### Step 7: Extract fetch\.py
Feed/Gutenberg fetching \+ caching\. Imports from config, sources, filter, terminal\.
### Step 8: Extract ntfy\.py and mic\.py
Refactor globals\+threads into classes\. Zero internal deps\.
### Step 9: Extract scroll\.py
The frame loop\. Last to extract because it depends on everything above\.
### Step 10: Extract app\.py
The `main()` function, boot sequence, signal handler\. Wire up all modules\.
### Step 11: Slim down mainline\.py
Keep only venv bootstrap \+ `from engine.app import main; main()`\.
### Step 12: Verify
Run `python3 mainline.py`, `python3 mainline.py --poetry`, and `python3 mainline.py --firehose` to confirm identical behavior\. No behavioral changes in this refactor\.
## What this enables
* **serve\.py** \(future\): `from engine.render import _render_line, _big_wrap` \+ `from engine.fetch import fetch_all` — imports the pipeline directly
* **Other visualizers**: `from engine.ntfy import NtfyPoller` — doorbell feature with no coupling to mainline's scroll engine
* **Rust port**: Clear boundaries for what to port first \(ntfy client, render pipeline\) vs what stays in Python \(fetching, caching — the server side\)

0
klubhaus-doorbell-hardware.md → docs/klubhaus-doorbell-hardware.md
View File

2
effects_plugins/fade.py
View File

@@ -36,7 +36,7 @@ class FadeEffect(EffectPlugin):
if fade >= 1.0:
return s
if fade <= 0.0:
return ""
return s # Preserve original line length - don't return empty
result = []
i = 0
while i < len(s):

28
effects_plugins/glitch.py
View File

@@ -21,17 +21,33 @@ class GlitchEffect(EffectPlugin):
n_hits = int(n_hits * intensity)
if random.random() < glitch_prob:
# Store original visible lengths before any modifications
# Strip ANSI codes to get visible length
import re
ansi_pattern = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]")
original_lengths = [len(ansi_pattern.sub("", line)) for line in result]
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)}"
original_line = result[gi]
target_len = original_lengths[gi] # Use stored original length
glitch_bar = self._glitch_bar(target_len)
result[gi] = glitch_bar
return result
def _glitch_bar(self, w: int) -> str:
def _glitch_bar(self, target_len: 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
n = random.randint(3, max(3, target_len // 2))
o = random.randint(0, max(0, target_len - n))
glitch_chars = c * n
trailing_spaces = target_len - o - n
trailing_spaces = max(0, trailing_spaces)
glitch_part = f"{G_LO}{DIM}" + glitch_chars + RST
result = " " * o + glitch_part + " " * trailing_spaces
return result
def configure(self, config: EffectConfig) -> None:
self.config = config

3
effects_plugins/noise.py
View File

@@ -19,7 +19,8 @@ class NoiseEffect(EffectPlugin):
for r in range(len(result)):
cy = ctx.scroll_cam + r
if random.random() < probability:
result[r] = self._generate_noise(ctx.terminal_width, cy)
original_line = result[r]
result[r] = self._generate_noise(len(original_line), cy)
return result
def _generate_noise(self, w: int, cy: int) -> str:

17
engine/display/backends/null.py
View File

@@ -9,11 +9,16 @@ class NullDisplay:
"""Headless/null display - discards all output.
This display does nothing - useful for headless benchmarking
or when no display output is needed.
or when no display output is needed. Captures last buffer
for testing purposes.
"""
width: int = 80
height: int = 24
_last_buffer: list[str] | None = None
def __init__(self):
self._last_buffer = None
def init(self, width: int, height: int, reuse: bool = False) -> None:
"""Initialize display with dimensions.
@@ -25,10 +30,12 @@ class NullDisplay:
"""
self.width = width
self.height = height
self._last_buffer = None
def show(self, buffer: list[str], border: bool = False) -> None:
from engine.display import get_monitor
self._last_buffer = buffer
monitor = get_monitor()
if monitor:
t0 = time.perf_counter()
@@ -49,3 +56,11 @@ class NullDisplay:
(width, height) in character cells
"""
return (self.width, self.height)
def is_quit_requested(self) -> bool:
"""Check if quit was requested (optional protocol method)."""
return False
def clear_quit_request(self) -> None:
"""Clear quit request (optional protocol method)."""
pass

32
engine/display/backends/terminal.py
View File

@@ -22,6 +22,7 @@ class TerminalDisplay:
self.target_fps = target_fps
self._frame_period = 1.0 / target_fps if target_fps > 0 else 0
self._last_frame_time = 0.0
self._cached_dimensions: tuple[int, int] | None = None
def init(self, width: int, height: int, reuse: bool = False) -> None:
"""Initialize display with dimensions.
@@ -62,14 +63,26 @@ class TerminalDisplay:
def get_dimensions(self) -> tuple[int, int]:
"""Get current terminal dimensions.
Returns cached dimensions to avoid querying terminal every frame,
which can cause inconsistent results. Dimensions are only refreshed
when they actually change.
Returns:
(width, height) in character cells
"""
try:
term_size = os.get_terminal_size()
return (term_size.columns, term_size.lines)
new_dims = (term_size.columns, term_size.lines)
except OSError:
return (self.width, self.height)
new_dims = (self.width, self.height)
# Only update cached dimensions if they actually changed
if self._cached_dimensions is None or self._cached_dimensions != new_dims:
self._cached_dimensions = new_dims
self.width = new_dims[0]
self.height = new_dims[1]
return self._cached_dimensions
def show(self, buffer: list[str], border: bool = False) -> None:
import sys
@@ -103,10 +116,9 @@ class TerminalDisplay:
if border:
buffer = render_border(buffer, self.width, self.height, fps, frame_time)
# Clear screen and home cursor before each frame
from engine.terminal import CLR
output = CLR + "".join(buffer)
# Write buffer with cursor home + erase down to avoid flicker
# \033[H = cursor home, \033[J = erase from cursor to end of screen
output = "\033[H\033[J" + "".join(buffer)
sys.stdout.buffer.write(output.encode())
sys.stdout.flush()
elapsed_ms = (time.perf_counter() - t0) * 1000
@@ -124,3 +136,11 @@ class TerminalDisplay:
from engine.terminal import CURSOR_ON
print(CURSOR_ON, end="", flush=True)
def is_quit_requested(self) -> bool:
"""Check if quit was requested (optional protocol method)."""
return False
def clear_quit_request(self) -> None:
"""Clear quit request (optional protocol method)."""
pass

9
engine/pipeline/adapters.py
View File

@@ -347,9 +347,12 @@ class CameraStage(Stage):
)
# Update camera's viewport dimensions so it knows its actual bounds
if hasattr(self._camera, "viewport_width"):
self._camera.viewport_width = viewport_width
self._camera.viewport_height = viewport_height
# Set canvas size to achieve desired viewport (viewport = canvas / zoom)
if hasattr(self._camera, "set_canvas_size"):
self._camera.set_canvas_size(
width=int(viewport_width * self._camera.zoom),
height=int(viewport_height * self._camera.zoom),
)
# Set canvas to full layout height so camera can scroll through all content
self._camera.set_canvas_size(width=canvas_width, height=total_layout_height)

8
engine/pipeline/params.py
View File

@@ -32,7 +32,7 @@ class PipelineParams:
# Effect config
effect_order: list[str] = field(
default_factory=lambda: ["noise", "fade", "glitch", "firehose", "hud"]
default_factory=lambda: ["noise", "fade", "glitch", "firehose"]
)
effect_enabled: dict[str, bool] = field(default_factory=dict)
effect_intensity: dict[str, float] = field(default_factory=dict)
@@ -127,19 +127,19 @@ DEFAULT_HEADLINE_PARAMS = PipelineParams(
source="headlines",
display="terminal",
camera_mode="vertical",
effect_order=["noise", "fade", "glitch", "firehose", "hud"],
effect_order=["noise", "fade", "glitch", "firehose"],
)
DEFAULT_PYGAME_PARAMS = PipelineParams(
source="headlines",
display="pygame",
camera_mode="vertical",
effect_order=["noise", "fade", "glitch", "firehose", "hud"],
effect_order=["noise", "fade", "glitch", "firehose"],
)
DEFAULT_PIPELINE_PARAMS = PipelineParams(
source="pipeline",
display="pygame",
camera_mode="trace",
effect_order=["hud"], # Just HUD for pipeline viz
effect_order=[], # No effects for pipeline viz
)

4
engine/pipeline/preset_loader.py
View File

@@ -19,7 +19,7 @@ DEFAULT_PRESET: dict[str, Any] = {
"source": "headlines",
"display": "terminal",
"camera": "vertical",
"effects": ["hud"],
"effects": [],
"viewport": {"width": 80, "height": 24},
"camera_speed": 1.0,
"firehose_enabled": False,
@@ -263,7 +263,7 @@ def generate_preset_toml(
"""
if effects is None:
effects = ["fade", "hud"]
effects = ["fade"]
output = []
output.append(f"[presets.{name}]")

12
engine/pipeline/presets.py
View File

@@ -80,7 +80,7 @@ DEMO_PRESET = PipelinePreset(
source="headlines",
display="pygame",
camera="scroll",
effects=["noise", "fade", "glitch", "firehose", "hud"],
effects=["noise", "fade", "glitch", "firehose"],
)
POETRY_PRESET = PipelinePreset(
@@ -89,7 +89,7 @@ POETRY_PRESET = PipelinePreset(
source="poetry",
display="pygame",
camera="scroll",
effects=["fade", "hud"],
effects=["fade"],
)
PIPELINE_VIZ_PRESET = PipelinePreset(
@@ -98,7 +98,7 @@ PIPELINE_VIZ_PRESET = PipelinePreset(
source="pipeline",
display="terminal",
camera="trace",
effects=["hud"],
effects=[],
)
WEBSOCKET_PRESET = PipelinePreset(
@@ -107,7 +107,7 @@ WEBSOCKET_PRESET = PipelinePreset(
source="headlines",
display="websocket",
camera="scroll",
effects=["noise", "fade", "glitch", "hud"],
effects=["noise", "fade", "glitch"],
)
SIXEL_PRESET = PipelinePreset(
@@ -116,7 +116,7 @@ SIXEL_PRESET = PipelinePreset(
source="headlines",
display="sixel",
camera="scroll",
effects=["noise", "fade", "glitch", "hud"],
effects=["noise", "fade", "glitch"],
)
FIREHOSE_PRESET = PipelinePreset(
@@ -125,7 +125,7 @@ FIREHOSE_PRESET = PipelinePreset(
source="headlines",
display="pygame",
camera="scroll",
effects=["noise", "fade", "glitch", "firehose", "hud"],
effects=["noise", "fade", "glitch", "firehose"],
)

16
mise.toml
View File

@@ -59,5 +59,21 @@ topics-init = "curl -s -d 'init' https://ntfy.sh/klubhaus_terminal_mainline_cc_c
pre-commit = "hk run pre-commit"
# =====================
# Diagrams
# =====================
# Render Mermaid diagrams to ASCII art
diagram-ascii = "python3 scripts/render-diagrams.py docs/ARCHITECTURE.md"
# Validate Mermaid syntax in docs (check all diagrams parse)
# Note: classDiagram not supported by mermaid-ascii but works in GitHub/GitLab
diagram-validate = """
python3 scripts/validate-diagrams.py
"""
# Render diagrams and check they match expected output
diagram-check = "mise run diagram-validate"
[env]
KAGI_API_KEY = "lOp6AGyX6TUB0kGzAli1BlAx5-VjlIN1OPCPYEXDdQc.FOKLieOa7NgWUUZi4mTZvHmrW2uNnOr8hfgv7jMvRQM"

54
presets.toml
View File

@@ -8,6 +8,60 @@
# - ~/.config/mainline/presets.toml
# - ./presets.toml (local override)
# ============================================
# TEST PRESETS
# ============================================
[presets.test-single-item]
description = "Test: Single item to isolate rendering stage issues"
source = "empty"
display = "terminal"
camera = "feed"
effects = []
camera_speed = 0.1
viewport_width = 80
viewport_height = 24
[presets.test-single-item-border]
description = "Test: Single item with border effect only"
source = "empty"
display = "terminal"
camera = "feed"
effects = ["border"]
camera_speed = 0.1
viewport_width = 80
viewport_height = 24
[presets.test-headlines]
description = "Test: Headlines from cache with border effect"
source = "headlines"
display = "terminal"
camera = "feed"
effects = ["border"]
camera_speed = 0.1
viewport_width = 80
viewport_height = 24
[presets.test-headlines-noise]
description = "Test: Headlines from cache with noise effect"
source = "headlines"
display = "terminal"
camera = "feed"
effects = ["noise"]
camera_speed = 0.1
viewport_width = 80
viewport_height = 24
[presets.test-demo-effects]
description = "Test: All demo effects with terminal display"
source = "headlines"
display = "terminal"
camera = "feed"
effects = ["noise", "fade", "firehose"]
camera_speed = 0.3
viewport_width = 80
viewport_height = 24
# ============================================
# DATA SOURCE GALLERY
# ============================================

4
requirements-dev.txt
View File

@@ -1,4 +0,0 @@
pytest>=8.0.0
pytest-cov>=4.1.0
pytest-mock>=3.12.0
ruff>=0.1.0

4
requirements.txt
View File

@@ -1,4 +0,0 @@
feedparser>=6.0.0
Pillow>=10.0.0
sounddevice>=0.4.0
numpy>=1.24.0

49
scripts/render-diagrams.py Normal file
View File

@@ -0,0 +1,49 @@
#!/usr/bin/env python3
"""Render Mermaid diagrams in markdown files to ASCII art."""
import re
import subprocess
import sys
def extract_mermaid_blocks(content: str) -> list[str]:
"""Extract mermaid blocks from markdown."""
return re.findall(r"```mermaid\n(.*?)\n```", content, re.DOTALL)
def render_diagram(block: str) -> str:
"""Render a single mermaid block to ASCII."""
result = subprocess.run(
["mermaid-ascii", "-f", "-"],
input=block,
capture_output=True,
text=True,
)
if result.returncode != 0:
return f"ERROR: {result.stderr}"
return result.stdout
def main():
if len(sys.argv) < 2:
print("Usage: render-diagrams.py <markdown-file>")
sys.exit(1)
filename = sys.argv[1]
content = open(filename).read()
blocks = extract_mermaid_blocks(content)
print(f"Found {len(blocks)} mermaid diagram(s) in {filename}")
print()
for i, block in enumerate(blocks):
# Skip if empty
if not block.strip():
continue
print(f"=== Diagram {i + 1} ===")
print(render_diagram(block))
if __name__ == "__main__":
main()

64
scripts/validate-diagrams.py Normal file
View File

@@ -0,0 +1,64 @@
#!/usr/bin/env python3
"""Validate Mermaid diagrams in markdown files."""
import glob
import re
import sys
# Diagram types that are valid in Mermaid
VALID_TYPES = {
"flowchart",
"graph",
"classDiagram",
"sequenceDiagram",
"stateDiagram",
"stateDiagram-v2",
"erDiagram",
"gantt",
"pie",
"mindmap",
"journey",
"gitGraph",
"requirementDiagram",
}
def extract_mermaid_blocks(content: str) -> list[tuple[int, str]]:
"""Extract mermaid blocks with their positions."""
blocks = []
for match in re.finditer(r"```mermaid\n(.*?)\n```", content, re.DOTALL):
blocks.append((match.start(), match.group(1)))
return blocks
def validate_block(block: str) -> bool:
"""Check if a mermaid block has a valid diagram type."""
if not block.strip():
return True # Empty block is OK
first_line = block.strip().split("\n")[0]
return any(first_line.startswith(t) for t in VALID_TYPES)
def main():
md_files = glob.glob("docs/*.md")
errors = []
for filepath in md_files:
content = open(filepath).read()
blocks = extract_mermaid_blocks(content)
for i, (_, block) in enumerate(blocks):
if not validate_block(block):
errors.append(f"{filepath}: invalid diagram type in block {i + 1}")
if errors:
for e in errors:
print(f"ERROR: {e}")
sys.exit(1)
print(f"Validated {len(md_files)} markdown files - all OK")
if __name__ == "__main__":
main()

0
kitty_test.py → tests/kitty_test.py
View File

99
tests/test_display.py
View File

@@ -2,6 +2,7 @@
Tests for engine.display module.
"""
import sys
from unittest.mock import MagicMock
from engine.display import DisplayRegistry, NullDisplay, TerminalDisplay
@@ -115,6 +116,83 @@ class TestTerminalDisplay:
display = TerminalDisplay()
display.cleanup()
def test_get_dimensions_returns_cached_value(self):
"""get_dimensions returns cached dimensions for stability."""
display = TerminalDisplay()
display.init(80, 24)
# First call should set cache
d1 = display.get_dimensions()
assert d1 == (80, 24)
def test_show_clears_screen_before_each_frame(self):
"""show clears previous frame to prevent visual wobble.
Regression test: Previously show() didn't clear the screen,
causing old content to remain and creating visual wobble.
The fix adds \\033[H\\033[J (cursor home + erase down) before each frame.
"""
from io import BytesIO
from unittest.mock import patch
display = TerminalDisplay()
display.init(80, 24)
buffer = ["line1", "line2", "line3"]
fake_buffer = BytesIO()
fake_stdout = MagicMock()
fake_stdout.buffer = fake_buffer
with patch.object(sys, "stdout", fake_stdout):
display.show(buffer)
output = fake_buffer.getvalue().decode("utf-8")
assert output.startswith("\033[H\033[J"), (
f"Output should start with clear sequence, got: {repr(output[:20])}"
)
def test_show_clears_screen_on_subsequent_frames(self):
"""show clears screen on every frame, not just the first.
Regression test: Ensures each show() call includes the clear sequence.
"""
from io import BytesIO
from unittest.mock import patch
# Use target_fps=0 to disable frame skipping in test
display = TerminalDisplay(target_fps=0)
display.init(80, 24)
buffer = ["line1", "line2"]
for i in range(3):
fake_buffer = BytesIO()
fake_stdout = MagicMock()
fake_stdout.buffer = fake_buffer
with patch.object(sys, "stdout", fake_stdout):
display.show(buffer)
output = fake_buffer.getvalue().decode("utf-8")
assert output.startswith("\033[H\033[J"), (
f"Frame {i} should start with clear sequence"
)
def test_get_dimensions_stable_across_rapid_calls(self):
"""get_dimensions should not fluctuate when called rapidly.
This test catches the bug where os.get_terminal_size() returns
inconsistent values, causing visual wobble.
"""
display = TerminalDisplay()
display.init(80, 24)
# Get dimensions 10 times rapidly (simulating frame loop)
dims = [display.get_dimensions() for _ in range(10)]
# All should be the same - this would fail if os.get_terminal_size()
# returns different values each call
assert len(set(dims)) == 1, f"Dimensions should be stable, got: {set(dims)}"
class TestNullDisplay:
"""Tests for NullDisplay class."""
@@ -141,6 +219,27 @@ class TestNullDisplay:
display = NullDisplay()
display.cleanup()
def test_show_stores_last_buffer(self):
"""show stores last buffer for testing inspection."""
display = NullDisplay()
display.init(80, 24)
buffer = ["line1", "line2", "line3"]
display.show(buffer)
assert display._last_buffer == buffer
def test_show_tracks_last_buffer_across_calls(self):
"""show updates last_buffer on each call."""
display = NullDisplay()
display.init(80, 24)
display.show(["first"])
assert display._last_buffer == ["first"]
display.show(["second"])
assert display._last_buffer == ["second"]
class TestMultiDisplay:
"""Tests for MultiDisplay class."""

238
tests/test_glitch_effect.py Normal file
View File

@@ -0,0 +1,238 @@
"""
Tests for Glitch effect - regression tests for stability issues.
"""
import re
import pytest
from engine.display import NullDisplay
from engine.effects.types import EffectConfig, EffectContext
def strip_ansi(s: str) -> str:
"""Remove ANSI escape sequences from string."""
return re.sub(r"\x1b\[[0-9;]*[a-zA-Z]", "", s)
class TestGlitchEffectStability:
"""Regression tests for Glitch effect stability."""
@pytest.fixture
def effect_context(self):
"""Create a consistent effect context for testing."""
return EffectContext(
terminal_width=80,
terminal_height=24,
scroll_cam=0,
ticker_height=20,
frame_number=0,
)
@pytest.fixture
def stable_buffer(self):
"""Create a stable buffer for testing."""
return ["line" + str(i).zfill(2) + " " * 60 for i in range(24)]
def test_glitch_preserves_line_count(self, effect_context, stable_buffer):
"""Glitch should not change the number of lines in buffer."""
from effects_plugins.glitch import GlitchEffect
effect = GlitchEffect()
result = effect.process(stable_buffer, effect_context)
assert len(result) == len(stable_buffer), (
f"Line count changed from {len(stable_buffer)} to {len(result)}"
)
def test_glitch_preserves_line_lengths(self, effect_context, stable_buffer):
"""Glitch should not change individual line lengths - prevents viewport jumping.
Note: Effects may add ANSI color codes, so we check VISIBLE length (stripped).
"""
from effects_plugins.glitch import GlitchEffect
effect = GlitchEffect()
# Run multiple times to catch randomness
for _ in range(10):
result = effect.process(stable_buffer, effect_context)
for i, (orig, new) in enumerate(zip(stable_buffer, result, strict=False)):
visible_new = strip_ansi(new)
assert len(visible_new) == len(orig), (
f"Line {i} visible length changed from {len(orig)} to {len(visible_new)}"
)
def test_glitch_no_cursor_positioning(self, effect_context, stable_buffer):
"""Glitch should not use cursor positioning escape sequences.
Regression test: Previously glitch used \\033[{row};1H which caused
conflicts with HUD and border rendering.
"""
from effects_plugins.glitch import GlitchEffect
effect = GlitchEffect()
result = effect.process(stable_buffer, effect_context)
# Check no cursor positioning in output
cursor_pos_pattern = re.compile(r"\033\[[0-9]+;[0-9]+H")
for i, line in enumerate(result):
match = cursor_pos_pattern.search(line)
assert match is None, (
f"Line {i} contains cursor positioning: {repr(line[:50])}"
)
def test_glitch_output_deterministic_given_seed(
self, effect_context, stable_buffer
):
"""Glitch output should be deterministic given the same random seed."""
from effects_plugins.glitch import GlitchEffect
effect = GlitchEffect()
effect.config = EffectConfig(enabled=True, intensity=1.0)
# With fixed random state, should get same result
import random
random.seed(42)
result1 = effect.process(stable_buffer, effect_context)
random.seed(42)
result2 = effect.process(stable_buffer, effect_context)
assert result1 == result2, (
"Glitch should be deterministic with fixed random seed"
)
class TestEffectViewportStability:
"""Tests to catch effects that cause viewport instability."""
def test_null_display_stable_without_effects(self):
"""NullDisplay should produce identical output without effects."""
display = NullDisplay()
display.init(80, 24)
buffer = ["test line " + "x" * 60 for _ in range(24)]
display.show(buffer)
output1 = display._last_buffer
display.show(buffer)
output2 = display._last_buffer
assert output1 == output2, (
"NullDisplay output should be identical for identical inputs"
)
def test_effect_chain_preserves_dimensions(self):
"""Effect chain should preserve buffer dimensions."""
from effects_plugins.fade import FadeEffect
from effects_plugins.glitch import GlitchEffect
from effects_plugins.noise import NoiseEffect
ctx = EffectContext(
terminal_width=80,
terminal_height=24,
scroll_cam=0,
ticker_height=20,
)
buffer = ["x" * 80 for _ in range(24)]
original_len = len(buffer)
original_widths = [len(line) for line in buffer]
effects = [NoiseEffect(), FadeEffect(), GlitchEffect()]
for effect in effects:
buffer = effect.process(buffer, ctx)
# Check dimensions preserved (check VISIBLE length, not raw)
# Effects may add ANSI codes which increase raw length but not visible width
assert len(buffer) == original_len, (
f"{effect.name} changed line count from {original_len} to {len(buffer)}"
)
for i, (orig_w, new_line) in enumerate(zip(original_widths, buffer, strict=False)):
visible_len = len(strip_ansi(new_line))
assert visible_len == orig_w, (
f"{effect.name} changed line {i} visible width from {orig_w} to {visible_len}"
)
class TestEffectTestMatrix:
"""Effect test matrix - test each effect for stability."""
@pytest.fixture
def effect_names(self):
"""List of all effect names to test."""
return ["noise", "fade", "glitch", "firehose", "border"]
@pytest.fixture
def stable_input_buffer(self):
"""A predictable buffer for testing."""
return [f"row{i:02d}" + " " * 70 for i in range(24)]
@pytest.mark.parametrize("effect_name", ["noise", "fade", "glitch"])
def test_effect_preserves_buffer_dimensions(self, effect_name, stable_input_buffer):
"""Each effect should preserve input buffer dimensions."""
try:
if effect_name == "border":
# Border is handled differently
pytest.skip("Border handled by display")
else:
effect_module = __import__(
f"effects_plugins.{effect_name}",
fromlist=[f"{effect_name.title()}Effect"],
)
effect_class = getattr(effect_module, f"{effect_name.title()}Effect")
effect = effect_class()
except ImportError:
pytest.skip(f"Effect {effect_name} not available")
ctx = EffectContext(
terminal_width=80,
terminal_height=24,
scroll_cam=0,
ticker_height=20,
)
result = effect.process(stable_input_buffer, ctx)
# Check dimensions preserved (check VISIBLE length)
assert len(result) == len(stable_input_buffer), (
f"{effect_name} changed line count"
)
for i, (orig, new) in enumerate(zip(stable_input_buffer, result, strict=False)):
visible_new = strip_ansi(new)
assert len(visible_new) == len(orig), (
f"{effect_name} changed line {i} visible length from {len(orig)} to {len(visible_new)}"
)
@pytest.mark.parametrize("effect_name", ["noise", "fade", "glitch"])
def test_effect_no_cursor_positioning(self, effect_name, stable_input_buffer):
"""Effects should not use cursor positioning (causes display conflicts)."""
try:
effect_module = __import__(
f"effects_plugins.{effect_name}",
fromlist=[f"{effect_name.title()}Effect"],
)
effect_class = getattr(effect_module, f"{effect_name.title()}Effect")
effect = effect_class()
except ImportError:
pytest.skip(f"Effect {effect_name} not available")
ctx = EffectContext(
terminal_width=80,
terminal_height=24,
scroll_cam=0,
ticker_height=20,
)
result = effect.process(stable_input_buffer, ctx)
cursor_pos_pattern = re.compile(r"\033\[[0-9]+;[0-9]+H")
for i, line in enumerate(result):
match = cursor_pos_pattern.search(line)
assert match is None, (
f"{effect_name} uses cursor positioning on line {i}: {repr(line[:50])}"
)

29
tests/test_pipeline.py
View File

@@ -483,7 +483,7 @@ class TestPipelineParams:
assert params.source == "headlines"
assert params.display == "terminal"
assert params.camera_mode == "vertical"
assert params.effect_order == ["noise", "fade", "glitch", "firehose", "hud"]
assert params.effect_order == ["noise", "fade", "glitch", "firehose"]
def test_effect_config(self):
"""PipelineParams effect config methods work."""
@@ -634,6 +634,33 @@ class TestStageAdapters:
assert "camera" in stage.capabilities
assert "render.output" in stage.dependencies # Depends on rendered content
def test_camera_stage_does_not_error_on_process(self):
"""CameraStage.process should not error when setting viewport.
Regression test: Previously CameraStage tried to set viewport_width
and viewport_height as writable properties, but they are computed
from canvas_size / zoom. This caused an AttributeError each frame.
"""
from engine.camera import Camera, CameraMode
from engine.pipeline.adapters import CameraStage
from engine.pipeline.core import PipelineContext
from engine.pipeline.params import PipelineParams
camera = Camera(mode=CameraMode.FEED)
stage = CameraStage(camera, name="vertical")
ctx = PipelineContext()
ctx.params = PipelineParams(viewport_width=80, viewport_height=24)
buffer = ["line" + str(i) for i in range(24)]
# This should not raise AttributeError
result = stage.process(buffer, ctx)
# Should return the buffer (unchanged for FEED mode)
assert result is not None
assert len(result) == 24
class TestDataSourceStage:
"""Tests for DataSourceStage adapter."""