From df2a003bd67f3cd96325ca84ac1a2765182da46a Mon Sep 17 00:00:00 2001 From: Rs4anti <44727222+Rs4anti@users.noreply.github.com> Date: Sun, 9 Aug 2026 18:32:20 +0200 Subject: [PATCH 1/3] Fix playlist clock refresh --- src/refresh_task.py | 42 +++++++- tests/test_refresh_task.py | 211 +++++++++++++++++++++++++++++++++++++ 2 files changed, 252 insertions(+), 1 deletion(-) create mode 100644 tests/test_refresh_task.py diff --git a/src/refresh_task.py b/src/refresh_task.py index f554e2adb..9dc9d02cb 100644 --- a/src/refresh_task.py +++ b/src/refresh_task.py @@ -73,7 +73,7 @@ def _run(self): while True: try: with self.condition: - sleep_time = self.device_config.get_config("plugin_cycle_interval_seconds", default=60*60) + sleep_time = self._get_sleep_time() # Wait for sleep_time or until notified self.condition.wait(timeout=sleep_time) @@ -160,6 +160,39 @@ def _get_current_datetime(self): tz_str = self.device_config.get_config("timezone", default="UTC") return datetime.now(pytz.timezone(tz_str)) + def _get_sleep_time(self): + """Determines how long to sleep before the next refresh check. + + Returns the minimum of the global plugin cycle interval and the currently + displayed plugin instance's refresh interval (if it has one), so that + plugins requiring frequent refreshes (e.g., a clock) are re-rendered + while they remain displayed. + """ + global_interval = self.device_config.get_config("plugin_cycle_interval_seconds", default=3600) + + playlist_manager = self.device_config.get_playlist_manager() + current_dt = self._get_current_datetime() + playlist = playlist_manager.determine_active_playlist(current_dt) + + if playlist and playlist.plugins and playlist.current_plugin_index is not None: + plugin_instance = playlist.plugins[playlist.current_plugin_index] + if "interval" in plugin_instance.refresh: + interval = plugin_instance.refresh.get("interval") + if interval: + # Calculate time remaining until the next refresh + latest_refresh_dt = plugin_instance.get_latest_refresh_dt() + if latest_refresh_dt: + # Ensure both datetimes are timezone-aware for comparison + if latest_refresh_dt.tzinfo is None: + latest_refresh_dt = latest_refresh_dt.replace(tzinfo=current_dt.tzinfo) + time_since_refresh = (current_dt - latest_refresh_dt).total_seconds() + time_until_refresh = interval - time_since_refresh + if time_until_refresh > 0: + return min(global_interval, time_until_refresh) + return min(global_interval, interval) + + return global_interval + def _determine_next_plugin(self, playlist_manager, latest_refresh_info, current_dt): """Determines the next plugin to refresh based on the active playlist, plugin cycle interval, and current time.""" playlist = playlist_manager.determine_active_playlist(current_dt) @@ -173,6 +206,13 @@ def _determine_next_plugin(self, playlist_manager, latest_refresh_info, current_ logger.info(f"Active playlist '{playlist.name}' has no plugins.") return None, None + # Check if the currently displayed plugin instance needs a refresh based on its own settings + if playlist.current_plugin_index is not None and playlist.current_plugin_index < len(playlist.plugins): + current_plugin = playlist.plugins[playlist.current_plugin_index] + if current_plugin.should_refresh(current_dt): + logger.info(f"Refreshing currently displayed plugin instance. | plugin_instance: {current_plugin.name}") + return playlist, current_plugin + latest_refresh_dt = latest_refresh_info.get_refresh_datetime() plugin_cycle_interval = self.device_config.get_config("plugin_cycle_interval_seconds", default=3600) should_refresh = PlaylistManager.should_refresh(latest_refresh_dt, plugin_cycle_interval, current_dt) diff --git a/tests/test_refresh_task.py b/tests/test_refresh_task.py new file mode 100644 index 000000000..0a32da416 --- /dev/null +++ b/tests/test_refresh_task.py @@ -0,0 +1,211 @@ +import pytest +import pytz +from datetime import datetime, timedelta +from unittest.mock import MagicMock, patch + +from src.model import Playlist, PlaylistManager, PluginInstance, RefreshInfo +from src.refresh_task import RefreshTask, PlaylistRefresh + + +class MockDeviceConfig: + """Mock device config for testing.""" + def __init__(self, config=None, playlist_manager=None, refresh_info=None): + self.config = config or { + "plugin_cycle_interval_seconds": 3600, + "timezone": "UTC", + } + self.playlist_manager = playlist_manager or PlaylistManager() + self.refresh_info = refresh_info or RefreshInfo( + refresh_type="Playlist", + plugin_id="clock", + refresh_time=datetime.now().isoformat(), + image_hash=0, + ) + self.plugin_image_dir = "/tmp" + self.current_image_file = "/tmp/current.png" + + def get_config(self, key=None, default=None): + if key is None: + return self.config + return self.config.get(key, default) + + def get_playlist_manager(self): + return self.playlist_manager + + def get_refresh_info(self): + return self.refresh_info + + def get_plugin(self, plugin_id): + return {"id": plugin_id, "image_settings": []} + + def write_config(self): + pass + + +class MockDisplayManager: + def display_image(self, image, image_settings=[]): + pass + + +def make_plugin_instance(plugin_id="clock", name="Clock", refresh=None, latest_refresh_time=None): + return PluginInstance( + plugin_id=plugin_id, + name=name, + settings={}, + refresh=refresh if refresh is not None else {"interval": 60}, + latest_refresh_time=latest_refresh_time, + ) + + +def make_playlist(plugins, current_plugin_index=0): + return Playlist( + name="Test Playlist", + start_time="00:00", + end_time="24:00", + plugins=[p.to_dict() for p in plugins], + current_plugin_index=current_plugin_index, + ) + + +class TestGetSleepTime: + def test_returns_global_interval_when_no_playlist(self): + device_config = MockDeviceConfig() + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == 3600 + + def test_returns_global_interval_when_no_plugin_interval(self): + plugin = make_plugin_instance(refresh={}) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == 3600 + + def test_returns_plugin_interval_when_shorter_than_global(self): + plugin = make_plugin_instance(refresh={"interval": 60}) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == 60 + + def test_returns_time_until_refresh_when_partially_elapsed(self): + # Plugin was refreshed 30 seconds ago with a 60 second interval + latest_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=30)).isoformat() + plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + # Should be ~30 seconds remaining + assert 25 <= sleep_time <= 35 + + def test_returns_global_interval_when_plugin_interval_larger(self): + plugin = make_plugin_instance(refresh={"interval": 7200}) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == 3600 + + +class TestDetermineNextPlugin: + def test_refreshes_current_plugin_when_its_interval_elapsed(self): + # Current plugin needs refresh (interval elapsed) + latest_refresh = (datetime.now() - timedelta(seconds=120)).isoformat() + plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + # Set global refresh info to be recent so global check would fail + device_config.refresh_info = RefreshInfo( + refresh_type="Playlist", + plugin_id="clock", + refresh_time=datetime.now().isoformat(), + image_hash=0, + ) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now() + ) + + assert result_playlist == playlist + assert result_plugin.name == plugin.name + # Should NOT have rotated to a different plugin + assert playlist.current_plugin_index == 0 + + def test_rotates_to_next_plugin_when_current_does_not_need_refresh(self): + # Current plugin doesn't need refresh (recently refreshed) + latest_refresh = datetime.now().isoformat() + plugin1 = make_plugin_instance(name="Clock1", refresh={"interval": 60}, latest_refresh_time=latest_refresh) + plugin2 = make_plugin_instance(name="Clock2", refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin1, plugin2], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + # Set global refresh info to be old so global check passes + device_config.refresh_info = RefreshInfo( + refresh_type="Playlist", + plugin_id="clock", + refresh_time=(datetime.now() - timedelta(hours=2)).isoformat(), + image_hash=0, + ) + + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now() + ) + + assert result_playlist == playlist + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 + + def test_returns_none_when_no_refresh_needed(self): + # Current plugin doesn't need refresh, and global interval hasn't elapsed + latest_refresh = datetime.now().isoformat() + plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + device_config.refresh_info = RefreshInfo( + refresh_type="Playlist", + plugin_id="clock", + refresh_time=datetime.now().isoformat(), + image_hash=0, + ) + + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now() + ) + + assert result_playlist is None + assert result_plugin is None + + def test_returns_none_when_no_active_playlist(self): + manager = PlaylistManager(playlists=[]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now() + ) + + assert result_playlist is None + assert result_plugin is None \ No newline at end of file From 239fb439b06bae48f4fcfdc70e53f1934f18a89b Mon Sep 17 00:00:00 2001 From: Rs4anti <44727222+Rs4anti@users.noreply.github.com> Date: Sun, 9 Aug 2026 20:36:19 +0200 Subject: [PATCH 2/3] fix bug get_next_plugin --- src/refresh_task.py | 94 ++++++++++++++------ tests/test_refresh_task.py | 171 ++++++++++++++++++++++++++++++------- 2 files changed, 208 insertions(+), 57 deletions(-) diff --git a/src/refresh_task.py b/src/refresh_task.py index 9dc9d02cb..d64ffcbe7 100644 --- a/src/refresh_task.py +++ b/src/refresh_task.py @@ -115,7 +115,17 @@ def _run(self): image_hash = compute_image_hash(image) refresh_info = refresh_action.get_refresh_info() - refresh_info.update({"refresh_time": current_dt.isoformat(), "image_hash": image_hash}) + refresh_info.update({"image_hash": image_hash}) + + # Only update the global rotation time when the plugin actually + # changes (rotation or manual update). In-place refreshes of the + # same plugin (e.g., a clock updating every 60 seconds) must not + # reset the playlist rotation timer. + if self._is_rotation_refresh(refresh_info, latest_refresh): + refresh_info["refresh_time"] = current_dt.isoformat() + else: + refresh_info["refresh_time"] = latest_refresh.refresh_time + # check if image is the same as current image if image_hash != latest_refresh.image_hash: logger.info(f"Updating display. | refresh_info: {refresh_info}") @@ -163,10 +173,10 @@ def _get_current_datetime(self): def _get_sleep_time(self): """Determines how long to sleep before the next refresh check. - Returns the minimum of the global plugin cycle interval and the currently - displayed plugin instance's refresh interval (if it has one), so that - plugins requiring frequent refreshes (e.g., a clock) are re-rendered - while they remain displayed. + Returns the minimum of the time until the next global playlist rotation + and the time until the currently active plugin instance's next refresh, + so that plugins requiring frequent refreshes (e.g., a clock) are re-rendered + while they remain displayed, and the playlist still advances on schedule. """ global_interval = self.device_config.get_config("plugin_cycle_interval_seconds", default=3600) @@ -174,27 +184,46 @@ def _get_sleep_time(self): current_dt = self._get_current_datetime() playlist = playlist_manager.determine_active_playlist(current_dt) - if playlist and playlist.plugins and playlist.current_plugin_index is not None: + if not playlist or not playlist.plugins: + return global_interval + + # Time until the next global rotation (plugin change) + time_until_rotation = global_interval + latest_refresh = self.device_config.get_refresh_info() + latest_refresh_dt = latest_refresh.get_refresh_datetime() + if latest_refresh_dt: + if latest_refresh_dt.tzinfo is None: + latest_refresh_dt = latest_refresh_dt.replace(tzinfo=current_dt.tzinfo) + time_since_rotation = (current_dt - latest_refresh_dt).total_seconds() + time_until_rotation = max(0, global_interval - time_since_rotation) + + # Time until the current plugin's next refresh + time_until_plugin_refresh = global_interval + if playlist.current_plugin_index is not None and playlist.current_plugin_index < len(playlist.plugins): plugin_instance = playlist.plugins[playlist.current_plugin_index] if "interval" in plugin_instance.refresh: interval = plugin_instance.refresh.get("interval") if interval: - # Calculate time remaining until the next refresh latest_refresh_dt = plugin_instance.get_latest_refresh_dt() if latest_refresh_dt: - # Ensure both datetimes are timezone-aware for comparison if latest_refresh_dt.tzinfo is None: latest_refresh_dt = latest_refresh_dt.replace(tzinfo=current_dt.tzinfo) time_since_refresh = (current_dt - latest_refresh_dt).total_seconds() - time_until_refresh = interval - time_since_refresh - if time_until_refresh > 0: - return min(global_interval, time_until_refresh) - return min(global_interval, interval) + time_until_plugin_refresh = max(0, interval - time_since_refresh) + else: + time_until_plugin_refresh = interval - return global_interval + return min(time_until_rotation, time_until_plugin_refresh) def _determine_next_plugin(self, playlist_manager, latest_refresh_info, current_dt): - """Determines the next plugin to refresh based on the active playlist, plugin cycle interval, and current time.""" + """Determines the next plugin to refresh based on the active playlist, plugin cycle interval, and current time. + + Priority: + 1. If the global rotation interval has elapsed, advance to the next plugin. + 2. Otherwise, if the currently displayed plugin instance has its own refresh + interval that has elapsed, refresh it in place. + 3. Otherwise, do nothing. + """ playlist = playlist_manager.determine_active_playlist(current_dt) if not playlist: playlist_manager.active_playlist = None @@ -206,27 +235,40 @@ def _determine_next_plugin(self, playlist_manager, latest_refresh_info, current_ logger.info(f"Active playlist '{playlist.name}' has no plugins.") return None, None - # Check if the currently displayed plugin instance needs a refresh based on its own settings + # 1. Check if it's time to rotate to the next plugin in the playlist. + latest_refresh_dt = latest_refresh_info.get_refresh_datetime() + plugin_cycle_interval = self.device_config.get_config("plugin_cycle_interval_seconds", default=3600) + should_rotate = PlaylistManager.should_refresh(latest_refresh_dt, plugin_cycle_interval, current_dt) + + if should_rotate: + plugin = playlist.get_next_plugin() + logger.info(f"Determined next plugin. | active_playlist: {playlist.name} | plugin_instance: {plugin.name}") + return playlist, plugin + + # 2. Otherwise, check if the currently displayed plugin instance needs a + # refresh based on its own settings (e.g., a clock refreshing every 60s). if playlist.current_plugin_index is not None and playlist.current_plugin_index < len(playlist.plugins): current_plugin = playlist.plugins[playlist.current_plugin_index] if current_plugin.should_refresh(current_dt): logger.info(f"Refreshing currently displayed plugin instance. | plugin_instance: {current_plugin.name}") return playlist, current_plugin - latest_refresh_dt = latest_refresh_info.get_refresh_datetime() - plugin_cycle_interval = self.device_config.get_config("plugin_cycle_interval_seconds", default=3600) - should_refresh = PlaylistManager.should_refresh(latest_refresh_dt, plugin_cycle_interval, current_dt) + # 3. Nothing to do. + latest_refresh_str = latest_refresh_dt.strftime('%Y-%m-%d %H:%M:%S') if latest_refresh_dt else "None" + logger.info(f"Not time to update display. | latest_update: {latest_refresh_str} | plugin_cycle_interval: {plugin_cycle_interval}") + return None, None - if not should_refresh: - latest_refresh_str = latest_refresh_dt.strftime('%Y-%m-%d %H:%M:%S') if latest_refresh_dt else "None" - logger.info(f"Not time to update display. | latest_update: {latest_refresh_str} | plugin_cycle_interval: {plugin_cycle_interval}") - return None, None + def _is_rotation_refresh(self, refresh_info, latest_refresh): + """Determines whether a refresh action represents a rotation (plugin change) + or a manual update, as opposed to an in-place refresh of the same plugin. - plugin = playlist.get_next_plugin() - logger.info(f"Determined next plugin. | active_playlist: {playlist.name} | plugin_instance: {plugin.name}") + In-place refreshes (e.g., a clock updating every 60 seconds) must not reset + the global playlist rotation timer. + """ + if refresh_info.get("refresh_type") != "Playlist": + return True + return refresh_info.get("plugin_instance") != latest_refresh.plugin_instance - return playlist, plugin - def log_system_stats(self): metrics = { 'cpu_percent': psutil.cpu_percent(interval=1), diff --git a/tests/test_refresh_task.py b/tests/test_refresh_task.py index 0a32da416..12c1d40cb 100644 --- a/tests/test_refresh_task.py +++ b/tests/test_refresh_task.py @@ -11,15 +11,17 @@ class MockDeviceConfig: """Mock device config for testing.""" def __init__(self, config=None, playlist_manager=None, refresh_info=None): self.config = config or { - "plugin_cycle_interval_seconds": 3600, + "plugin_cycle_interval_seconds": 600, "timezone": "UTC", } self.playlist_manager = playlist_manager or PlaylistManager() self.refresh_info = refresh_info or RefreshInfo( refresh_type="Playlist", plugin_id="clock", - refresh_time=datetime.now().isoformat(), + refresh_time=datetime.now(pytz.UTC).isoformat(), image_hash=0, + playlist="Test Playlist", + plugin_instance="Clock", ) self.plugin_image_dir = "/tmp" self.current_image_file = "/tmp/current.png" @@ -67,13 +69,26 @@ def make_playlist(plugins, current_plugin_index=0): ) +def make_refresh_info(plugin_instance="Clock", refresh_time=None, image_hash=0): + if refresh_time is None: + refresh_time = datetime.now(pytz.UTC).isoformat() + return RefreshInfo( + refresh_type="Playlist", + plugin_id="clock", + refresh_time=refresh_time, + image_hash=image_hash, + playlist="Test Playlist", + plugin_instance=plugin_instance, + ) + + class TestGetSleepTime: def test_returns_global_interval_when_no_playlist(self): device_config = MockDeviceConfig() task = RefreshTask(device_config, MockDisplayManager()) sleep_time = task._get_sleep_time() - assert sleep_time == 3600 + assert sleep_time == 600 def test_returns_global_interval_when_no_plugin_interval(self): plugin = make_plugin_instance(refresh={}) @@ -83,7 +98,7 @@ def test_returns_global_interval_when_no_plugin_interval(self): task = RefreshTask(device_config, MockDisplayManager()) sleep_time = task._get_sleep_time() - assert sleep_time == 3600 + assert sleep_time == pytest.approx(600, abs=1) def test_returns_plugin_interval_when_shorter_than_global(self): plugin = make_plugin_instance(refresh={"interval": 60}) @@ -116,13 +131,14 @@ def test_returns_global_interval_when_plugin_interval_larger(self): task = RefreshTask(device_config, MockDisplayManager()) sleep_time = task._get_sleep_time() - assert sleep_time == 3600 + assert sleep_time == pytest.approx(600, abs=1) class TestDetermineNextPlugin: def test_refreshes_current_plugin_when_its_interval_elapsed(self): + """A clock with interval 60 is refreshed in place while it's the current plugin.""" # Current plugin needs refresh (interval elapsed) - latest_refresh = (datetime.now() - timedelta(seconds=120)).isoformat() + latest_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=120)).isoformat() plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) playlist = make_playlist([plugin], current_plugin_index=0) manager = PlaylistManager(playlists=[playlist]) @@ -131,15 +147,10 @@ def test_refreshes_current_plugin_when_its_interval_elapsed(self): task = RefreshTask(device_config, MockDisplayManager()) # Set global refresh info to be recent so global check would fail - device_config.refresh_info = RefreshInfo( - refresh_type="Playlist", - plugin_id="clock", - refresh_time=datetime.now().isoformat(), - image_hash=0, - ) + device_config.refresh_info = make_refresh_info(refresh_time=datetime.now(pytz.UTC).isoformat()) result_playlist, result_plugin = task._determine_next_plugin( - manager, device_config.get_refresh_info(), datetime.now() + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) ) assert result_playlist == playlist @@ -147,9 +158,11 @@ def test_refreshes_current_plugin_when_its_interval_elapsed(self): # Should NOT have rotated to a different plugin assert playlist.current_plugin_index == 0 - def test_rotates_to_next_plugin_when_current_does_not_need_refresh(self): + def test_rotates_to_next_plugin_when_global_interval_elapsed(self): + """When plugin_cycle_interval_seconds has elapsed, the playlist advances + to the next plugin even if the current plugin has a refresh interval.""" # Current plugin doesn't need refresh (recently refreshed) - latest_refresh = datetime.now().isoformat() + latest_refresh = datetime.now(pytz.UTC).isoformat() plugin1 = make_plugin_instance(name="Clock1", refresh={"interval": 60}, latest_refresh_time=latest_refresh) plugin2 = make_plugin_instance(name="Clock2", refresh={"interval": 60}, latest_refresh_time=latest_refresh) playlist = make_playlist([plugin1, plugin2], current_plugin_index=0) @@ -157,42 +170,103 @@ def test_rotates_to_next_plugin_when_current_does_not_need_refresh(self): device_config = MockDeviceConfig(playlist_manager=manager) # Set global refresh info to be old so global check passes - device_config.refresh_info = RefreshInfo( - refresh_type="Playlist", - plugin_id="clock", - refresh_time=(datetime.now() - timedelta(hours=2)).isoformat(), - image_hash=0, + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock1", + refresh_time=(datetime.now(pytz.UTC) - timedelta(seconds=700)).isoformat(), + ) + + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + + assert result_playlist == playlist + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 + + def test_rotates_to_next_plugin_even_when_current_plugin_needs_refresh(self): + """When the global rotation interval has elapsed, the playlist advances + to the next plugin even if the current plugin's own refresh interval + has also elapsed. Rotation takes priority.""" + # Current plugin needs refresh (interval elapsed) + current_plugin_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=120)).isoformat() + plugin1 = make_plugin_instance(name="Clock1", refresh={"interval": 60}, latest_refresh_time=current_plugin_refresh) + plugin2 = make_plugin_instance(name="Clock2", refresh={"interval": 60}, latest_refresh_time=datetime.now(pytz.UTC).isoformat()) + playlist = make_playlist([plugin1, plugin2], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + # Set global refresh info to be old so rotation check passes + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock1", + refresh_time=(datetime.now(pytz.UTC) - timedelta(seconds=700)).isoformat(), ) task = RefreshTask(device_config, MockDisplayManager()) result_playlist, result_plugin = task._determine_next_plugin( - manager, device_config.get_refresh_info(), datetime.now() + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) ) assert result_playlist == playlist assert result_plugin.name == plugin2.name assert playlist.current_plugin_index == 1 + def test_new_plugin_can_refresh_after_rotation(self): + """After rotation, the new plugin can be refreshed according to its own interval.""" + # Current plugin (Clock1) doesn't need refresh, but global rotation is due + current_plugin_refresh = datetime.now(pytz.UTC).isoformat() + plugin1 = make_plugin_instance(name="Clock1", refresh={"interval": 60}, latest_refresh_time=current_plugin_refresh) + # New plugin (Clock2) needs refresh (interval elapsed) + new_plugin_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=120)).isoformat() + plugin2 = make_plugin_instance(name="Clock2", refresh={"interval": 60}, latest_refresh_time=new_plugin_refresh) + playlist = make_playlist([plugin1, plugin2], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock1", + refresh_time=(datetime.now(pytz.UTC) - timedelta(seconds=700)).isoformat(), + ) + + task = RefreshTask(device_config, MockDisplayManager()) + + # First call: rotation happens + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 + + # Simulate the rotation refresh completing: update refresh_info to point to Clock2 + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock2", + refresh_time=datetime.now(pytz.UTC).isoformat(), + ) + + # Second call: Clock2 needs refresh (its interval elapsed) + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 # Still on Clock2 + def test_returns_none_when_no_refresh_needed(self): + """No unnecessary refreshes before expiry.""" # Current plugin doesn't need refresh, and global interval hasn't elapsed - latest_refresh = datetime.now().isoformat() + latest_refresh = datetime.now(pytz.UTC).isoformat() plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) playlist = make_playlist([plugin], current_plugin_index=0) manager = PlaylistManager(playlists=[playlist]) device_config = MockDeviceConfig(playlist_manager=manager) - device_config.refresh_info = RefreshInfo( - refresh_type="Playlist", - plugin_id="clock", - refresh_time=datetime.now().isoformat(), - image_hash=0, - ) + device_config.refresh_info = make_refresh_info(refresh_time=datetime.now(pytz.UTC).isoformat()) task = RefreshTask(device_config, MockDisplayManager()) result_playlist, result_plugin = task._determine_next_plugin( - manager, device_config.get_refresh_info(), datetime.now() + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) ) assert result_playlist is None @@ -204,8 +278,43 @@ def test_returns_none_when_no_active_playlist(self): task = RefreshTask(device_config, MockDisplayManager()) result_playlist, result_plugin = task._determine_next_plugin( - manager, device_config.get_refresh_info(), datetime.now() + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) ) assert result_playlist is None - assert result_plugin is None \ No newline at end of file + assert result_plugin is None + + +class TestIsRotationRefresh: + def test_rotation_detected_when_plugin_changes(self): + """A rotation to a different plugin is detected.""" + refresh_info = { + "refresh_type": "Playlist", + "plugin_instance": "Clock2", + } + latest_refresh = make_refresh_info(plugin_instance="Clock1") + task = RefreshTask(MockDeviceConfig(), MockDisplayManager()) + + assert task._is_rotation_refresh(refresh_info, latest_refresh) is True + + def test_in_place_refresh_not_rotation(self): + """An in-place refresh of the same plugin is not a rotation.""" + refresh_info = { + "refresh_type": "Playlist", + "plugin_instance": "Clock1", + } + latest_refresh = make_refresh_info(plugin_instance="Clock1") + task = RefreshTask(MockDeviceConfig(), MockDisplayManager()) + + assert task._is_rotation_refresh(refresh_info, latest_refresh) is False + + def test_manual_update_is_rotation(self): + """A manual update is always treated as a rotation (resets rotation timer).""" + refresh_info = { + "refresh_type": "Manual Update", + "plugin_instance": "Clock1", + } + latest_refresh = make_refresh_info(plugin_instance="Clock1") + task = RefreshTask(MockDeviceConfig(), MockDisplayManager()) + + assert task._is_rotation_refresh(refresh_info, latest_refresh) is True From 77aaceed63efa4a68dd6fa5e46f284d356f9adac Mon Sep 17 00:00:00 2001 From: Rs4anti <44727222+Rs4anti@users.noreply.github.com> Date: Sun, 9 Aug 2026 21:45:07 +0200 Subject: [PATCH 3/3] Add server status plugin --- src/plugins/server_status/icon.png | Bin 0 -> 4596 bytes src/plugins/server_status/plugin-info.json | 9 + src/plugins/server_status/server_status.py | 242 +++++++++++++++++++++ 3 files changed, 251 insertions(+) create mode 100644 src/plugins/server_status/icon.png create mode 100644 src/plugins/server_status/plugin-info.json create mode 100644 src/plugins/server_status/server_status.py diff --git a/src/plugins/server_status/icon.png b/src/plugins/server_status/icon.png new file mode 100644 index 0000000000000000000000000000000000000000..25ad14ceacbe0292dc986f2f835d41462e2b013b GIT binary patch literal 4596 zcmb_A_dgVl`wAs{lv$n1=qh_hE_+kfMHzR<$mVQL5w5II=r}uM?|mmkcylu2a8`6@ z-5DLu`0D#Ne4f|qnXhO3_AJB`V;vSoenuJ^8WufWO;Z}03z&b%KznXE8v;H)XRc2S z&9%E7hq}7D z005w;r^n9D&dA7UVq&7Cq(nnQqot)~U|>K=Nhvlqwx*`W)YKFPgQcaVxx2eFF)^K2 zWAn+Vr=j7{*3(oo56aq{BLtK+b6y(BeGP2P6m?*3**;6=5zXdGx+2`!^#=zE-M3;N z_*^uK6F86k-$19|M*D?}b6!^{)4`i;Pm~+C6y7QaWbl zg~l(Wzm*IDK^nGrVweKdTuzS;VCSAi`Lgh0R>LA{!lL1VuJtkFxnIw^?of`?6&=d` zZoYKsBWk=|$Jog*sr!o@yqJ&I{f$*JZN;IoT;!k52XQ9+9XY!G_S`~3^6>EE>l?H- z@e5J>=m@$bBg0gttAm<`o9I0j`;uR;pN8wuIwqz9Ve8M`C`CM_F#E5wYS=Zj`+cXy zX2!FW+5oE2DkFXWV3YJc>9^n)AZmSdXN57uHSu8xh(oobq)+F4c5foXf*2KZP0(_D z{I<*es~GeJAqlnmw+F|0f{{(Ld&&%1DlK785IOtC@pcoWZRl`gMn}uaP8~0)-?Wbh8YK50bR zP%N(KI1APMWB#}Lq3rJWeWu%ypxRI*Sab1MNn^<(xWma57*>pDz)c44J2T8_ggA!* z8%xwNzbKTOKGP@jNu^m1-JXn>^vKo{-o5B;$V_n;xA~wWUGZ5IUp=e}dcY&Bdo7)!5jp zZM6rlY-jguOO)+#{`jQQ(~r!`Uipak^M|RU$HkAG_O~s%ML&EFOMx!49aZMLU?NtL zVPy}W@x}~_oQ0&ND?|}pUXq^jPBq17MeVM59~E9f7OkrEIr@G+Ds81k#XMJ$m5ZDw zr=#cgo#&`d(tD|i-Q_DIG2IH9nf#r(1@8Roxcp+x?Om%$5@3=5$0K<=?aQhOkW5|~ ziAdX~&!p@b=dp{r`aCsZS(y!`uFKem%t5MkF&t5sS*yAWZT}=ndI#~X(k08EvoFkd zEM>i$dlGoH%iR{d`{QUvIXBz6hhgx!!-`i6fxqFcn5|>hRt!1bv6#B9s#ZI5@h&G< z?L&`w7RXOekywTgsCcbAa9aBk{dP~dSX5~NCT;BB3V%%o z9}1SGI8%I7udrMVlMHIb~ZZdoQjkXmC0#efX<)I+6cL+~jy3KIz8E zR({TQAMS%0*s2W-S1B$Ewk>{q_?Rh))4J+6$|mzJn5LPow&`hhwxc1TGD!d~%y;~o zg*m$!o!iQIt9X!?-ESiBF_ot*J7%FBRlT+q@a|)zZ0O$Umn8F$?cXNnK5@Ugm$NCo^y_Q!tzCl?3;Y#sgbR2r8M<+Qr+>NI&pWnUfA?s1IuN0kF+pe5lrsw>0tb~9sd6}CqtVRD?WB%H(r&pYm zhMrhN78b56O(J;IW>NzKZ!|QC5=B?~$o#|%?E{g*%@(!c|ryrX3{bVGfiz{(E zdwO>&GG<~CzFf{V{RZ;b0ru)w2(R80J?<1c1MPBE;FMNUectd_WHYmFxvCNz8bd>D zN6C_CM18*82_f3vTH3#=BI^mDuSxIwTpW?dQp4sp4`hG>7N^CS$=>MTLk*XxK=R&- zcm2g5??o{vPb-a9h}XPDY$$X_b|zNs2CqVyKHTK+A^$KR9ZTt(j{am&C&;ZWYDhO^X{ z>YtfBepN?vu-vuSCBtW0pAQbd4&-h6(>v(rQF|VWWz!Xdybl$30 ze}9?!D&KJ1SLe6*m-})x74Dr?&=xZWyVy&Oh-_X zHPf+;ugQ>BI)q9{sjzKm>vaeIV(x39xq%%>eo(2DUZ+nhsQcFe;f1xI2=#t^0$`0_ z;MA@4(#ZIM>kzgXDGn+vC67%XcXSv@T|9=m9q*w`P+PTrX3LaHn9|GEGFz4e09%9j zcgNH@sq>jH>t%#D3II3ek-2m_aQf^R9~8!iH0ntE_Iq4%!5*_M`#Fx~g9K@?_y1hgo} ze5;_opB<-`Bf;`?GyH zv*lvoVNW{E8lZP6G=bcGSU~nZHHiXZ@ooloq`#}n$D1zYKmUT8Q#WshH5{~rbupSh zr!Dc#p1o6b4Wl5Mwk#tNj;MF6Qg#(ZYH>c@g%Z@tG;nXZ9I>N8TGa;VTtdV~p-b4> zK4;dpe88%IUVKk8wv)-o0((GAMd-?(-m2@EGK5mNgHSe;Lw(I}N{TP@D>{f^g8*$w z9yT-c2ExOw5^*@ks;rCVW7n_)gmqP}$Q_T4TIBc5D2x%)dsJF-tTmeMFr@!Xy@{_XD~huI*{=73 zi%8@*K<$03O%6F}Th;IkF(0p^?}87kq_q_phgn~;%n1i%hsmdamIKkdrR%Iq2iBp_ zzo@qtQ1>+8>Ri84lH#Qjl&`~T``eIKilIhYSwTu2=DQ)G&qWpwS6CRK%66qGu*(9Dpf&t;{v1UvDNf zMTx6d){8waG|PDx=+`i`^O6UHnsIh;pP^}Ou+D$*>2r-8spgqjH^-Wp@yoXCC6Be% zu~hL$A|KewaDVg7sHsY6J~iNr@DFn~k$u=7p*w0-5#TU-TizCLf8Y~U;r;^0{{<%Dn_HV_4 zYX3Fo`aol8d811*;Rgz;8ZQ8Ew4E-f{1_wa86V7(F8(q6u=HBb3Rj5wPkgs1>vjjC zafNp<`NOc!sroyIcD`pxA;^0Ok+!LCD;&q3U33oWF|cC_8LV)BUGno*&Fl(aeYgK1 zUj^1sDPjBiR~-w&a3$-ol+&G1!zq&qA4NEye7_?BfdPFm?zusGlK`nD zY2CV3NwtozOT}^qXx86_T+Rl$!hJ>JNI94-nlzqemi5hliYOMVHs&%BO@ThRgh$q< z-_8h|hH%IV0Vf<18rBdl2^reEB5h;AkY-5;doyy&c(e$Tcfp1ix-^;2+3;^uODx-H z4_nOJs#3-fDpzc8x<2N2uX^rE+4V2SMXTg~3n;p9wUm#x(?V?S)INc4aXps3PpK)& zwTshFq}s+#Y_9uF=2di4Qv#;;0MU{XB9_)x3CRW8qyx(MwHczgy)YFnybxQ0avsZ_ zgK`~S?MH(dZ!!1hNf0{cys?}n+shM!2>0kZ=xrv=EgiG$?fH!r49iDAzX=`i0fiDfIW0CLGO)^cQzv@cAe2t9o8 zwTdD{7a%6mKj)jc`^T6ybLcN+Jet7{&}pYHxJG%-g9Vn8`%@!sb|0v=V?yg@&qj6R zVGG^_vc=ME*v+jU$_EK@MqX)%lSz1nM%O+-)OeK%ySpofwN8Zhjc=NpS zZ6?;kbX#uaep{)qzV_N!_e|C!)omDu=G5=m)vq5oAWA_JN5jfrE%E{L3t4YHBP{4R z&;~r+4BS6CeM_>7iajRllh*d@5lq?`zp-s>cd)*SPW97wIfAGRu`#$={`qxtRK+u3 zr3>mU@aI>~7xD1H_+=NY1!J9QGGXS5Jq7+k+M^1%Pkm-`671-3}q#`%Ner;(8Z3@+_a~v1qNQ`)-)|2iJjHZET2e_^&}5lG}~tRE*Wh9Z+;m6vXIKo YXdlxHS#D`DPW{vMw2UAEcHFp#T5? literal 0 HcmV?d00001 diff --git a/src/plugins/server_status/plugin-info.json b/src/plugins/server_status/plugin-info.json new file mode 100644 index 000000000..9ef6b1467 --- /dev/null +++ b/src/plugins/server_status/plugin-info.json @@ -0,0 +1,9 @@ +{ + "name": "Server Status", + "id": "server_status", + "description": "Shows WireGuard and Pi-hole status", + "version": "1.0.0", + "author": "robi", + "entry_point": "server_status.py", + "class": "ServerStatus" +} diff --git a/src/plugins/server_status/server_status.py b/src/plugins/server_status/server_status.py new file mode 100644 index 000000000..1b7d1124d --- /dev/null +++ b/src/plugins/server_status/server_status.py @@ -0,0 +1,242 @@ +import subprocess +import logging +from datetime import datetime + +from PIL import Image, ImageDraw +from plugins.base_plugin.base_plugin import BasePlugin +import socket +import psutil + + + +logger = logging.getLogger(__name__) + + +class ServerStatus(BasePlugin): + + def __init__(self, config): + super().__init__(config) + self.config = config + + def generate_settings_template(self): + return super().generate_settings_template() + + def generate_image(self, settings, device_config): + + status = self.get_status() + + w, h = device_config.get_resolution() + + img = Image.new("RGB", (w, h), "white") + draw = ImageDraw.Draw(img) + + # Font opzionale + try: + from PIL import ImageFont + font = ImageFont.truetype( + "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + 18 + ) + title_font = ImageFont.truetype( + "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", + 22 + ) + except Exception: + font = None + title_font = None + + + # Titolo + draw.text( + (20, 15), + status["title"], + fill="black", + font=title_font + ) + + + # Linea separazione + draw.line( + (w // 2, 45, w // 2, h - 30), + fill="black", + width=1 + ) + + + # Colonna sinistra - Sistema + x1 = 15 + y1 = 55 + + system_lines = [ + f"Temp : {status['temperature']} C", + f"CPU : {status['cpu']:.0f} %", + f"RAM : {status['ram']:.0f} %", + f"Disk : {status['disk']:.0f} %", + f"Up : {status['uptime']}", + ] + + for text in system_lines: + draw.text( + (x1, y1), + text, + fill="black", + font=font + ) + y1 += 25 + + + # Colonna destra - Servizi + x2 = w // 2 + 15 + y2 = 55 + + service_lines = [ + f"Net: {status['network']}", + f"IP: {status['ip']}", + "", + f"Pi-hole:", + f" {status['pihole']}", + "", + f"WireGuard:", + f" {status['wireguard']}", + f"Peers: {status['peers']}", + ] + + for text in service_lines: + draw.text( + (x2, y2), + text, + fill="black", + font=font + ) + y2 += 23 + + + # Ora aggiornamento + draw.text( + (20, h - 25), + f"Aggiornato: {status['time']}", + fill="black", + font=font + ) + + + return img + + + + def get_status(self): + + system = self.get_system_info() + + return { + "title": "RASPI STATUS", + + "temperature": system["temperature"], + "cpu": system["cpu"], + "ram": system["ram"], + "disk": system["disk"], + "uptime": system["uptime"], + + "network": system["network"], + "ip": system["ip"], + + "pihole": self.service_status("pihole-FTL"), + "wireguard": self.service_status("wg-quick@wg0"), + "peers": self.wg_peers(), + + "time": datetime.now().strftime("%H:%M") + } + + + def get_system_info(self): + + # Temperatura CPU + try: + with open("/sys/class/thermal/thermal_zone0/temp") as f: + temperature = round(int(f.read()) / 1000, 1) + except Exception: + temperature = None + + # CPU + cpu = psutil.cpu_percent(interval=0.5) + + # RAM + ram = psutil.virtual_memory().percent + + # Disco (/) + disk = psutil.disk_usage("/").percent + + # Uptime + boot = datetime.fromtimestamp(psutil.boot_time()) + uptime = datetime.now() - boot + + days = uptime.days + hours = uptime.seconds // 3600 + + if days > 0: + uptime_str = f"{days}d {hours}h" + else: + uptime_str = f"{hours}h" + + # IP locale + ip = "-" + + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + ip = s.getsockname()[0] + s.close() + network = "OK" + except Exception: + network = "DOWN" + + return { + "temperature": temperature, + "cpu": cpu, + "ram": ram, + "disk": disk, + "uptime": uptime_str, + "network": network, + "ip": ip + } + + + + def service_status(self, service): + try: + result = subprocess.check_output( + ["systemctl", "is-active", service], + text=True + ).strip() + + return "OK" if result == "active" else "DOWN" + + except Exception: + return "DOWN" + + + def wg_peers(self): + try: + data = subprocess.check_output( + ["wg", "show", "wg0", "latest-handshakes"], + text=True + ) + + now = int(datetime.now().timestamp()) + active = 0 + + for line in data.strip().splitlines(): + parts = line.split() + + if len(parts) == 2: + timestamp = int(parts[1]) + + if timestamp > now - 300: + active += 1 + + return active + + except Exception as e: + logger.error(e) + return -1 +