refactor: phase 2 - modularization of scroll engine
Split monolithic scroll.py into focused modules: - viewport.py: terminal size (tw/th), ANSI positioning helpers - frame.py: FrameTimer class, scroll step calculation - layers.py: message overlay, ticker zone, firehose rendering - scroll.py: simplified orchestrator, imports from new modules Add stream controller and event types for future event-driven architecture: - controller.py: StreamController for source initialization and stream lifecycle - events.py: EventType enum and event dataclasses (HeadlineEvent, FrameTickEvent, etc.) Added tests for new modules: - test_viewport.py: 8 tests for viewport utilities - test_frame.py: 10 tests for frame timing - test_layers.py: 13 tests for layer compositing - test_events.py: 11 tests for event types - test_controller.py: 6 tests for stream controller This enables: - Testable chunks with clear responsibilities - Reusable viewport utilities across modules - Better separation of concerns in render pipeline - Foundation for future event-driven architecture Also includes Phase 1 documentation updates in code comments.
This commit is contained in:
57
engine/frame.py
Normal file
57
engine/frame.py
Normal file
@@ -0,0 +1,57 @@
|
||||
"""
|
||||
Frame timing utilities — FPS control and precise timing.
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
|
||||
class FrameTimer:
|
||||
"""Frame timer for consistent render loop timing."""
|
||||
|
||||
def __init__(self, target_frame_dt: float = 0.05):
|
||||
self.target_frame_dt = target_frame_dt
|
||||
self._frame_count = 0
|
||||
self._start_time = time.monotonic()
|
||||
self._last_frame_time = self._start_time
|
||||
|
||||
@property
|
||||
def fps(self) -> float:
|
||||
"""Current FPS based on elapsed frames."""
|
||||
elapsed = time.monotonic() - self._start_time
|
||||
if elapsed > 0:
|
||||
return self._frame_count / elapsed
|
||||
return 0.0
|
||||
|
||||
def sleep_until_next_frame(self) -> float:
|
||||
"""Sleep to maintain target frame rate. Returns actual elapsed time."""
|
||||
now = time.monotonic()
|
||||
elapsed = now - self._last_frame_time
|
||||
self._last_frame_time = now
|
||||
self._frame_count += 1
|
||||
|
||||
sleep_time = max(0, self.target_frame_dt - elapsed)
|
||||
if sleep_time > 0:
|
||||
time.sleep(sleep_time)
|
||||
return elapsed
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Reset frame counter and start time."""
|
||||
self._frame_count = 0
|
||||
self._start_time = time.monotonic()
|
||||
self._last_frame_time = self._start_time
|
||||
|
||||
|
||||
def calculate_scroll_step(
|
||||
scroll_dur: float, view_height: int, padding: int = 15
|
||||
) -> float:
|
||||
"""Calculate scroll step interval for smooth scrolling.
|
||||
|
||||
Args:
|
||||
scroll_dur: Duration in seconds for one headline to scroll through view
|
||||
view_height: Terminal height in rows
|
||||
padding: Extra rows for off-screen content
|
||||
|
||||
Returns:
|
||||
Time in seconds between scroll steps
|
||||
"""
|
||||
return scroll_dur / (view_height + padding) * 2
|
||||
Reference in New Issue
Block a user