David Gwilliam
e4b143ff36
This commit implements the Sideline/Mainline split with a clean plugin architecture: ## Core Changes ### Sideline Framework (New Directory) - Created directory containing the reusable pipeline framework - Moved pipeline core, controllers, adapters, and registry to - Moved display system to - Moved effects system to - Created plugin system with security and compatibility management in - Created preset pack system with ASCII art encoding in - Added default font (Corptic) to - Added terminal ANSI constants to ### Mainline Application (Updated) - Created for Mainline stage component registration - Updated to register Mainline stages at startup - Updated as a compatibility shim re-exporting from sideline ### Terminology Consistency - : Base class for all pipeline components (sources, effects, displays, cameras) - : Base class for distributable plugin packages (was ) - : Base class for visual effects (was ) - Backward compatibility aliases maintained for existing code ## Key Features - Plugin discovery via entry points and explicit registration - Security permissions system for plugins - Compatibility management with semantic version constraints - Preset pack system for distributable configurations - Default font bundled with Sideline (Corptic.otf) ## Testing - Updated tests to register Mainline stages before discovery - All StageRegistry tests passing Note: This is a major refactoring that separates the framework (Sideline) from the application (Mainline), enabling Sideline to be used by other applications.
294 lines
8.7 KiB
Python
294 lines
8.7 KiB
Python
"""Adapters for transform stages (viewport, font, image, canvas)."""
|
|
|
|
from typing import Any
|
|
|
|
import sideline.render
|
|
from sideline.data_sources import SourceItem
|
|
from sideline.pipeline.core import DataType, PipelineContext, Stage
|
|
|
|
|
|
def estimate_simple_height(text: str, width: int) -> int:
|
|
"""Estimate height in terminal rows using simple word wrap.
|
|
|
|
Uses conservative estimation suitable for headlines.
|
|
Each wrapped line is approximately 6 terminal rows (big block rendering).
|
|
"""
|
|
words = text.split()
|
|
if not words:
|
|
return 6
|
|
|
|
lines = 1
|
|
current_len = 0
|
|
for word in words:
|
|
word_len = len(word)
|
|
if current_len + word_len + 1 > width - 4: # -4 for margins
|
|
lines += 1
|
|
current_len = word_len
|
|
else:
|
|
current_len += word_len + 1
|
|
|
|
return lines * 6 # 6 rows per line for big block rendering
|
|
|
|
|
|
class ViewportFilterStage(Stage):
|
|
"""Filter items to viewport height based on rendered height."""
|
|
|
|
def __init__(self, name: str = "viewport-filter"):
|
|
self.name = name
|
|
self.category = "render"
|
|
self.optional = True
|
|
self._layout: list[int] = []
|
|
|
|
@property
|
|
def stage_type(self) -> str:
|
|
return "render"
|
|
|
|
@property
|
|
def capabilities(self) -> set[str]:
|
|
return {"source.filtered"}
|
|
|
|
@property
|
|
def dependencies(self) -> set[str]:
|
|
# Always requires camera.state for viewport filtering
|
|
# CameraUpdateStage provides this (auto-injected if missing)
|
|
return {"source", "camera.state"}
|
|
|
|
@property
|
|
def inlet_types(self) -> set:
|
|
return {DataType.SOURCE_ITEMS}
|
|
|
|
@property
|
|
def outlet_types(self) -> set:
|
|
return {DataType.SOURCE_ITEMS}
|
|
|
|
def process(self, data: Any, ctx: PipelineContext) -> Any:
|
|
"""Filter items to viewport height based on rendered height."""
|
|
if data is None:
|
|
return data
|
|
|
|
if not isinstance(data, list):
|
|
return data
|
|
|
|
if not data:
|
|
return []
|
|
|
|
# Get viewport parameters from context
|
|
viewport_height = ctx.params.viewport_height if ctx.params else 24
|
|
viewport_width = ctx.params.viewport_width if ctx.params else 80
|
|
camera_y = ctx.get("camera_y", 0)
|
|
|
|
# Estimate height for each item and cache layout
|
|
self._layout = []
|
|
cumulative_heights = []
|
|
current_height = 0
|
|
|
|
for item in data:
|
|
title = item.content if isinstance(item, SourceItem) else str(item)
|
|
# Use simple height estimation (not PIL-based)
|
|
estimated_height = estimate_simple_height(title, viewport_width)
|
|
self._layout.append(estimated_height)
|
|
current_height += estimated_height
|
|
cumulative_heights.append(current_height)
|
|
|
|
# Find visible range based on camera_y and viewport_height
|
|
# camera_y is the scroll offset (how many rows are scrolled up)
|
|
start_y = camera_y
|
|
end_y = camera_y + viewport_height
|
|
|
|
# Find start index (first item that intersects with visible range)
|
|
start_idx = 0
|
|
start_item_y = 0 # Y position where the first visible item starts
|
|
for i, total_h in enumerate(cumulative_heights):
|
|
if total_h > start_y:
|
|
start_idx = i
|
|
# Calculate the Y position of the start of this item
|
|
if i > 0:
|
|
start_item_y = cumulative_heights[i - 1]
|
|
break
|
|
|
|
# Find end index (first item that extends beyond visible range)
|
|
end_idx = len(data)
|
|
for i, total_h in enumerate(cumulative_heights):
|
|
if total_h >= end_y:
|
|
end_idx = i + 1
|
|
break
|
|
|
|
# Adjust camera_y for the filtered buffer
|
|
# The filtered buffer starts at row 0, but the camera position
|
|
# needs to be relative to where the first visible item starts
|
|
filtered_camera_y = camera_y - start_item_y
|
|
|
|
# Update context with the filtered camera position
|
|
# This ensures CameraStage can correctly slice the filtered buffer
|
|
ctx.set_state("camera_y", filtered_camera_y)
|
|
ctx.set_state("camera_x", ctx.get("camera_x", 0)) # Keep camera_x unchanged
|
|
|
|
# Return visible items
|
|
return data[start_idx:end_idx]
|
|
|
|
|
|
class FontStage(Stage):
|
|
"""Render items using font."""
|
|
|
|
def __init__(self, name: str = "font"):
|
|
self.name = name
|
|
self.category = "render"
|
|
self.optional = False
|
|
|
|
@property
|
|
def stage_type(self) -> str:
|
|
return "render"
|
|
|
|
@property
|
|
def capabilities(self) -> set[str]:
|
|
return {"render.output"}
|
|
|
|
@property
|
|
def stage_dependencies(self) -> set[str]:
|
|
# Must connect to viewport_filter stage to get filtered source
|
|
return {"viewport_filter"}
|
|
|
|
@property
|
|
def dependencies(self) -> set[str]:
|
|
# Depend on source.filtered (provided by viewport_filter)
|
|
# This ensures we get the filtered/processed source, not raw source
|
|
return {"source.filtered"}
|
|
|
|
@property
|
|
def inlet_types(self) -> set:
|
|
return {DataType.SOURCE_ITEMS}
|
|
|
|
@property
|
|
def outlet_types(self) -> set:
|
|
return {DataType.TEXT_BUFFER}
|
|
|
|
def process(self, data: Any, ctx: PipelineContext) -> Any:
|
|
"""Render items to text buffer using font."""
|
|
if data is None:
|
|
return []
|
|
|
|
if not isinstance(data, list):
|
|
return [str(data)]
|
|
|
|
import os
|
|
|
|
if os.environ.get("DEBUG_CAMERA"):
|
|
print(f"FontStage: input items={len(data)}")
|
|
|
|
viewport_width = ctx.params.viewport_width if ctx.params else 80
|
|
|
|
result = []
|
|
for item in data:
|
|
if isinstance(item, SourceItem):
|
|
title = item.content
|
|
src = item.source
|
|
ts = item.timestamp
|
|
content_lines, _, _ = engine.render.make_block(
|
|
title, src, ts, viewport_width
|
|
)
|
|
result.extend(content_lines)
|
|
elif hasattr(item, "content"):
|
|
title = str(item.content)
|
|
content_lines, _, _ = engine.render.make_block(
|
|
title, "", "", viewport_width
|
|
)
|
|
result.extend(content_lines)
|
|
else:
|
|
result.append(str(item))
|
|
return result
|
|
|
|
|
|
class ImageToTextStage(Stage):
|
|
"""Convert image items to text."""
|
|
|
|
def __init__(self, name: str = "image-to-text"):
|
|
self.name = name
|
|
self.category = "render"
|
|
self.optional = True
|
|
|
|
@property
|
|
def stage_type(self) -> str:
|
|
return "render"
|
|
|
|
@property
|
|
def capabilities(self) -> set[str]:
|
|
return {"render.output"}
|
|
|
|
@property
|
|
def dependencies(self) -> set[str]:
|
|
return {"source"}
|
|
|
|
@property
|
|
def inlet_types(self) -> set:
|
|
return {DataType.SOURCE_ITEMS}
|
|
|
|
@property
|
|
def outlet_types(self) -> set:
|
|
return {DataType.TEXT_BUFFER}
|
|
|
|
def process(self, data: Any, ctx: PipelineContext) -> Any:
|
|
"""Convert image items to text representation."""
|
|
if data is None:
|
|
return []
|
|
|
|
if not isinstance(data, list):
|
|
return [str(data)]
|
|
|
|
result = []
|
|
for item in data:
|
|
# Check if item is an image
|
|
if hasattr(item, "image_path") or hasattr(item, "image_data"):
|
|
# Placeholder: would normally render image to ASCII art
|
|
result.append(f"[Image: {getattr(item, 'image_path', 'data')}]")
|
|
elif isinstance(item, SourceItem):
|
|
result.extend(item.content.split("\n"))
|
|
else:
|
|
result.append(str(item))
|
|
return result
|
|
|
|
|
|
class CanvasStage(Stage):
|
|
"""Render items to canvas."""
|
|
|
|
def __init__(self, name: str = "canvas"):
|
|
self.name = name
|
|
self.category = "render"
|
|
self.optional = False
|
|
|
|
@property
|
|
def stage_type(self) -> str:
|
|
return "render"
|
|
|
|
@property
|
|
def capabilities(self) -> set[str]:
|
|
return {"render.output"}
|
|
|
|
@property
|
|
def dependencies(self) -> set[str]:
|
|
return {"source"}
|
|
|
|
@property
|
|
def inlet_types(self) -> set:
|
|
return {DataType.SOURCE_ITEMS}
|
|
|
|
@property
|
|
def outlet_types(self) -> set:
|
|
return {DataType.TEXT_BUFFER}
|
|
|
|
def process(self, data: Any, ctx: PipelineContext) -> Any:
|
|
"""Render items to canvas."""
|
|
if data is None:
|
|
return []
|
|
|
|
if not isinstance(data, list):
|
|
return [str(data)]
|
|
|
|
# Simple canvas rendering
|
|
result = []
|
|
for item in data:
|
|
if isinstance(item, SourceItem):
|
|
result.extend(item.content.split("\n"))
|
|
else:
|
|
result.append(str(item))
|
|
return result
|