art-dag/test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add streaming video compositor with sexp interpreter

Browse Source 
- New streaming/ module for real-time video processing:
  - compositor.py: Main streaming compositor with cycle-crossfade
  - sexp_executor.py: Executes compiled sexp recipes in real-time
  - sexp_interp.py: Full S-expression interpreter for SLICE_ON Lambda
  - recipe_adapter.py: Bridges recipes to streaming compositor
  - sources.py: Video source with ffmpeg streaming
  - audio.py: Real-time audio analysis (energy, beats)
  - output.py: Preview (mpv) and file output with audio muxing

- New templates/:
  - cycle-crossfade.sexp: Smooth zoom-based video cycling
  - process-pair.sexp: Dual-clip processing with effects

- Key features:
  - Videos cycle in input-videos order (not definition order)
  - Cumulative whole-spin rotation
  - Zero-weight sources skip processing
  - Live audio-reactive effects

- New effects: blend_multi for weighted layer compositing
- Updated primitives and interpreter for streaming compatibility

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-29 01:27:39 +00:00
parent 17e3e23f06
commit d241e2a663
31 changed files with 5143 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
sexp_effects/primitives.py
View File

@@ -1444,42 +1444,80 @@ CHAR_ALPHABETS = {
"digits": " 0123456789",
}
# Global atlas cache
# Global atlas cache: keyed on (frozenset(chars), cell_size) ->
# (atlas_array, char_to_idx) where atlas_array is (N, cell_size, cell_size) uint8.
_char_atlas_cache = {}
_CHAR_ATLAS_CACHE_MAX = 32
def _get_char_atlas(alphabet: str, cell_size: int) -> dict:
"""Get or create character atlas for alphabet."""
cache_key = f"{alphabet}_{cell_size}"
if cache_key in _char_atlas_cache:
return _char_atlas_cache[cache_key]
"""Get or create character atlas for alphabet (legacy dict version)."""
atlas_arr, char_to_idx = _get_render_atlas(alphabet, cell_size)
# Build legacy dict from array
idx_to_char = {v: k for k, v in char_to_idx.items()}
return {idx_to_char[i]: atlas_arr[i] for i in range(len(atlas_arr))}
def _get_render_atlas(unique_chars_or_alphabet, cell_size: int):
"""Get or build a stacked numpy atlas for vectorised rendering.
Args:
unique_chars_or_alphabet: Either an alphabet name (str looked up in
CHAR_ALPHABETS), a literal character string, or a set/frozenset
of characters.
cell_size: Pixel size of each cell.
Returns:
(atlas_array, char_to_idx) where
atlas_array: (num_chars, cell_size, cell_size) uint8 masks
char_to_idx: dict mapping character -> index in atlas_array
"""
if isinstance(unique_chars_or_alphabet, (set, frozenset)):
chars_tuple = tuple(sorted(unique_chars_or_alphabet))
else:
resolved = CHAR_ALPHABETS.get(unique_chars_or_alphabet, unique_chars_or_alphabet)
chars_tuple = tuple(resolved)
cache_key = (chars_tuple, cell_size)
cached = _char_atlas_cache.get(cache_key)
if cached is not None:
return cached
chars = CHAR_ALPHABETS.get(alphabet, alphabet) # Use as literal if not found
font = cv2.FONT_HERSHEY_SIMPLEX
font_scale = cell_size / 20.0
thickness = max(1, int(cell_size / 10))
atlas = {}
for char in chars:
char_img = np.zeros((cell_size, cell_size), dtype=np.uint8)
if char != ' ':
n = len(chars_tuple)
atlas = np.zeros((n, cell_size, cell_size), dtype=np.uint8)
char_to_idx = {}
for i, char in enumerate(chars_tuple):
char_to_idx[char] = i
if char and char != ' ':
try:
(text_w, text_h), baseline = cv2.getTextSize(char, font, font_scale, thickness)
(text_w, text_h), _ = cv2.getTextSize(char, font, font_scale, thickness)
text_x = max(0, (cell_size - text_w) // 2)
text_y = (cell_size + text_h) // 2
cv2.putText(char_img, char, (text_x, text_y), font, font_scale, 255, thickness, cv2.LINE_AA)
except:
cv2.putText(atlas[i], char, (text_x, text_y),
font, font_scale, 255, thickness, cv2.LINE_AA)
except Exception:
pass
atlas[char] = char_img
_char_atlas_cache[cache_key] = atlas
return atlas
# Evict oldest entry if cache is full
if len(_char_atlas_cache) >= _CHAR_ATLAS_CACHE_MAX:
_char_atlas_cache.pop(next(iter(_char_atlas_cache)))
_char_atlas_cache[cache_key] = (atlas, char_to_idx)
return atlas, char_to_idx
def prim_cell_sample(img: np.ndarray, cell_size: int) -> Tuple[np.ndarray, np.ndarray]:
"""
Sample image into cell grid, returning average colors and luminances.
Uses cv2.resize with INTER_AREA (pixel-area averaging) which is
~25x faster than numpy reshape+mean for block downsampling.
Args:
img: source image
cell_size: size of each cell in pixels
@@ -1497,13 +1535,10 @@ def prim_cell_sample(img: np.ndarray, cell_size: int) -> Tuple[np.ndarray, np.nd
return (np.zeros((1, 1, 3), dtype=np.uint8),
np.zeros((1, 1), dtype=np.float32))
# Crop to grid
# Crop to exact grid then block-average via cv2 area interpolation.
grid_h, grid_w = rows * cell_size, cols * cell_size
cropped = img[:grid_h, :grid_w]
# Reshape and average
reshaped = cropped.reshape(rows, cell_size, cols, cell_size, 3)
colors = reshaped.mean(axis=(1, 3)).astype(np.uint8)
colors = cv2.resize(cropped, (cols, rows), interpolation=cv2.INTER_AREA)
# Compute luminance
luminances = (0.299 * colors[:, :, 0] +
@@ -1628,16 +1663,11 @@ def prim_luminance_to_chars(luminances: np.ndarray, alphabet: str, contrast: flo
indices = ((lum / 255) * (num_chars - 1)).astype(np.int32)
indices = np.clip(indices, 0, num_chars - 1)
# Convert to character array
rows, cols = indices.shape
result = []
for r in range(rows):
row = []
for c in range(cols):
row.append(chars[indices[r, c]])
result.append(row)
# Vectorised conversion via numpy char array lookup
chars_arr = np.array(list(chars))
char_grid = chars_arr[indices.ravel()].reshape(indices.shape)
return result
return char_grid.tolist()
def prim_render_char_grid(img: np.ndarray, chars: List[List[str]], colors: np.ndarray,
@@ -1647,6 +1677,10 @@ def prim_render_char_grid(img: np.ndarray, chars: List[List[str]], colors: np.nd
"""
Render a grid of characters onto an image.
Uses vectorised numpy operations instead of per-cell Python loops:
the character atlas is looked up via fancy indexing and the full
mask + colour image are assembled in bulk.
Args:
img: source image (for dimensions)
chars: 2D list of single characters
@@ -1664,12 +1698,11 @@ def prim_render_char_grid(img: np.ndarray, chars: List[List[str]], colors: np.nd
# Parse background_color
if isinstance(background_color, (list, tuple)):
# Legacy: accept RGB list
bg_color = tuple(int(c) for c in background_color[:3])
else:
bg_color = parse_color(background_color)
if bg_color is None:
bg_color = (0, 0, 0) # Default to black
bg_color = (0, 0, 0)
# Handle invert_colors - swap fg and bg
if invert_colors and fg_color is not None:
@@ -1686,58 +1719,66 @@ def prim_render_char_grid(img: np.ndarray, chars: List[List[str]], colors: np.nd
bg = list(bg_color)
result = np.full((h, w, 3), bg, dtype=np.uint8)
# Collect all unique characters to build minimal atlas
# --- Build atlas & index grid ---
unique_chars = set()
for row in chars:
for ch in row:
unique_chars.add(ch)
# Build atlas for unique chars
font = cv2.FONT_HERSHEY_SIMPLEX
font_scale = cell_size / 20.0
thickness = max(1, int(cell_size / 10))
atlas, char_to_idx = _get_render_atlas(unique_chars, cell_size)
atlas = {}
for char in unique_chars:
char_img = np.zeros((cell_size, cell_size), dtype=np.uint8)
if char and char != ' ':
try:
(text_w, text_h), _ = cv2.getTextSize(char, font, font_scale, thickness)
text_x = max(0, (cell_size - text_w) // 2)
text_y = (cell_size + text_h) // 2
cv2.putText(char_img, char, (text_x, text_y), font, font_scale, 255, thickness, cv2.LINE_AA)
except:
pass
atlas[char] = char_img
# Convert 2D char list to index array using ordinal lookup table
# (avoids per-cell Python dict lookup).
space_idx = char_to_idx.get(' ', 0)
max_ord = max(ord(ch) for ch in char_to_idx) + 1
ord_lookup = np.full(max_ord, space_idx, dtype=np.int32)
for ch, idx in char_to_idx.items():
if ch:
ord_lookup[ord(ch)] = idx
# Render characters
for r in range(rows):
for c in range(cols):
char = chars[r][c]
if not char or char == ' ':
continue
flat = [ch for row in chars for ch in row]
ords = np.frombuffer(np.array(flat, dtype='U1'), dtype=np.uint32)
char_indices = ord_lookup[ords].reshape(rows, cols)
y1, x1 = r * cell_size, c * cell_size
char_mask = atlas.get(char)
# --- Vectorised mask assembly ---
# atlas[char_indices] -> (rows, cols, cell_size, cell_size)
# Transpose to (rows, cell_size, cols, cell_size) then reshape to full image.
all_masks = atlas[char_indices]
full_mask = all_masks.transpose(0, 2, 1, 3).reshape(h, w)
if char_mask is None:
continue
# Expand per-cell colours to per-pixel (only when needed).
need_color_full = (color_mode in ("color", "invert")
or (fg_color is None and color_mode != "mono"))
if fg_color is not None:
# Use fixed color (named color or hex value)
color = np.array(fg_color, dtype=np.uint8)
elif color_mode == "mono":
color = np.array([255, 255, 255], dtype=np.uint8)
elif color_mode == "invert":
result[y1:y1+cell_size, x1:x1+cell_size] = colors[r, c]
color = np.array([0, 0, 0], dtype=np.uint8)
else: # color
color = colors[r, c]
if need_color_full:
color_full = np.repeat(
np.repeat(colors[:rows, :cols], cell_size, axis=0),
cell_size, axis=1)
mask = char_mask > 0
result[y1:y1+cell_size, x1:x1+cell_size][mask] = color
# --- Vectorised colour composite ---
# Use element-wise multiply/np.where instead of boolean-indexed scatter
# for much better memory access patterns.
mask_u8 = (full_mask > 0).astype(np.uint8)[:, :, np.newaxis]
if color_mode == "invert":
# Background is source colour; characters are black.
# result = color_full * (1 - mask)
result = color_full * (1 - mask_u8)
elif fg_color is not None:
# Fixed foreground colour on background.
fg = np.array(fg_color, dtype=np.uint8)
bg_arr = np.array(bg, dtype=np.uint8)
result = np.where(mask_u8, fg, bg_arr).astype(np.uint8)
elif color_mode == "mono":
bg_arr = np.array(bg, dtype=np.uint8)
result = np.where(mask_u8, np.uint8(255), bg_arr).astype(np.uint8)
else:
# "color" mode – each cell uses its source colour on bg.
if bg == [0, 0, 0]:
result = color_full * mask_u8
else:
bg_arr = np.array(bg, dtype=np.uint8)
result = np.where(mask_u8, color_full, bg_arr).astype(np.uint8)
# Resize to match original if needed
orig_h, orig_w = img.shape[:2]