scroll priority to determine data refreshes

This commit is contained in:
ChuckBuilds
2025-08-18 15:21:39 -05:00
parent a5ce721733
commit e63198dc49
7 changed files with 565 additions and 66 deletions

View File

@@ -530,14 +530,33 @@ class DisplayController:
def _update_modules(self):
"""Call update methods on active managers."""
if self.weather: self.weather.get_weather()
if self.stocks: self.stocks.update_stock_data()
if self.news: self.news.update_news_data()
if self.odds_ticker: self.odds_ticker.update()
if self.calendar: self.calendar.update(time.time())
if self.youtube: self.youtube.update()
if self.text_display: self.text_display.update()
if self.of_the_day: self.of_the_day.update(time.time())
# Check if we're currently scrolling and defer updates if so
if self.display_manager.is_currently_scrolling():
logger.debug("Display is currently scrolling, deferring module updates")
# Defer updates for modules that might cause lag during scrolling
if self.odds_ticker:
self.display_manager.defer_update(self.odds_ticker.update, priority=1)
if self.stocks:
self.display_manager.defer_update(self.stocks.update_stock_data, priority=2)
if self.news:
self.display_manager.defer_update(self.news.update_news_data, priority=2)
# Continue with non-scrolling-sensitive updates
if self.weather: self.weather.get_weather()
if self.calendar: self.calendar.update(time.time())
if self.youtube: self.youtube.update()
if self.text_display: self.text_display.update()
if self.of_the_day: self.of_the_day.update(time.time())
else:
# Not scrolling, perform all updates normally
if self.weather: self.weather.get_weather()
if self.stocks: self.stocks.update_stock_data()
if self.news: self.news.update_news_data()
if self.odds_ticker: self.odds_ticker.update()
if self.calendar: self.calendar.update(time.time())
if self.youtube: self.youtube.update()
if self.text_display: self.text_display.update()
if self.of_the_day: self.of_the_day.update(time.time())
# News manager fetches data when displayed, not during updates
# if self.news_manager: self.news_manager.fetch_news_data()
@@ -940,6 +959,9 @@ class DisplayController:
# Update data for all modules first
self._update_modules()
# Process any deferred updates that may have accumulated
self.display_manager.process_deferred_updates()
# Update live modes in rotation if needed
self._update_live_modes_in_rotation()

View File

@@ -30,6 +30,15 @@ class DisplayManager:
self._snapshot_path = "/tmp/led_matrix_preview.png"
self._snapshot_min_interval_sec = 0.2 # max ~5 fps
self._last_snapshot_ts = 0.0
# Scrolling state tracking for graceful updates
self._scrolling_state = {
'is_scrolling': False,
'last_scroll_activity': 0,
'scroll_inactivity_threshold': 2.0, # seconds of inactivity before considering "not scrolling"
'deferred_updates': []
}
self._setup_matrix()
logger.info("Matrix setup completed in %.3f seconds", time.time() - start_time)
@@ -634,6 +643,77 @@ class DisplayManager:
return dt.strftime(f"%b %-d{suffix}")
def set_scrolling_state(self, is_scrolling: bool):
"""Set the current scrolling state. Call this when a display starts/stops scrolling."""
current_time = time.time()
self._scrolling_state['is_scrolling'] = is_scrolling
if is_scrolling:
self._scrolling_state['last_scroll_activity'] = current_time
logger.debug(f"Scrolling state set to: {is_scrolling}")
def is_currently_scrolling(self) -> bool:
"""Check if the display is currently in a scrolling state."""
current_time = time.time()
# If explicitly not scrolling, return False
if not self._scrolling_state['is_scrolling']:
return False
# If we've been inactive for the threshold period, consider it not scrolling
if current_time - self._scrolling_state['last_scroll_activity'] > self._scrolling_state['scroll_inactivity_threshold']:
self._scrolling_state['is_scrolling'] = False
return False
return True
def defer_update(self, update_func, priority: int = 0):
"""Defer an update function to be called when not scrolling.
Args:
update_func: Function to call when not scrolling
priority: Priority level (lower numbers = higher priority)
"""
self._scrolling_state['deferred_updates'].append({
'func': update_func,
'priority': priority,
'timestamp': time.time()
})
# Sort by priority (lower numbers first)
self._scrolling_state['deferred_updates'].sort(key=lambda x: x['priority'])
logger.debug(f"Deferred update added. Total deferred: {len(self._scrolling_state['deferred_updates'])}")
def process_deferred_updates(self):
"""Process any deferred updates if not currently scrolling."""
if self.is_currently_scrolling():
return
if not self._scrolling_state['deferred_updates']:
return
# Process all deferred updates
updates_to_process = self._scrolling_state['deferred_updates'].copy()
self._scrolling_state['deferred_updates'].clear()
logger.debug(f"Processing {len(updates_to_process)} deferred updates")
for update_info in updates_to_process:
try:
update_info['func']()
logger.debug("Deferred update executed successfully")
except Exception as e:
logger.error(f"Error executing deferred update: {e}")
# Re-add failed updates for retry
self._scrolling_state['deferred_updates'].append(update_info)
def get_scrolling_stats(self) -> dict:
"""Get current scrolling statistics for debugging."""
return {
'is_scrolling': self._scrolling_state['is_scrolling'],
'last_activity': self._scrolling_state['last_scroll_activity'],
'deferred_count': len(self._scrolling_state['deferred_updates']),
'inactivity_threshold': self._scrolling_state['scroll_inactivity_threshold']
}
def _write_snapshot_if_due(self) -> None:
"""Write the current image to a PNG snapshot file at a limited frequency."""
try:

View File

@@ -937,6 +937,16 @@ class OddsTickerManager:
logger.debug("Odds ticker is disabled, skipping update")
return
# Check if we're currently scrolling and defer the update if so
if self.display_manager.is_currently_scrolling():
logger.debug("Odds ticker is currently scrolling, deferring update")
self.display_manager.defer_update(self._perform_update, priority=1)
return
self._perform_update()
def _perform_update(self):
"""Internal method to perform the actual update."""
current_time = time.time()
if current_time - self.last_update < self.update_interval:
logger.debug(f"Odds ticker update interval not reached. Next update in {self.update_interval - (current_time - self.last_update)} seconds")
@@ -992,8 +1002,18 @@ class OddsTickerManager:
try:
current_time = time.time()
# Check if we should be scrolling
should_scroll = current_time - self.last_scroll_time >= self.scroll_delay
# Signal scrolling state to display manager
if should_scroll:
self.display_manager.set_scrolling_state(True)
else:
# If we're not scrolling, check if we should process deferred updates
self.display_manager.process_deferred_updates()
# Scroll the image
if current_time - self.last_scroll_time >= self.scroll_delay:
if should_scroll:
self.scroll_position += self.scroll_speed
self.last_scroll_time = current_time
@@ -1010,6 +1030,8 @@ class OddsTickerManager:
# Stop scrolling when we reach the end
if self.scroll_position >= self.ticker_image.width - width:
self.scroll_position = self.ticker_image.width - width
# Signal that scrolling has stopped
self.display_manager.set_scrolling_state(False)
# Create the visible part of the image by pasting from the ticker_image
visible_image = Image.new('RGB', (width, height))

View File

@@ -366,33 +366,53 @@ class StockManager:
logger.info("Stock display settings changed, clearing cache")
def update_stock_data(self):
"""Update stock and crypto data for all configured symbols."""
"""Update stock data from API."""
current_time = time.time()
update_interval = self.stocks_config.get('update_interval', 300)
update_interval = self.stocks_config.get('update_interval', 600)
# Check if we need to update based on time
if current_time - self.last_update > update_interval:
stock_symbols = self.stocks_config.get('symbols', [])
crypto_symbols = self.crypto_config.get('symbols', []) if self.crypto_config.get('enabled', False) else []
# Check if we're currently scrolling and defer the update if so
if self.display_manager.is_currently_scrolling():
logger.debug("Stock display is currently scrolling, deferring update")
self.display_manager.defer_update(self._perform_stock_update, priority=2)
return
if not stock_symbols and not crypto_symbols:
logger.warning("No stock or crypto symbols configured")
self._perform_stock_update()
def _perform_stock_update(self):
"""Internal method to perform the actual stock update."""
current_time = time.time()
update_interval = self.stocks_config.get('update_interval', 600)
if current_time - self.last_update < update_interval:
return
try:
logger.debug("Updating stock data")
symbols = self.stocks_config.get('symbols', [])
if not symbols:
logger.warning("No stock symbols configured")
return
# Update stocks
for symbol in stock_symbols:
data = self._fetch_stock_data(symbol, is_crypto=False)
if data:
self.stock_data[symbol] = data
# Update crypto
for symbol in crypto_symbols:
data = self._fetch_stock_data(symbol, is_crypto=True)
if data:
self.stock_data[symbol] = data
# Fetch stock data
for symbol in symbols:
try:
data = self._fetch_stock_data(symbol)
if data:
self.stock_data[symbol] = data
logger.debug(f"Updated data for {symbol}: {data}")
except Exception as e:
logger.error(f"Error fetching data for {symbol}: {e}")
self.last_update = current_time
# Clear cached text to force regeneration
self.cached_text = None
self.cached_text_image = None
except Exception as e:
logger.error(f"Error updating stock data: {e}")
def _get_stock_logo(self, symbol: str, is_crypto: bool = False) -> Image.Image:
"""Get stock or crypto logo image from local directory."""
try:
@@ -696,6 +716,15 @@ class StockManager:
width = self.display_manager.matrix.width
total_width = self.cached_text_image.width
# Check if we should be scrolling
should_scroll = True # Stock display always scrolls continuously
# Signal scrolling state to display manager
self.display_manager.set_scrolling_state(True)
# Process any deferred updates (though stocks are always scrolling)
self.display_manager.process_deferred_updates()
# Update scroll position with small increments
self.scroll_position = (self.scroll_position + self.scroll_speed) % total_width

View File

@@ -169,50 +169,52 @@ class StockNewsManager:
return []
def update_news_data(self):
"""Update news data for all configured stock symbols."""
"""Update news data from API."""
current_time = time.time()
update_interval = self.stock_news_config.get('update_interval', 300)
update_interval = self.stock_news_config.get('update_interval', 3600)
# Check if we need to update based on time
if current_time - self.last_update > update_interval:
# Check if we're currently scrolling and defer the update if so
if self.display_manager.is_currently_scrolling():
logger.debug("Stock news display is currently scrolling, deferring update")
self.display_manager.defer_update(self._perform_news_update, priority=2)
return
self._perform_news_update()
def _perform_news_update(self):
"""Internal method to perform the actual news update."""
current_time = time.time()
update_interval = self.stock_news_config.get('update_interval', 3600)
if current_time - self.last_update < update_interval:
return
try:
logger.debug("Updating stock news data")
symbols = self.stocks_config.get('symbols', [])
if not symbols:
logger.warning("No stock symbols configured for news")
return
# Get cached data
cached_data = self.cache_manager.get('stock_news')
# Update each symbol
new_data = {}
success = False
# Fetch news for each symbol
for symbol in symbols:
# Check if data has changed before fetching
if cached_data and symbol in cached_data:
current_state = cached_data[symbol]
if not self.cache_manager.has_data_changed('stock_news', current_state):
logger.info(f"News data hasn't changed for {symbol}, using existing data")
new_data[symbol] = current_state
success = True
continue
try:
news = self._fetch_news_for_symbol(symbol)
if news:
self.news_data[symbol] = news
logger.debug(f"Updated news for {symbol}: {len(news)} headlines")
except Exception as e:
logger.error(f"Error fetching news for {symbol}: {e}")
# Add a longer delay between requests to avoid rate limiting
time.sleep(random.uniform(1.0, 2.0)) # increased delay between requests
news_items = self._fetch_news(symbol)
if news_items:
new_data[symbol] = news_items
success = True
self.last_update = current_time
if success:
# Cache the new data
self.cache_manager.update_cache('stock_news', new_data)
# Only update the displayed data when we have new data
self.news_data = new_data
self.last_update = current_time
logger.info(f"Updated news data for {len(new_data)} symbols")
else:
logger.error("Failed to fetch news for any configured stocks")
# Clear cached text to force regeneration
self.cached_text = None
self.cached_text_image = None
except Exception as e:
logger.error(f"Error updating stock news data: {e}")
def _create_text_image(self, text: str, color: Tuple[int, int, int] = (255, 255, 255)) -> Image.Image:
"""Create an image containing the text for efficient scrolling."""
@@ -405,12 +407,23 @@ class StockNewsManager:
# If total_width is somehow less than screen width, don't scroll
if total_width <= width:
# Signal that we're not scrolling for this frame
self.display_manager.set_scrolling_state(False)
# Process any deferred updates
self.display_manager.process_deferred_updates()
self.display_manager.image.paste(self.cached_text_image, (0, 0))
self.display_manager.update_display()
time.sleep(self.stock_news_config.get('item_display_duration', 5)) # Hold static image
self.cached_text_image = None # Force recreation next cycle
return True
# Signal that we're scrolling
self.display_manager.set_scrolling_state(True)
# Process any deferred updates (though news is usually always scrolling)
self.display_manager.process_deferred_updates()
# Update scroll position
self.scroll_position += self.scroll_speed
if self.scroll_position >= total_width:

187
test_graceful_updates.py Normal file
View File

@@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""
Test script to demonstrate the graceful update system for scrolling displays.
This script shows how updates are deferred during scrolling periods to prevent lag.
"""
import time
import logging
import sys
import os
# Add the project root directory to Python path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# Configure logging first
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s.%(msecs)03d - %(levelname)s:%(name)s:%(message)s',
datefmt='%H:%M:%S',
stream=sys.stdout
)
logger = logging.getLogger(__name__)
# Mock rgbmatrix module for testing on non-Raspberry Pi systems
try:
from rgbmatrix import RGBMatrix, RGBMatrixOptions
except ImportError:
logger.info("rgbmatrix module not available, using mock for testing")
class MockRGBMatrixOptions:
def __init__(self):
self.rows = 32
self.cols = 64
self.chain_length = 2
self.parallel = 1
self.hardware_mapping = 'adafruit-hat-pwm'
self.brightness = 90
self.pwm_bits = 10
self.pwm_lsb_nanoseconds = 150
self.led_rgb_sequence = 'RGB'
self.pixel_mapper_config = ''
self.row_address_type = 0
self.multiplexing = 0
self.disable_hardware_pulsing = False
self.show_refresh_rate = False
self.limit_refresh_rate_hz = 90
self.gpio_slowdown = 2
class MockRGBMatrix:
def __init__(self, options=None):
self.width = 128 # 64 * 2 chain length
self.height = 32
def CreateFrameCanvas(self):
return MockCanvas()
def SwapOnVSync(self, canvas, dont_wait=False):
pass
def Clear(self):
pass
class MockCanvas:
def __init__(self):
self.width = 128
self.height = 32
def SetImage(self, image):
pass
def Clear(self):
pass
RGBMatrix = MockRGBMatrix
RGBMatrixOptions = MockRGBMatrixOptions
from src.display_manager import DisplayManager
from src.config_manager import ConfigManager
def simulate_scrolling_display(display_manager, duration=10):
"""Simulate a scrolling display for testing."""
logger.info(f"Starting scrolling simulation for {duration} seconds")
start_time = time.time()
while time.time() - start_time < duration:
# Signal that we're scrolling
display_manager.set_scrolling_state(True)
# Simulate some scrolling work
time.sleep(0.1)
# Every 2 seconds, try to defer an update
if int(time.time() - start_time) % 2 == 0:
logger.info("Attempting to defer an update during scrolling")
display_manager.defer_update(
lambda: logger.info("This update was deferred and executed later!"),
priority=1
)
# Signal that scrolling has stopped
display_manager.set_scrolling_state(False)
logger.info("Scrolling simulation completed")
def test_graceful_updates():
"""Test the graceful update system."""
logger.info("Testing graceful update system")
# Load config
config_manager = ConfigManager()
config = config_manager.load_config()
# Initialize display manager
display_manager = DisplayManager(config, force_fallback=True)
try:
# Test 1: Defer updates during scrolling
logger.info("=== Test 1: Defer updates during scrolling ===")
# Add some deferred updates
display_manager.defer_update(
lambda: logger.info("Update 1: High priority update"),
priority=1
)
display_manager.defer_update(
lambda: logger.info("Update 2: Medium priority update"),
priority=2
)
display_manager.defer_update(
lambda: logger.info("Update 3: Low priority update"),
priority=3
)
# Start scrolling simulation
simulate_scrolling_display(display_manager, duration=5)
# Check scrolling stats
stats = display_manager.get_scrolling_stats()
logger.info(f"Scrolling stats: {stats}")
# Test 2: Process deferred updates when not scrolling
logger.info("=== Test 2: Process deferred updates when not scrolling ===")
# Process any remaining deferred updates
display_manager.process_deferred_updates()
# Test 3: Test inactivity threshold
logger.info("=== Test 3: Test inactivity threshold ===")
# Signal scrolling started
display_manager.set_scrolling_state(True)
logger.info(f"Is scrolling: {display_manager.is_currently_scrolling()}")
# Wait longer than the inactivity threshold
time.sleep(3)
logger.info(f"Is scrolling after inactivity: {display_manager.is_currently_scrolling()}")
# Test 4: Test priority ordering
logger.info("=== Test 4: Test priority ordering ===")
# Add updates in reverse priority order
display_manager.defer_update(
lambda: logger.info("Priority 3 update"),
priority=3
)
display_manager.defer_update(
lambda: logger.info("Priority 1 update"),
priority=1
)
display_manager.defer_update(
lambda: logger.info("Priority 2 update"),
priority=2
)
# Process them (should execute in priority order: 1, 2, 3)
display_manager.process_deferred_updates()
logger.info("All tests completed successfully!")
except Exception as e:
logger.error(f"Test failed: {e}", exc_info=True)
finally:
# Cleanup
display_manager.cleanup()
if __name__ == "__main__":
test_graceful_updates()

146
wiki/GRACEFUL_UPDATES.md Normal file
View File

@@ -0,0 +1,146 @@
# Graceful Update System
The LED Matrix project now includes a graceful update system that prevents lag during scrolling displays by deferring updates until the display is not actively scrolling.
## Overview
When displays like the odds ticker, stock ticker, or news ticker are actively scrolling, performing API updates or data fetching can cause visual lag or stuttering. The graceful update system solves this by:
1. **Tracking scrolling state** - The system monitors when displays are actively scrolling
2. **Deferring updates** - Updates that might cause lag are deferred during scrolling periods
3. **Processing when safe** - Deferred updates are processed when scrolling stops or during non-scrolling periods
4. **Priority-based execution** - Updates are executed in priority order when processed
## How It Works
### Scrolling State Tracking
The `DisplayManager` class now includes scrolling state tracking:
```python
# Signal when scrolling starts
display_manager.set_scrolling_state(True)
# Signal when scrolling stops
display_manager.set_scrolling_state(False)
# Check if currently scrolling
if display_manager.is_currently_scrolling():
# Defer updates
pass
```
### Deferred Updates
Updates can be deferred using the `defer_update` method:
```python
# Defer an update with priority
display_manager.defer_update(
lambda: self._perform_update(),
priority=1 # Lower numbers = higher priority
)
```
### Automatic Processing
Deferred updates are automatically processed when:
- A display signals it's not scrolling
- The main loop processes updates during non-scrolling periods
- The inactivity threshold is reached (default: 2 seconds)
## Implementation Details
### Display Manager Changes
The `DisplayManager` class now includes:
- `set_scrolling_state(is_scrolling)` - Signal scrolling state changes
- `is_currently_scrolling()` - Check if display is currently scrolling
- `defer_update(update_func, priority)` - Defer an update function
- `process_deferred_updates()` - Process all pending deferred updates
- `get_scrolling_stats()` - Get current scrolling statistics
### Manager Updates
The following managers have been updated to use the graceful update system:
#### Odds Ticker Manager
- Defers API updates during scrolling
- Signals scrolling state during display
- Processes deferred updates when not scrolling
#### Stock Manager
- Defers stock data updates during scrolling
- Always signals scrolling state (continuous scrolling)
- Priority 2 for stock updates
#### Stock News Manager
- Defers news data updates during scrolling
- Signals scrolling state during display
- Priority 2 for news updates
### Display Controller Changes
The main display controller now:
- Checks scrolling state before updating modules
- Defers scrolling-sensitive updates during scrolling periods
- Processes deferred updates in the main loop
- Continues non-scrolling-sensitive updates normally
## Configuration
The system uses these default settings:
- **Inactivity threshold**: 2.0 seconds
- **Update priorities**:
- Priority 1: Odds ticker updates
- Priority 2: Stock and news updates
- Priority 3+: Other updates
## Benefits
1. **Smoother Scrolling** - No more lag during ticker scrolling
2. **Better User Experience** - Displays remain responsive during updates
3. **Efficient Resource Usage** - Updates happen when the system is idle
4. **Priority-Based** - Important updates are processed first
5. **Automatic** - No manual intervention required
## Testing
You can test the graceful update system using the provided test script:
```bash
python test_graceful_updates.py
```
This script demonstrates:
- Deferring updates during scrolling
- Processing updates when not scrolling
- Priority-based execution
- Inactivity threshold behavior
## Debugging
To debug the graceful update system, enable debug logging:
```python
import logging
logging.getLogger('src.display_manager').setLevel(logging.DEBUG)
```
The system will log:
- When scrolling state changes
- When updates are deferred
- When deferred updates are processed
- Current scrolling statistics
## Future Enhancements
Potential improvements to the system:
1. **Configurable thresholds** - Allow users to adjust inactivity thresholds
2. **More granular priorities** - Add more priority levels for different update types
3. **Update batching** - Group similar updates to reduce processing overhead
4. **Performance metrics** - Track and report update deferral statistics
5. **Web interface integration** - Show deferred update status in the web UI