-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmanager.py
More file actions
3156 lines (2842 loc) · 158 KB
/
Copy pathmanager.py
File metadata and controls
3156 lines (2842 loc) · 158 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
MLB Scoreboard plugin for ChuckBuilds/LEDMatrix.
Layout:
- Left half: two team columns side by side, each full panel height.
Logo fills nearly the whole column; a darkened bar across the
bottom holds the bold "ABBR SCORE" text for contrast.
- Right half (black background):
- upper-left: inning indicator (anti-aliased triangle + number)
- upper-right: diamond of bases (anti-aliased, configurable colors)
- lower-left: ball-strike count
- lower-right: outs indicator (configurable colors)
By default this cycles through every currently-live MLB game leaguewide
every `game_rotation_seconds`. Set `show_favorite_teams_only: true` to
restrict rotation to your favorite teams' live games. Falls back to
your favorite team's most recent/upcoming game if nothing is live.
FONT: rather than hardcoding a guessed filename, this scans
assets/fonts/ at startup and picks a real font shipped with your
LEDMatrix install (preferring anything that looks like a pixel/arcade
font, e.g. Press Start 2P, since that's the style the rest of the
project's plugins use). Falls back to a system font if that folder
isn't found. Team abbreviation/score text size is fit dynamically to
the column width so it can never overflow regardless of which font
gets picked up.
Data comes from ESPN's public scoreboard API:
https://site.api.espn.com/apis/site/v2/sports/baseball/mlb/scoreboard
NOTE ON display_manager: different LEDMatrix versions have exposed the
PIL image slightly differently over time. This plugin builds its own
RGB PIL.Image internally and then tries, in order:
1. display_manager.image.paste(...) + display_manager.update_display()
2. display_manager.set_image(...)
"""
import datetime
import logging
import math
import os
import random
import re
import threading
import time
from io import BytesIO
from typing import Optional, Dict, Any, Tuple, List
import requests
from PIL import Image, ImageDraw, ImageFont
try:
from src.plugin_system.base_plugin import BasePlugin
except ImportError:
class BasePlugin: # type: ignore
def __init__(self, plugin_id, config, display_manager, cache_manager, plugin_manager):
self.plugin_id = plugin_id
self.config = config
self.display_manager = display_manager
self.cache_manager = cache_manager
self.plugin_manager = plugin_manager
self.logger = logging.getLogger(plugin_id)
ESPN_SCOREBOARD_URL = "https://site.api.espn.com/apis/site/v2/sports/baseball/mlb/scoreboard"
ESPN_SUMMARY_URL = "https://site.api.espn.com/apis/site/v2/sports/baseball/mlb/summary"
DEFAULT_AWAY_COLOR = (255, 255, 255)
DEFAULT_HOME_COLOR = (255, 255, 255)
# Fonts bundled directly with this plugin (in ./fonts/), pulled from the
# same assets/fonts/ folder the core LEDMatrix project ships with, so
# the look matches the rest of your display without depending on
# discovering files from the main install at runtime.
#
# Measured glyph widths at various sizes (see plugin README) show
# Press Start 2P is quite wide per character -- it doesn't comfortably
# fit "ABBR SCORE" in a 32px-wide column even at very small sizes
# without becoming unreadably tiny. 5by7 and 4x6 are much more compact
# pixel fonts and are the better fit for that particular row; Press
# Start 2P remains a selectable option since it may suit other layouts
# or wider panels.
PLUGIN_DIR = os.path.dirname(os.path.abspath(__file__))
FONT_CHOICES = {
"5by7": os.path.join(PLUGIN_DIR, "fonts", "5by7_regular.ttf"),
"4x6": os.path.join(PLUGIN_DIR, "fonts", "4x6-font.ttf"),
"press_start_2p": os.path.join(PLUGIN_DIR, "fonts", "PressStart2P-Regular.ttf"),
"tom_thumb": os.path.join(PLUGIN_DIR, "fonts", "tom-thumb.bdf"),
"system": None,
}
# Font choices backed by a real bitmap format (BDF) rather than a
# scalable TrueType outline. These render every pixel exactly as
# designed with zero anti-aliasing/rasterization softness -- no
# FreeType involved at all -- and are NOT resizable (BDF is a single
# fixed pixel size), so they skip the shrink-to-fit sizing logic used
# for the TTF options.
BDF_FONT_CHOICES = {"tom_thumb"}
# Routine pitch-by-pitch play types to SKIP when last_play_filter is
# "significant" -- i.e., don't flash for these, only for actual outcomes
# (hits, walks, strikeouts, outs, runs, etc.). This is a denylist rather
# than an allowlist of "big moment" types on purpose: only two real
# samples of situation.lastPlay.type.type have been confirmed so far
# ("ball" and "start-batterpitcher"), so a denylist means an unknown
# type defaults to SHOWING (safer for not missing a real highlight)
# rather than defaulting to hidden. Add more here if routine updates
# still slip through once you see this against more live data.
NON_SIGNIFICANT_PLAY_TYPES = {
"ball", "strike", "strike-looking", "strike-swinging",
"foul", "foul-ball", "foul-tip",
"start-batterpitcher", "pitch", "no-pitch",
"automatic-ball", "automatic-strike", "warmup",
}
# Best-effort guess at ESPN's home-run type code -- confirmed WRONG in
# practice (real home runs were observed not triggering the animation).
# Kept as one signal, but no longer the only one -- see
# _is_home_run_play, which also checks the play's actual narrative text
# for "home run"/"homers", a far more reliable signal since that text
# is human-readable and ESPN's phrasing for a home run call
# ("Judge homers to right field") is predictable regardless of
# whatever internal type code they use.
HOME_RUN_PLAY_TYPES = {"home-run", "homerun", "home_run", "hr", "home run"}
HOME_RUN_TEXT_KEYWORDS = ("home run", "homers", "homered")
# Preference order for auto-discovering a bundled font from the main
# LEDMatrix install, used only when font_choice is "system" or the
# selected bundled file is missing for some reason.
FONT_NAME_PREFERENCE = ["press", "pixel", "matrix", "arcade", "8x8", "4x6", "retro"]
class BDFFont:
"""Minimal BDF (Glyph Bitmap Distribution Format) parser and
renderer. Pillow's ImageFont.truetype() can't load .bdf files at
all, and BDF glyphs are exact per-pixel bitmaps rather than vector
outlines -- so drawing them is just copying 1-bit pixel data
directly, with no rasterization/anti-aliasing step to introduce any
softness or halo. This is intentionally tiny: it only implements
enough of BDF to render basic Latin text (letters, digits, and the
handful of punctuation marks this plugin actually uses)."""
def __init__(self, path: str):
self.glyphs: Dict[int, Dict[str, Any]] = {}
self.ascent = 0
self.descent = 0
self._parse(path)
def _parse(self, path: str):
with open(path, "r", errors="replace") as f:
lines = f.read().splitlines()
i, n = 0, len(lines)
cur: Optional[Dict[str, Any]] = None
while i < n:
line = lines[i].strip()
if line.startswith("FONT_ASCENT"):
self.ascent = int(line.split()[1])
elif line.startswith("FONT_DESCENT"):
self.descent = int(line.split()[1])
elif line.startswith("STARTCHAR"):
cur = {}
elif line.startswith("ENCODING") and cur is not None:
cur["encoding"] = int(line.split()[1])
elif line.startswith("DWIDTH") and cur is not None:
cur["dwidth"] = int(line.split()[1])
elif line.startswith("BBX") and cur is not None:
p = line.split()
cur["bbw"], cur["bbh"] = int(p[1]), int(p[2])
cur["bbxoff"], cur["bbyoff"] = int(p[3]), int(p[4])
elif line.startswith("BITMAP") and cur is not None:
rows = []
for _ in range(cur.get("bbh", 0)):
i += 1
hexrow = lines[i].strip()
nbits = len(hexrow) * 4
val = int(hexrow, 16) if hexrow else 0
bits = [(val >> (nbits - 1 - b)) & 1 for b in range(cur["bbw"])]
rows.append(bits)
cur["rows"] = rows
elif line.startswith("ENDCHAR") and cur is not None:
if "encoding" in cur:
self.glyphs[cur["encoding"]] = cur
cur = None
i += 1
def _glyph(self, ch: str) -> Optional[Dict[str, Any]]:
return self.glyphs.get(ord(ch))
def textbbox(self, text: str) -> Tuple[int, int, int, int]:
"""Mimics ImageDraw.textbbox((0,0), text, font=...) closely
enough for this plugin's centering/width-fit math: returns
(left, top, right, bottom) with (0,0) as the text origin."""
cursor_x = 0
min_top: Optional[int] = None
max_bottom: Optional[int] = None
for ch in text:
g = self._glyph(ch)
if g is None:
cursor_x += 4
continue
glyph_top = self.ascent - (g["bbyoff"] + g["bbh"])
glyph_bottom = glyph_top + g["bbh"]
min_top = glyph_top if min_top is None else min(min_top, glyph_top)
max_bottom = glyph_bottom if max_bottom is None else max(max_bottom, glyph_bottom)
cursor_x += g.get("dwidth", 4)
if min_top is None:
min_top, max_bottom = 0, 0
return (0, min_top, cursor_x, max_bottom)
def draw(self, image: Image.Image, xy: Tuple[int, int], text: str, fill: Tuple[int, int, int]):
x0, y0 = xy
cursor_x = x0
img_w, img_h = image.size
for ch in text:
g = self._glyph(ch)
if g is None:
cursor_x += 4
continue
glyph_top = self.ascent - (g["bbyoff"] + g["bbh"])
for row_idx, row in enumerate(g.get("rows", [])):
py = y0 + glyph_top + row_idx
if py < 0 or py >= img_h:
continue
for col_idx, bit in enumerate(row):
if not bit:
continue
px = cursor_x + g["bbxoff"] + col_idx
if 0 <= px < img_w:
image.putpixel((px, py), fill)
cursor_x += g.get("dwidth", 4)
class TidbytBaseballPlugin(BasePlugin):
def __init__(self, plugin_id, config, display_manager, cache_manager, plugin_manager):
super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
self.logger = logging.getLogger(f"plugin.{plugin_id}")
self._derive_settings()
self.session = requests.Session()
self.session.headers.update({"User-Agent": "LEDMatrix-TidbytBaseball/1.0"})
self.live_games: List[Dict[str, Any]] = []
self.past_games: List[Dict[str, Any]] = []
self.upcoming_games: List[Dict[str, Any]] = []
self.rotation_games: List[Dict[str, Any]] = [] # what _current_game()/_maybe_rotate() actually cycle through
# ESPN's scoreboard endpoint defaults to TODAY's games only with
# no date parameter -- a favorite team's actual next game is
# almost always tomorrow or later (not today), so relying on the
# main scoreboard fetch alone means upcoming_games is nearly
# always empty. This cache holds results from explicitly querying
# the next few days (see _fetch_future_upcoming_games), refreshed
# on its own slower timer since schedules barely change.
self._cached_future_upcoming_games: List[Dict[str, Any]] = []
self._upcoming_last_fetch_time: float = 0.0
# Same issue, opposite direction: ESPN's main scoreboard call
# only covers TODAY, so a favorite team's most recent completed
# game (if it was yesterday or earlier -- an off-day today, or
# today's game hasn't finished) never shows up as a past game
# without explicitly looking backward too.
self._cached_past_lookback_games: List[Dict[str, Any]] = []
# Tracks which final games have already had their hits/errors
# enriched from the summary endpoint -- since a completed game's
# stats can't change, this ensures we only ever fetch it once
# per game rather than re-fetching every poll it stays in rotation.
self._enriched_boxscore_event_ids: set = set()
self._past_last_fetch_time: float = 0.0
self.fallback_game: Optional[Dict[str, Any]] = None
self.current_index: int = 0
self.last_switch_time: float = time.time()
self.last_fetch_time: float = 0.0
# Per-game (keyed by event_id) last-play flash state. Games get
# entirely new dicts each poll (live_games is rebuilt from
# scratch), so this has to live on self, not on the game dict,
# to persist across polls.
#
# IMPORTANT: this is a QUEUE + single "active" slot, not just a
# per-game expiry timestamp. A plain "flash_until per event_id"
# design (the original version) let a significant play's flash
# window start counting down the moment it was DETECTED,
# completely independent of whether that game was actually on
# screen -- normal rotation runs on its own separate timer with
# no awareness of pending flashes. With 2+ live games rotating,
# a flash could easily expire before rotation ever got around to
# showing that game, so the person never saw it -- exactly the
# "works intermittently" symptom. The queue below is serviced by
# _service_flash_queue(), called before rotation each frame, so
# a pending flash can force-jump the display to the right game
# (pausing normal rotation) and guarantee it's actually seen.
self._last_shown_play_id: Dict[str, str] = {}
self._pending_flash_event_ids: List[str] = []
self._active_flash: Optional[Dict[str, Any]] = None
self._logo_cache: Dict[str, Optional[Image.Image]] = {}
self._font_cache: Dict[Tuple[str, int], ImageFont.FreeTypeFont] = {}
self._fit_font_cache: Dict[Tuple[str, int, bool], ImageFont.FreeTypeFont] = {}
self._repo_font_path = self._discover_repo_font()
# Verify the bundled font up front, not just when a render call
# happens to fail into the fallback -- this makes it immediately
# visible in the logs whether the plugin's own fonts/ folder
# actually made it into the install correctly.
fonts_dir = os.path.join(PLUGIN_DIR, "fonts")
try:
actual_files = os.listdir(fonts_dir) if os.path.isdir(fonts_dir) else None
except Exception as e:
actual_files = f"<could not list: {e}>"
self.logger.info(f"Plugin fonts/ directory ({fonts_dir}) actually contains: {actual_files}")
expected_bundled_path = FONT_CHOICES.get(self.font_choice)
if expected_bundled_path and os.path.isfile(expected_bundled_path):
self.logger.info(f"font_choice '{self.font_choice}' -> bundled file found OK at {expected_bundled_path}")
elif self.font_choice != "system":
self.logger.error(
f"font_choice '{self.font_choice}' -> bundled file NOT FOUND at "
f"{expected_bundled_path}. This means the plugin's fonts/ folder "
f"didn't make it into your install correctly (see the directory "
f"listing logged just above -- if it's missing entirely or empty, "
f"that confirms it). Text will fall back to auto-discovery, other "
f"bundled fonts, system fonts, or worst case PIL's crude default "
f"bitmap font."
)
if self._repo_font_path:
self.logger.info(f"Also found a font in the main LEDMatrix install: {self._repo_font_path}")
self.font_small = self._load_font(9)
self.font_tiny = self._load_font(7)
self.font_count = self._load_font(6)
# Log the ACTUAL resolved font type/size for the font_choice
# selected -- this is the most direct way to confirm from logs
# alone whether BDF loaded correctly, fell back to a TTF, or
# fell all the way back to PIL's default bitmap font.
resolved = self._load_font(9, bold=True)
if isinstance(resolved, BDFFont):
self.logger.info(f"font_choice '{self.font_choice}' resolved to: BDFFont (correct)")
elif isinstance(resolved, ImageFont.FreeTypeFont):
self.logger.info(
f"font_choice '{self.font_choice}' resolved to a TrueType font "
f"(path={getattr(resolved, 'path', '?')}, size={getattr(resolved, 'size', '?')}). "
f"{'This is expected if you picked a TTF font_choice.' if self.font_choice not in BDF_FONT_CHOICES else 'WARNING: you picked tom_thumb but got a TTF font back -- BDF parsing failed, see errors above.'}"
)
else:
self.logger.error(
f"font_choice '{self.font_choice}' resolved to {type(resolved)} -- "
f"this is almost certainly PIL's crude default bitmap font, meaning "
f"EVERY candidate failed to load. Text will look wrong and ignore "
f"requested sizes. Check all the errors logged above for why."
)
# Guards concurrent access to live_games/fallback_game/current_index
# between the background thread below and display()/update() being
# called from whatever thread the core scheduler uses.
self._data_lock = threading.Lock()
# IMPORTANT: this plugin no longer relies solely on the core
# calling update() often enough. Earlier debugging (score
# staying frozen after the first successful load, even after
# fixing an unrelated exception-swallowing bug) pointed at the
# core scheduler likely only calling update() when this
# plugin's rotation slot is active on screen -- not
# continuously in the background. Rather than depend on that,
# this background thread polls ESPN on its own schedule,
# completely independent of how often update()/display() get
# invoked externally. update() still exists and still works if
# the core DOES call it regularly (both paths share the same
# underlying _maybe_refresh() logic, gated by the same
# last_fetch_time check, so there's no duplicate-fetch risk).
self._stop_background_thread = threading.Event()
self._background_thread = threading.Thread(
target=self._background_update_loop, daemon=True, name=f"{plugin_id}-updater"
)
self._background_thread.start()
def _background_update_loop(self):
"""Runs for the lifetime of the process, independent of
whatever the core scheduler's calling pattern for update() is.
Sleeps briefly between checks so it responds quickly once a
game goes live, but the actual fetch still only happens as
often as live_update_interval_seconds/update_interval_seconds
allow (via _maybe_refresh's own interval check) -- this thread
just guarantees SOMETHING is checking regularly."""
while not self._stop_background_thread.is_set():
try:
self._maybe_refresh()
except Exception as e:
self.logger.error(f"Background updater thread hit an unexpected error: {e}", exc_info=True)
self._stop_background_thread.wait(timeout=5)
# ------------------------------------------------------------------
# Config handling
# ------------------------------------------------------------------
def _derive_settings(self):
cfg = self.config or {}
self.favorite_teams = [t.upper() for t in cfg.get("favorite_teams", ["PHI"])]
self.update_interval = cfg.get("update_interval_seconds", 300)
self.live_update_interval = cfg.get("live_update_interval_seconds", 15)
self.game_rotation_seconds = cfg.get("game_rotation_seconds", 8)
self.show_favorite_teams_only = cfg.get("show_favorite_teams_only", False)
self.display_duration = cfg.get("display_duration", 20)
self.away_color_fallback = tuple(cfg.get("away_color", DEFAULT_AWAY_COLOR))
self.home_color_fallback = tuple(cfg.get("home_color", DEFAULT_HOME_COLOR))
self.use_team_colors = cfg.get("use_team_colors", True)
self.show_logos = cfg.get("show_logos", True)
self.logo_dir = cfg.get("logo_dir", "assets/sports/mlb_logos")
self.base_fill_color = tuple(cfg.get("base_fill_color", [255, 255, 255]))
self.base_empty_color = tuple(cfg.get("base_empty_color", [95, 95, 95]))
self.out_fill_color = tuple(cfg.get("out_fill_color", [255, 140, 0]))
self.out_empty_color = tuple(cfg.get("out_empty_color", [120, 120, 120]))
self.font_choice = "tom_thumb" # no longer user-configurable -- see FONT_CHOICES for fallback chain if this fails to load
self.show_batter_name = cfg.get("show_batter_name", True)
self.show_pitcher_name = cfg.get("show_pitcher_name", True)
self.show_pitch_count = cfg.get("show_pitch_count", True)
self.show_delayed_overlay = cfg.get("show_delayed_overlay", True)
self.show_home_run_animation = cfg.get("show_home_run_animation", True)
self.show_last_play = cfg.get("show_last_play", True)
self.last_play_display_seconds = cfg.get("last_play_display_seconds", 5)
self.home_run_display_seconds = cfg.get("home_run_display_seconds", 10)
self.last_play_filter = cfg.get("last_play_filter", "significant")
self.last_play_favorites_only = cfg.get("last_play_favorites_only", False)
self.show_past_games = cfg.get("show_past_games", False)
self.show_upcoming_games = cfg.get("show_upcoming_games", False)
self.max_past_games = cfg.get("max_past_games", 3)
self.max_upcoming_games = cfg.get("max_upcoming_games", 3)
self.past_upcoming_all_teams = cfg.get("past_upcoming_all_teams", False)
self.upcoming_games_lookahead_days = cfg.get("upcoming_games_lookahead_days", 5)
self.upcoming_games_refresh_seconds = cfg.get("upcoming_games_refresh_seconds", 1800)
self.past_games_lookback_days = cfg.get("past_games_lookback_days", 3)
self.past_games_refresh_seconds = cfg.get("past_games_refresh_seconds", 1800)
self.test_mode = cfg.get("test_mode", False)
def on_config_change(self, new_config):
self.config = new_config
self._derive_settings()
with self._data_lock:
self.last_fetch_time = 0
def cleanup(self):
"""Called by the core on plugin unload/disable, if it supports
that -- stops the background updater thread cleanly. Harmless
no-op risk if the core never calls this (the thread is a daemon
thread anyway, so it won't block process exit either way)."""
self._stop_background_thread.set()
def validate_config(self) -> bool:
if not self.favorite_teams:
self.logger.error("No favorite_teams configured")
return False
return True
# ------------------------------------------------------------------
# Fonts
# ------------------------------------------------------------------
def _discover_repo_font(self) -> Optional[str]:
"""Scans assets/fonts/ (relative to the LEDMatrix install root)
for a real bundled font instead of guessing a filename. Prefers
anything that looks like a pixel/arcade font so team text
matches the aesthetic the rest of the project's plugins use."""
fonts_dir = "assets/fonts"
if not os.path.isdir(fonts_dir):
return None
try:
files = [f for f in os.listdir(fonts_dir) if f.lower().endswith((".ttf", ".otf"))]
except Exception as e:
self.logger.warning(f"Could not list {fonts_dir}: {e}")
return None
if not files:
return None
for keyword in FONT_NAME_PREFERENCE:
for f in files:
if keyword in f.lower():
return os.path.join(fonts_dir, f)
return os.path.join(fonts_dir, sorted(files)[0])
def _load_font(self, size: int, bold: bool = False) -> Any:
cache_key = (self.font_choice, size)
if cache_key in self._font_cache:
return self._font_cache[cache_key]
if self.font_choice in BDF_FONT_CHOICES:
# BDF is a fixed-pixel bitmap format -- there's no "size" to
# request, so every size maps to the same single instance.
# Cache it once under a size-independent key too.
bdf_key = (self.font_choice, "bdf")
if bdf_key in self._font_cache:
font = self._font_cache[bdf_key]
else:
bdf_path = FONT_CHOICES[self.font_choice]
try:
font = BDFFont(bdf_path)
except Exception as e:
self.logger.error(f"Failed to parse BDF font at {bdf_path}: {e}", exc_info=True)
font = None
self._font_cache[bdf_key] = font
if font is not None:
self._font_cache[cache_key] = font
return font
# fall through to TTF/system candidates below if BDF parsing failed
candidates = []
bundled_path = FONT_CHOICES.get(self.font_choice)
if bundled_path and os.path.isfile(bundled_path) and self.font_choice not in BDF_FONT_CHOICES:
candidates.append(bundled_path)
elif self.font_choice != "system" and self.font_choice not in BDF_FONT_CHOICES:
self.logger.warning(
f"font_choice '{self.font_choice}' bundled file not found at "
f"expected path ({bundled_path}); falling back to auto-discovery / system font."
)
if self._repo_font_path:
candidates.append(self._repo_font_path)
# Try every OTHER bundled TTF this plugin ships with before
# falling back to system fonts -- these are guaranteed to be
# sitting right next to manager.py (assuming the plugin's own
# fonts/ folder made it into the install at all), so they're
# more likely to actually be there than OS-level font packages,
# which a minimal Raspberry Pi OS Lite install may not include.
for choice, path in FONT_CHOICES.items():
if choice in BDF_FONT_CHOICES or choice == self.font_choice or path is None:
continue
if os.path.isfile(path):
candidates.append(path)
if bold:
candidates += [
"/usr/share/fonts/truetype/dejavu/DejaVuSansMono-Bold.ttf",
"/usr/share/fonts/truetype/liberation/LiberationMono-Bold.ttf",
]
else:
candidates.append("/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf")
font = None
for path in candidates:
try:
font = ImageFont.truetype(path, size)
break
except Exception:
continue
if font is None:
# IMPORTANT: PIL's load_default() renders a fixed, crude
# bitmap font that IGNORES the requested `size` entirely on
# a lot of Pillow versions. If you're seeing blocky,
# oddly-large, or generic-looking text regardless of the
# font_choice you picked, THIS is almost certainly why --
# every candidate above failed to load. Check the plugin
# logs for this exact error to confirm.
self.logger.error(
f"ALL font candidates failed to load for size={size}, bold={bold}: "
f"{candidates}. Falling back to PIL's built-in default bitmap font, "
f"which ignores the requested size -- this is very likely why text "
f"looks wrong. Check that the plugin's fonts/ folder actually made it "
f"into your install (should be at {os.path.join(PLUGIN_DIR, 'fonts')})."
)
font = ImageFont.load_default()
self._font_cache[cache_key] = font
return font
def _measure(self, font: Any, text: str) -> Tuple[int, int, int, int]:
"""Unified text bounding-box measurement for either a BDFFont or
a normal PIL font, so the rest of the code doesn't need to care
which one is active."""
if isinstance(font, BDFFont):
return font.textbbox(text)
tmp_draw = ImageDraw.Draw(Image.new("RGB", (1, 1)))
return tmp_draw.textbbox((0, 0), text, font=font)
def _render_text(self, image: Image.Image, xy: Tuple[int, int], text: str, font: Any, fill: Tuple[int, int, int]):
"""Unified text drawing for either a BDFFont (direct pixel
writes, no anti-aliasing) or a normal PIL font (draw.text)."""
if isinstance(font, BDFFont):
font.draw(image, xy, text, fill)
else:
ImageDraw.Draw(image).text(xy, text, font=font, fill=fill)
def _ink_extent(self, font: Any, text: str) -> Tuple[int, int]:
"""Renders `text` to a small scratch image and returns the
actual leftmost/rightmost columns containing ink (non-background
pixels) -- as opposed to the font's nominal advance width, which
for punctuation like ":" or "." often includes several columns
of blank design space the font author left for normal spacing.
Measuring real ink is what lets tightening work correctly
regardless of which font is active, rather than guessing a
fixed pixel offset tuned for one specific font."""
bbox = self._measure(font, text)
w = max(bbox[2] - bbox[0], 1) + 6
h = max(bbox[3] - bbox[1], 1) + 6
scratch = Image.new("RGB", (w, h), (0, 0, 0))
self._render_text(scratch, (3, 3), text, font, (255, 255, 255))
cols = [x for x in range(w) for y in range(h) if scratch.getpixel((x, y)) != (0, 0, 0)]
if not cols:
return (3, 3)
return (min(cols), max(cols))
def _draw_tight_join(self, image, x, y, font, fill, text_a: str, text_b: str, ink_gap: int = 1) -> int:
"""Draws text_a then text_b immediately after it, with only
`ink_gap` background pixels between their actual rendered ink
-- not their nominal advance widths. This is what actually
tightens up spacing like "P:" or "T. Lastname": the blank space
people see isn't extra spacing added between characters, it's
blank design space baked into narrow glyphs (colons, periods)
that a font author left for normal-width spacing. Returns the
total pixel width used, for cursor advancement."""
self._render_text(image, (x, y), text_a, font, fill)
_, a_right_scratch = self._ink_extent(font, text_a)
a_right_actual = x + (a_right_scratch - 3)
b_left_scratch, b_right_scratch = self._ink_extent(font, text_b)
# Target: B's first ink column should land at (A's last ink
# column + 1 + ink_gap) -- the "+1" is because a_right_actual
# IS the last ink pixel, so the very next column is already 0
# gap; ink_gap blank columns after that is where B's ink starts.
b_x = (a_right_actual + 1 + ink_gap) - (b_left_scratch - 3)
self._render_text(image, (b_x, y), text_b, font, fill)
b_bbox = self._measure(font, text_b)
return (b_x + (b_bbox[2] - b_bbox[0])) - x
def _draw_name_tightened(self, image, xy, font, fill, name: str, ink_gap: int = 1) -> int:
"""Draws a 'F. Lastname'-style string with the gap after the
initial+period tightened to `ink_gap` real pixels instead of
whatever blank space the space character/font design normally
leaves (measured as 6px of pure blank for tom_thumb's "T. " --
see the investigation that led to this). Falls back to a plain
render if the string doesn't match that pattern. Returns the
pixel width used."""
x, y = xy
m = re.match(r"^([A-Za-z]{1,2}\.) (.+)$", name)
if not m:
self._render_text(image, (x, y), name, font, fill)
bbox = self._measure(font, name)
return bbox[2] - bbox[0]
prefix, rest = m.group(1), m.group(2)
return self._draw_tight_join(image, x, y, font, fill, prefix, rest, ink_gap=ink_gap)
def _measure_name_tightened(self, font, name: str, ink_gap: int = 1) -> int:
"""Width the tightened name would actually take up, WITHOUT
drawing to the real image. Reuses _draw_name_tightened itself
against a scratch canvas rather than reimplementing the
positioning math separately -- that duplication is exactly what
let measurement and final rendering drift apart before (fit
checks used the untightened width, so text that only fit
because of the tightening savings was truncating anyway, and
then even the truncated fallback skipped tightening entirely)."""
scratch = Image.new("RGB", (400, 30), (0, 0, 0))
return self._draw_name_tightened(scratch, (2, 2), font, (255, 255, 255), name, ink_gap=ink_gap)
def _fit_font_for_width(self, draw, text: str, max_width: int, start_size: int, min_size: int = 4) -> Any:
"""Shrinks the font size until `text` fits within max_width.
Works regardless of which font got auto-discovered, since
different fonts have very different glyph widths (this is what
caused double-digit scores to overflow before).
BDF fonts are a fixed size, so this skips shrinking for them --
but only after confirming _load_font() actually returned a
BDFFont. If font_choice is "tom_thumb" but the BDF file failed
to parse for some reason, _load_font() silently falls back to a
full-size TTF font -- and skipping the shrink loop in that case
would render that TTF at `start_size` with ZERO shrinking,
which is exactly what caused oversized, overflowing text. Only
the confirmed-BDF case skips the loop; any fallback still goes
through normal shrink-to-fit."""
candidate = self._load_font(start_size, bold=True)
if isinstance(candidate, BDFFont):
return candidate
cache_key = (self.font_choice, text, max_width)
if cache_key in self._fit_font_cache:
return self._fit_font_cache[cache_key]
size = start_size
chosen = None
while size >= min_size:
font = self._load_font(size, bold=True)
bbox = self._measure(font, text)
if bbox[2] - bbox[0] <= max_width:
chosen = font
break
size -= 1
if chosen is None:
chosen = self._load_font(min_size, bold=True)
self._fit_font_cache[cache_key] = chosen
return chosen
def _fit_font_for_pair(self, draw, text_a: str, text_b: str, max_width: int, start_size: int, min_size: int = 4) -> Any:
"""Like _fit_font_for_width, but sizes for whichever of the two
strings is wider, so both team columns render at the SAME font
size rather than each shrinking independently based on its own
text length (that mismatch was the original bug). See
_fit_font_for_width's docstring for why this checks
isinstance(..., BDFFont) rather than trusting font_choice."""
candidate = self._load_font(start_size, bold=True)
if isinstance(candidate, BDFFont):
return candidate
cache_key = (self.font_choice, text_a, text_b, max_width)
if cache_key in self._fit_font_cache:
return self._fit_font_cache[cache_key]
size = start_size
chosen = None
while size >= min_size:
font = self._load_font(size, bold=True)
bbox_a = self._measure(font, text_a)
bbox_b = self._measure(font, text_b)
widest = max(bbox_a[2] - bbox_a[0], bbox_b[2] - bbox_b[0])
if widest <= max_width:
chosen = font
break
size -= 1
if chosen is None:
chosen = self._load_font(min_size, bold=True)
self._fit_font_cache[cache_key] = chosen
return chosen
# ------------------------------------------------------------------
# Data fetching
# ------------------------------------------------------------------
def update(self):
"""Called by the core scheduler, if/whenever it calls it. Just
delegates to _maybe_refresh() -- the actual polling now also
happens independently via the background thread started in
__init__, so data refreshes either way regardless of the core's
calling cadence for this method."""
self._maybe_refresh()
def _maybe_refresh(self):
now = time.time()
with self._data_lock:
has_data = bool(self.live_games) or self.fallback_game is not None
interval = self.live_update_interval if self.live_games else self.update_interval
seconds_since_last = now - self.last_fetch_time
should_skip = has_data and (seconds_since_last < interval)
if should_skip:
self.logger.debug(
f"Skipping fetch -- only {seconds_since_last:.1f}s since last fetch "
f"(interval is {interval}s)."
)
return
with self._data_lock:
self.last_fetch_time = now
if self.test_mode:
game = self._fake_game()
self._resolve_logos(game)
with self._data_lock:
self.live_games = [game]
self.fallback_game = None
self.rotation_games = [game]
if self.current_index >= len(self.rotation_games):
self.current_index = 0
return
try:
resp = self.session.get(ESPN_SCOREBOARD_URL, timeout=10)
resp.raise_for_status()
data = resp.json()
except Exception as e:
self.logger.error(f"Failed to fetch MLB scoreboard: {e}", exc_info=True)
return
# IMPORTANT: this whole block used to be unprotected. If any
# single game had an unusual shape ESPN sometimes sends
# (pitching change, extra innings, a null field mid-play, etc.)
# that our parsing code didn't handle, the exception would
# propagate straight out uncaught. Wrapping this means a single
# bad game/response degrades gracefully (keeps last-known-good
# data, tries again next interval) instead of permanently
# freezing everything.
try:
live_games, past_games, upcoming_games, fallback_game = self._process_scoreboard(data)
# ESPN's main scoreboard call only covers today -- merge in
# the separately-cached multi-day lookahead so upcoming_games
# isn't nearly always empty (see _fetch_future_upcoming_games
# for why). Only re-queries the future days on its own slower
# timer, since schedules barely change within a day.
now_ts = time.time()
if self.show_upcoming_games and (now_ts - self._upcoming_last_fetch_time >= self.upcoming_games_refresh_seconds):
try:
self._cached_future_upcoming_games = self._fetch_future_upcoming_games()
self._upcoming_last_fetch_time = now_ts
except Exception as e:
self.logger.warning(f"Upcoming-games lookahead failed, keeping previous cache: {e}")
if self.show_upcoming_games:
combined_upcoming = {g["event_id"]: g for g in upcoming_games}
for g in self._cached_future_upcoming_games:
combined_upcoming.setdefault(g["event_id"], g)
upcoming_games = sorted(combined_upcoming.values(), key=lambda g: g.get("event_date_raw") or "")
upcoming_games = upcoming_games[: self.max_upcoming_games]
# Mirror image of the upcoming-games fix: ESPN's main call
# only covers today, so a favorite team's most recent
# completed game (yesterday or earlier) never shows up
# without explicitly looking backward too.
if self.show_past_games and (now_ts - self._past_last_fetch_time >= self.past_games_refresh_seconds):
try:
self._cached_past_lookback_games = self._fetch_past_games_lookback()
self._past_last_fetch_time = now_ts
except Exception as e:
self.logger.warning(f"Past-games lookback failed, keeping previous cache: {e}")
if self.show_past_games:
combined_past = {g["event_id"]: g for g in past_games}
for g in self._cached_past_lookback_games:
combined_past.setdefault(g["event_id"], g)
# Most recent first -- reverse chronological, unlike
# upcoming games which sort soonest-first.
past_games = sorted(combined_past.values(), key=lambda g: g.get("event_date_raw") or "", reverse=True)
past_games = past_games[: self.max_past_games]
for g in past_games:
try:
self._enrich_boxscore_stats(g)
except Exception as e:
self.logger.warning(f"Could not fetch box score stats for {g['away_abbr']}@{g['home_abbr']}: {e}")
for g in live_games + past_games + upcoming_games:
self._resolve_logos(g)
if fallback_game:
self._resolve_logos(fallback_game)
# --- Favorite-team priority cascade ---
# 1. If ANY favorite team has a live game right now, show
# ONLY that (those) live game(s) -- past/upcoming and
# every other team's live game are fully suppressed
# while a favorite is live.
# 2. Otherwise, if show_favorite_teams_only is OFF: show
# every live game (any team) plus past/upcoming (scope
# controlled separately by past_upcoming_all_teams).
# 2s. Otherwise (show_favorite_teams_only is ON, strict
# mode): show only favorites' past/upcoming -- no other
# team's live game ever appears.
# 3. Falls out naturally: if the live portion is empty in
# either branch, rotation is just past/upcoming; if ALL
# of that is empty too, fall back to the single
# best-guess favorite game (preserves pre-this-feature
# behavior for anyone with everything else off).
def _is_favorite_game(g):
return g["away_abbr"] in self.favorite_teams or g["home_abbr"] in self.favorite_teams
favorite_live_games = [g for g in live_games if _is_favorite_game(g)]
if favorite_live_games:
# Live favorite game(s) still get priority (listed
# first, so they're what shows first each time rotation
# cycles back around), but past/upcoming toggles are
# still honored here too -- previously this branch
# suppressed past/upcoming entirely whenever a favorite
# was live, even for the SAME favorite team. Scoped
# strictly to favorite teams regardless of
# past_upcoming_all_teams -- mixing in some OTHER team's
# past/upcoming game here would defeat the point of
# favorite-team prioritization.
rotation_games = list(favorite_live_games)
if self.show_past_games:
rotation_games += [g for g in past_games if _is_favorite_game(g)]
if self.show_upcoming_games:
rotation_games += [g for g in upcoming_games if _is_favorite_game(g)]
cascade_state = "favorite team(s) live -- showing that + favorites' past/upcoming"
elif self.show_favorite_teams_only:
rotation_games = []
if self.show_past_games:
rotation_games += past_games
if self.show_upcoming_games:
rotation_games += upcoming_games
cascade_state = "strict mode, no favorite live -- favorites' past/upcoming only"
elif live_games:
# NEW: some OTHER team is live (not a favorite). Explicit
# request: past/upcoming games should be restricted to
# favorites ONLY here, regardless of past_upcoming_all_teams
# -- that setting only matters when NOTHING is live
# anywhere (see the final branch below). Otherwise a
# non-favorite team's past/upcoming game shows up right
# alongside live games happening right now, which is
# exactly the cluttered experience being avoided.
rotation_games = list(live_games)
if self.show_past_games:
rotation_games += [g for g in past_games if _is_favorite_game(g)]
if self.show_upcoming_games:
rotation_games += [g for g in upcoming_games if _is_favorite_game(g)]
cascade_state = "other team(s) live, no favorite live -- showing those + favorites' past/upcoming only"
else:
# Nothing live ANYWHERE -- past_upcoming_all_teams now
# applies as designed (past_games/upcoming_games were
# already scoped at fetch time per that setting).
rotation_games = []
if self.show_past_games:
rotation_games += past_games
if self.show_upcoming_games:
rotation_games += upcoming_games
cascade_state = "nothing live anywhere -- past/upcoming per past_upcoming_all_teams setting"
if not rotation_games and fallback_game:
rotation_games = [fallback_game]
cascade_state += " (nothing available -- using single fallback game)"
# Real pitch count isn't in the lightweight scoreboard
# response (confirmed from actual captured data) -- fetch
# it from ESPN's more detailed per-game summary endpoint
# instead. Wrapped in its own try/except per game so one
# game's summary failing (or ESPN changing that endpoint's
# shape) can't take down the main scoreboard update.
#
# IMPORTANT: this only runs on the LIVE games actually
# selected into rotation_games, not every live game
# leaguewide -- "show all live games" mode could otherwise
# mean fetching a summary for a dozen simultaneous MLB
# games every poll, which is a lot of extra requests for
# data that never even gets displayed.
enrich_targets = [g for g in rotation_games if g.get("game_type") == "live"]
for g in enrich_targets:
try:
self._enrich_pitch_count(g)
except Exception as e:
self.logger.warning(f"Could not fetch pitch count for {g['away_abbr']}@{g['home_abbr']}: {e}")
for g in enrich_targets:
try:
self._maybe_trigger_last_play_flash(g)
except Exception as e:
self.logger.warning(f"Error checking last-play flash for {g['away_abbr']}@{g['home_abbr']}: {e}")
self.logger.info(
f"Fetched scoreboard OK: {len(live_games)} live game(s) leaguewide, "
f"{len(favorite_live_games)} involving a favorite team, "
f"{len(past_games)} past, {len(upcoming_games)} upcoming. "
f"Cascade: {cascade_state}. Rotation has {len(rotation_games)} game(s)."
)
with self._data_lock:
self.live_games = live_games
self.past_games = past_games
self.upcoming_games = upcoming_games
self.fallback_game = fallback_game
self.rotation_games = rotation_games
if self.current_index >= len(self.rotation_games):
self.current_index = 0
except Exception as e:
self.logger.error(
f"Fetched scoreboard successfully but failed to parse/process it: {e}. "
f"Keeping last-known-good data instead of crashing -- will try again "
f"next update cycle. If this repeats every time, something about the "
f"CURRENT game state (extra innings, pitching change, etc.) is hitting "
f"a parsing bug -- please share this traceback.",
exc_info=True,
)
def _maybe_trigger_last_play_flash(self, game: Dict[str, Any]):
"""Detects when a game has a NEW play (by comparing lastPlay's
id against what we last saw for this specific game, keyed by
event_id since game dicts are rebuilt fresh every poll) and, if
it's a "significant" play type, QUEUES it to be flashed. The
actual timing/expiry is handled by _service_flash_queue(),
called from display() before rotation -- this function only
decides WHETHER something should flash, never when or for how
long, since that's what guarantees it's actually shown (see the
big comment on _pending_flash_event_ids in __init__).
Deliberately does NOT flash the very first time we ever see a
given game (i.e., when we have no previous play id to compare
against) -- otherwise every game would flash immediately on
first load / plugin startup for whatever play happened to be
current already, which isn't really a "new" play from the
person watching the display's perspective."""
if not self.show_last_play:
return
event_id = game.get("event_id")
play_id = game.get("last_play_id")
if not event_id or not play_id:
return
with self._data_lock:
previous_id = self._last_shown_play_id.get(event_id)
self._last_shown_play_id[event_id] = play_id
if previous_id is None or play_id == previous_id:
return # first time seeing this game, or no change
play_type = (game.get("last_play_type") or "").lower()
is_significant = (
self.last_play_filter != "significant"
or play_type not in NON_SIGNIFICANT_PLAY_TYPES
)