8-task plan covering SVG rasterization, overlay rendering, FigmentEffect plugin, trigger protocol, and scroll loop integration. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1111 lines
31 KiB
Markdown
1111 lines
31 KiB
Markdown
# 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
|
|
<svg xmlns="http://www.w3.org/2000/svg" width="100" height="100" viewBox="0 0 100 100">
|
|
<rect x="10" y="10" width="80" height="80" fill="black"/>
|
|
</svg>
|
|
```
|
|
|
|
- [ ] **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.)
|