forked from genewildish/Mainline
David Gwilliam
238bac1bb2
- Implements pipeline hot-rebuild with state preservation (issue #43) - Adds auto-injection of MVP stages for missing capabilities - Adds radial camera mode for polar coordinate scanning - Adds afterimage and motionblur effects using framebuffer history - Adds comprehensive acceptance tests for camera modes and pipeline rebuild - Updates presets.toml with new effect configurations Related to: #35 (Pipeline Mutation API epic) Closes: #43, #44, #45
120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
"""Motion blur effect using frame history."""
|
|
|
|
from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
|
|
|
|
|
class MotionBlurEffect(EffectPlugin):
|
|
"""Apply motion blur by blending current frame with previous frames.
|
|
|
|
This effect requires a FrameBufferStage to be present in the pipeline.
|
|
The framebuffer provides frame history which is blended with the current
|
|
frame based on intensity.
|
|
|
|
Attributes:
|
|
name: "motionblur"
|
|
config: EffectConfig with intensity parameter (0.0-1.0)
|
|
param_bindings: Optional sensor bindings for intensity modulation
|
|
|
|
Example:
|
|
>>> effect = MotionBlurEffect()
|
|
>>> effect.configure(EffectConfig(intensity=0.5))
|
|
>>> result = effect.process(buffer, ctx)
|
|
"""
|
|
|
|
name = "motionblur"
|
|
config: EffectConfig = EffectConfig(enabled=True, intensity=0.5)
|
|
param_bindings: dict[str, dict[str, str | float]] = {}
|
|
supports_partial_updates = False
|
|
|
|
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
|
"""Apply motion blur by blending with previous frames.
|
|
|
|
Args:
|
|
buf: Current text buffer (list of strings)
|
|
ctx: Effect context with access to framebuffer history
|
|
|
|
Returns:
|
|
Blended buffer with motion blur effect applied
|
|
"""
|
|
if not buf:
|
|
return buf
|
|
|
|
# Get framebuffer history from context
|
|
# We'll look for the first available framebuffer history
|
|
history = None
|
|
|
|
for key in ctx.state:
|
|
if key.startswith("framebuffer.") and key.endswith(".history"):
|
|
history = ctx.state[key]
|
|
break
|
|
|
|
if not history:
|
|
# No framebuffer available, return unchanged
|
|
return buf
|
|
|
|
# Get intensity from config
|
|
intensity = self.config.params.get("intensity", self.config.intensity)
|
|
intensity = max(0.0, min(1.0, intensity))
|
|
|
|
if intensity <= 0.0:
|
|
return buf
|
|
|
|
# Get decay factor (how quickly older frames fade)
|
|
decay = self.config.params.get("decay", 0.7)
|
|
|
|
# Build output buffer
|
|
result = []
|
|
viewport_height = ctx.terminal_height - ctx.ticker_height
|
|
|
|
# Determine how many frames to blend (up to history depth)
|
|
max_frames = min(len(history), 5) # Cap at 5 frames for performance
|
|
|
|
for row in range(len(buf)):
|
|
if row >= viewport_height:
|
|
# Beyond viewport, just copy
|
|
result.append(buf[row])
|
|
continue
|
|
|
|
# Start with current frame
|
|
blended = buf[row]
|
|
|
|
# Blend with historical frames
|
|
weight_sum = 1.0
|
|
if max_frames > 0 and intensity > 0:
|
|
for i in range(max_frames):
|
|
frame_weight = intensity * (decay**i)
|
|
if frame_weight < 0.01: # Skip negligible weights
|
|
break
|
|
|
|
hist_row = history[i][row] if row < len(history[i]) else ""
|
|
# Simple string blending: we'll concatenate with space
|
|
# For a proper effect, we'd need to blend ANSI colors
|
|
# This is a simplified version that just adds the frames
|
|
blended = self._blend_strings(blended, hist_row, frame_weight)
|
|
weight_sum += frame_weight
|
|
|
|
result.append(blended)
|
|
|
|
return result
|
|
|
|
def _blend_strings(self, current: str, historical: str, weight: float) -> str:
|
|
"""Blend two strings with given weight.
|
|
|
|
This is a simplified blending that works with ANSI codes.
|
|
For proper blending we'd need to parse colors, but for now
|
|
we use a heuristic: if strings are identical, return one.
|
|
If they differ, we alternate or concatenate based on weight.
|
|
"""
|
|
if current == historical:
|
|
return current
|
|
|
|
# If weight is high, show current; if low, show historical
|
|
if weight > 0.5:
|
|
return current
|
|
else:
|
|
return historical
|
|
|
|
def configure(self, config: EffectConfig) -> None:
|
|
"""Configure the effect."""
|
|
self.config = config
|