diff --git a/.gitignore b/.gitignore
index 590c496..573af53 100644
--- a/.gitignore
+++ b/.gitignore
@@ -9,3 +9,4 @@ htmlcov/
.coverage
.pytest_cache/
*.egg-info/
+.DS_Store
diff --git a/README.md b/README.md
index cd549ba..0dcbd3f 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,34 @@
> *Digital consciousness stream. Matrix aesthetic · THX-1138 hue.*
-A full-screen terminal news ticker that renders live global headlines in large OTF-font block characters with a white-hot → deep green gradient. Headlines auto-translate into the native script of their subject region. Ambient mic input warps the glitch rate in real time. A `--poetry` mode replaces the feed with public-domain literary passages. Live messages can be pushed to the display over [ntfy.sh](https://ntfy.sh).
+A full-screen terminal news ticker that renders live global headlines in large OTF-font block characters with selectable color gradients (Verdant Green, Molten Orange, or Violet Purple). Headlines auto-translate into the native script of their subject region. Ambient mic input warps the glitch rate in real time. A `--poetry` mode replaces the feed with public-domain literary passages. Live messages can be pushed to the display over [ntfy.sh](https://ntfy.sh). **Figment mode** overlays flickery, theme-colored SVG glyphs on the running stream at timed intervals — controllable from any input source via an extensible trigger protocol.
+
+---
+
+## Contents
+
+- [Using](#using)
+ - [Run](#run)
+ - [Config](#config)
+ - [Display Modes](#display-modes)
+ - [Feeds](#feeds)
+ - [Fonts](#fonts)
+ - [ntfy.sh](#ntfysh)
+ - [Figment Mode](#figment-mode)
+ - [Command & Control](#command--control-cc)
+- [Internals](#internals)
+ - [How it works](#how-it-works)
+ - [Architecture](#architecture)
+- [Development](#development)
+ - [Setup](#setup)
+ - [Tasks](#tasks)
+ - [Testing](#testing)
+ - [Linting](#linting)
+- [Roadmap](#roadmap)
+ - [Performance](#performance)
+ - [Graphics](#graphics)
+ - [Cyberpunk Vibes](#cyberpunk-vibes)
+ - [Extensibility](#extensibility)
---
@@ -15,8 +42,11 @@ python3 mainline.py # news stream
python3 mainline.py --poetry # literary consciousness mode
python3 mainline.py -p # same
python3 mainline.py --firehose # dense rapid-fire headline mode
+python3 mainline.py --figment # enable periodic SVG glyph overlays
+python3 mainline.py --figment-interval 30 # figment every 30 seconds (default: 60)
python3 mainline.py --display websocket # web browser display only
python3 mainline.py --display both # terminal + web browser
+python3 mainline.py --refresh # force re-fetch (bypass cache)
python3 mainline.py --no-font-picker # skip interactive font picker
python3 mainline.py --font-file path.otf # use a specific font file
python3 mainline.py --font-dir ~/fonts # scan a different font folder
@@ -68,6 +98,7 @@ All constants live in `engine/config.py`:
| `FRAME_DT` | `0.05` | Frame interval in seconds (20 FPS) |
| `FIREHOSE_H` | `12` | Firehose zone height (terminal rows) |
| `GRAD_SPEED` | `0.08` | Gradient sweep speed |
+| `FIGMENT_INTERVAL` | `60` | Seconds between figment appearances (set by `--figment-interval`) |
### Display Modes
@@ -104,20 +135,56 @@ To push a message:
curl -d "Body text" -H "Title: Alert title" https://ntfy.sh/your_topic
```
+Update `NTFY_TOPIC` in `engine/config.py` to point at your own topic.
+
+### Figment Mode
+
+Figment mode periodically overlays a full-screen SVG glyph on the running ticker — flickering through a reveal → hold (strobe) → dissolve cycle, colored with a randomly selected theme gradient.
+
+**Enable it** with the `--figment` flag:
+
+```bash
+uv run mainline.py --figment # glyph every 60 seconds (default)
+uv run mainline.py --figment --figment-interval 30 # every 30 seconds
+```
+
+**Figment assets** live in `figments/` — drop any `.svg` file there and it will be picked up automatically. The bundled set contains Mayan and Aztec glyphs. Figments are selected randomly, avoiding immediate repeats, and rasterized into half-block terminal art at display time.
+
+**Triggering manually** — any object with a `poll() -> FigmentCommand | None` method satisfies the `FigmentTrigger` protocol and can be passed to the plugin:
+
+```python
+from engine.figment_trigger import FigmentAction, FigmentCommand
+
+class MyTrigger:
+ def poll(self):
+ if some_condition:
+ return FigmentCommand(action=FigmentAction.TRIGGER)
+ return None
+```
+
+Built-in commands: `TRIGGER`, `SET_INTENSITY`, `SET_INTERVAL`, `SET_COLOR`, `STOP`.
+
+**System dependency:** Figment mode requires the Cairo C library (`brew install cairo` on macOS) in addition to the `figment` extras group:
+
+```bash
+uv sync --extra figment # adds cairosvg
+```
+
---
## Internals
### How it works
-- On launch, the font picker scans `fonts/` and presents a live-rendered TUI for face selection
-- Feeds are fetched and filtered on startup; results are cached for fast restarts
-- Headlines are rasterized via Pillow with 4× SSAA into half-block characters
-- The ticker uses a sweeping white-hot → deep green gradient
-- Subject-region detection triggers Google Translate and font swap for non-Latin scripts
-- The mic stream runs in a background thread, feeding RMS dB into glitch probability
-- The viewport scrolls through pre-rendered blocks with fade zones
-- An ntfy.sh SSE stream runs in a background thread for messages and C&C commands
+- 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 SSE stream runs in a background thread for messages and C&C commands; incoming messages interrupt the scroll and render full-screen until dismissed or expired
+- Figment mode rasterizes SVGs via cairosvg → PIL → greyscale → half-block encode, then overlays them with ANSI cursor-positioning commands between the effect chain and the ntfy message layer
### Architecture
@@ -138,32 +205,40 @@ engine/
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
+ 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
+ viewport.py terminal dimension tracking (tw/th)
frame.py scroll step calculation, timing
- layers.py ticker zone, firehose, message overlay
- eventbus.py thread-safe event publishing
+ layers.py ticker zone, firehose, message + figment overlay rendering
+ figment_render.py SVG → cairosvg → PIL → half-block rasterizer with cache
+ figment_trigger.py FigmentTrigger protocol, FigmentAction enum, FigmentCommand
+ eventbus.py thread-safe event publishing for decoupled communication
events.py event types and definitions
- controller.py coordinates ntfy/mic monitoring
- emitters.py background emitters
- types.py type definitions
+ controller.py coordinates ntfy/mic monitoring and event publishing
+ emitters.py background emitters for ntfy and mic
+ types.py type definitions and dataclasses
+ themes.py THEME_REGISTRY — gradient color 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
+ sixel.py Sixel graphics (pure Python)
+ null.py headless display for testing
+ multi.py forwards to multiple displays
benchmark.py performance benchmarking tool
+
+effects_plugins/
+ __init__.py plugin discovery (ABC issubclass scan)
+ noise.py NoiseEffect — random character noise
+ glitch.py GlitchEffect — horizontal glitch bars
+ fade.py FadeEffect — edge fade zones
+ firehose.py FirehoseEffect — dense bottom ticker strip
+ figment.py FigmentEffect — periodic SVG glyph overlay (state machine)
+
+figments/ SVG assets for figment mode
```
---
@@ -175,11 +250,15 @@ engine/
Requires Python 3.10+ and [uv](https://docs.astral.sh/uv/).
```bash
-uv sync # minimal (no mic)
-uv sync --all-extras # with mic support
+uv sync # minimal (no mic, no figment)
+uv sync --extra mic # with mic support (sounddevice + numpy)
+uv sync --extra figment # with figment mode (cairosvg + system Cairo)
+uv sync --all-extras # all optional features
uv sync --all-extras --group dev # full dev environment
```
+Figment mode also requires the Cairo C library: `brew install cairo` (macOS).
+
### Tasks
With [mise](https://mise.jdx.dev/):
@@ -209,6 +288,8 @@ mise run topics-init # initialize ntfy topics
### Testing
+Tests live in `tests/` and cover `config`, `filter`, `mic`, `ntfy`, `sources`, `terminal`, and the full figment pipeline (`figment_render`, `figment_trigger`, `figment`, `figment_overlay`). Figment tests are automatically skipped if Cairo is not installed.
+
```bash
uv run pytest
uv run pytest --cov=engine --cov-report=term-missing
@@ -252,12 +333,19 @@ Pre-commit hooks run lint automatically via `hk`.
- 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)
+- **Figment intensity wiring** — `config.intensity` currently stored but not yet applied to reveal/dissolve speed or strobe frequency
+- **ntfy figment trigger** — built-in `NtfyFigmentTrigger` that listens on a dedicated topic to fire figments on demand
+- **Keyword watch list** — highlight or strobe any headline matching tracked terms (names, topics, tickers)
+- **Breaking interrupt** — full-screen flash + synthesized blip when a high-priority keyword hits
+- **Live data overlay** — secondary ticker strip at screen edge: BTC price, ISS position, geomagnetic index
+- **Theme switcher** — `--amber` (phosphor), `--ice` (electric cyan), `--red` (alert state) palette modes via CLI flag
+- **Persona modes** — `--surveillance`, `--oracle`, `--underground` as feed presets with matching color themes and boot copy
+- **Synthesized audio** — short static bursts tied to glitch events, independent of mic input
+
+### Extensibility
+- **serve.py** — HTTP server that imports `engine.render` and `engine.fetch` directly to stream 1-bit bitmaps to an ESP32 display
+- **Rust port** — `ntfy.py` and `render.py` are the natural first targets; clear module boundaries make incremental porting viable
---
-*Python 3.10+. Primary display font is user-selectable via bundled `fonts/` picker.*
\ No newline at end of file
+*Python 3.10+. Primary display font is user-selectable via bundled `fonts/` picker.*
diff --git a/docs/superpowers/plans/2026-03-19-figment-mode.md b/docs/superpowers/plans/2026-03-19-figment-mode.md
new file mode 100644
index 0000000..0d5c9c7
--- /dev/null
+++ b/docs/superpowers/plans/2026-03-19-figment-mode.md
@@ -0,0 +1,1110 @@
+# Figment Mode 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:** Add a periodic full-screen SVG glyph overlay ("figment mode") that renders flickery, theme-colored half-block art on top of the running news ticker.
+
+**Architecture:** Hybrid EffectPlugin + overlay. `FigmentEffect` (effect plugin) owns the lifecycle, timer, and state machine. `render_figment_overlay()` (in layers.py) handles ANSI overlay rendering. `engine/figment_render.py` handles SVG→half-block rasterization. `engine/figment_trigger.py` defines the extensible input protocol.
+
+**Tech Stack:** Python 3.10+, cairosvg (SVG→PNG), Pillow (image processing), existing effect plugin system (ABC-based), existing theme gradients.
+
+**Spec:** `docs/superpowers/specs/2026-03-19-figment-mode-design.md`
+
+---
+
+## Chunk 1: Foundation
+
+### Task 1: Merge main and add cairosvg dependency
+
+The `feat/figment` branch is behind `main` by 2 commits (the ABC plugin migration). Must merge first so `EffectPlugin` is ABC-based.
+
+**Files:**
+- Modify: `pyproject.toml:28-38`
+
+- [ ] **Step 1: Merge main into feat/figment**
+
+```bash
+git merge main
+```
+
+Expected: Fast-forward or clean merge. No conflicts (branch only added docs).
+
+- [ ] **Step 2: Add cairosvg optional dependency**
+
+In `pyproject.toml`, add a `figment` extras group after the `mic` group (line 32):
+
+```toml
+figment = [
+ "cairosvg>=2.7.0",
+]
+```
+
+- [ ] **Step 3: Sync dependencies**
+
+```bash
+uv sync --all-extras
+```
+
+Expected: cairosvg installs successfully.
+
+- [ ] **Step 4: Verify cairosvg works**
+
+```bash
+uv run python -c "import cairosvg; print('cairosvg OK')"
+```
+
+Expected: prints `cairosvg OK`
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add pyproject.toml uv.lock
+git commit -m "build: add cairosvg optional dependency for figment mode"
+```
+
+---
+
+### Task 2: Test fixture SVG and event types
+
+**Files:**
+- Create: `tests/fixtures/test.svg`
+- Modify: `engine/events.py:12-21` (add FIGMENT_TRIGGER), `engine/events.py:62-68` (add FigmentTriggerEvent)
+
+- [ ] **Step 1: Create minimal test SVG**
+
+Create `tests/fixtures/test.svg` — a simple 100x100 black rectangle on white:
+
+```xml
+
+```
+
+- [ ] **Step 2: Add FIGMENT_TRIGGER event type**
+
+In `engine/events.py`, add to the `EventType` enum (after `STREAM_END = auto()` at line 20):
+
+```python
+ FIGMENT_TRIGGER = auto()
+```
+
+And add the event dataclass at the end of the file (after `StreamEvent`):
+
+```python
+@dataclass
+class FigmentTriggerEvent:
+ """Event emitted when a figment is triggered."""
+
+ action: str
+ value: float | str | None = None
+ timestamp: datetime | None = None
+```
+
+- [ ] **Step 3: Run existing tests to verify no breakage**
+
+```bash
+uv run pytest tests/test_events.py -v
+```
+
+Expected: All existing event tests pass.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add tests/fixtures/test.svg engine/events.py
+git commit -m "feat(figment): add test fixture SVG and FIGMENT_TRIGGER event type"
+```
+
+---
+
+### Task 3: Trigger protocol and command types
+
+**Files:**
+- Create: `engine/figment_trigger.py`
+- Create: `tests/test_figment_trigger.py`
+
+- [ ] **Step 1: Write failing tests for FigmentCommand and FigmentAction**
+
+Create `tests/test_figment_trigger.py`:
+
+```python
+"""Tests for engine.figment_trigger module."""
+
+from enum import Enum
+
+from engine.figment_trigger import FigmentAction, FigmentCommand
+
+
+class TestFigmentAction:
+ def test_is_enum(self):
+ assert issubclass(FigmentAction, Enum)
+
+ def test_has_trigger(self):
+ assert FigmentAction.TRIGGER.value == "trigger"
+
+ def test_has_set_intensity(self):
+ assert FigmentAction.SET_INTENSITY.value == "set_intensity"
+
+ def test_has_set_interval(self):
+ assert FigmentAction.SET_INTERVAL.value == "set_interval"
+
+ def test_has_set_color(self):
+ assert FigmentAction.SET_COLOR.value == "set_color"
+
+ def test_has_stop(self):
+ assert FigmentAction.STOP.value == "stop"
+
+
+class TestFigmentCommand:
+ def test_trigger_command(self):
+ cmd = FigmentCommand(action=FigmentAction.TRIGGER)
+ assert cmd.action == FigmentAction.TRIGGER
+ assert cmd.value is None
+
+ def test_set_intensity_command(self):
+ cmd = FigmentCommand(action=FigmentAction.SET_INTENSITY, value=0.8)
+ assert cmd.value == 0.8
+
+ def test_set_color_command(self):
+ cmd = FigmentCommand(action=FigmentAction.SET_COLOR, value="orange")
+ assert cmd.value == "orange"
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+```bash
+uv run pytest tests/test_figment_trigger.py -v
+```
+
+Expected: FAIL — `ModuleNotFoundError: No module named 'engine.figment_trigger'`
+
+- [ ] **Step 3: Write FigmentTrigger protocol, FigmentAction, FigmentCommand**
+
+Create `engine/figment_trigger.py`:
+
+```python
+"""
+Figment trigger protocol and command types.
+
+Defines the extensible input abstraction for triggering figment displays
+from any control surface (ntfy, MQTT, serial, etc.).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Protocol
+
+
+class FigmentAction(Enum):
+ TRIGGER = "trigger"
+ SET_INTENSITY = "set_intensity"
+ SET_INTERVAL = "set_interval"
+ SET_COLOR = "set_color"
+ STOP = "stop"
+
+
+@dataclass
+class FigmentCommand:
+ action: FigmentAction
+ value: float | str | None = None
+
+
+class FigmentTrigger(Protocol):
+ """Protocol for figment trigger sources.
+
+ Any input source (ntfy, MQTT, serial) can implement this
+ to trigger and control figment displays.
+ """
+
+ def poll(self) -> FigmentCommand | None: ...
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+```bash
+uv run pytest tests/test_figment_trigger.py -v
+```
+
+Expected: All 8 tests pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add engine/figment_trigger.py tests/test_figment_trigger.py
+git commit -m "feat(figment): add trigger protocol and command types"
+```
+
+---
+
+## Chunk 2: SVG Rasterization
+
+### Task 4: SVG to half-block rasterizer
+
+**Files:**
+- Create: `engine/figment_render.py`
+- Create: `tests/test_figment_render.py`
+
+- [ ] **Step 1: Write failing tests for rasterize_svg**
+
+Create `tests/test_figment_render.py`:
+
+```python
+"""Tests for engine.figment_render module."""
+
+import os
+
+from engine.figment_render import rasterize_svg
+
+FIXTURE_SVG = os.path.join(os.path.dirname(__file__), "fixtures", "test.svg")
+
+
+class TestRasterizeSvg:
+ def test_returns_list_of_strings(self):
+ rows = rasterize_svg(FIXTURE_SVG, 40, 20)
+ assert isinstance(rows, list)
+ assert all(isinstance(r, str) for r in rows)
+
+ def test_output_height_matches_terminal_height(self):
+ rows = rasterize_svg(FIXTURE_SVG, 40, 20)
+ assert len(rows) == 20
+
+ def test_output_contains_block_characters(self):
+ rows = rasterize_svg(FIXTURE_SVG, 40, 20)
+ all_chars = "".join(rows)
+ block_chars = {"█", "▀", "▄"}
+ assert any(ch in all_chars for ch in block_chars)
+
+ def test_different_sizes_produce_different_output(self):
+ rows_small = rasterize_svg(FIXTURE_SVG, 20, 10)
+ rows_large = rasterize_svg(FIXTURE_SVG, 80, 40)
+ assert len(rows_small) == 10
+ assert len(rows_large) == 40
+
+ def test_nonexistent_file_raises(self):
+ import pytest
+ with pytest.raises(Exception):
+ rasterize_svg("/nonexistent/file.svg", 40, 20)
+
+
+class TestRasterizeCache:
+ def test_cache_returns_same_result(self):
+ rows1 = rasterize_svg(FIXTURE_SVG, 40, 20)
+ rows2 = rasterize_svg(FIXTURE_SVG, 40, 20)
+ assert rows1 == rows2
+
+ def test_cache_invalidated_by_size_change(self):
+ rows1 = rasterize_svg(FIXTURE_SVG, 40, 20)
+ rows2 = rasterize_svg(FIXTURE_SVG, 60, 30)
+ assert len(rows1) != len(rows2)
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+```bash
+uv run pytest tests/test_figment_render.py -v
+```
+
+Expected: FAIL — `ModuleNotFoundError: No module named 'engine.figment_render'`
+
+- [ ] **Step 3: Implement rasterize_svg**
+
+Create `engine/figment_render.py`:
+
+```python
+"""
+SVG to half-block terminal art rasterization.
+
+Pipeline: SVG -> cairosvg -> PIL -> greyscale threshold -> half-block encode.
+Follows the same pixel-pair approach as engine/render.py for OTF fonts.
+"""
+
+from __future__ import annotations
+
+from io import BytesIO
+
+import cairosvg
+from PIL import Image
+
+_cache: dict[tuple[str, int, int], list[str]] = {}
+
+
+def rasterize_svg(svg_path: str, width: int, height: int) -> list[str]:
+ """Convert SVG file to list of half-block terminal rows (uncolored).
+
+ Args:
+ svg_path: Path to SVG file.
+ width: Target terminal width in columns.
+ height: Target terminal height in rows.
+
+ Returns:
+ List of strings, one per terminal row, containing block characters.
+ """
+ cache_key = (svg_path, width, height)
+ if cache_key in _cache:
+ return _cache[cache_key]
+
+ # SVG -> PNG in memory
+ png_bytes = cairosvg.svg2png(
+ url=svg_path,
+ output_width=width,
+ output_height=height * 2, # 2 pixel rows per terminal row
+ )
+
+ # PNG -> greyscale PIL image
+ img = Image.open(BytesIO(png_bytes)).convert("L")
+ img = img.resize((width, height * 2), Image.Resampling.LANCZOS)
+
+ data = img.tobytes()
+ pix_w = width
+ pix_h = height * 2
+ threshold = 80
+
+ # Half-block encode: walk pixel pairs
+ rows: list[str] = []
+ for y in range(0, pix_h, 2):
+ row: list[str] = []
+ for x in range(pix_w):
+ top = data[y * pix_w + x] > threshold
+ bot = data[(y + 1) * pix_w + x] > threshold if y + 1 < pix_h else False
+ if top and bot:
+ row.append("█")
+ elif top:
+ row.append("▀")
+ elif bot:
+ row.append("▄")
+ else:
+ row.append(" ")
+ rows.append("".join(row))
+
+ _cache[cache_key] = rows
+ return rows
+
+
+def clear_cache() -> None:
+ """Clear the rasterization cache (e.g., on terminal resize)."""
+ _cache.clear()
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+```bash
+uv run pytest tests/test_figment_render.py -v
+```
+
+Expected: All 7 tests pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add engine/figment_render.py tests/test_figment_render.py
+git commit -m "feat(figment): add SVG to half-block rasterization pipeline"
+```
+
+---
+
+## Chunk 3: FigmentEffect Plugin
+
+### Task 5: FigmentEffect state machine and lifecycle
+
+This is the core plugin. It manages the timer, SVG selection, state machine, and exposes `get_figment_state()`.
+
+**Files:**
+- Create: `effects_plugins/figment.py`
+- Create: `tests/test_figment.py`
+
+- [ ] **Step 1: Write failing tests for FigmentState, FigmentPhase, and state machine**
+
+Create `tests/test_figment.py`:
+
+```python
+"""Tests for the FigmentEffect plugin."""
+
+import os
+from enum import Enum
+from unittest.mock import patch
+
+import pytest
+
+from effects_plugins.figment import FigmentEffect, FigmentPhase, FigmentState
+from engine.effects.types import EffectConfig, EffectContext
+
+
+FIXTURE_SVG = os.path.join(
+ os.path.dirname(__file__), "fixtures", "test.svg"
+)
+FIGMENTS_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
+
+
+class TestFigmentPhase:
+ def test_is_enum(self):
+ assert issubclass(FigmentPhase, Enum)
+
+ def test_has_all_phases(self):
+ assert hasattr(FigmentPhase, "REVEAL")
+ assert hasattr(FigmentPhase, "HOLD")
+ assert hasattr(FigmentPhase, "DISSOLVE")
+
+
+class TestFigmentState:
+ def test_creation(self):
+ state = FigmentState(
+ phase=FigmentPhase.REVEAL,
+ progress=0.5,
+ rows=["█▀▄", " █ "],
+ gradient=[46, 40, 34, 28, 22, 22, 34, 40, 46, 82, 118, 231],
+ center_row=5,
+ center_col=10,
+ )
+ assert state.phase == FigmentPhase.REVEAL
+ assert state.progress == 0.5
+ assert len(state.rows) == 2
+
+
+class TestFigmentEffectInit:
+ def test_name(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ assert effect.name == "figment"
+
+ def test_default_config(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ assert effect.config.enabled is False
+ assert effect.config.intensity == 1.0
+ assert effect.config.params["interval_secs"] == 60
+ assert effect.config.params["display_secs"] == 4.5
+
+ def test_process_is_noop(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ buf = ["line1", "line2"]
+ ctx = EffectContext(
+ terminal_width=80,
+ terminal_height=24,
+ scroll_cam=0,
+ ticker_height=20,
+ )
+ result = effect.process(buf, ctx)
+ assert result == buf
+ assert result is buf
+
+ def test_configure(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ new_cfg = EffectConfig(enabled=True, intensity=0.5)
+ effect.configure(new_cfg)
+ assert effect.config.enabled is True
+ assert effect.config.intensity == 0.5
+
+
+class TestFigmentStateMachine:
+ def test_idle_initially(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ state = effect.get_figment_state(0, 80, 24)
+ # Timer hasn't fired yet, should be None (idle)
+ assert state is None
+
+ def test_trigger_starts_reveal(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ effect.trigger(80, 24)
+ state = effect.get_figment_state(1, 80, 24)
+ assert state is not None
+ assert state.phase == FigmentPhase.REVEAL
+
+ def test_full_cycle(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ effect.config.params["display_secs"] = 0.15 # 3 phases x 0.05s
+
+ effect.trigger(40, 20)
+
+ # Advance through reveal (30 frames at 0.05s = 1.5s, but we shrunk it)
+ # With display_secs=0.15, each phase is 0.05s = 1 frame
+ state = effect.get_figment_state(1, 40, 20)
+ assert state is not None
+ assert state.phase == FigmentPhase.REVEAL
+
+ # Advance enough frames to get through all phases
+ last_state = None
+ for frame in range(2, 100):
+ state = effect.get_figment_state(frame, 40, 20)
+ if state is None:
+ break
+ last_state = state
+
+ # Should have completed the full cycle back to idle
+ assert state is None
+
+ def test_timer_fires_at_interval(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ effect.config.params["interval_secs"] = 0.1 # 2 frames at 20fps
+
+ # Frame 0: idle
+ state = effect.get_figment_state(0, 40, 20)
+ assert state is None
+
+ # Advance past interval (0.1s = 2 frames)
+ state = effect.get_figment_state(1, 40, 20)
+ state = effect.get_figment_state(2, 40, 20)
+ state = effect.get_figment_state(3, 40, 20)
+ # Timer should have fired by now
+ assert state is not None
+
+
+class TestFigmentEdgeCases:
+ def test_empty_figment_dir(self, tmp_path):
+ effect = FigmentEffect(figment_dir=str(tmp_path))
+ effect.config.enabled = True
+ effect.trigger(40, 20)
+ state = effect.get_figment_state(1, 40, 20)
+ # No SVGs available — should stay idle
+ assert state is None
+
+ def test_missing_figment_dir(self):
+ effect = FigmentEffect(figment_dir="/nonexistent/path")
+ effect.config.enabled = True
+ effect.trigger(40, 20)
+ state = effect.get_figment_state(1, 40, 20)
+ assert state is None
+
+ def test_disabled_ignores_trigger(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = False
+ effect.trigger(80, 24)
+ state = effect.get_figment_state(1, 80, 24)
+ assert state is None
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+```bash
+uv run pytest tests/test_figment.py -v
+```
+
+Expected: FAIL — `ImportError`
+
+- [ ] **Step 3: Implement FigmentEffect**
+
+Create `effects_plugins/figment.py`:
+
+```python
+"""
+Figment effect plugin — periodic SVG glyph overlay.
+
+Owns the figment lifecycle: timer, SVG selection, state machine.
+Delegates rendering to render_figment_overlay() in engine/layers.py.
+
+Named FigmentEffect (not FigmentPlugin) to match the *Effect discovery
+convention in effects_plugins/__init__.py.
+
+NOT added to the EffectChain order — process() is a no-op. The overlay
+rendering is handled by scroll.py calling get_figment_state().
+"""
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass
+from enum import Enum, auto
+from pathlib import Path
+
+from engine import config
+from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
+from engine.figment_render import rasterize_svg
+from engine.figment_trigger import FigmentAction, FigmentCommand, FigmentTrigger
+from engine.themes import THEME_REGISTRY
+
+
+class FigmentPhase(Enum):
+ REVEAL = auto()
+ HOLD = auto()
+ DISSOLVE = auto()
+
+
+@dataclass
+class FigmentState:
+ phase: FigmentPhase
+ progress: float
+ rows: list[str]
+ gradient: list[int]
+ center_row: int
+ center_col: int
+
+
+class FigmentEffect(EffectPlugin):
+ name = "figment"
+ config = EffectConfig(
+ enabled=False,
+ intensity=1.0,
+ params={
+ "interval_secs": 60,
+ "display_secs": 4.5,
+ "figment_dir": "figments",
+ },
+ )
+
+ def __init__(self, figment_dir: str | None = None, triggers: list[FigmentTrigger] | None = None):
+ self.config = EffectConfig(
+ enabled=False,
+ intensity=1.0,
+ params={
+ "interval_secs": 60,
+ "display_secs": 4.5,
+ "figment_dir": figment_dir or "figments",
+ },
+ )
+ self._triggers = triggers or []
+ self._phase: FigmentPhase | None = None
+ self._progress: float = 0.0
+ self._rows: list[str] = []
+ self._gradient: list[int] = []
+ self._center_row: int = 0
+ self._center_col: int = 0
+ self._timer: float = 0.0
+ self._last_svg: str | None = None
+ self._svg_files: list[str] = []
+ self._scan_svgs()
+
+ def _scan_svgs(self) -> None:
+ figment_dir = Path(self.config.params["figment_dir"])
+ if figment_dir.is_dir():
+ self._svg_files = sorted(str(p) for p in figment_dir.glob("*.svg"))
+
+ def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
+ return buf
+
+ def configure(self, cfg: EffectConfig) -> None:
+ self.config = cfg
+ self._scan_svgs()
+
+ def trigger(self, w: int, h: int) -> None:
+ """Manually trigger a figment display."""
+ if not self._svg_files:
+ return
+
+ # Pick a random SVG, avoid repeating
+ candidates = [s for s in self._svg_files if s != self._last_svg]
+ if not candidates:
+ candidates = self._svg_files
+ svg_path = random.choice(candidates)
+ self._last_svg = svg_path
+
+ # Rasterize
+ try:
+ self._rows = rasterize_svg(svg_path, w, h)
+ except Exception:
+ return
+
+ # Pick random theme gradient
+ theme_key = random.choice(list(THEME_REGISTRY.keys()))
+ self._gradient = THEME_REGISTRY[theme_key].main_gradient
+
+ # Center in viewport
+ figment_h = len(self._rows)
+ figment_w = max((len(r) for r in self._rows), default=0)
+ self._center_row = max(0, (h - figment_h) // 2)
+ self._center_col = max(0, (w - figment_w) // 2)
+
+ # Start reveal phase
+ self._phase = FigmentPhase.REVEAL
+ self._progress = 0.0
+
+ def get_figment_state(self, frame_number: int, w: int, h: int) -> FigmentState | None:
+ """Tick the state machine and return current state, or None if idle."""
+ if not self.config.enabled:
+ return None
+
+ # Poll triggers
+ for trig in self._triggers:
+ cmd = trig.poll()
+ if cmd is not None:
+ self._handle_command(cmd, w, h)
+
+ # Tick timer when idle
+ if self._phase is None:
+ self._timer += config.FRAME_DT
+ interval = self.config.params.get("interval_secs", 60)
+ if self._timer >= interval:
+ self._timer = 0.0
+ self.trigger(w, h)
+
+ # Tick animation
+ if self._phase is not None:
+ display_secs = self.config.params.get("display_secs", 4.5)
+ phase_duration = display_secs / 3.0
+ self._progress += config.FRAME_DT / phase_duration
+
+ if self._progress >= 1.0:
+ self._progress = 0.0
+ if self._phase == FigmentPhase.REVEAL:
+ self._phase = FigmentPhase.HOLD
+ elif self._phase == FigmentPhase.HOLD:
+ self._phase = FigmentPhase.DISSOLVE
+ elif self._phase == FigmentPhase.DISSOLVE:
+ self._phase = None
+ return None
+
+ return FigmentState(
+ phase=self._phase,
+ progress=self._progress,
+ rows=self._rows,
+ gradient=self._gradient,
+ center_row=self._center_row,
+ center_col=self._center_col,
+ )
+
+ return None
+
+ def _handle_command(self, cmd: FigmentCommand, w: int, h: int) -> None:
+ if cmd.action == FigmentAction.TRIGGER:
+ self.trigger(w, h)
+ elif cmd.action == FigmentAction.SET_INTENSITY and isinstance(cmd.value, (int, float)):
+ self.config.intensity = float(cmd.value)
+ elif cmd.action == FigmentAction.SET_INTERVAL and isinstance(cmd.value, (int, float)):
+ self.config.params["interval_secs"] = float(cmd.value)
+ elif cmd.action == FigmentAction.SET_COLOR and isinstance(cmd.value, str):
+ if cmd.value in THEME_REGISTRY:
+ self._gradient = THEME_REGISTRY[cmd.value].main_gradient
+ elif cmd.action == FigmentAction.STOP:
+ self._phase = None
+ self._progress = 0.0
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+```bash
+uv run pytest tests/test_figment.py -v
+```
+
+Expected: All tests pass.
+
+- [ ] **Step 5: Verify plugin discovery finds FigmentEffect**
+
+```bash
+uv run python -c "
+from engine.effects.registry import EffectRegistry, set_registry
+set_registry(EffectRegistry())
+from effects_plugins import discover_plugins
+plugins = discover_plugins()
+print('Discovered:', list(plugins.keys()))
+assert 'figment' in plugins, 'FigmentEffect not discovered!'
+print('OK')
+"
+```
+
+Expected: Prints `Discovered: ['noise', 'glitch', 'fade', 'firehose', 'figment']` and `OK`.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add effects_plugins/figment.py tests/test_figment.py
+git commit -m "feat(figment): add FigmentEffect plugin with state machine and timer"
+```
+
+---
+
+## Chunk 4: Overlay Rendering and Scroll Integration
+
+### Task 6: Figment overlay renderer in layers.py
+
+**Files:**
+- Modify: `engine/layers.py:1-4` (add import), append `render_figment_overlay()` function
+- Create: `tests/test_figment_overlay.py`
+
+- [ ] **Step 1: Write failing tests for render_figment_overlay**
+
+Create `tests/test_figment_overlay.py`:
+
+```python
+"""Tests for render_figment_overlay in engine.layers."""
+
+from effects_plugins.figment import FigmentPhase, FigmentState
+from engine.layers import render_figment_overlay
+
+
+def _make_state(phase=FigmentPhase.HOLD, progress=0.5):
+ return FigmentState(
+ phase=phase,
+ progress=progress,
+ rows=["█▀▄ █", " ▄█▀ ", "█ █"],
+ gradient=[46, 40, 34, 28, 22, 22, 34, 40, 46, 82, 118, 231],
+ center_row=10,
+ center_col=37,
+ )
+
+
+class TestRenderFigmentOverlay:
+ def test_returns_list_of_strings(self):
+ state = _make_state()
+ result = render_figment_overlay(state, 80, 24)
+ assert isinstance(result, list)
+ assert all(isinstance(s, str) for s in result)
+
+ def test_contains_ansi_positioning(self):
+ state = _make_state()
+ result = render_figment_overlay(state, 80, 24)
+ # Should contain cursor positioning escape codes
+ assert any("\033[" in s for s in result)
+
+ def test_reveal_phase_partial(self):
+ state = _make_state(phase=FigmentPhase.REVEAL, progress=0.0)
+ result = render_figment_overlay(state, 80, 24)
+ # At progress 0.0, very few cells should be visible
+ # Result should still be a valid list
+ assert isinstance(result, list)
+
+ def test_hold_phase_full(self):
+ state = _make_state(phase=FigmentPhase.HOLD, progress=0.5)
+ result = render_figment_overlay(state, 80, 24)
+ # During hold, content should be present
+ assert len(result) > 0
+
+ def test_dissolve_phase(self):
+ state = _make_state(phase=FigmentPhase.DISSOLVE, progress=0.9)
+ result = render_figment_overlay(state, 80, 24)
+ # At high dissolve progress, most cells are gone
+ assert isinstance(result, list)
+
+ def test_empty_rows(self):
+ state = FigmentState(
+ phase=FigmentPhase.HOLD,
+ progress=0.5,
+ rows=[],
+ gradient=[46] * 12,
+ center_row=0,
+ center_col=0,
+ )
+ result = render_figment_overlay(state, 80, 24)
+ assert result == []
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+```bash
+uv run pytest tests/test_figment_overlay.py -v
+```
+
+Expected: FAIL — `ImportError: cannot import name 'render_figment_overlay' from 'engine.layers'`
+
+- [ ] **Step 3: Implement render_figment_overlay**
+
+Add to the end of `engine/layers.py` (after `get_effect_chain()`):
+
+```python
+def render_figment_overlay(
+ figment_state,
+ w: int,
+ h: int,
+) -> list[str]:
+ """Render figment overlay as ANSI cursor-positioning commands.
+
+ Args:
+ figment_state: FigmentState with phase, progress, rows, gradient, centering.
+ w: terminal width
+ h: terminal height
+
+ Returns:
+ List of ANSI strings to append to display buffer.
+ """
+ from engine.render import lr_gradient, _color_codes_to_ansi
+
+ rows = figment_state.rows
+ if not rows:
+ return []
+
+ phase = figment_state.phase
+ progress = figment_state.progress
+ gradient = figment_state.gradient
+ center_row = figment_state.center_row
+ center_col = figment_state.center_col
+
+ cols = _color_codes_to_ansi(gradient)
+
+ # Determine cell visibility based on phase
+ # Build a visibility mask for non-space cells
+ cell_positions = []
+ for r_idx, row in enumerate(rows):
+ for c_idx, ch in enumerate(row):
+ if ch != " ":
+ cell_positions.append((r_idx, c_idx))
+
+ n_cells = len(cell_positions)
+ if n_cells == 0:
+ return []
+
+ # Use a deterministic seed so the reveal/dissolve pattern is stable per-figment
+ rng = random.Random(hash(tuple(rows[0][:10])) if rows[0] else 42)
+ shuffled = list(cell_positions)
+ rng.shuffle(shuffled)
+
+ # Phase-dependent visibility
+ from effects_plugins.figment import FigmentPhase
+
+ if phase == FigmentPhase.REVEAL:
+ visible_count = int(n_cells * progress)
+ visible = set(shuffled[:visible_count])
+ elif phase == FigmentPhase.HOLD:
+ visible = set(cell_positions)
+ # Strobe: dim some cells periodically
+ if int(progress * 20) % 3 == 0:
+ # Dim ~30% of cells for strobe effect
+ dim_count = int(n_cells * 0.3)
+ visible -= set(shuffled[:dim_count])
+ elif phase == FigmentPhase.DISSOLVE:
+ remaining_count = int(n_cells * (1.0 - progress))
+ visible = set(shuffled[:remaining_count])
+ else:
+ visible = set(cell_positions)
+
+ # Build overlay commands
+ overlay: list[str] = []
+ n_cols = len(cols)
+ max_x = max((len(r.rstrip()) for r in rows if r.strip()), default=1)
+
+ for r_idx, row in enumerate(rows):
+ scr_row = center_row + r_idx + 1 # 1-indexed
+ if scr_row < 1 or scr_row > h:
+ continue
+
+ line_buf: list[str] = []
+ has_content = False
+
+ for c_idx, ch in enumerate(row):
+ scr_col = center_col + c_idx + 1
+ if scr_col < 1 or scr_col > w:
+ continue
+
+ if ch != " " and (r_idx, c_idx) in visible:
+ # Apply gradient color
+ shifted = (c_idx / max(max_x - 1, 1)) % 1.0
+ idx = min(round(shifted * (n_cols - 1)), n_cols - 1)
+ line_buf.append(f"{cols[idx]}{ch}{RST}")
+ has_content = True
+ else:
+ line_buf.append(" ")
+
+ if has_content:
+ # Trim trailing spaces
+ line_str = "".join(line_buf).rstrip()
+ if line_str.strip():
+ overlay.append(
+ f"\033[{scr_row};{center_col + 1}H{line_str}{RST}"
+ )
+
+ return overlay
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+```bash
+uv run pytest tests/test_figment_overlay.py -v
+```
+
+Expected: All 6 tests pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add engine/layers.py tests/test_figment_overlay.py
+git commit -m "feat(figment): add render_figment_overlay() to layers.py"
+```
+
+---
+
+### Task 7: Scroll loop integration
+
+**Files:**
+- Modify: `engine/scroll.py:18-24` (add import), `engine/scroll.py:30` (setup), `engine/scroll.py:125-127` (frame loop)
+
+- [ ] **Step 1: Add figment import and setup to stream()**
+
+In `engine/scroll.py`, add the import for `render_figment_overlay` to the existing layers import block (line 18-24):
+
+```python
+from engine.layers import (
+ apply_glitch,
+ process_effects,
+ render_firehose,
+ render_figment_overlay,
+ render_message_overlay,
+ render_ticker_zone,
+)
+```
+
+Then add the figment setup inside `stream()`, after the `frame_number = 0` line (line 54):
+
+```python
+ # Figment overlay (optional — requires cairosvg)
+ try:
+ from effects_plugins.figment import FigmentEffect
+ from engine.effects.registry import get_registry
+
+ _fg_plugin = get_registry().get("figment")
+ figment = _fg_plugin if isinstance(_fg_plugin, FigmentEffect) else None
+ except ImportError:
+ figment = None
+```
+
+- [ ] **Step 2: Add figment overlay to frame loop**
+
+In the frame loop, insert the figment overlay block between the effects processing (line 120) and the message overlay (line 126). Insert after the `else:` block at line 124:
+
+```python
+ # Figment overlay (between effects and ntfy message)
+ if figment and figment.config.enabled:
+ figment_state = figment.get_figment_state(frame_number, w, h)
+ if figment_state is not None:
+ figment_buf = render_figment_overlay(figment_state, w, h)
+ buf.extend(figment_buf)
+```
+
+- [ ] **Step 3: Run full test suite**
+
+```bash
+uv run pytest tests/ -v
+```
+
+Expected: All tests pass (existing + new). The 3 pre-existing `warmup_topics` failures remain.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add engine/scroll.py
+git commit -m "feat(figment): integrate figment overlay into scroll loop"
+```
+
+---
+
+### Task 8: Run lint and final verification
+
+- [ ] **Step 1: Run ruff linter**
+
+```bash
+uv run ruff check .
+```
+
+Fix any issues found.
+
+- [ ] **Step 2: Run ruff formatter**
+
+```bash
+uv run ruff format .
+```
+
+- [ ] **Step 3: Run full test suite one more time**
+
+```bash
+uv run pytest tests/ -v
+```
+
+Expected: All tests pass (except the 3 pre-existing `warmup_topics` failures).
+
+- [ ] **Step 4: Commit any lint/format fixes**
+
+```bash
+git add -u
+git commit -m "style: apply ruff formatting to figment modules"
+```
+
+(Skip this commit if ruff made no changes.)
diff --git a/docs/superpowers/specs/2026-03-19-figment-mode-design.md b/docs/superpowers/specs/2026-03-19-figment-mode-design.md
new file mode 100644
index 0000000..c6bd3f9
--- /dev/null
+++ b/docs/superpowers/specs/2026-03-19-figment-mode-design.md
@@ -0,0 +1,308 @@
+# Figment Mode Design Spec
+
+> Periodic full-screen SVG glyph overlay with flickery animation, theme-aware coloring, and extensible physical device control.
+
+## Overview
+
+Figment mode displays a randomly selected SVG from the `figments/` directory as a flickery, glitchy half-block terminal overlay on top of the running ticker. It appears once per minute (configurable), holds for ~4.5 seconds with a three-phase animation (progressive reveal, strobing hold, dissolve), then fades back to the ticker. Colors are randomly chosen from the existing theme gradients.
+
+The feature is designed for extensibility: a generic input protocol allows MQTT, ntfy, serial, or any other control surface to trigger figments and adjust parameters in real time.
+
+## Goals
+
+- Display SVG figments as half-block terminal art overlaid on the running ticker
+- Three-phase animation: progressive reveal, strobing hold, dissolve
+- Random color from existing theme gradients (green, orange, purple)
+- Configurable interval and duration via C&C
+- Extensible input abstraction for physical device control (MQTT, serial, etc.)
+
+## Out of Scope
+
+- Multi-figment simultaneous display (one at a time)
+- SVG animation support (static SVGs only; animation comes from the overlay phases)
+- Custom color palettes beyond existing themes
+- MQTT and serial adapters (v1 ships with ntfy C&C only; protocol is ready for future adapters)
+
+## Architecture: Hybrid Plugin + Overlay
+
+The figment is an **EffectPlugin** for lifecycle, discovery, and configuration, but delegates rendering to a **layers-style overlay helper**. This avoids stretching the `EffectPlugin.process()` contract (which transforms line buffers) while still benefiting from the plugin system for C&C, auto-discovery, and config management.
+
+**Important**: The plugin class is named `FigmentEffect` (not `FigmentPlugin`) to match the `*Effect` naming convention required by `discover_plugins()` in `effects_plugins/__init__.py`. The plugin is **not** added to the `EffectChain` order list — its `process()` is a no-op that returns the buffer unchanged. The chain only processes effects that transform buffers (noise, fade, glitch, firehose). Figment's rendering happens via the overlay path in `scroll.py`, outside the chain.
+
+### Component Diagram
+
+```
+ +-------------------+
+ | FigmentTrigger | (Protocol)
+ | - NtfyTrigger | (v1)
+ | - MqttTrigger | (future)
+ | - SerialTrigger | (future)
+ +--------+----------+
+ |
+ | FigmentCommand
+ v
++------------------+ +-----------------+ +----------------------+
+| figment_render |<---| FigmentEffect |--->| render_figment_ |
+| .py | | (EffectPlugin) | | overlay() in |
+| | | | | layers.py |
+| SVG -> PIL -> | | Timer, state | | |
+| half-block cache | | machine, SVG | | ANSI cursor-position |
+| | | selection | | commands for overlay |
++------------------+ +-----------------+ +----------------------+
+ |
+ | get_figment_state()
+ v
+ +-------------------+
+ | scroll.py |
+ +-------------------+
+```
+
+## Section 1: SVG Rasterization
+
+**File: `engine/figment_render.py`**
+
+Reuses the same PIL-based half-block encoding that `engine/render.py` uses for OTF fonts.
+
+### Pipeline
+
+1. **Load**: `cairosvg.svg2png()` converts SVG to PNG bytes in memory (no temp files)
+2. **Resize**: PIL scales to fit terminal — width = `tw()`, height = `th() * 2` pixels (each terminal row encodes 2 pixel rows via half-blocks)
+3. **Threshold**: Convert to greyscale ("L" mode), apply binary threshold to get visible/not-visible
+4. **Half-block encode**: Walk pixel pairs top-to-bottom. For each 2-row pair, emit `█` (both lit), `▀` (top only), `▄` (bottom only), or space (neither)
+5. **Cache**: Results cached per `(svg_path, terminal_width, terminal_height)` — invalidated on terminal resize
+
+### Dependency
+
+`cairosvg` added as an optional dependency in `pyproject.toml` (like `sounddevice`). If `cairosvg` is not installed, the `FigmentEffect` class will fail to import, and `discover_plugins()` will silently skip it (the existing `except Exception: pass` in discovery handles this). The plugin simply won't appear in the registry.
+
+### Key Function
+
+```python
+def rasterize_svg(svg_path: str, width: int, height: int) -> list[str]:
+ """Convert SVG file to list of half-block terminal rows (uncolored)."""
+```
+
+## Section 2: Figment Overlay Rendering
+
+**Integration point: `engine/layers.py`**
+
+New function following the `render_message_overlay()` pattern.
+
+### FigmentState Dataclass
+
+Defined in `effects_plugins/figment.py`, passed between the plugin and the overlay renderer:
+
+```python
+@dataclass
+class FigmentState:
+ phase: FigmentPhase # enum: REVEAL, HOLD, DISSOLVE
+ progress: float # 0.0 to 1.0 within current phase
+ rows: list[str] # rasterized half-block rows (uncolored)
+ gradient: list[int] # 12-color ANSI 256 gradient from chosen theme
+ center_row: int # top row for centering in viewport
+ center_col: int # left column for centering in viewport
+```
+
+### Function Signature
+
+```python
+def render_figment_overlay(figment_state: FigmentState, w: int, h: int) -> list[str]:
+ """Return ANSI cursor-positioning commands for the current figment frame."""
+```
+
+### Animation Phases (~4.5 seconds total)
+
+Progress advances each frame as: `progress += config.FRAME_DT / phase_duration`. At 20 FPS (FRAME_DT=0.05s), a 1.5s phase takes 30 frames to complete.
+
+| Phase | Duration | Behavior |
+|-------|----------|----------|
+| **Reveal** | ~1.5s | Progressive scanline fill. Each frame, a percentage of the figment's non-empty cells become visible in random block order. Intensity scales reveal speed. |
+| **Hold** | ~1.5s | Full figment visible. Strobes between full brightness and dimmed/partial visibility every few frames. Intensity scales strobe frequency. |
+| **Dissolve** | ~1.5s | Inverse of reveal. Cells randomly drop out, replaced by spaces. Intensity scales dissolve speed. |
+
+### Color
+
+A random theme gradient is selected from `THEME_REGISTRY` at trigger time. Applied via `lr_gradient()` — the same function that colors headlines and messages.
+
+### Positioning
+
+Figment is centered in the viewport. Each visible row is an ANSI `\033[row;colH` command appended to the buffer, identical to how the message overlay works.
+
+## Section 3: FigmentEffect (Effect Plugin)
+
+**File: `effects_plugins/figment.py`**
+
+An `EffectPlugin(ABC)` subclass named `FigmentEffect` to match the `*Effect` discovery convention.
+
+### Chain Exclusion
+
+`FigmentEffect` is registered in the `EffectRegistry` (for C&C access and config management) but is **not** added to the `EffectChain` order list. Its `process()` returns the buffer unchanged. The `enabled` flag is checked directly by `scroll.py` when deciding whether to call `get_figment_state()`, not by the chain.
+
+### Responsibilities
+
+- **Timer**: Tracks elapsed time via `config.FRAME_DT` accumulation. At the configured interval (default 60s), triggers a new figment.
+- **SVG selection**: Randomly picks from `figments/*.svg`. Avoids repeating the last shown.
+- **State machine**: `idle -> reveal -> hold -> dissolve -> idle`. Tracks phase progress (0.0 to 1.0).
+- **Color selection**: Picks a random theme key (`"green"`, `"orange"`, `"purple"`) at trigger time.
+- **Rasterization**: Calls `rasterize_svg()` on trigger, caches result for the display duration.
+
+### State Machine
+
+```
+idle ──(timer fires or trigger received)──> reveal
+reveal ──(progress >= 1.0)──> hold
+hold ──(progress >= 1.0)──> dissolve
+dissolve ──(progress >= 1.0)──> idle
+```
+
+### Interface
+
+The `process()` method returns the buffer unchanged (no-op). The plugin exposes state via:
+
+```python
+def get_figment_state(self, frame_number: int) -> FigmentState | None:
+ """Tick the state machine and return current state, or None if idle."""
+```
+
+This mirrors the `ntfy_poller.get_active_message()` pattern.
+
+### Scroll Loop Access
+
+`scroll.py` imports `FigmentEffect` directly and uses `isinstance()` to safely downcast from the registry:
+
+```python
+from effects_plugins.figment import FigmentEffect
+
+plugin = registry.get("figment")
+figment = plugin if isinstance(plugin, FigmentEffect) else None
+```
+
+This is a one-time setup check, not per-frame. If `cairosvg` is missing, the import is wrapped in a try/except and `figment` stays `None`.
+
+### EffectConfig
+
+- `enabled`: bool (default `False` — opt-in)
+- `intensity`: float — scales strobe frequency and reveal/dissolve speed
+- `params`:
+ - `interval_secs`: 60 (time between figments)
+ - `display_secs`: 4.5 (total animation duration)
+ - `figment_dir`: "figments" (SVG source directory)
+
+Controllable via C&C: `/effects figment on`, `/effects figment intensity 0.7`.
+
+## Section 4: Input Abstraction (FigmentTrigger)
+
+**File: `engine/figment_trigger.py`**
+
+### Protocol
+
+```python
+class FigmentTrigger(Protocol):
+ def poll(self) -> FigmentCommand | None: ...
+```
+
+### FigmentCommand
+
+```python
+class FigmentAction(Enum):
+ TRIGGER = "trigger"
+ SET_INTENSITY = "set_intensity"
+ SET_INTERVAL = "set_interval"
+ SET_COLOR = "set_color"
+ STOP = "stop"
+
+@dataclass
+class FigmentCommand:
+ action: FigmentAction
+ value: float | str | None = None
+```
+
+Uses an enum for consistency with `EventType` in `engine/events.py`.
+
+### Adapters
+
+| Adapter | Transport | Dependency | Status |
+|---------|-----------|------------|--------|
+| `NtfyTrigger` | Existing C&C ntfy topic | None (reuses ntfy) | v1 |
+| `MqttTrigger` | MQTT broker | `paho-mqtt` (optional) | Future |
+| `SerialTrigger` | USB serial | `pyserial` (optional) | Future |
+
+**NtfyTrigger v1**: Subscribes as a callback on the existing `NtfyPoller`. Parses messages with a `/figment` prefix (e.g., `/figment trigger`, `/figment intensity 0.8`). This is separate from the `/effects figment on` C&C path — the trigger protocol allows external devices to send commands without knowing the effects controller API.
+
+### Integration
+
+The `FigmentEffect` accepts a list of triggers. Each frame, it polls all triggers and acts on commands. Triggers are optional — if none are configured, the plugin runs on its internal timer alone.
+
+### EventBus Bridge
+
+A new `FIGMENT_TRIGGER` variant is added to the `EventType` enum in `engine/events.py`, with a corresponding `FigmentTriggerEvent` dataclass. Triggers publish to the EventBus for other components to react (logging, multi-display sync).
+
+## Section 5: Scroll Loop Integration
+
+Minimal change to `engine/scroll.py`:
+
+```python
+# In stream() setup (with safe import):
+try:
+ from effects_plugins.figment import FigmentEffect
+ _plugin = registry.get("figment")
+ figment = _plugin if isinstance(_plugin, FigmentEffect) else None
+except ImportError:
+ figment = None
+
+# In frame loop, after effects processing, before ntfy message overlay:
+if figment and figment.config.enabled:
+ figment_state = figment.get_figment_state(frame_number)
+ if figment_state is not None:
+ figment_overlay = render_figment_overlay(figment_state, w, h)
+ buf.extend(figment_overlay)
+```
+
+### Overlay Priority
+
+Figment overlay appends **after** effects processing but **before** the ntfy message overlay. This means:
+- Ntfy messages always appear on top of figments (higher priority)
+- Existing glitch/noise effects run over the ticker underneath the figment
+
+Note: If more overlay types are added in the future, a priority-based overlay system should replace the current positional ordering.
+
+## Section 6: Error Handling
+
+| Scenario | Behavior |
+|----------|----------|
+| `cairosvg` not installed | `FigmentEffect` fails to import; `discover_plugins()` silently skips it; `scroll.py` import guard sets `figment = None` |
+| `figments/` directory missing | Plugin logs warning at startup, stays in permanent `idle` state |
+| `figments/` contains zero `.svg` files | Same as above: warning, permanent `idle` |
+| Malformed SVG | `cairosvg` raises exception; plugin catches it, skips that SVG, picks another. If all SVGs fail, enters permanent `idle` with warning |
+| Terminal resize during animation | Re-rasterize on next frame using new dimensions. Cache miss triggers fresh rasterization. Animation phase/progress are preserved; only the rendered rows update |
+
+## Section 7: File Summary
+
+### New Files
+
+| File | Purpose |
+|------|---------|
+| `effects_plugins/figment.py` | FigmentEffect — lifecycle, timer, state machine, SVG selection, FigmentState/FigmentPhase |
+| `engine/figment_render.py` | SVG to half-block rasterization pipeline |
+| `engine/figment_trigger.py` | FigmentTrigger protocol, FigmentAction enum, FigmentCommand, NtfyTrigger adapter |
+| `figments/` | SVG source directory (ships with sample SVGs) |
+| `tests/test_figment.py` | FigmentEffect lifecycle, state machine transitions, timer |
+| `tests/test_figment_render.py` | SVG rasterization, caching, edge cases |
+| `tests/test_figment_trigger.py` | FigmentCommand parsing, NtfyTrigger adapter |
+| `tests/fixtures/test.svg` | Minimal SVG for deterministic rasterization tests |
+
+### Modified Files
+
+| File | Change |
+|------|--------|
+| `engine/scroll.py` | Figment overlay integration (setup + per-frame block) |
+| `engine/layers.py` | Add `render_figment_overlay()` function |
+| `engine/events.py` | Add `FIGMENT_TRIGGER` to `EventType` enum, add `FigmentTriggerEvent` dataclass |
+| `pyproject.toml` | Add `cairosvg` as optional dependency |
+
+## Testing Strategy
+
+- **Unit**: State machine transitions (idle→reveal→hold→dissolve→idle), timer accuracy (fires at interval_secs), SVG rasterization output dimensions, FigmentCommand parsing, FigmentAction enum coverage
+- **Integration**: Plugin discovery (verify `FigmentEffect` is found by `discover_plugins()`), overlay rendering with mock terminal dimensions, C&C command handling via `/effects figment on`
+- **Edge cases**: Missing figments dir, empty dir, malformed SVG, cairosvg unavailable, terminal resize mid-animation
+- **Fixture**: Minimal `test.svg` (simple rectangle) for deterministic rasterization tests
diff --git a/effects_plugins/figment.py b/effects_plugins/figment.py
new file mode 100644
index 0000000..bf9ca14
--- /dev/null
+++ b/effects_plugins/figment.py
@@ -0,0 +1,200 @@
+"""
+Figment effect plugin — periodic SVG glyph overlay.
+
+Owns the figment lifecycle: timer, SVG selection, state machine.
+Delegates rendering to render_figment_overlay() in engine/layers.py.
+
+Named FigmentEffect (not FigmentPlugin) to match the *Effect discovery
+convention in effects_plugins/__init__.py.
+
+NOT added to the EffectChain order — process() is a no-op. The overlay
+rendering is handled by scroll.py calling get_figment_state().
+"""
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass
+from enum import Enum, auto
+from pathlib import Path
+
+from engine import config
+from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
+from engine.figment_render import rasterize_svg
+from engine.figment_trigger import FigmentAction, FigmentCommand, FigmentTrigger
+from engine.themes import THEME_REGISTRY
+
+
+class FigmentPhase(Enum):
+ REVEAL = auto()
+ HOLD = auto()
+ DISSOLVE = auto()
+
+
+@dataclass
+class FigmentState:
+ phase: FigmentPhase
+ progress: float
+ rows: list[str]
+ gradient: list[int]
+ center_row: int
+ center_col: int
+
+
+class FigmentEffect(EffectPlugin):
+ name = "figment"
+ config = EffectConfig(
+ enabled=False,
+ intensity=1.0,
+ params={
+ "interval_secs": 60,
+ "display_secs": 4.5,
+ "figment_dir": "figments",
+ },
+ )
+
+ def __init__(
+ self,
+ figment_dir: str | None = None,
+ triggers: list[FigmentTrigger] | None = None,
+ ):
+ self.config = EffectConfig(
+ enabled=False,
+ intensity=1.0,
+ params={
+ "interval_secs": 60,
+ "display_secs": 4.5,
+ "figment_dir": figment_dir or "figments",
+ },
+ )
+ self._triggers = triggers or []
+ self._phase: FigmentPhase | None = None
+ self._progress: float = 0.0
+ self._rows: list[str] = []
+ self._gradient: list[int] = []
+ self._center_row: int = 0
+ self._center_col: int = 0
+ self._timer: float = 0.0
+ self._last_svg: str | None = None
+ self._svg_files: list[str] = []
+ self._scan_svgs()
+
+ def _scan_svgs(self) -> None:
+ figment_dir = Path(self.config.params["figment_dir"])
+ if figment_dir.is_dir():
+ self._svg_files = sorted(str(p) for p in figment_dir.glob("*.svg"))
+
+ def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
+ return buf
+
+ def configure(self, cfg: EffectConfig) -> None:
+ # Preserve figment_dir if the new config doesn't supply one
+ figment_dir = cfg.params.get(
+ "figment_dir", self.config.params.get("figment_dir", "figments")
+ )
+ self.config = cfg
+ if "figment_dir" not in self.config.params:
+ self.config.params["figment_dir"] = figment_dir
+ self._scan_svgs()
+
+ def trigger(self, w: int, h: int) -> None:
+ """Manually trigger a figment display."""
+ if not self._svg_files:
+ return
+
+ # Pick a random SVG, avoid repeating
+ candidates = [s for s in self._svg_files if s != self._last_svg]
+ if not candidates:
+ candidates = self._svg_files
+ svg_path = random.choice(candidates)
+ self._last_svg = svg_path
+
+ # Rasterize
+ try:
+ self._rows = rasterize_svg(svg_path, w, h)
+ except Exception:
+ return
+
+ # Pick random theme gradient
+ theme_key = random.choice(list(THEME_REGISTRY.keys()))
+ self._gradient = THEME_REGISTRY[theme_key].main_gradient
+
+ # Center in viewport
+ figment_h = len(self._rows)
+ figment_w = max((len(r) for r in self._rows), default=0)
+ self._center_row = max(0, (h - figment_h) // 2)
+ self._center_col = max(0, (w - figment_w) // 2)
+
+ # Start reveal phase
+ self._phase = FigmentPhase.REVEAL
+ self._progress = 0.0
+
+ def get_figment_state(
+ self, frame_number: int, w: int, h: int
+ ) -> FigmentState | None:
+ """Tick the state machine and return current state, or None if idle."""
+ if not self.config.enabled:
+ return None
+
+ # Poll triggers
+ for trig in self._triggers:
+ cmd = trig.poll()
+ if cmd is not None:
+ self._handle_command(cmd, w, h)
+
+ # Tick timer when idle
+ if self._phase is None:
+ self._timer += config.FRAME_DT
+ interval = self.config.params.get("interval_secs", 60)
+ if self._timer >= interval:
+ self._timer = 0.0
+ self.trigger(w, h)
+
+ # Tick animation — snapshot current phase/progress, then advance
+ if self._phase is not None:
+ # Capture the state at the start of this frame
+ current_phase = self._phase
+ current_progress = self._progress
+
+ # Advance for next frame
+ display_secs = self.config.params.get("display_secs", 4.5)
+ phase_duration = display_secs / 3.0
+ self._progress += config.FRAME_DT / phase_duration
+
+ if self._progress >= 1.0:
+ self._progress = 0.0
+ if self._phase == FigmentPhase.REVEAL:
+ self._phase = FigmentPhase.HOLD
+ elif self._phase == FigmentPhase.HOLD:
+ self._phase = FigmentPhase.DISSOLVE
+ elif self._phase == FigmentPhase.DISSOLVE:
+ self._phase = None
+
+ return FigmentState(
+ phase=current_phase,
+ progress=current_progress,
+ rows=self._rows,
+ gradient=self._gradient,
+ center_row=self._center_row,
+ center_col=self._center_col,
+ )
+
+ return None
+
+ def _handle_command(self, cmd: FigmentCommand, w: int, h: int) -> None:
+ if cmd.action == FigmentAction.TRIGGER:
+ self.trigger(w, h)
+ elif cmd.action == FigmentAction.SET_INTENSITY and isinstance(
+ cmd.value, (int, float)
+ ):
+ self.config.intensity = float(cmd.value)
+ elif cmd.action == FigmentAction.SET_INTERVAL and isinstance(
+ cmd.value, (int, float)
+ ):
+ self.config.params["interval_secs"] = float(cmd.value)
+ elif cmd.action == FigmentAction.SET_COLOR and isinstance(cmd.value, str):
+ if cmd.value in THEME_REGISTRY:
+ self._gradient = THEME_REGISTRY[cmd.value].main_gradient
+ elif cmd.action == FigmentAction.STOP:
+ self._phase = None
+ self._progress = 0.0
diff --git a/engine/app.py b/engine/app.py
index 10eea6d..8ba2803 100644
--- a/engine/app.py
+++ b/engine/app.py
@@ -413,6 +413,14 @@ def main():
if config.FIREHOSE:
boot_ln("Firehose", "ENGAGED", True)
+ if config.FIGMENT:
+ try:
+ from effects_plugins.figment import FigmentEffect # noqa: F401
+
+ boot_ln("Figment", f"ARMED [{config.FIGMENT_INTERVAL}s interval]", True)
+ except (ImportError, OSError):
+ boot_ln("Figment", "UNAVAILABLE — run: brew install cairo", False)
+
time.sleep(0.4)
slow_print(" > STREAMING...\n")
time.sleep(0.2)
diff --git a/engine/config.py b/engine/config.py
index c5ce46c..8ca8191 100644
--- a/engine/config.py
+++ b/engine/config.py
@@ -196,6 +196,8 @@ MODE = (
else "news"
)
FIREHOSE = "--firehose" in sys.argv
+FIGMENT = "--figment" in sys.argv
+FIGMENT_INTERVAL = _arg_int("--figment-interval", 60) # seconds between appearances
# ─── NTFY MESSAGE QUEUE ──────────────────────────────────
NTFY_TOPIC = "https://ntfy.sh/klubhaus_terminal_mainline/json"
diff --git a/engine/events.py b/engine/events.py
index d686285..61fcfc0 100644
--- a/engine/events.py
+++ b/engine/events.py
@@ -18,6 +18,7 @@ class EventType(Enum):
NTFY_MESSAGE = auto()
STREAM_START = auto()
STREAM_END = auto()
+ FIGMENT_TRIGGER = auto()
@dataclass
@@ -65,3 +66,12 @@ class StreamEvent:
event_type: EventType
headline_count: int = 0
timestamp: datetime | None = None
+
+
+@dataclass
+class FigmentTriggerEvent:
+ """Event emitted when a figment is triggered."""
+
+ action: str
+ value: float | str | None = None
+ timestamp: datetime | None = None
diff --git a/engine/figment_render.py b/engine/figment_render.py
new file mode 100644
index 0000000..0b9e0ea
--- /dev/null
+++ b/engine/figment_render.py
@@ -0,0 +1,90 @@
+"""
+SVG to half-block terminal art rasterization.
+
+Pipeline: SVG -> cairosvg -> PIL -> greyscale threshold -> half-block encode.
+Follows the same pixel-pair approach as engine/render.py for OTF fonts.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from io import BytesIO
+
+# cairocffi (used by cairosvg) calls dlopen() to find the Cairo C library.
+# On macOS with Homebrew, Cairo lives in /opt/homebrew/lib (Apple Silicon) or
+# /usr/local/lib (Intel), which are not in dyld's default search path.
+# Setting DYLD_LIBRARY_PATH before the import directs dlopen() to those paths.
+if sys.platform == "darwin" and not os.environ.get("DYLD_LIBRARY_PATH"):
+ for _brew_lib in ("/opt/homebrew/lib", "/usr/local/lib"):
+ if os.path.exists(os.path.join(_brew_lib, "libcairo.2.dylib")):
+ os.environ["DYLD_LIBRARY_PATH"] = _brew_lib
+ break
+
+import cairosvg
+from PIL import Image
+
+_cache: dict[tuple[str, int, int], list[str]] = {}
+
+
+def rasterize_svg(svg_path: str, width: int, height: int) -> list[str]:
+ """Convert SVG file to list of half-block terminal rows (uncolored).
+
+ Args:
+ svg_path: Path to SVG file.
+ width: Target terminal width in columns.
+ height: Target terminal height in rows.
+
+ Returns:
+ List of strings, one per terminal row, containing block characters.
+ """
+ cache_key = (svg_path, width, height)
+ if cache_key in _cache:
+ return _cache[cache_key]
+
+ # SVG -> PNG in memory
+ png_bytes = cairosvg.svg2png(
+ url=svg_path,
+ output_width=width,
+ output_height=height * 2, # 2 pixel rows per terminal row
+ )
+
+ # PNG -> greyscale PIL image
+ # Composite RGBA onto white background so transparent areas become white (255)
+ # and drawn pixels retain their luminance values.
+ img_rgba = Image.open(BytesIO(png_bytes)).convert("RGBA")
+ img_rgba = img_rgba.resize((width, height * 2), Image.Resampling.LANCZOS)
+ background = Image.new("RGBA", img_rgba.size, (255, 255, 255, 255))
+ background.paste(img_rgba, mask=img_rgba.split()[3])
+ img = background.convert("L")
+
+ data = img.tobytes()
+ pix_w = width
+ pix_h = height * 2
+ # White (255) = empty space, dark (< threshold) = filled pixel
+ threshold = 128
+
+ # Half-block encode: walk pixel pairs
+ rows: list[str] = []
+ for y in range(0, pix_h, 2):
+ row: list[str] = []
+ for x in range(pix_w):
+ top = data[y * pix_w + x] < threshold
+ bot = data[(y + 1) * pix_w + x] < threshold if y + 1 < pix_h else False
+ if top and bot:
+ row.append("█")
+ elif top:
+ row.append("▀")
+ elif bot:
+ row.append("▄")
+ else:
+ row.append(" ")
+ rows.append("".join(row))
+
+ _cache[cache_key] = rows
+ return rows
+
+
+def clear_cache() -> None:
+ """Clear the rasterization cache (e.g., on terminal resize)."""
+ _cache.clear()
diff --git a/engine/figment_trigger.py b/engine/figment_trigger.py
new file mode 100644
index 0000000..d3aac9c
--- /dev/null
+++ b/engine/figment_trigger.py
@@ -0,0 +1,36 @@
+"""
+Figment trigger protocol and command types.
+
+Defines the extensible input abstraction for triggering figment displays
+from any control surface (ntfy, MQTT, serial, etc.).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Protocol
+
+
+class FigmentAction(Enum):
+ TRIGGER = "trigger"
+ SET_INTENSITY = "set_intensity"
+ SET_INTERVAL = "set_interval"
+ SET_COLOR = "set_color"
+ STOP = "stop"
+
+
+@dataclass
+class FigmentCommand:
+ action: FigmentAction
+ value: float | str | None = None
+
+
+class FigmentTrigger(Protocol):
+ """Protocol for figment trigger sources.
+
+ Any input source (ntfy, MQTT, serial) can implement this
+ to trigger and control figment displays.
+ """
+
+ def poll(self) -> FigmentCommand | None: ...
diff --git a/engine/layers.py b/engine/layers.py
index aa8fd59..a3cc0d5 100644
--- a/engine/layers.py
+++ b/engine/layers.py
@@ -57,9 +57,7 @@ def render_message_overlay(
else:
msg_rows = msg_cache[1]
- msg_rows = msg_gradient(
- msg_rows, (time.monotonic() * config.GRAD_SPEED) % 1.0
- )
+ msg_rows = msg_gradient(msg_rows, (time.monotonic() * config.GRAD_SPEED) % 1.0)
elapsed_s = int(time.monotonic() - m_ts)
remaining = max(0, config.MESSAGE_DISPLAY_SECS - elapsed_s)
@@ -258,3 +256,101 @@ def get_effect_chain() -> EffectChain | None:
if _effect_chain is None:
init_effects()
return _effect_chain
+
+
+def render_figment_overlay(
+ figment_state,
+ w: int,
+ h: int,
+) -> list[str]:
+ """Render figment overlay as ANSI cursor-positioning commands.
+
+ Args:
+ figment_state: FigmentState with phase, progress, rows, gradient, centering.
+ w: terminal width
+ h: terminal height
+
+ Returns:
+ List of ANSI strings to append to display buffer.
+ """
+ from engine.render import _color_codes_to_ansi
+
+ rows = figment_state.rows
+ if not rows:
+ return []
+
+ phase = figment_state.phase
+ progress = figment_state.progress
+ gradient = figment_state.gradient
+ center_row = figment_state.center_row
+ center_col = figment_state.center_col
+
+ cols = _color_codes_to_ansi(gradient)
+
+ # Build a list of non-space cell positions
+ cell_positions = []
+ for r_idx, row in enumerate(rows):
+ for c_idx, ch in enumerate(row):
+ if ch != " ":
+ cell_positions.append((r_idx, c_idx))
+
+ n_cells = len(cell_positions)
+ if n_cells == 0:
+ return []
+
+ # Use a deterministic seed so the reveal/dissolve pattern is stable per-figment
+ rng = random.Random(hash(tuple(rows[0][:10])) if rows[0] else 42)
+ shuffled = list(cell_positions)
+ rng.shuffle(shuffled)
+
+ # Phase-dependent visibility
+ from effects_plugins.figment import FigmentPhase
+
+ if phase == FigmentPhase.REVEAL:
+ visible_count = int(n_cells * progress)
+ visible = set(shuffled[:visible_count])
+ elif phase == FigmentPhase.HOLD:
+ visible = set(cell_positions)
+ # Strobe: dim some cells periodically
+ if int(progress * 20) % 3 == 0:
+ dim_count = int(n_cells * 0.3)
+ visible -= set(shuffled[:dim_count])
+ elif phase == FigmentPhase.DISSOLVE:
+ remaining_count = int(n_cells * (1.0 - progress))
+ visible = set(shuffled[:remaining_count])
+ else:
+ visible = set(cell_positions)
+
+ # Build overlay commands
+ overlay: list[str] = []
+ n_cols = len(cols)
+ max_x = max((len(r.rstrip()) for r in rows if r.strip()), default=1)
+
+ for r_idx, row in enumerate(rows):
+ scr_row = center_row + r_idx + 1 # 1-indexed
+ if scr_row < 1 or scr_row > h:
+ continue
+
+ line_buf: list[str] = []
+ has_content = False
+
+ for c_idx, ch in enumerate(row):
+ scr_col = center_col + c_idx + 1
+ if scr_col < 1 or scr_col > w:
+ continue
+
+ if ch != " " and (r_idx, c_idx) in visible:
+ # Apply gradient color
+ shifted = (c_idx / max(max_x - 1, 1)) % 1.0
+ idx = min(round(shifted * (n_cols - 1)), n_cols - 1)
+ line_buf.append(f"{cols[idx]}{ch}{RST}")
+ has_content = True
+ else:
+ line_buf.append(" ")
+
+ if has_content:
+ line_str = "".join(line_buf).rstrip()
+ if line_str.strip():
+ overlay.append(f"\033[{scr_row};{center_col + 1}H{line_str}{RST}")
+
+ return overlay
diff --git a/engine/scroll.py b/engine/scroll.py
index d13408b..1bb90a2 100644
--- a/engine/scroll.py
+++ b/engine/scroll.py
@@ -18,6 +18,7 @@ from engine.frame import calculate_scroll_step
from engine.layers import (
apply_glitch,
process_effects,
+ render_figment_overlay,
render_firehose,
render_message_overlay,
render_ticker_zone,
@@ -53,6 +54,18 @@ def stream(items, ntfy_poller, mic_monitor, display: Display | None = None):
msg_cache = (None, None)
frame_number = 0
+ # Figment overlay (optional — requires cairosvg)
+ figment = None
+ if config.FIGMENT:
+ try:
+ from effects_plugins.figment import FigmentEffect
+
+ figment = FigmentEffect()
+ figment.config.enabled = True
+ figment.config.params["interval_secs"] = config.FIGMENT_INTERVAL
+ except (ImportError, OSError):
+ pass
+
while True:
if queued >= config.HEADLINE_LIMIT and not active:
break
@@ -123,6 +136,13 @@ def stream(items, ntfy_poller, mic_monitor, display: Display | None = None):
firehose_buf = render_firehose(items, w, fh, h)
buf.extend(firehose_buf)
+ # Figment overlay (between effects and ntfy message)
+ if figment and figment.config.enabled:
+ figment_state = figment.get_figment_state(frame_number, w, h)
+ if figment_state is not None:
+ figment_buf = render_figment_overlay(figment_state, w, h)
+ buf.extend(figment_buf)
+
if msg_overlay:
buf.extend(msg_overlay)
diff --git a/figments/animal-head-symbol-of-mexico-antique-cultures-svgrepo-com.svg b/figments/animal-head-symbol-of-mexico-antique-cultures-svgrepo-com.svg
new file mode 100644
index 0000000..264eb08
--- /dev/null
+++ b/figments/animal-head-symbol-of-mexico-antique-cultures-svgrepo-com.svg
@@ -0,0 +1,32 @@
+
+
+
+
\ No newline at end of file
diff --git a/figments/mayan-mask-of-mexico-svgrepo-com.svg b/figments/mayan-mask-of-mexico-svgrepo-com.svg
new file mode 100644
index 0000000..75fca60
--- /dev/null
+++ b/figments/mayan-mask-of-mexico-svgrepo-com.svg
@@ -0,0 +1,60 @@
+
+
+
+
\ No newline at end of file
diff --git a/figments/mayan-symbol-of-mexico-svgrepo-com.svg b/figments/mayan-symbol-of-mexico-svgrepo-com.svg
new file mode 100644
index 0000000..a396536
--- /dev/null
+++ b/figments/mayan-symbol-of-mexico-svgrepo-com.svg
@@ -0,0 +1,110 @@
+
+
+
+
\ No newline at end of file
diff --git a/mise.toml b/mise.toml
index 32f7c59..5129e28 100644
--- a/mise.toml
+++ b/mise.toml
@@ -1,3 +1,7 @@
+[env]
+_.path = ["/opt/homebrew/lib"]
+DYLD_LIBRARY_PATH = "/opt/homebrew/lib"
+
[tools]
python = "3.12"
hk = "latest"
diff --git a/pyproject.toml b/pyproject.toml
index f52a05a..84f1b52 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -30,6 +30,9 @@ mic = [
"sounddevice>=0.4.0",
"numpy>=1.24.0",
]
+figment = [
+ "cairosvg>=2.7.0",
+]
dev = [
"pytest>=8.0.0",
"pytest-cov>=4.1.0",
diff --git a/tests/fixtures/test.svg b/tests/fixtures/test.svg
new file mode 100644
index 0000000..f35f4b3
--- /dev/null
+++ b/tests/fixtures/test.svg
@@ -0,0 +1,3 @@
+
diff --git a/tests/test_figment.py b/tests/test_figment.py
new file mode 100644
index 0000000..6774b1a
--- /dev/null
+++ b/tests/test_figment.py
@@ -0,0 +1,151 @@
+"""Tests for the FigmentEffect plugin."""
+
+import os
+from enum import Enum
+
+import pytest
+
+pytest.importorskip("cairosvg", reason="cairosvg requires system Cairo library")
+
+from effects_plugins.figment import FigmentEffect, FigmentPhase, FigmentState
+from engine.effects.types import EffectConfig, EffectContext
+
+FIXTURE_SVG = os.path.join(os.path.dirname(__file__), "fixtures", "test.svg")
+FIGMENTS_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
+
+
+class TestFigmentPhase:
+ def test_is_enum(self):
+ assert issubclass(FigmentPhase, Enum)
+
+ def test_has_all_phases(self):
+ assert hasattr(FigmentPhase, "REVEAL")
+ assert hasattr(FigmentPhase, "HOLD")
+ assert hasattr(FigmentPhase, "DISSOLVE")
+
+
+class TestFigmentState:
+ def test_creation(self):
+ state = FigmentState(
+ phase=FigmentPhase.REVEAL,
+ progress=0.5,
+ rows=["█▀▄", " █ "],
+ gradient=[46, 40, 34, 28, 22, 22, 34, 40, 46, 82, 118, 231],
+ center_row=5,
+ center_col=10,
+ )
+ assert state.phase == FigmentPhase.REVEAL
+ assert state.progress == 0.5
+ assert len(state.rows) == 2
+
+
+class TestFigmentEffectInit:
+ def test_name(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ assert effect.name == "figment"
+
+ def test_default_config(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ assert effect.config.enabled is False
+ assert effect.config.intensity == 1.0
+ assert effect.config.params["interval_secs"] == 60
+ assert effect.config.params["display_secs"] == 4.5
+
+ def test_process_is_noop(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ buf = ["line1", "line2"]
+ ctx = EffectContext(
+ terminal_width=80,
+ terminal_height=24,
+ scroll_cam=0,
+ ticker_height=20,
+ )
+ result = effect.process(buf, ctx)
+ assert result == buf
+ assert result is buf
+
+ def test_configure(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ new_cfg = EffectConfig(enabled=True, intensity=0.5)
+ effect.configure(new_cfg)
+ assert effect.config.enabled is True
+ assert effect.config.intensity == 0.5
+
+
+class TestFigmentStateMachine:
+ def test_idle_initially(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ state = effect.get_figment_state(0, 80, 24)
+ # Timer hasn't fired yet, should be None (idle)
+ assert state is None
+
+ def test_trigger_starts_reveal(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ effect.trigger(80, 24)
+ state = effect.get_figment_state(1, 80, 24)
+ assert state is not None
+ assert state.phase == FigmentPhase.REVEAL
+
+ def test_full_cycle(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ effect.config.params["display_secs"] = 0.15 # 3 phases x 0.05s
+
+ effect.trigger(40, 20)
+
+ # Advance through reveal (30 frames at 0.05s = 1.5s, but we shrunk it)
+ # With display_secs=0.15, each phase is 0.05s = 1 frame
+ state = effect.get_figment_state(1, 40, 20)
+ assert state is not None
+ assert state.phase == FigmentPhase.REVEAL
+
+ # Advance enough frames to get through all phases
+ for frame in range(2, 100):
+ state = effect.get_figment_state(frame, 40, 20)
+ if state is None:
+ break
+
+ # Should have completed the full cycle back to idle
+ assert state is None
+
+ def test_timer_fires_at_interval(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = True
+ effect.config.params["interval_secs"] = 0.1 # 2 frames at 20fps
+
+ # Frame 0: idle
+ state = effect.get_figment_state(0, 40, 20)
+ assert state is None
+
+ # Advance past interval (0.1s = 2 frames)
+ state = effect.get_figment_state(1, 40, 20)
+ state = effect.get_figment_state(2, 40, 20)
+ state = effect.get_figment_state(3, 40, 20)
+ # Timer should have fired by now
+ assert state is not None
+
+
+class TestFigmentEdgeCases:
+ def test_empty_figment_dir(self, tmp_path):
+ effect = FigmentEffect(figment_dir=str(tmp_path))
+ effect.config.enabled = True
+ effect.trigger(40, 20)
+ state = effect.get_figment_state(1, 40, 20)
+ # No SVGs available — should stay idle
+ assert state is None
+
+ def test_missing_figment_dir(self):
+ effect = FigmentEffect(figment_dir="/nonexistent/path")
+ effect.config.enabled = True
+ effect.trigger(40, 20)
+ state = effect.get_figment_state(1, 40, 20)
+ assert state is None
+
+ def test_disabled_ignores_trigger(self):
+ effect = FigmentEffect(figment_dir=FIGMENTS_DIR)
+ effect.config.enabled = False
+ effect.trigger(80, 24)
+ state = effect.get_figment_state(1, 80, 24)
+ assert state is None
diff --git a/tests/test_figment_overlay.py b/tests/test_figment_overlay.py
new file mode 100644
index 0000000..99152be
--- /dev/null
+++ b/tests/test_figment_overlay.py
@@ -0,0 +1,64 @@
+"""Tests for render_figment_overlay in engine.layers."""
+
+import pytest
+
+pytest.importorskip("cairosvg", reason="cairosvg requires system Cairo library")
+
+from effects_plugins.figment import FigmentPhase, FigmentState
+from engine.layers import render_figment_overlay
+
+
+def _make_state(phase=FigmentPhase.HOLD, progress=0.5):
+ return FigmentState(
+ phase=phase,
+ progress=progress,
+ rows=["█▀▄ █", " ▄█▀ ", "█ █"],
+ gradient=[46, 40, 34, 28, 22, 22, 34, 40, 46, 82, 118, 231],
+ center_row=10,
+ center_col=37,
+ )
+
+
+class TestRenderFigmentOverlay:
+ def test_returns_list_of_strings(self):
+ state = _make_state()
+ result = render_figment_overlay(state, 80, 24)
+ assert isinstance(result, list)
+ assert all(isinstance(s, str) for s in result)
+
+ def test_contains_ansi_positioning(self):
+ state = _make_state()
+ result = render_figment_overlay(state, 80, 24)
+ # Should contain cursor positioning escape codes
+ assert any("\033[" in s for s in result)
+
+ def test_reveal_phase_partial(self):
+ state = _make_state(phase=FigmentPhase.REVEAL, progress=0.0)
+ result = render_figment_overlay(state, 80, 24)
+ # At progress 0.0, very few cells should be visible
+ # Result should still be a valid list
+ assert isinstance(result, list)
+
+ def test_hold_phase_full(self):
+ state = _make_state(phase=FigmentPhase.HOLD, progress=0.5)
+ result = render_figment_overlay(state, 80, 24)
+ # During hold, content should be present
+ assert len(result) > 0
+
+ def test_dissolve_phase(self):
+ state = _make_state(phase=FigmentPhase.DISSOLVE, progress=0.9)
+ result = render_figment_overlay(state, 80, 24)
+ # At high dissolve progress, most cells are gone
+ assert isinstance(result, list)
+
+ def test_empty_rows(self):
+ state = FigmentState(
+ phase=FigmentPhase.HOLD,
+ progress=0.5,
+ rows=[],
+ gradient=[46] * 12,
+ center_row=0,
+ center_col=0,
+ )
+ result = render_figment_overlay(state, 80, 24)
+ assert result == []
diff --git a/tests/test_figment_render.py b/tests/test_figment_render.py
new file mode 100644
index 0000000..fffb62f
--- /dev/null
+++ b/tests/test_figment_render.py
@@ -0,0 +1,52 @@
+"""Tests for engine.figment_render module."""
+
+import os
+
+import pytest
+
+pytest.importorskip("cairosvg", reason="cairosvg requires system Cairo library")
+
+from engine.figment_render import rasterize_svg
+
+FIXTURE_SVG = os.path.join(os.path.dirname(__file__), "fixtures", "test.svg")
+
+
+class TestRasterizeSvg:
+ def test_returns_list_of_strings(self):
+ rows = rasterize_svg(FIXTURE_SVG, 40, 20)
+ assert isinstance(rows, list)
+ assert all(isinstance(r, str) for r in rows)
+
+ def test_output_height_matches_terminal_height(self):
+ rows = rasterize_svg(FIXTURE_SVG, 40, 20)
+ assert len(rows) == 20
+
+ def test_output_contains_block_characters(self):
+ rows = rasterize_svg(FIXTURE_SVG, 40, 20)
+ all_chars = "".join(rows)
+ block_chars = {"█", "▀", "▄"}
+ assert any(ch in all_chars for ch in block_chars)
+
+ def test_different_sizes_produce_different_output(self):
+ rows_small = rasterize_svg(FIXTURE_SVG, 20, 10)
+ rows_large = rasterize_svg(FIXTURE_SVG, 80, 40)
+ assert len(rows_small) == 10
+ assert len(rows_large) == 40
+
+ def test_nonexistent_file_raises(self):
+ import pytest
+
+ with pytest.raises((FileNotFoundError, OSError)):
+ rasterize_svg("/nonexistent/file.svg", 40, 20)
+
+
+class TestRasterizeCache:
+ def test_cache_returns_same_result(self):
+ rows1 = rasterize_svg(FIXTURE_SVG, 40, 20)
+ rows2 = rasterize_svg(FIXTURE_SVG, 40, 20)
+ assert rows1 == rows2
+
+ def test_cache_invalidated_by_size_change(self):
+ rows1 = rasterize_svg(FIXTURE_SVG, 40, 20)
+ rows2 = rasterize_svg(FIXTURE_SVG, 60, 30)
+ assert len(rows1) != len(rows2)
diff --git a/tests/test_figment_trigger.py b/tests/test_figment_trigger.py
new file mode 100644
index 0000000..989a0bb
--- /dev/null
+++ b/tests/test_figment_trigger.py
@@ -0,0 +1,40 @@
+"""Tests for engine.figment_trigger module."""
+
+from enum import Enum
+
+from engine.figment_trigger import FigmentAction, FigmentCommand
+
+
+class TestFigmentAction:
+ def test_is_enum(self):
+ assert issubclass(FigmentAction, Enum)
+
+ def test_has_trigger(self):
+ assert FigmentAction.TRIGGER.value == "trigger"
+
+ def test_has_set_intensity(self):
+ assert FigmentAction.SET_INTENSITY.value == "set_intensity"
+
+ def test_has_set_interval(self):
+ assert FigmentAction.SET_INTERVAL.value == "set_interval"
+
+ def test_has_set_color(self):
+ assert FigmentAction.SET_COLOR.value == "set_color"
+
+ def test_has_stop(self):
+ assert FigmentAction.STOP.value == "stop"
+
+
+class TestFigmentCommand:
+ def test_trigger_command(self):
+ cmd = FigmentCommand(action=FigmentAction.TRIGGER)
+ assert cmd.action == FigmentAction.TRIGGER
+ assert cmd.value is None
+
+ def test_set_intensity_command(self):
+ cmd = FigmentCommand(action=FigmentAction.SET_INTENSITY, value=0.8)
+ assert cmd.value == 0.8
+
+ def test_set_color_command(self):
+ cmd = FigmentCommand(action=FigmentAction.SET_COLOR, value="orange")
+ assert cmd.value == "orange"