David Gwilliam
affafe810c
Two critical fixes: 1. ListDataSource Cache Bug - Previously, ListDataSource.__init__ cached raw tuples directly - get_items() would return cached raw tuples without converting to SourceItem - This caused SourceItemsToBufferStage to receive tuples and stringify them - Results: ugly tuple representations in terminal/pygame instead of formatted text - Fix: Store raw items in _raw_items, let fetch() convert to SourceItem - Cache now contains proper SourceItem objects 2. Camera Dependency Resolution - CameraStage declared dependency on 'source.items' exactly - DataSourceStage provides 'source.headlines' (or 'source.poetry', etc.) - Capability matching didn't trigger prefix match for exact dependency - Fix: Change CameraStage dependency to 'source' for prefix matching 3. Added app.py Camera Stage Support - Pipeline now adds camera stage from preset.camera config - Supports vertical, horizontal, omni, floating, bounce modes - Tests now passing with proper data flow through all stages Tests: All 502 tests passing, 16 skipped
491 lines
14 KiB
Python
491 lines
14 KiB
Python
"""
|
|
Data sources for the pipeline architecture.
|
|
|
|
This module contains all DataSource implementations:
|
|
- DataSource: Abstract base class
|
|
- SourceItem, ImageItem: Data containers
|
|
- HeadlinesDataSource, PoetryDataSource, ImageDataSource: Concrete sources
|
|
- SourceRegistry: Registry for source discovery
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class SourceItem:
|
|
"""A single item from a data source."""
|
|
|
|
content: str
|
|
source: str
|
|
timestamp: str
|
|
metadata: dict[str, Any] | None = None
|
|
|
|
|
|
@dataclass
|
|
class ImageItem:
|
|
"""An image item from a data source - wraps a PIL Image."""
|
|
|
|
image: Any # PIL Image
|
|
source: str
|
|
timestamp: str
|
|
path: str | None = None # File path or URL if applicable
|
|
metadata: dict[str, Any] | None = None
|
|
|
|
|
|
class DataSource(ABC):
|
|
"""Abstract base class for data sources.
|
|
|
|
Static sources: Data fetched once and cached. Safe to call fetch() multiple times.
|
|
Dynamic sources: Data changes over time. fetch() should be idempotent.
|
|
"""
|
|
|
|
@property
|
|
@abstractmethod
|
|
def name(self) -> str:
|
|
"""Display name for this source."""
|
|
...
|
|
|
|
@property
|
|
def is_dynamic(self) -> bool:
|
|
"""Whether this source updates dynamically while the app runs. Default False."""
|
|
return False
|
|
|
|
@abstractmethod
|
|
def fetch(self) -> list[SourceItem]:
|
|
"""Fetch fresh data from the source. Must be idempotent."""
|
|
...
|
|
|
|
def get_items(self) -> list[SourceItem]:
|
|
"""Get current items. Default implementation returns cached fetch results."""
|
|
if not hasattr(self, "_items") or self._items is None:
|
|
self._items = self.fetch()
|
|
return self._items
|
|
|
|
def refresh(self) -> list[SourceItem]:
|
|
"""Force refresh - clear cache and fetch fresh data."""
|
|
self._items = self.fetch()
|
|
return self._items
|
|
|
|
def stream(self):
|
|
"""Optional: Yield items continuously. Override for streaming sources."""
|
|
raise NotImplementedError
|
|
|
|
def __post_init__(self):
|
|
self._items: list[SourceItem] | None = None
|
|
|
|
|
|
class HeadlinesDataSource(DataSource):
|
|
"""Data source for RSS feed headlines."""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "headlines"
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
from engine.fetch import fetch_all
|
|
|
|
items, _, _ = fetch_all()
|
|
return [SourceItem(content=t, source=s, timestamp=ts) for t, s, ts in items]
|
|
|
|
|
|
class EmptyDataSource(DataSource):
|
|
"""Empty data source that produces blank lines for testing.
|
|
|
|
Useful for testing display borders, effects, and other pipeline
|
|
components without needing actual content.
|
|
"""
|
|
|
|
def __init__(self, width: int = 80, height: int = 24):
|
|
self.width = width
|
|
self.height = height
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "empty"
|
|
|
|
@property
|
|
def is_dynamic(self) -> bool:
|
|
return False
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
# Return empty lines as content
|
|
content = "\n".join([" " * self.width for _ in range(self.height)])
|
|
return [SourceItem(content=content, source="empty", timestamp="0")]
|
|
|
|
|
|
class ListDataSource(DataSource):
|
|
"""Data source that wraps a pre-fetched list of items.
|
|
|
|
Used for bootstrap loading when items are already available in memory.
|
|
This is a simple wrapper for already-fetched data.
|
|
"""
|
|
|
|
def __init__(self, items, name: str = "list"):
|
|
self._raw_items = items # Store raw items separately
|
|
self._items = None # Cache for converted SourceItem objects
|
|
self._name = name
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@property
|
|
def is_dynamic(self) -> bool:
|
|
return False
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
# Convert tuple items to SourceItem if needed
|
|
result = []
|
|
for item in self._raw_items:
|
|
if isinstance(item, SourceItem):
|
|
result.append(item)
|
|
elif isinstance(item, tuple) and len(item) >= 3:
|
|
# Assume (content, source, timestamp) tuple format
|
|
result.append(
|
|
SourceItem(content=item[0], source=item[1], timestamp=str(item[2]))
|
|
)
|
|
else:
|
|
# Fallback: treat as string content
|
|
result.append(
|
|
SourceItem(content=str(item), source="list", timestamp="0")
|
|
)
|
|
return result
|
|
|
|
|
|
class PoetryDataSource(DataSource):
|
|
"""Data source for Poetry DB."""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "poetry"
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
from engine.fetch import fetch_poetry
|
|
|
|
items, _, _ = fetch_poetry()
|
|
return [SourceItem(content=t, source=s, timestamp=ts) for t, s, ts in items]
|
|
|
|
|
|
class ImageDataSource(DataSource):
|
|
"""Data source that loads PNG images from file paths or URLs.
|
|
|
|
Supports:
|
|
- Local file paths (e.g., /path/to/image.png)
|
|
- URLs (e.g., https://example.com/image.png)
|
|
|
|
Yields ImageItem objects containing PIL Image objects that can be
|
|
converted to text buffers by an ImageToTextTransform stage.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
path: str | list[str] | None = None,
|
|
urls: str | list[str] | None = None,
|
|
):
|
|
"""
|
|
Args:
|
|
path: Single path or list of paths to PNG files
|
|
urls: Single URL or list of URLs to PNG images
|
|
"""
|
|
self._paths = [path] if isinstance(path, str) else (path or [])
|
|
self._urls = [urls] if isinstance(urls, str) else (urls or [])
|
|
self._images: list[ImageItem] = []
|
|
self._load_images()
|
|
|
|
def _load_images(self) -> None:
|
|
"""Load all images from paths and URLs."""
|
|
from datetime import datetime
|
|
from io import BytesIO
|
|
from urllib.request import urlopen
|
|
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
for path in self._paths:
|
|
try:
|
|
from PIL import Image
|
|
|
|
img = Image.open(path)
|
|
if img.mode != "RGBA":
|
|
img = img.convert("RGBA")
|
|
self._images.append(
|
|
ImageItem(
|
|
image=img,
|
|
source=f"file:{path}",
|
|
timestamp=timestamp,
|
|
path=path,
|
|
)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
for url in self._urls:
|
|
try:
|
|
from PIL import Image
|
|
|
|
with urlopen(url) as response:
|
|
img = Image.open(BytesIO(response.read()))
|
|
if img.mode != "RGBA":
|
|
img = img.convert("RGBA")
|
|
self._images.append(
|
|
ImageItem(
|
|
image=img,
|
|
source=f"url:{url}",
|
|
timestamp=timestamp,
|
|
path=url,
|
|
)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "image"
|
|
|
|
@property
|
|
def is_dynamic(self) -> bool:
|
|
return False # Static images, not updating
|
|
|
|
def fetch(self) -> list[ImageItem]:
|
|
"""Return loaded images as ImageItem list."""
|
|
return self._images
|
|
|
|
def get_items(self) -> list[ImageItem]:
|
|
"""Return current image items."""
|
|
return self._images
|
|
|
|
|
|
class MetricsDataSource(DataSource):
|
|
"""Data source that renders live pipeline metrics as ASCII art.
|
|
|
|
Wraps a Pipeline and displays active stages with their average execution
|
|
time and approximate FPS impact. Updates lazily when camera is about to
|
|
focus on a new node (frame % 15 == 12).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
pipeline: Any,
|
|
viewport_width: int = 80,
|
|
viewport_height: int = 24,
|
|
):
|
|
self.pipeline = pipeline
|
|
self.viewport_width = viewport_width
|
|
self.viewport_height = viewport_height
|
|
self.frame = 0
|
|
self._cached_metrics: dict | None = None
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "metrics"
|
|
|
|
@property
|
|
def is_dynamic(self) -> bool:
|
|
return True
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
if self.frame % 15 == 12:
|
|
self._cached_metrics = None
|
|
|
|
if self._cached_metrics is None:
|
|
self._cached_metrics = self._fetch_metrics()
|
|
|
|
buffer = self._render_metrics(self._cached_metrics)
|
|
self.frame += 1
|
|
content = "\n".join(buffer)
|
|
return [
|
|
SourceItem(content=content, source="metrics", timestamp=f"f{self.frame}")
|
|
]
|
|
|
|
def _fetch_metrics(self) -> dict:
|
|
if hasattr(self.pipeline, "get_metrics_summary"):
|
|
metrics = self.pipeline.get_metrics_summary()
|
|
if "error" not in metrics:
|
|
return metrics
|
|
return {"stages": {}, "pipeline": {"avg_ms": 0}}
|
|
|
|
def _render_metrics(self, metrics: dict) -> list[str]:
|
|
stages = metrics.get("stages", {})
|
|
|
|
if not stages:
|
|
return self._render_empty()
|
|
|
|
active_stages = {
|
|
name: stats for name, stats in stages.items() if stats.get("avg_ms", 0) > 0
|
|
}
|
|
|
|
if not active_stages:
|
|
return self._render_empty()
|
|
|
|
total_avg = sum(s["avg_ms"] for s in active_stages.values())
|
|
if total_avg == 0:
|
|
total_avg = 1
|
|
|
|
lines: list[str] = []
|
|
lines.append("═" * self.viewport_width)
|
|
lines.append(" PIPELINE METRICS ".center(self.viewport_width, "─"))
|
|
lines.append("─" * self.viewport_width)
|
|
|
|
header = f"{'STAGE':<20} {'AVG_MS':>8} {'FPS %':>8}"
|
|
lines.append(header)
|
|
lines.append("─" * self.viewport_width)
|
|
|
|
for name, stats in sorted(active_stages.items()):
|
|
avg_ms = stats.get("avg_ms", 0)
|
|
fps_impact = (avg_ms / 16.67) * 100 if avg_ms > 0 else 0
|
|
|
|
row = f"{name:<20} {avg_ms:>7.2f} {fps_impact:>7.1f}%"
|
|
lines.append(row[: self.viewport_width])
|
|
|
|
lines.append("─" * self.viewport_width)
|
|
total_row = (
|
|
f"{'TOTAL':<20} {total_avg:>7.2f} {(total_avg / 16.67) * 100:>7.1f}%"
|
|
)
|
|
lines.append(total_row[: self.viewport_width])
|
|
lines.append("─" * self.viewport_width)
|
|
lines.append(
|
|
f" Frame:{self.frame:04d} Cache:{'HIT' if self._cached_metrics else 'MISS'}"
|
|
)
|
|
|
|
while len(lines) < self.viewport_height:
|
|
lines.append(" " * self.viewport_width)
|
|
|
|
return lines[: self.viewport_height]
|
|
|
|
def _render_empty(self) -> list[str]:
|
|
lines = [" " * self.viewport_width for _ in range(self.viewport_height)]
|
|
msg = "No metrics available"
|
|
y = self.viewport_height // 2
|
|
x = (self.viewport_width - len(msg)) // 2
|
|
lines[y] = " " * x + msg + " " * (self.viewport_width - x - len(msg))
|
|
return lines
|
|
|
|
def get_items(self) -> list[SourceItem]:
|
|
return self.fetch()
|
|
|
|
|
|
class CachedDataSource(DataSource):
|
|
"""Data source that wraps another source with caching."""
|
|
|
|
def __init__(self, source: DataSource, max_items: int = 100):
|
|
self.source = source
|
|
self.max_items = max_items
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return f"cached:{self.source.name}"
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
items = self.source.fetch()
|
|
return items[: self.max_items]
|
|
|
|
def get_items(self) -> list[SourceItem]:
|
|
if not hasattr(self, "_items") or self._items is None:
|
|
self._items = self.fetch()
|
|
return self._items
|
|
|
|
|
|
class TransformDataSource(DataSource):
|
|
"""Data source that transforms items from another source.
|
|
|
|
Applies optional filter and map functions to each item.
|
|
This enables chaining: source → transform → transformed output.
|
|
|
|
Args:
|
|
source: The source to fetch items from
|
|
filter_fn: Optional function(item: SourceItem) -> bool
|
|
map_fn: Optional function(item: SourceItem) -> SourceItem
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
source: DataSource,
|
|
filter_fn: Callable[[SourceItem], bool] | None = None,
|
|
map_fn: Callable[[SourceItem], SourceItem] | None = None,
|
|
):
|
|
self.source = source
|
|
self.filter_fn = filter_fn
|
|
self.map_fn = map_fn
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return f"transform:{self.source.name}"
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
items = self.source.fetch()
|
|
|
|
if self.filter_fn:
|
|
items = [item for item in items if self.filter_fn(item)]
|
|
|
|
if self.map_fn:
|
|
items = [self.map_fn(item) for item in items]
|
|
|
|
return items
|
|
|
|
|
|
class CompositeDataSource(DataSource):
|
|
"""Data source that combines multiple sources."""
|
|
|
|
def __init__(self, sources: list[DataSource]):
|
|
self.sources = sources
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "composite"
|
|
|
|
def fetch(self) -> list[SourceItem]:
|
|
items = []
|
|
for source in self.sources:
|
|
items.extend(source.fetch())
|
|
return items
|
|
|
|
|
|
class SourceRegistry:
|
|
"""Registry for data sources."""
|
|
|
|
def __init__(self):
|
|
self._sources: dict[str, DataSource] = {}
|
|
self._default: str | None = None
|
|
|
|
def register(self, source: DataSource, default: bool = False) -> None:
|
|
self._sources[source.name] = source
|
|
if default or self._default is None:
|
|
self._default = source.name
|
|
|
|
def get(self, name: str) -> DataSource | None:
|
|
return self._sources.get(name)
|
|
|
|
def list_all(self) -> dict[str, DataSource]:
|
|
return dict(self._sources)
|
|
|
|
def default(self) -> DataSource | None:
|
|
if self._default:
|
|
return self._sources.get(self._default)
|
|
return None
|
|
|
|
def create_headlines(self) -> HeadlinesDataSource:
|
|
return HeadlinesDataSource()
|
|
|
|
def create_poetry(self) -> PoetryDataSource:
|
|
return PoetryDataSource()
|
|
|
|
|
|
_global_registry: SourceRegistry | None = None
|
|
|
|
|
|
def get_source_registry() -> SourceRegistry:
|
|
global _global_registry
|
|
if _global_registry is None:
|
|
_global_registry = SourceRegistry()
|
|
return _global_registry
|
|
|
|
|
|
def init_default_sources() -> SourceRegistry:
|
|
"""Initialize the default source registry with standard sources."""
|
|
registry = get_source_registry()
|
|
registry.register(HeadlinesDataSource(), default=True)
|
|
registry.register(PoetryDataSource())
|
|
return registry
|