rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-04 06:15:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(sports): live Vegas cards for the scoreboards (shared layer) (#698)

Browse Source 
* perf(timing): say which render-thread work a late frame followed

The soak already says how often a moving frame reached the panel late, but
not what the render thread was doing just before it. Vegas does two kinds of
work there between frames -- building its strip (compose, extend) and, with
live elements, patching changed pixels into it -- and deciding whether either
is affordable needs their own numbers.

- FrameTimingRecorder.note_op(kind, nbytes) tags the next presented frame.
  Totals gain op_frames, late_op_frames, op_freezes and op_bytes per kind;
  aggregate() still takes frames without ops. The file schema is unchanged.
- Vegas tags compose and every strip extension (with the bytes it copied).
- frame_soak prints an "after work" table: frames, late %, freezes and MB
  moved per kind, only when something tagged its work.
- render_bench gains --strip-screens (Vegas-sized strips), --patch-bytes /
  --patch-every / --patch-where (in-place column writes, as a live element
  update does) and --extend-every-screens / --extend-width (append + trim on
  a fixed cadence that holds the strip's width).

No runtime behaviour changes: this is the measurement gate for live Vegas
elements.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* docs(changelog): note the frame-op attribution and bench modes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* perf(scroll): build the strip's PIL image only when something reads it

Every Vegas strip extension rebuilt ScrollHelper.cached_image from
cached_array in full, twice (append, then trim), on the render thread:
Image.fromarray is 1.7ms for an 8,000px strip and 3.8ms for 20,000px on a
Pi 4 (measured on ledpi), about two thirds of an extension's render-thread
cost. Nothing on the frame path reads the image's pixels; every frame is cut
from the array.

cached_image is now a property. append_content and drop_scrolled_prefix
defer it; the first read builds it from the array it started with and keeps
it only if the strip has not changed meanwhile, so a sync push racing an
extension cannot leave a stale image cached. Assigning cached_image stores
exactly what was assigned, as before. has_strip() says whether there is a
strip without building its image; the helper's frame path, Vegas and the
adapter's scroll-cache invalidation use it. The strip is also no longer held
in memory twice.

In Vegas the image is now built only by a multi-display sync push.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(vegas): live elements -- a plugin API for content that changes while it scrolls

Vegas bakes each plugin's pictures into one strip, so a card already on its
way across the panel keeps what it showed when it was drawn. This adds the
API and bookkeeping for content that can be updated in place; the worker
that redraws and swaps it follows separately. No shipped plugin implements
the hook yet, so nothing changes for users.

Plugin API (core 3.8.0), all no-ops by default:
- BasePlugin.get_vegas_elements() -> [VegasElement(key, image, version,
  live, refresh_hz)]: named, fixed-width pieces of Vegas content.
- BasePlugin.redraw_vegas_element(key, width, height, at): a lock-free
  redraw for content that changes with time.
- BasePlugin.notify_vegas_data_changed(): data that lands outside update().
- src/plugin_system/vegas_elements.py (VegasElement, re-exported from
  base_plugin).

Core:
- PluginAdapter asks a plugin that implements the hook for elements on the
  background fetch only (under its lock, on its own canvas); every other
  path keeps get_vegas_content(). Live elements are pinned (padded with
  content_padding, never trimmed), tagged with their key, digest and data
  epoch in Image.info so the existing cache and group plumbing carry them
  unchanged, and untagged if a width budget crops them.
- RenderPipeline records where each live element lands (ElementRecord), in
  absolute strip columns a trim does not move; the block-start arithmetic
  is shared with the STATIC markers.
- PluginManager update listeners (add/remove_update_listener,
  notify_data_changed): told the moment update() completes, not at the
  next ~4s Vegas poll. The coordinator uses one to move each plugin's data
  epoch on.
- vegas_scroll.live_refresh (kill switch), live_max_hz, live_min_interval,
  live_lead_screens; per-plugin core-owned vegas_live. Live elements are
  off under multi-display sync, in swap mode and with offscreen_prefetch off.
- scripts/check_plugin.py checks the element contract
  (src/plugin_system/testing/vegas.py); test/fixtures/plugins/vegas-live-stub
  is a working example.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(vegas): live elements update in place while they scroll

One background worker (src/vegas_mode/live_worker.py) redraws a plugin's
live elements when its data epoch moves on (update listener) or on their
refresh_hz, nearest the screen first, and hands changed pixels lock-free to
the render thread, which copies them into the strip between frames
(RenderPipeline.apply_live_patches, ScrollHelper.patch_columns): at most
four patches or two screens of bytes a frame, no drawing or locks there.
The worker takes over group prefetch once a live element is placed, runs
inside the render gate, and is supervised. Update tick 1s while live
elements exist. Web UI switch for live_refresh. OFFSCREEN_RENDERING.md
describes what was built and why SegmentStrip was not needed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(sports): live Vegas cards for the scoreboards (shared layer)

One live element per game, drawn only when what the card shows changes, so
a score changes on a card already crossing the panel. The shared part, so
each scoreboard adopts it in a few lines:

- src/common/sports_vegas.py: game_key, game_fingerprint (the whole game
  dict, frozen: no drawn field can be missed), dedupe_games, VegasCardCache,
  StickyOdds (odds a live poll left out stay drawn), finished_games /
  with_finished_games (a game that just went final keeps its card, after its
  league's live games; one a heuristic only judged over keeps its live
  state, so a tied end of regulation never shows FINAL early).
- SportsScrollDisplay.make_vegas_renderer() is the override point;
  build_vegas_elements() and SportsScrollDisplayManager
  .get_vegas_elements_for() do the rest. A card's version includes its
  teams' ranks, which the renderer draws from the rankings cache.
- SportsLiveSharedMixin._record_finished_game() / finished_games_snapshot():
  held for FINISHED_GAME_TTL after it leaves the live list.

A sport that does not implement make_vegas_renderer keeps its ordinary Vegas
content, so no scoreboard changes until it opts in.

scripts/render_plugin.py --vegas renders a plugin's Vegas block as the
ticker lays it out, and --timeline stacks it at successive moments as
the ticker would update it in place; the join is now
render_pipeline.join_plugin_rows().

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(sports): a default _determine_game_type on SportsScrollDisplay

render_vegas_card looked the method up with getattr and a None default, which
static analysis (Codacy) reports as calling something that may not be
callable. The base class now has the default -- the card type from the game's
state -- and the plugins that define their own override it as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: review follow-ups on the shared live-card layer

- The reused Vegas renderer always gets the current rankings, empty
  included, so ranks cleared since are not kept drawn.
- render_plugin.py: --timeline refuses --no-live (a timeline shows live
  elements changing), --timeline/--no-live need --vegas, and the Vegas
  paths create the output's directory like the display path does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-30 21:32:07 -04:00
committed by GitHub
co-authored by Claude Opus 5.5
parent f4bda50710
commit a12be7c3c5
11 changed files with 1150 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+22
View File
@@ -384,6 +384,28 @@ read any of them:
the Vegas update tick runs every second instead of every four.
- Web UI: "Update live content while it scrolls" under Vegas mode's Cycle
Pacing (`display.vegas_scroll.live_refresh`).
- **Live cards for the scoreboards (shared code).** New module
`src/common/sports_vegas.py`: `game_key()`, `dedupe_games()`,
`VegasCardCache` (draws a card only when its fingerprint changes) and
`StickyOdds` (keeps a card's odds through a live poll that left them out),
`finished_games()` and `with_finished_games()` (a game that just went final
keeps its card, showing FINAL, where its live card was).
`SportsScrollDisplay` gains `make_vegas_renderer()` (the override point; a
sport that does not implement it keeps its ordinary Vegas content),
`render_vegas_card()`, `vegas_separator()` and `build_vegas_elements()`,
and `SportsScrollDisplayManager` gains `get_vegas_elements_for()`.
`SportsLiveSharedMixin` gains `_record_finished_game()` /
`finished_games_snapshot()`, so a game that goes final keeps a card to show
FINAL on until the hourly recent list takes it over.
- `scripts/render_plugin.py --vegas` renders a plugin's block of the Vegas
strip as the ticker lays it out (live elements, or with `--no-live` its
ordinary content) and writes the live elements' keys and columns beside
it. `--timeline ROWS` stacks the block at successive moments as the
ticker would update it in place (`--timeline-step`, and
`--timeline-update` to run `update()` between rows).
`render_vegas_strip()` and `render_vegas_timeline()` in
`src/plugin_system/testing/vegas.py`; the join is now
`render_pipeline.join_plugin_rows()`.
### Scrolling
mypy-clean.txt
+1
View File
@@ -37,6 +37,7 @@ src/common/sports_celebration.py
src/common/sports_fetch.py
src/common/sports_scroll.py
src/common/sports_timezone.py
src/common/sports_vegas.py
src/config_service.py
src/core_config_keys.py
src/deprecation.py
scripts/render_plugin.py
+57
View File
@@ -56,9 +56,33 @@ def main() -> int:
help='Display mode to render, for plugins that declare '
'more than one in their manifest (e.g. nrl_live). '
'Omitted, the plugin picks its own default.')
parser.add_argument('--vegas', action='store_true',
help="Render the plugin's block of the Vegas ticker strip "
"instead of display(): its live elements if it has "
"them, else its Vegas content, laid out as the "
"ticker lays them out. Also writes the live "
"elements' keys and columns to <output>.json")
parser.add_argument('--no-live', action='store_true',
help="With --vegas: ignore live elements and render the "
"plugin's ordinary Vegas content (for before/after)")
parser.add_argument('--timeline', type=int, default=0, metavar='ROWS',
help="With --vegas: render ROWS rows, each the block a "
"--timeline-step later as the ticker would update it "
"in place (animated elements redrawn for that moment)")
parser.add_argument('--timeline-step', type=float, default=0.25, metavar='SECONDS',
help="Seconds between --timeline rows (default 0.25)")
parser.add_argument('--timeline-update', action='store_true',
help="With --timeline: run update() before each row and "
"redraw every live element from the new data")
args = parser.parse_args()
if args.timeline > 1 and args.no_live:
# A timeline shows live elements changing; plain content never does.
parser.error("--timeline shows live elements; it cannot be combined with --no-live")
if (args.timeline or args.no_live) and not args.vegas:
parser.error("--timeline and --no-live need --vegas")
if not (MIN_DIMENSION <= args.width <= MAX_DIMENSION):
print(f"Error: --width must be between {MIN_DIMENSION} and {MAX_DIMENSION} (got {args.width})")
raise SystemExit(1)
@@ -145,6 +169,39 @@ def main() -> int:
except Exception as e:
logger.warning("update() raised: %s — continuing to display()", e)
if args.vegas:
Path(args.output).parent.mkdir(parents=True, exist_ok=True)
if args.vegas and args.timeline > 1:
from src.plugin_system.testing.vegas import render_vegas_timeline
image, rows = render_vegas_timeline(
plugin_instance, args.plugin, display_manager, steps=args.timeline,
step_seconds=args.timeline_step, run_update=args.timeline_update)
if image is None:
logger.error("Plugin '%s' has no Vegas content", args.plugin)
return 1
image.save(args.output)
logger.info("Saved a %d-row Vegas timeline (%dx%d) to %s",
rows, image.width, image.height, args.output)
return 0
if args.vegas:
from src.plugin_system.testing.vegas import render_vegas_strip
block, layout = render_vegas_strip(
plugin_instance, args.plugin, display_manager, live=not args.no_live)
if block is None:
logger.error("Plugin '%s' has no Vegas content", args.plugin)
return 1
block.save(args.output)
sidecar = Path(args.output).with_suffix('.json')
sidecar.write_text(json.dumps(
{"width": block.width, "height": block.height,
"live_elements": [{"key": k, "x": x, "width": w} for x, k, w in layout]},
indent=2) + "\n", encoding="utf-8")
logger.info("Saved Vegas strip %dx%d (%d live element(s)) to %s and %s",
block.width, block.height, len(layout), args.output, sidecar)
return 0
# A plugin that declares several display modes usually renders nothing
# useful without being told which one to draw: the scoreboards keep their
# state on per-mode sub-managers and their no-argument path returns False.
src/common/README.md
+12
View File
@@ -45,6 +45,7 @@ Rules for the package:
| [`sports_helpers`](#sports_helpers) | Small helpers every scoreboard `sports.py` copies | Yes (scoreboards) | 3.5.0 |
| [`sports_scroll`](#sports_scroll) | Scoreboard scroll-display orchestration | Yes (scoreboards) | 3.2.0 |
| [`sports_shared`](#sports_shared) | Sport-independent `sports.py` methods | Yes (scoreboards) | 3.3.0 |
| [`sports_vegas`](#sports_vegas) | Live Vegas cards: keys, card cache, sticky odds, finished games | Yes (scoreboards) | 3.8.0 |
| [`sports_timezone`](#sports_timezone) | Which timezone a scoreboard draws start times in | Yes (scoreboards) | 3.6.0 |
| [`sync_manager`](#sync_manager) | Leader/follower sync between two displays | No, core-internal | n/a |
| [`text_helper`](#text_helper) | Outlined text, wrapping, measurement | Yes | — |
@@ -280,6 +281,17 @@ fonts, colours, dates, the switch-mode upcoming card). The docstring lists
the attributes the host class must have and the three methods deliberately
left out.
### sports_vegas
[`sports_vegas.py`](sports_vegas.py). What a scoreboard needs for live Vegas
cards (one element per game, swapped in place while it scrolls):
`game_key()`, `game_fingerprint()`, `dedupe_games()`, `VegasCardCache` (draws
a card only when its fingerprint changes), `StickyOdds` (keeps a card's odds
through a live poll that left them out), and `finished_games()` /
`with_finished_games()` (a game that just went final keeps its card, showing
FINAL). `SportsScrollDisplay.build_vegas_elements()` in `sports_scroll` puts
them together; a scoreboard not built on it (UFC) uses them directly.
### sports_timezone
[`sports_timezone.py`](sports_timezone.py).
src/common/sports_scroll.py
+184 -2
View File
@@ -42,17 +42,19 @@ Usage::
from __future__ import annotations
import functools
import logging
import time
from typing import Any, Dict, List, Optional
from typing import Any, Callable, Dict, List, Optional, Tuple
from PIL import Image
from src.common import scroll_config
from src.common import scroll_config, sports_vegas
from src.common.scroll_helper import ScrollHelper
logger = logging.getLogger(__name__)
#: Defaults every copy agreed on. A subclass overrides
#: :meth:`SportsScrollDisplay.scroll_settings_defaults` to change them —
#: the soccer lineage uses a 24px gap and min/max duration keys instead.
@@ -413,6 +415,160 @@ class SportsScrollDisplay:
"""Whether content is prepared and ready to scroll."""
return bool(self.scroll_helper.cached_image)
# ------------------------------------------------------------------
# Live Vegas cards
# ------------------------------------------------------------------
#
# One live element per game (src/plugin_system/vegas_elements.py): the
# ticker swaps a card in place when its game changes. A sport opts in by
# implementing make_vegas_renderer(); everything else is here.
def make_vegas_renderer(self, card_width: int,
rankings_cache: Optional[Dict[str, int]] = None) -> Any:
"""The renderer this sport draws one game card with, at ``card_width``.
**Override point.** Return the object whose ``render_game_card(game,
game_type)`` draws one card exactly ``card_width`` wide at the display's
height -- the one prepare_scroll_content already builds -- without the
black padding prepare_scroll_content adds around each card (the ticker
adds its own). Raising NotImplementedError, the default, keeps the
plugin on its ordinary Vegas content.
"""
raise NotImplementedError(
f"{type(self).__name__} has no live Vegas cards (make_vegas_renderer)")
def _determine_game_type(self, game: Dict[str, Any]) -> str:
"""The card a game is drawn as: 'live', 'recent' or 'upcoming'.
From the game's state; a sport whose scroll display decides it
differently (most define their own) overrides this.
"""
return {'in': 'live', 'post': 'recent'}.get(sports_vegas._state(game), 'upcoming')
def render_vegas_card(self, renderer: Any, game: Dict[str, Any]) -> Image.Image:
"""Draw one game's card. Override only if the renderer is called differently."""
card: Image.Image = renderer.render_game_card(game, self._determine_game_type(game))
return card
def vegas_separator(self, league: str) -> Optional[Image.Image]:
"""The league separator shown before a league's cards, if there is an icon."""
icon = self._separator_icons.get(league)
if icon is None:
return None
gap = self._vegas_settings(league).get("gap_between_games", 48)
pad = max(4, int(gap) // 2)
image = Image.new('RGB', (icon.width + pad * 2, self.display_height), (0, 0, 0))
mask = icon if icon.mode == 'RGBA' else None
image.paste(icon, (pad, (self.display_height - icon.height) // 2), mask)
return image
def _vegas_memo(self) -> Dict[Any, Any]:
"""Per-size, per-config memo for the live path; emptied when either changes."""
stamp = (self.display_width, self.display_height, id(self.config))
memo: Optional[Tuple[Any, Dict[Any, Any]]] = getattr(self, '_vegas_memo_store', None)
if memo is None or memo[0] != stamp:
memo = (stamp, {})
self._vegas_memo_store = memo
store: Dict[Any, Any] = memo[1]
return store
def _vegas_settings(self, league: Optional[str]) -> Dict[str, Any]:
"""A league's scroll settings, looked up once per size and config.
The live path asks after every update; a sport's settings lookup can
be expensive (sizing the default card width builds probe renderers).
"""
memo = self._vegas_memo()
key = ('settings', league)
if key not in memo:
memo[key] = dict(self._get_scroll_settings(league))
settings: Dict[str, Any] = memo[key]
return settings
def _vegas_renderer(self, card_width: int,
rankings_cache: Optional[Dict[str, int]]) -> Any:
"""The sport's renderer for one card width, built once rather than per slate.
Building one loads fonts and, for the default card width, probes the
layout; the scroll path pays that on every prepare, which the live
path would repeat on every update.
"""
memo = self._vegas_memo()
key = ('renderer', card_width)
if key not in memo:
memo[key] = self.make_vegas_renderer(card_width, rankings_cache)
renderer = memo[key]
if hasattr(renderer, 'set_rankings_cache'):
# Every time, empty included: the renderer is reused across
# slates, and ranks cleared since must not stay drawn.
renderer.set_rankings_cache(rankings_cache or {})
return renderer
def build_vegas_elements(
self,
games: List[Dict[str, Any]],
leagues: List[str],
rankings_cache: Optional[Dict[str, int]] = None,
fingerprint: Optional[Callable[[Dict[str, Any]], Any]] = None,
now: Optional[float] = None,
) -> Optional[List[Any]]:
"""The slate as live Vegas elements: one card per game, separators between leagues.
Only cards whose fingerprint changed are drawn; the rest come from the
cache. ``fingerprint(game)`` should return what the card draws (the
plugin's own signature fields, the clock included for live games); by
default the whole game dict is used, which redraws on any change. The
teams' ranks from ``rankings_cache`` count too: the renderer draws
them from there, not from the game.
Raises NotImplementedError when the sport has no make_vegas_renderer.
"""
from src.plugin_system.vegas_elements import VegasElement
games = sports_vegas.dedupe_games(games)
if not games:
return None
# Settings follow each game's own league, not the slate's first one:
# a card's width must not change because another league has no games
# today (the ticker refuses a redraw of another width).
first = self._vegas_settings(leagues[0] if leagues else None)
cards = getattr(self, '_vegas_cards', None)
if cards is None:
cards = self._vegas_cards = sports_vegas.VegasCardCache()
odds = getattr(self, '_vegas_odds', None)
if odds is None:
odds = self._vegas_odds = sports_vegas.StickyOdds()
fingerprint = fingerprint or sports_vegas.game_fingerprint
elements: List[Any] = []
keys: List[str] = []
current_league = None
separators = 0
for game in games:
league = game.get("league")
settings = self._vegas_settings(league) if league else first
card_width = int(settings.get("game_card_width", self.display_width))
if settings.get("show_league_separators", True) and league != current_league:
separator = self.vegas_separator(league) if league else None
if separator is not None:
elements.append(VegasElement(
key=f"sep:{separators}:{league}", image=separator, live=False))
separators += 1
current_league = league
key = sports_vegas.game_key(game)
drawn = odds.apply(key, game, now)
ranks = (rankings_cache.get(str(drawn.get("home_abbr"))),
rankings_cache.get(str(drawn.get("away_abbr")))) if rankings_cache else None
renderer = self._vegas_renderer(card_width, rankings_cache)
elements.append(cards.element(
key, (fingerprint(drawn), ranks, card_width, self.display_height),
functools.partial(self.render_vegas_card, renderer, drawn)))
keys.append(key)
cards.retain(keys)
odds.retain(keys)
return elements
def get_current_game_count(self) -> int:
return len(self._current_games)
@@ -525,6 +681,32 @@ class SportsScrollDisplayManager:
scroll_display.clear()
self._current_game_type = ""
def get_vegas_elements_for(
self,
game_type: str,
games: List[Dict[str, Any]],
leagues: List[str],
rankings_cache: Optional[Dict[str, int]] = None,
fingerprint: Optional[Callable[[Dict[str, Any]], Any]] = None,
) -> Optional[List[Any]]:
"""Live Vegas cards for a slate, built on the ``game_type`` display.
None when the sport has no live cards (it does not implement
make_vegas_renderer) or building them failed, so the plugin's
get_vegas_elements() can return it and the ticker falls back to the
plugin's ordinary Vegas content.
"""
scroll_display = self.get_scroll_display(game_type)
try:
return scroll_display.build_vegas_elements(
games, leagues, rankings_cache, fingerprint)
except NotImplementedError:
return None
except Exception:
# Built straight from feed data, like prepare_scroll_content.
self.logger.exception("Error building live Vegas cards")
return None
def get_all_vegas_content_items(self) -> List[Image.Image]:
"""Every display's Vegas items, for splicing into the marquee."""
items: List[Image.Image] = []
src/common/sports_shared.py
+49
View File
@@ -1348,6 +1348,55 @@ class SportsLiveSharedMixin:
or candidate < current):
self._next_scheduled_start_ts = candidate
#: How long a game that finished live is still reported by
#: finished_games_snapshot(): long enough for the recent-games list, which
#: refreshes about hourly, to take it over well before most slates would.
FINISHED_GAME_TTL = 900.0
def _record_finished_game(self, details: Dict) -> None:
"""Remember a game that was live and has just gone final (or looks over).
A finished game leaves ``live_games`` at the next poll, and the recent
list that will show it refreshes about hourly, so in between nothing
holds the game's final score -- and a live Vegas card for it would keep
its last live score. Call this wherever a poll drops a game as final
or over. Only a game this manager had as live is taken; one already
held takes the newer details (a game dropped by an "is it over"
heuristic, then marked final by the feed) but keeps its expiry, so a
feed that lists finals all day cannot keep one here all day.
"""
game_id = details.get("id") if isinstance(details, dict) else None
if not game_id:
return
finished = self.__dict__.setdefault("_finished_games", {})
held = finished.get(game_id)
if held is not None:
finished[game_id] = (held[0], dict(details))
return
if not any(g.get("id") == game_id for g in getattr(self, "live_games", ()) or ()):
return
finished[game_id] = (time.monotonic(), dict(details))
def finished_games_snapshot(self) -> List[Dict]:
"""Games that went final here within FINISHED_GAME_TTL, newest data first.
Copies, safe to decorate. The caller dedupes them against its other
lists (src/common/sports_vegas.dedupe_games keeps the liveliest copy,
and a final beats nothing but a live one).
"""
finished = self.__dict__.get("_finished_games")
if not finished:
return []
now = time.monotonic()
# A copy first: a manager finishing its update in the background (off
# the plugin's lock) may record a game while the ticker reads these.
held = list(finished.items())
for game_id, (seen, _game) in held:
if now - seen > self.FINISHED_GAME_TTL:
finished.pop(game_id, None)
return [dict(game) for _id, (seen, game) in held
if now - seen <= self.FINISHED_GAME_TTL]
def _note_live_fetch(self, found_live: bool) -> None:
"""Record whether a look for live games found any."""
if found_live:
src/common/sports_vegas.py
+250
View File
@@ -0,0 +1,250 @@
"""Live Vegas cards for the sports scoreboards.
A scoreboard hands the Vegas ticker one card per game. As live elements
(src/plugin_system/vegas_elements.py) those cards change on the panel while
they scroll: a goal redraws its game's card and the ticker swaps it in place.
This module is what every scoreboard needs for that and would otherwise write
nine times:
- :func:`game_key` -- a stable key per game, so the ticker can tell which card
a redraw belongs to however the slate is re-sorted.
- :class:`VegasCardCache` -- draws a card only when what it shows changed
(its fingerprint), so an unchanged slate costs a dictionary lookup per game
and a changed one only the cards that changed.
- :class:`StickyOdds` -- live odds are fetched only for games near the front
of the rotation, so a card's odds come and go between polls; this keeps the
last odds for a while instead of redrawing the card without them.
- :func:`dedupe_games` -- a game present in two managers' lists (live and
recent, around the final whistle) appears once, its liveliest copy.
- :func:`finished_games` / :func:`with_finished_games` -- a game that has just
gone final keeps its card, now showing FINAL, where its live card was,
until the recent list (refreshed about hourly) takes it over.
- :func:`game_fingerprint` -- what a card is redrawn on by default: the whole
game dict, frozen hashable.
SportsScrollDisplay.build_vegas_elements (src/common/sports_scroll.py) puts
them together; a plugin adopts it by implementing make_vegas_renderer().
"""
from __future__ import annotations
import time
from collections import OrderedDict
from typing import Any, Callable, Dict, Hashable, Iterable, List, Optional, Tuple
from PIL import Image
#: Which copy of a duplicated game wins: the liveliest.
_STATE_PRIORITY = {'in': 3, 'post': 2, 'pre': 1}
def _state(game: Dict[str, Any]) -> str:
status = game.get('status')
state = status.get('state') if isinstance(status, dict) else status
if isinstance(state, str):
return state
if game.get('is_live'):
return 'in'
if game.get('is_final'):
return 'post'
return 'pre'
def _freeze(value: Any) -> Any:
"""A hashable, order-stable copy of feed data."""
if isinstance(value, dict):
return tuple(sorted((str(k), _freeze(v)) for k, v in value.items()))
if isinstance(value, (list, tuple)):
return tuple(_freeze(v) for v in value)
if isinstance(value, (str, int, float, bool)) or value is None:
return value
return repr(value)
def game_fingerprint(game: Dict[str, Any]) -> Hashable:
"""Everything in a game dict, hashable: a card drawn from it changes only if this does.
The default card version. Nothing a card could draw is left out, so no
field is ever frozen on the panel; the cost is a redraw when a field the
card does not draw changes too, which feed data rarely does between polls.
"""
frozen: Hashable = _freeze(game)
return frozen
def game_key(game: Dict[str, Any]) -> str:
"""A key that names this game and nothing else, across polls.
``game:<league>:<id>`` from the feed's own id. A game without one falls
back to its teams and start time, which is stable for the life of a game.
"""
league = game.get('league') or 'game'
game_id = game.get('id') or game.get('game_id')
if game_id not in (None, ''):
return f"game:{league}:{game_id}"
away = game.get('away_abbr') or game.get('away_team') or '?'
home = game.get('home_abbr') or game.get('home_team') or '?'
start = game.get('start_time_utc') or game.get('start_time') or ''
return f"game:{league}:{away}@{home}:{start}"
def dedupe_games(games: Iterable[Dict[str, Any]],
key_fn: Callable[[Dict[str, Any]], str] = game_key) -> List[Dict[str, Any]]:
"""Each game once, in first-seen order, keeping its liveliest copy.
Around a final whistle a game can be in the live list (last poll) and the
recent list (next poll) at once; two cards with one key would be refused
by the ticker, and showing the game twice is wrong anyway.
"""
chosen: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
for game in games:
key = key_fn(game)
current = chosen.get(key)
if current is None or _STATE_PRIORITY.get(_state(game), 0) > \
_STATE_PRIORITY.get(_state(current), 0):
chosen[key] = game
return list(chosen.values())
def finished_games(
live_managers: Iterable[Tuple[str, Any]]) -> List[Dict[str, Any]]:
"""Games that just left these live managers' lists, final ones as recent games.
``live_managers`` pairs each league with its live manager (None is
skipped). Each manager reports what SportsLiveSharedMixin recorded
(finished_games_snapshot, copies), with its league. A final game is
drawn as a recent card. One a poll only judged over -- a tied end of
regulation looks like that too -- keeps its last live state, so its card
never says FINAL early; if play resumes the live list has it again, and
dedupe_games keeps that copy.
"""
finished: List[Dict[str, Any]] = []
for league, manager in live_managers:
snapshot = getattr(manager, 'finished_games_snapshot', None)
if not callable(snapshot):
continue
for game in snapshot():
game['league'] = league
if game.get('is_final'):
status = game.get('status')
status = dict(status) if isinstance(status, dict) else {}
status['state'] = 'post'
game.update(status=status, is_live=False)
finished.append(game)
return finished
def with_finished_games(
games: List[Dict[str, Any]], leagues: List[str],
finished: List[Dict[str, Any]],
) -> Tuple[List[Dict[str, Any]], List[str]]:
"""The slate with games that just went final where their live cards were.
A slate lists each league's games together, live ones first. Each
finished game goes after its league's live games, ahead of the rest; a
league with no games left in the slate is added at the end. A finished
game the slate also has (the recent list caught up) is left for
dedupe_games, which keeps one copy.
"""
if not finished:
return list(games), list(leagues)
pending: "OrderedDict[Any, List[Dict[str, Any]]]" = OrderedDict()
for game in finished:
pending.setdefault(game.get('league'), []).append(game)
merged: List[Dict[str, Any]] = []
for index, game in enumerate(games):
league = game.get('league')
if league in pending and _state(game) != 'in':
merged.extend(pending.pop(league))
merged.append(game)
following = games[index + 1] if index + 1 < len(games) else None
if league in pending and (following is None or following.get('league') != league):
merged.extend(pending.pop(league)) # the league's games were all live
leagues = list(leagues)
for league, rest in pending.items():
merged.extend(rest)
if league not in leagues:
leagues.append(league)
return merged, leagues
class VegasCardCache:
"""Cards drawn once per fingerprint, kept for as long as their game is.
``element(key, fingerprint, render)`` returns a VegasElement whose image is
``render()``'s -- called only when the fingerprint differs from the one the
cached card was drawn for. The fingerprint is also the element's version,
so the ticker skips unchanged cards without comparing pixels.
Bounded: keys not passed to :meth:`retain` after a slate are dropped, and
at most ``max_entries`` are ever held (oldest first).
"""
def __init__(self, max_entries: int = 96) -> None:
self.max_entries = max(1, int(max_entries))
self._cards: "OrderedDict[str, Tuple[Hashable, Image.Image]]" = OrderedDict()
self.renders = 0
def element(self, key: str, fingerprint: Hashable,
render: Callable[[], Image.Image], live: bool = True) -> Any:
from src.plugin_system.vegas_elements import VegasElement
cached = self._cards.get(key)
if cached is not None and cached[0] == fingerprint:
self._cards.move_to_end(key)
image = cached[1]
else:
image = render()
self.renders += 1
self._cards[key] = (fingerprint, image)
self._cards.move_to_end(key)
while len(self._cards) > self.max_entries:
self._cards.popitem(last=False)
return VegasElement(key=key, image=image, version=fingerprint, live=live)
def retain(self, keys: Iterable[str]) -> None:
"""Forget every card whose key is not in ``keys``."""
keep = set(keys)
for key in [k for k in self._cards if k not in keep]:
self._cards.pop(key, None)
def clear(self) -> None:
self._cards.clear()
def __len__(self) -> int:
return len(self._cards)
class StickyOdds:
"""Keep a game's last odds on its card while a live poll leaves them out.
Live odds are fetched only for games near the front of the rotation
(src/common/sports_fetch.py), so the same game's dict has odds on one poll
and none on the next. Drawn as-is that redraws the card every poll with
the odds flickering in and out. ``apply`` returns the game with its last
non-empty odds put back, for up to ``ttl_s`` seconds after they were seen.
"""
def __init__(self, ttl_s: float = 600.0) -> None:
self.ttl_s = float(ttl_s)
self._seen: Dict[str, Tuple[float, Any]] = {}
def apply(self, key: str, game: Dict[str, Any],
now: Optional[float] = None) -> Dict[str, Any]:
now = time.monotonic() if now is None else now
odds = game.get('odds')
if odds:
self._seen[key] = (now, odds)
return game
seen = self._seen.get(key)
if seen is None or now - seen[0] > self.ttl_s:
self._seen.pop(key, None)
return game
refilled = dict(game)
refilled['odds'] = seen[1]
return refilled
def retain(self, keys: Iterable[str]) -> None:
keep = set(keys)
for key in [k for k in self._seen if k not in keep]:
self._seen.pop(key, None)
src/plugin_system/testing/vegas.py
+85
View File
@@ -266,6 +266,91 @@ def _base_redraw():
return BasePlugin.redraw_vegas_element
def render_vegas_strip(plugin: Any, plugin_id: str, display_manager: Any,
live: bool = True) -> Any:
"""The plugin's block of the Vegas strip, laid out exactly as the ticker would.
Fetched through the ticker's own adapter (trimming, pinning, width
budget) and joined with its own spacing, so what this draws is what
scrolls. ``live=False`` shows the ordinary get_vegas_content() instead.
Returns ``(block, layout)`` -- layout a list of ``(x, key, width)`` for the
live elements in the block -- or ``(None, [])`` when there is nothing.
"""
_adapter, block, layout = _vegas_block(plugin, plugin_id, display_manager, live)
return block, [(x, meta.key, width) for x, meta, width in layout]
def _vegas_block(plugin: Any, plugin_id: str, display_manager: Any, live: bool) -> Any:
"""(adapter, block, layout) for a plugin's Vegas block, through the ticker's own code."""
from src.vegas_mode.config import VegasModeConfig
from src.vegas_mode.elements import LiveEpochs
from src.vegas_mode.plugin_adapter import PluginAdapter
from src.vegas_mode.render_pipeline import join_plugin_rows
config = VegasModeConfig()
adapter = PluginAdapter(display_manager, config)
adapter.live_elements_enabled = live
adapter.live_epochs = LiveEpochs()
images = adapter.get_content(plugin, plugin_id, offscreen_only=True)
if not images:
return adapter, None, []
block, layout = join_plugin_rows(images, config)
return adapter, block, layout
def render_vegas_timeline(plugin: Any, plugin_id: str, display_manager: Any,
steps: int = 8, step_seconds: float = 0.25,
run_update: bool = False) -> Any:
"""The plugin's Vegas block at successive moments, one row per step.
Row 0 is the block as placed (render_vegas_strip). Each later row is the
same block ``step_seconds`` later, changed the way the ticker would change
it in place: every live element with ``refresh_hz`` redrawn for that
moment through redraw_vegas_element(), and -- with ``run_update`` -- the
plugin's update() run first and every live element redrawn from the new
data. A redraw of another width is left out, as the ticker refuses it.
Rows are separated by a grey line.
Returns ``(image, rows)``, or ``(None, 0)`` when there is nothing.
"""
import time
import numpy as np
adapter, block, layout = _vegas_block(plugin, plugin_id, display_manager, True)
if block is None:
return None, 0
base = np.array(block.convert('RGB'))
rows = [base.copy()]
height = base.shape[0]
start = time.monotonic()
for step in range(1, max(1, int(steps))):
frame = rows[-1].copy()
if run_update:
plugin.update()
adapter.live_epochs.bump(plugin_id)
batch = adapter.render_live_elements(plugin, plugin_id, lock_timeout=5.0)
rendered = batch[1] if batch is not None else {}
for x, meta, width in layout:
element = rendered.get(meta.key)
if element is not None and element.width == width:
frame[:, x:x + width] = element.pixels
at = start + step * float(step_seconds)
for x, meta, width in layout:
if meta.refresh_hz > 0:
element = adapter.redraw_live_element(plugin, plugin_id, meta.key,
width, height, at)
if element is not None and element.width == width:
frame[:, x:x + width] = element.pixels
rows.append(frame)
divider = np.full((1, base.shape[1], 3), 60, dtype=np.uint8)
stacked = [rows[0]]
for row in rows[1:]:
stacked.extend([divider, row])
return Image.fromarray(np.concatenate(stacked, axis=0)), len(rows)
def check_plugin_vegas_elements(plugin_id: str, plugin_dir: Any, config: dict,
mock_data: dict, width: int, height: int,
run_update: bool = True) -> VegasElementReport:
src/vegas_mode/render_pipeline.py
+47 -35
View File
@@ -32,6 +32,51 @@ logger = logging.getLogger(__name__)
SYNC_SEND_INTERVAL = 1.0 / 90
def join_plugin_rows(
images: List[Image.Image], config: VegasModeConfig
) -> Tuple[Image.Image, List[Tuple[int, ElementMeta, int]]]:
"""Join one plugin's images into the block the strip will hold.
Returns ``(block, layout)``, layout being ``(x, meta, width)`` for every
image tagged as a live element (src/vegas_mode/elements.py), x measured
from the block's left edge. A single image is returned as it is.
A module function so tooling (scripts/render_plugin.py --vegas) lays a
plugin out exactly as the ticker does.
"""
if len(images) == 1:
meta = meta_of(images[0])
layout = [(0, meta, images[0].width)] if meta is not None else []
return images[0], layout
floor = max(0, config.intra_plugin_gap)
target = max(0, config.min_content_separation)
threshold = config.trim_threshold
# Space by measured separation, not a flat gap. Rows drawn flush to their
# own edges (sports score cards) would otherwise end up nearly touching,
# while rows that already carry wide margins would be pushed needlessly
# further apart.
gaps = [
separation_gap(images[i], images[i + 1], target, floor, threshold)
for i in range(len(images) - 1)
]
width = sum(img.width for img in images) + sum(gaps)
height = max(img.height for img in images)
block = Image.new('RGB', (width, height), (0, 0, 0))
layout: List[Tuple[int, ElementMeta, int]] = []
x = 0
for i, img in enumerate(images):
block.paste(img, (x, 0))
meta = meta_of(img)
if meta is not None:
layout.append((x, meta, img.width))
x += img.width + (gaps[i] if i < len(gaps) else 0)
return block, layout
class RenderPipeline:
"""
High-performance render pipeline for Vegas scroll mode.
@@ -787,42 +832,9 @@ class RenderPipeline:
) -> Tuple[Image.Image, List[Tuple[int, ElementMeta, int]]]:
"""_join_plugin_rows, plus where each live element landed in the block.
Returns ``(block, layout)``, layout being ``(x, meta, width)`` for
every image tagged as a live element (src/vegas_mode/elements.py), x
measured from the block's left edge. The offsets were always computed
here; they used to be thrown away.
See join_plugin_rows.
"""
if len(images) == 1:
meta = meta_of(images[0])
layout = [(0, meta, images[0].width)] if meta is not None else []
return images[0], layout
floor = max(0, self.config.intra_plugin_gap)
target = max(0, self.config.min_content_separation)
threshold = self.config.trim_threshold
# Space by measured separation, not a flat gap. Rows drawn flush to
# their own edges (sports score cards) would otherwise end up nearly
# touching, while rows that already carry wide margins would be pushed
# needlessly further apart.
gaps = [
separation_gap(images[i], images[i + 1], target, floor, threshold)
for i in range(len(images) - 1)
]
width = sum(img.width for img in images) + sum(gaps)
height = max(img.height for img in images)
block = Image.new('RGB', (width, height), (0, 0, 0))
layout: List[Tuple[int, ElementMeta, int]] = []
x = 0
for i, img in enumerate(images):
block.paste(img, (x, 0))
meta = meta_of(img)
if meta is not None:
layout.append((x, meta, img.width))
x += img.width + (gaps[i] if i < len(gaps) else 0)
return block, layout
return join_plugin_rows(images, self.config)
# -- live element records ---------------------------------------------
#
test/test_harness_vegas_elements.py
+28
View File
@@ -15,6 +15,7 @@ from src.plugin_system.testing.harness import _instantiate # noqa: E402
from src.plugin_system.testing.loading import build_full_config, load_harness_spec, load_manifest # noqa: E402
from src.plugin_system.testing.vegas import ( # noqa: E402
check_vegas_elements, implements_vegas_elements, render_vegas_elements,
render_vegas_strip, render_vegas_timeline,
)
from src.plugin_system.testing.visual_display_manager import VisualTestDisplayManager # noqa: E402
from src.plugin_system.vegas_elements import VegasElement # noqa: E402
@@ -178,3 +179,30 @@ def test_check_plugin_reports_a_failing_element_check_and_carries_on(monkeypatch
False, None, None)
vegas = [r for r in results if r.mode == "vegas elements"]
assert len(vegas) == 1 and "boom" in vegas[0].error
def test_a_timeline_moves_what_animates_and_nothing_else():
import numpy as np
plugin, dm = _stub(dot_speed=200)
_block, layout = render_vegas_strip(plugin, "vegas-live-stub", dm)
image, rows = render_vegas_timeline(plugin, "vegas-live-stub", dm, steps=3,
step_seconds=0.5)
assert rows == 3 and image.height == 3 * H + 2
pixels = np.asarray(image)
first, last = pixels[:H], pixels[2 * (H + 1):]
columns = {key: (x, width) for x, key, width in layout}
x, width = columns["map"]
assert (first[:, x:x + width] != last[:, x:x + width]).any()
x, width = columns["card:0"]
assert (first[:, x:x + width] == last[:, x:x + width]).all()
def test_a_timeline_with_updates_redraws_the_cards_in_place():
import numpy as np
plugin, dm = _stub(map_hz=0)
_block, layout = render_vegas_strip(plugin, "vegas-live-stub", dm)
image, rows = render_vegas_timeline(plugin, "vegas-live-stub", dm, steps=2,
run_update=True)
pixels = np.asarray(image)
x, width = {key: (x, width) for x, key, width in layout}["card:0"]
assert (pixels[:H, x:x + width] != pixels[H + 1:, x:x + width]).any()
test/test_sports_vegas_elements.py
+415
View File
@@ -0,0 +1,415 @@
"""Live Vegas cards for the scoreboards (src/common/sports_vegas.py, sports_scroll).
A scoreboard adopts live cards by implementing make_vegas_renderer(); the
shared code then draws one card per game, only when what the card shows
changed, keyed by the game so the ticker can swap it in place.
"""
import sys
import time
from pathlib import Path
from types import SimpleNamespace
import pytest
from PIL import Image
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from src.common import sports_vegas # noqa: E402
from src.common.sports_scroll import SportsScrollDisplay, SportsScrollDisplayManager # noqa: E402
from src.common.sports_shared import SportsLiveSharedMixin # noqa: E402
from src.plugin_system.vegas_elements import VegasElement # noqa: E402
W, H = 128, 32
def _game(gid, league="nfl", home=0, away=0, state="in", clock="10:00", odds=None):
return {"id": gid, "league": league, "home_abbr": "HOM", "away_abbr": "AWY",
"home_score": home, "away_score": away,
"status": {"state": state}, "clock": clock, "odds": odds}
class _Renderer:
def __init__(self, width):
self.width = width
self.calls = []
self.rankings = None
def render_game_card(self, game, game_type):
self.calls.append((game["id"], game_type, game.get("odds")))
return Image.new("RGB", (self.width, H), (game["home_score"] * 10 % 255, 0, 0))
def set_rankings_cache(self, rankings):
self.rankings = rankings
class _Display(SportsScrollDisplay):
made = 0
def make_vegas_renderer(self, card_width, rankings_cache=None):
type(self).made += 1
self.renderer = _Renderer(card_width)
return self.renderer
def _load_separator_icons(self):
self._separator_icons = {"nfl": Image.new("RGBA", (10, 10), (255, 255, 255, 255)),
"ncaa_fb": Image.new("RGBA", (10, 10), (0, 255, 0, 255))}
class _Plain(SportsScrollDisplay):
pass
def _dm():
return SimpleNamespace(width=W, height=H, matrix=None, refresh_hz=100.0,
set_scrolling_state=lambda *a, **k: None)
def _display(cls=_Display, **config):
return cls(_dm(), config)
def _fp(game):
return (game["home_score"], game["away_score"], game["clock"],
bool(game.get("odds")))
# -- building blocks ------------------------------------------------------------
def test_game_keys_name_the_game():
assert sports_vegas.game_key(_game("401")) == "game:nfl:401"
assert sports_vegas.game_key({"league": "mlb", "away_abbr": "A", "home_abbr": "B",
"start_time": "2026-09-30T19:00Z"}) \
== "game:mlb:A@B:2026-09-30T19:00Z"
def test_dedupe_keeps_the_liveliest_copy_in_first_seen_order():
games = [_game("1", state="post", home=3), _game("2", state="pre"),
_game("1", state="in", home=2)]
kept = sports_vegas.dedupe_games(games)
assert [g["id"] for g in kept] == ["1", "2"]
assert kept[0]["status"]["state"] == "in"
def test_the_card_cache_draws_only_on_a_new_fingerprint():
cache = sports_vegas.VegasCardCache()
drawn = []
def render(value):
drawn.append(value)
return Image.new("RGB", (4, H))
first = cache.element("k", 1, lambda: render(1))
again = cache.element("k", 1, lambda: render(2))
changed = cache.element("k", 2, lambda: render(3))
assert drawn == [1, 3]
assert isinstance(first, VegasElement) and first.version == 1
assert again.image is first.image and changed.version == 2
def test_the_card_cache_forgets_games_not_retained_and_is_bounded():
cache = sports_vegas.VegasCardCache(max_entries=3)
for key in "abcde":
cache.element(key, 0, lambda: Image.new("RGB", (4, H)))
assert len(cache) == 3
cache.retain(["e"])
assert len(cache) == 1
def test_sticky_odds_fill_a_gap_and_then_expire():
odds = sports_vegas.StickyOdds(ttl_s=60)
game = _game("1", odds={"spread": -3})
assert odds.apply("k", game, now=0)["odds"] == {"spread": -3}
later = _game("1", odds=None)
refilled = odds.apply("k", later, now=30)
assert refilled["odds"] == {"spread": -3}
assert later["odds"] is None # the feed's dict is untouched
assert odds.apply("k", later, now=100)["odds"] is None
# -- SportsScrollDisplay.build_vegas_elements ---------------------------------------
def test_a_slate_becomes_keyed_cards_with_league_separators():
display = _display()
games = [_game("1"), _game("2"), _game("3", league="ncaa_fb")]
elements = display.build_vegas_elements(games, ["nfl", "ncaa_fb"], fingerprint=_fp)
keys = [e.key for e in elements]
assert keys == ["sep:0:nfl", "game:nfl:1", "game:nfl:2",
"sep:1:ncaa_fb", "game:ncaa_fb:3"]
assert [e.live for e in elements] == [False, True, True, False, True]
card = elements[1]
assert card.image.size == (display._get_scroll_settings()["game_card_width"], H)
def test_only_changed_games_are_redrawn():
display = _display()
games = [_game("1"), _game("2"), _game("3")]
display.build_vegas_elements(games, ["nfl"], fingerprint=_fp)
renderer = display.renderer
assert len(renderer.calls) == 3
display.build_vegas_elements(games, ["nfl"], fingerprint=_fp)
assert len(renderer.calls) == 3 # nothing changed
games[1] = _game("2", home=7)
display.build_vegas_elements(games, ["nfl"], fingerprint=_fp)
assert [c[0] for c in renderer.calls[3:]] == ["2"]
def test_a_clock_change_redraws_that_card_when_the_fingerprint_has_the_clock():
display = _display()
display.build_vegas_elements([_game("1"), _game("2")], ["nfl"], fingerprint=_fp)
display.build_vegas_elements([_game("1", clock="9:41"), _game("2")], ["nfl"],
fingerprint=_fp)
assert [c[0] for c in display.renderer.calls[2:]] == ["1"]
def test_the_renderer_is_built_once_not_per_slate():
_Display.made = 0
display = _display()
for _ in range(3):
display.build_vegas_elements([_game("1")], ["nfl"], fingerprint=_fp)
assert _Display.made == 1
def test_a_card_keeps_its_odds_through_a_poll_without_them():
display = _display()
display.build_vegas_elements([_game("1", odds={"spread": -3})], ["nfl"], fingerprint=_fp)
display.build_vegas_elements([_game("1", odds=None)], ["nfl"], fingerprint=_fp)
assert len(display.renderer.calls) == 1 # no redraw without the odds
def test_game_dicts_are_never_mutated():
display = _display()
game = _game("1", odds=None)
snapshot = dict(game)
display.build_vegas_elements([game], ["nfl"], fingerprint=_fp)
assert game == snapshot
def test_a_sport_without_a_renderer_stays_on_its_ordinary_content():
manager = SportsScrollDisplayManager(_dm(), {})
manager.display_class = _Plain
assert manager.get_vegas_elements_for("mixed", [_game("1")], ["nfl"]) is None
def test_a_renderer_that_fails_falls_back_instead_of_raising():
class Broken(_Display):
def render_vegas_card(self, renderer, game):
raise KeyError("home_logo")
manager = SportsScrollDisplayManager(_dm(), {})
manager.display_class = Broken
assert manager.get_vegas_elements_for("mixed", [_game("1")], ["nfl"]) is None
def test_the_manager_hands_back_the_elements():
manager = SportsScrollDisplayManager(_dm(), {})
manager.display_class = _Display
elements = manager.get_vegas_elements_for("mixed", [_game("1")], ["nfl"],
fingerprint=_fp)
assert [e.key for e in elements] == ["sep:0:nfl", "game:nfl:1"]
# -- the finished-game capture ------------------------------------------------------
class _Live(SportsLiveSharedMixin):
def __init__(self):
self.live_games = [_game("1"), _game("2")]
def test_a_game_that_was_live_is_remembered_when_it_ends():
live = _Live()
live._record_finished_game(_game("1", state="post", home=24))
live._record_finished_game(_game("99", state="post")) # never live here
finished = live.finished_games_snapshot()
assert [g["id"] for g in finished] == ["1"]
assert finished[0]["home_score"] == 24
finished[0]["league"] = "changed"
assert live.finished_games_snapshot()[0]["league"] == "nfl" # copies
def test_a_finished_game_is_forgotten_after_its_ttl(monkeypatch):
live = _Live()
live._record_finished_game(_game("1", state="post"))
real = time.monotonic
monkeypatch.setattr(time, "monotonic", lambda: real() + live.FINISHED_GAME_TTL + 1)
assert live.finished_games_snapshot() == []
def test_final_replaces_the_live_card_in_place():
# The same key, so the ticker swaps the card rather than adding one.
display = _display()
before = display.build_vegas_elements([_game("1", home=7)], ["nfl"], fingerprint=_fp)
after = display.build_vegas_elements(
sports_vegas.dedupe_games([_game("1", state="post", home=14)]), ["nfl"],
fingerprint=_fp)
assert before[-1].key == after[-1].key
assert before[-1].version != after[-1].version
def test_settings_are_looked_up_once_per_league_not_per_update(monkeypatch):
display = _display()
calls = []
real = display._get_scroll_settings
monkeypatch.setattr(display, "_get_scroll_settings",
lambda league=None: calls.append(league) or real(league))
slate = [_game("1"), _game("2", league="ncaa_fb")]
for _ in range(3):
display.build_vegas_elements(slate, ["nfl", "ncaa_fb"], fingerprint=_fp)
assert sorted(calls) == ["ncaa_fb", "nfl"]
def test_each_card_takes_its_own_leagues_width():
display = _display(nfl={"scroll_settings": {"game_card_width": 100}},
ncaa_fb={"scroll_settings": {"game_card_width": 140}})
display.SCROLL_LEAGUE_KEYS = ("nfl", "ncaa_fb")
slate = [_game("1"), _game("2", league="ncaa_fb")]
widths = {e.key: e.image.width
for e in display.build_vegas_elements(slate, ["nfl", "ncaa_fb"], fingerprint=_fp)
if e.live}
# ...and keeps it when the other league has nothing to show.
alone = display.build_vegas_elements(slate[1:], ["ncaa_fb"], fingerprint=_fp)
assert widths == {"game:nfl:1": 100, "game:ncaa_fb:2": 140}
assert [e.image.width for e in alone if e.live] == [140]
def test_a_rank_change_redraws_the_card():
display = _display()
display.build_vegas_elements([_game("1")], ["nfl"], {"HOM": 5}, fingerprint=_fp)
drawn = display._vegas_cards.renders
display.build_vegas_elements([_game("1")], ["nfl"], {"HOM": 5}, fingerprint=_fp)
assert display._vegas_cards.renders == drawn
display.build_vegas_elements([_game("1")], ["nfl"], {"HOM": 3}, fingerprint=_fp)
assert display._vegas_cards.renders == drawn + 1
def test_finished_games_come_back_as_recent_games_of_their_league():
live = _Live()
live._record_finished_game(dict(_game("1", home=24), is_final=True))
finished = sports_vegas.finished_games([("ncaa_fb", live), ("nfl", None)])
assert [(g["id"], g["league"], g["status"]["state"], g["is_live"])
for g in finished] == [("1", "ncaa_fb", "post", False)]
def test_a_game_only_judged_over_is_not_drawn_final():
# A tied end of regulation looks over to the heuristics; overtime may follow.
live = _Live()
live._record_finished_game(_game("1", home=24, clock="0:00"))
finished = sports_vegas.finished_games([("nfl", live)])
assert finished[0]["status"]["state"] == "in"
assert not finished[0].get("is_final")
# ...and when play resumes, the live list's copy is the one kept.
resumed = _game("1", home=24, clock="15:00")
merged, _ = sports_vegas.with_finished_games([resumed], ["nfl"], finished)
assert sports_vegas.dedupe_games(merged) == [resumed]
def test_the_snapshot_copes_with_games_recorded_while_it_reads():
# A baseball league finishing its update in the background records games
# off the plugin's lock while the ticker takes snapshots.
import sys
import threading
live = _Live()
live.live_games = [_game(str(i)) for i in range(400)]
errors = []
stop = threading.Event()
def record():
for i in range(400):
live._record_finished_game(dict(_game(str(i)), is_final=True))
stop.set()
def read():
while not stop.is_set():
try:
live.finished_games_snapshot()
except RuntimeError as exc:
errors.append(exc)
return
previous = sys.getswitchinterval()
sys.setswitchinterval(1e-6)
try:
readers = [threading.Thread(target=read) for _ in range(2)]
for t in readers:
t.start()
record()
for t in readers:
t.join(10)
finally:
sys.setswitchinterval(previous)
assert not errors
assert len(live.finished_games_snapshot()) == 400
def test_a_held_game_takes_the_final_details_but_keeps_its_expiry(monkeypatch):
live = _Live()
live._record_finished_game(_game("1", home=24, clock="0:00")) # judged over
live.live_games = [] # dropped
now = time.monotonic()
monkeypatch.setattr(time, "monotonic", lambda: now + 60)
live._record_finished_game(dict(_game("1", home=27), is_final=True))
finished = sports_vegas.finished_games([("nfl", live)])
assert finished[0]["home_score"] == 27 and finished[0]["status"]["state"] == "post"
monkeypatch.setattr(time, "monotonic", lambda: now + live.FINISHED_GAME_TTL + 1)
assert live.finished_games_snapshot() == []
def _slate(*games):
return [_game(gid, league=league, state=state) for gid, league, state in games]
@pytest.mark.parametrize("games,leagues,expected_ids,expected_leagues", [
# After the league's live games, ahead of its recent and upcoming ones.
(_slate(("2", "nfl", "in"), ("3", "nfl", "post"), ("4", "nfl", "pre"),
("5", "ncaa_fb", "in")),
["nfl", "ncaa_fb"], ["2", "F", "3", "4", "5"], ["nfl", "ncaa_fb"]),
# A league whose games were all live.
(_slate(("2", "nfl", "in"), ("5", "ncaa_fb", "in")),
["nfl", "ncaa_fb"], ["2", "F", "5"], ["nfl", "ncaa_fb"]),
# No live games: first in its league.
(_slate(("5", "ncaa_fb", "pre"), ("3", "nfl", "post")),
["ncaa_fb", "nfl"], ["5", "F", "3"], ["ncaa_fb", "nfl"]),
# Its league has nothing else to show.
(_slate(("5", "ncaa_fb", "pre")), ["ncaa_fb"], ["5", "F"], ["ncaa_fb", "nfl"]),
])
def test_a_finished_game_goes_where_its_live_card_was(games, leagues, expected_ids,
expected_leagues):
final = _game("F", league="nfl", state="post")
merged, merged_leagues = sports_vegas.with_finished_games(games, leagues, [final])
assert [g["id"] for g in merged] == expected_ids
assert merged_leagues == expected_leagues
def test_the_default_fingerprint_follows_every_field_and_ignores_key_order():
game = _game("1", odds={"spread": -3, "details": ["a", "b"]})
same = dict(reversed(list(game.items())))
assert sports_vegas.game_fingerprint(game) == sports_vegas.game_fingerprint(same)
hash(sports_vegas.game_fingerprint(game))
for field, value in [("clock", "9:59"), ("odds", {"spread": -3, "details": ["a"]}),
("status", {"state": "post"})]:
assert sports_vegas.game_fingerprint(dict(game, **{field: value})) != \
sports_vegas.game_fingerprint(game)
def test_no_finished_games_leaves_the_slate_alone():
games = _slate(("2", "nfl", "in"))
merged, leagues = sports_vegas.with_finished_games(games, ["nfl"], [])
assert merged == games and merged is not games and leagues == ["nfl"]
@pytest.mark.parametrize("state,expected", [("in", "live"), ("post", "recent"),
("pre", "upcoming")])
def test_the_default_card_type_follows_the_game_state(state, expected):
display = _display()
display.build_vegas_elements([_game("1", state=state)], ["nfl"], fingerprint=_fp)
assert display.renderer.calls[0][1] == expected
def test_ranks_cleared_since_are_not_kept_by_the_reused_renderer():
display = _display()
display.build_vegas_elements([_game("1")], ["nfl"], {"HOM": 5}, fingerprint=_fp)
assert display.renderer.rankings == {"HOM": 5}
display.build_vegas_elements([_game("1")], ["nfl"], None, fingerprint=_fp)
assert display.renderer.rankings == {}