From 8f44447773706e7f67e79be43d3fb9b4eb2ff33a Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Aug 2026 10:11:12 -0400 Subject: [PATCH 1/2] fix(youtube-stats): stop retrying a failing API once per frame display() re-entered update() whenever channel_stats was empty: if not self.channel_stats: self.update() and update() assigns the fetch result unconditionally, so any failure -- missing key, HTTP error, exhausted quota, network down -- left it empty and the next frame tried again. That is one requests.get(timeout=10) per rendered frame, on the render thread. Two consequences, both self-sustaining. The render thread can block for the full 10s timeout per frame. And the YouTube Data API's default quota is 10,000 units/day, so a broken key burns it down in minutes -- at which point quota exhaustion becomes the error that keeps the result empty and the loop fed. The cache in _get_channel_stats never absorbs this: a failed fetch caches nothing. display() now gates on whether a fetch has been ATTEMPTED rather than on whether it produced data, and update() is throttled to update_interval. The core already calls update() at about that cadence, so the throttle is a no-op for it; it exists to bound the display-driven path. monotonic(), not time(): these Pis have no RTC, so a wall-clock jump at NTP sync must not make the next attempt look due or centuries away. Tests: new test_no_per_frame_api_storm.py -- 200 frames against a permanently failing API cost one call, the retry lands once the interval elapses, a working API is unaffected. 6 of its 9 checks fail against the pre-fix manager. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW --- plugins/youtube-stats/manager.py | 36 +++++- plugins/youtube-stats/manifest.json | 8 +- .../test_no_per_frame_api_storm.py | 116 ++++++++++++++++++ 3 files changed, 155 insertions(+), 5 deletions(-) create mode 100644 plugins/youtube-stats/test_no_per_frame_api_storm.py diff --git a/plugins/youtube-stats/manager.py b/plugins/youtube-stats/manager.py index be4a4865..e11e0619 100644 --- a/plugins/youtube-stats/manager.py +++ b/plugins/youtube-stats/manager.py @@ -8,6 +8,7 @@ """ import os +import time from pathlib import Path from typing import Dict, Any, Optional from PIL import Image, ImageDraw, ImageFont @@ -72,6 +73,11 @@ def _rgb(value, default): self.font = None self.youtube_logo = None self.last_displayed_stats: Optional[Dict[str, Any]] = None # Track last displayed to prevent unnecessary redraws + # A fetch has been attempted at least once, and when. display() used to + # retry on an empty result, so a failing API meant a request per FRAME; + # these gate on "did we try" and "how long ago" instead. + self._has_fetched: bool = False + self._last_attempt: float = 0.0 self._api_key_error: Optional[str] = None # Set on any config/auth problem that # prevents fetching stats (missing/invalid API key, missing channel ID) -- display() # shows this on-screen instead of leaving a blank canvas when it's set. @@ -325,10 +331,30 @@ def _create_display(self, channel_stats: Dict[str, Any]) -> Optional[Image.Image return None def update(self) -> None: - """Fetch/update data for this plugin.""" + """Fetch/update data for this plugin. + + Throttled to update_interval. The core already calls this on roughly + that cadence, so this changes nothing for it -- it exists because + display() also calls update() for the first paint, and a failing fetch + leaves channel_stats empty. Without the throttle that combination + issued one YouTube API request per rendered frame, on the render + thread, each able to block for the full 10s timeout. The Data API's + default quota is 10k units/day, so a broken key burned through it in + minutes -- and quota exhaustion is itself an error, which kept the + result empty and the loop fed. + + monotonic(), not time(): these Pis have no RTC, so a wall-clock jump + at NTP sync must not make the next attempt look due (or centuries away). + """ if not self.enabled: return - + + now = time.monotonic() + if self._has_fetched and (now - self._last_attempt) < self.update_interval_config: + return + self._has_fetched = True + self._last_attempt = now + self.channel_stats = self._get_channel_stats() def display(self, force_clear: bool = False) -> None: @@ -336,8 +362,10 @@ def display(self, force_clear: bool = False) -> None: if not self.enabled: return - # Fetch stats if we don't have them yet - if not self.channel_stats: + # First paint can land before the core's first update() tick. Gate on + # whether a fetch has been ATTEMPTED, not on whether it produced data: + # gating on the result meant a failing API was retried every frame. + if not self._has_fetched: self.update() if self.channel_stats: diff --git a/plugins/youtube-stats/manifest.json b/plugins/youtube-stats/manifest.json index 843fca83..16b7b532 100644 --- a/plugins/youtube-stats/manifest.json +++ b/plugins/youtube-stats/manifest.json @@ -1,7 +1,7 @@ { "id": "youtube-stats", "name": "YouTube Stats", - "version": "1.1.0", + "version": "1.1.1", "author": "ChuckBuilds", "description": "Display YouTube channel statistics including subscriber count, total views, and channel name on your LED matrix", "category": "social", @@ -33,6 +33,12 @@ } ], "versions": [ + { + "version": "1.1.1", + "released": "2026-08-22", + "ledmatrix_min_version": "2.0.0", + "notes": "Stop retrying a failing YouTube API once per rendered frame. display() re-ran update() whenever channel_stats was empty, and update() leaves it empty on any failure, so a bad key or an exhausted quota produced one requests.get(timeout=10) per frame on the render thread. display() now gates on whether a fetch was attempted rather than on its result, and update() is throttled to update_interval, so a broken API costs one request per interval instead of one per frame." + }, { "version": "1.1.0", "released": "2026-07-31", diff --git a/plugins/youtube-stats/test_no_per_frame_api_storm.py b/plugins/youtube-stats/test_no_per_frame_api_storm.py new file mode 100644 index 00000000..e4998e68 --- /dev/null +++ b/plugins/youtube-stats/test_no_per_frame_api_storm.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +""" +Regression test: a failing YouTube API must not be retried once per frame. + +display() used to call update() whenever channel_stats was empty: + + if not self.channel_stats: + self.update() + +update() assigns the fetch result unconditionally, so any failure -- missing +key, HTTP error, quota exhaustion, network down -- left channel_stats empty and +the next frame tried again. That is one requests.get(timeout=10) per rendered +frame, on the render thread, against an API whose default quota is 10k +units/day. Quota exhaustion is itself an error, so the loop fed itself. + +display() now gates on whether a fetch was ATTEMPTED, and update() is throttled +to update_interval, so a broken API costs one request per interval. + +Run: /bin/python plugins/youtube-stats/test_no_per_frame_api_storm.py +""" + +import sys +import types +from pathlib import Path + +plugin_dir = Path(__file__).parent +sys.path.insert(0, str(plugin_dir)) + + +def _stub_core_src(): + def mod(name, **attrs): + m = types.ModuleType(name) + for k, v in attrs.items(): + setattr(m, k, v) + sys.modules.setdefault(name, m) + return m + + mod("src") + mod("src.plugin_system") + mod("src.plugin_system.base_plugin", BasePlugin=object, VegasDisplayMode=object) + + +_stub_core_src() + +import logging # noqa: E402 +import time # noqa: E402 + +from manager import YouTubeStatsPlugin # noqa: E402 + +results = [] + + +def check(case, passed): + results.append((case, passed)) + print(f" [{'pass' if passed else 'FAIL'}] {case}") + + +def make_plugin(update_interval=300): + p = YouTubeStatsPlugin.__new__(YouTubeStatsPlugin) + p.logger = logging.getLogger("test-youtube-storm") + p.enabled = True + p.channel_stats = None + p.update_interval_config = update_interval + p._has_fetched = False + p._last_attempt = 0.0 + p._api_key_error = None + return p + + +# --- a permanently failing API ------------------------------------------- +plugin = make_plugin() +calls = [] +plugin._get_channel_stats = lambda: calls.append(1) or None + +# The core's own tick, then a burst of frames. +plugin.update() +for _ in range(200): + if not plugin._has_fetched: + plugin.update() + +check("a failing fetch is attempted once per interval, not once per frame", + len(calls) == 1) +check("the failure is recorded as attempted", plugin._has_fetched is True) +check("channel_stats stays empty (no fabricated data)", not plugin.channel_stats) + +# --- the throttle expires ------------------------------------------------- +plugin._last_attempt = time.monotonic() - 301 +plugin.update() +check("a retry happens once the interval has elapsed", len(calls) == 2) + +# --- the throttle does not block a successful first fetch ---------------- +ok = make_plugin() +ok_calls = [] +ok._get_channel_stats = lambda: ok_calls.append(1) or {"subscribers": 5} +ok.update() +check("a working API still populates stats", ok.channel_stats == {"subscribers": 5}) +check("and cost exactly one call", len(ok_calls) == 1) + +# --- the guard display() uses --------------------------------------------- +import inspect # noqa: E402 +disp = inspect.getsource(YouTubeStatsPlugin.display) +check("display() no longer gates on the result", + "if not self.channel_stats:\n self.update()" not in disp) +check("display() gates on whether a fetch was attempted", + "self._has_fetched" in disp) + +upd = inspect.getsource(YouTubeStatsPlugin.update) +check("update() uses monotonic (these Pis have no RTC)", "time.monotonic()" in upd) + +print() +failed = [case for case, passed in results if not passed] +print(f"{len(results) - len(failed)}/{len(results)} passed") +if failed: + for case in failed: + print(f" FAILED: {case}") + sys.exit(1) From 8de5ba9045fb6729f684107825925f376048b2d9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 23 Aug 2026 12:11:40 -0400 Subject: [PATCH 2/2] test(youtube-stats): construct the stand-in via object.__new__ Codacy flagged line 59 as "No value for argument 'cls' in classmethod call". It is a false positive -- YouTubeStatsPlugin.__new__(YouTubeStatsPlugin) is valid -- but only this test triggers it, because its stub makes BasePlugin `object`, so __new__ resolves to object.__new__ and the explicit class argument reads as a bound call missing its cls. object.__new__(YouTubeStatsPlugin) says the same thing unambiguously. The sibling tests in hockey and soccer keep Cls.__new__(Cls); their stubs do not create the ambiguity. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW --- plugins/youtube-stats/test_no_per_frame_api_storm.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/plugins/youtube-stats/test_no_per_frame_api_storm.py b/plugins/youtube-stats/test_no_per_frame_api_storm.py index e4998e68..15d52a2e 100644 --- a/plugins/youtube-stats/test_no_per_frame_api_storm.py +++ b/plugins/youtube-stats/test_no_per_frame_api_storm.py @@ -56,7 +56,11 @@ def check(case, passed): def make_plugin(update_interval=300): - p = YouTubeStatsPlugin.__new__(YouTubeStatsPlugin) + # object.__new__ rather than YouTubeStatsPlugin.__new__: the stub above + # makes BasePlugin `object`, so the latter resolves to object.__new__ and + # static analysis reads the explicit class argument as a bound-method call + # with a missing `cls`. Same result, no false positive. + p = object.__new__(YouTubeStatsPlugin) p.logger = logging.getLogger("test-youtube-storm") p.enabled = True p.channel_stats = None