diff --git a/client.py b/client.py index d87abc4..672e5f5 100644 --- a/client.py +++ b/client.py @@ -3,6 +3,7 @@ import asyncio import json import logging +import math import time from typing import Any, Dict, Optional @@ -23,8 +24,9 @@ PING_MSG = json.dumps({"action": "Ping"}) -RECONNECT_DELAY = 2 # seconds -PING_INTERVAL = 30 # seconds +RECONNECT_DELAY = 2 # seconds +PING_INTERVAL = 30 # seconds +PENALTY_BOX_DURATION_S = 30 # seconds _PREFIX = "ScoreBoard.CurrentGame." _VERSION_KEY = "ScoreBoard.Version(release)" @@ -111,6 +113,9 @@ def __init__(self, host: str = "localhost", port: int = 8000) -> None: self._connected = False self._last_update: Optional[float] = None self._task: Optional[asyncio.Task] = None + self._box_entry_times: Dict[str, int] = {} # key: "." -> epoch ms + self._box_entry_mono: Dict[str, float] = {} # key: "." -> monotonic s + self._was_in_box: Dict[str, bool] = {} # key: "." -> last known in_box @property def connected(self) -> bool: @@ -152,6 +157,31 @@ def _get(self, suffix: str, target_type: type = str) -> Any: return None return _coerce(raw, target_type) + def _update_box_entry_times(self) -> None: + """Record entry time on observed false→True penalty-box transitions. + + Called after every state update. Entry times are only set when a + false→True transition is observed; skaters already in the box when + the client first connects keep None until an exit+re-entry occurs. + The monotonic clock is stored for stable countdown computation; + epoch ms is also stored for client overlay usage. + """ + for team_n in (1, 2): + for crg_pos in POSITION_MAP.values(): + key = f"{team_n}.{crg_pos}" + in_box = ( + self._get(f"Team({team_n}).Position({crg_pos}).PenaltyBox", bool) + or False + ) + was_in_box = self._was_in_box.get(key, False) + if in_box and not was_in_box: + self._box_entry_times[key] = int(time.time() * 1000) + self._box_entry_mono[key] = time.monotonic() + elif not in_box: + self._box_entry_times.pop(key, None) + self._box_entry_mono.pop(key, None) + self._was_in_box[key] = in_box + def _team(self, n: int) -> TeamState: """Build a TeamState for team n using TEAM_FIELD_MAP and POSITION_MAP.""" prefix = f"Team({n})." @@ -159,14 +189,22 @@ def _team(self, n: int) -> TeamState: field: self._get(prefix + suffix, typ) for field, (suffix, typ) in TEAM_FIELD_MAP.items() } - positions = { - field: SkaterPosition( + positions = {} + for field, crg_pos in POSITION_MAP.items(): + entered_ms = self._box_entry_times.get(f"{n}.{crg_pos}") + entered_mono = self._box_entry_mono.get(f"{n}.{crg_pos}") + if entered_mono is not None: + elapsed_s = time.monotonic() - entered_mono + remaining: Optional[int] = max(0, math.ceil(PENALTY_BOX_DURATION_S - elapsed_s)) + else: + remaining = None + positions[field] = SkaterPosition( name=self._get(f"{prefix}Position({crg_pos}).Name"), number=self._get(f"{prefix}Position({crg_pos}).RosterNumber"), in_box=self._get(f"{prefix}Position({crg_pos}).PenaltyBox", bool) or False, + box_entered_at_ms=entered_ms, + box_time_remaining_s=remaining, ) - for field, crg_pos in POSITION_MAP.items() - } return TeamState(**flat_fields, **positions) def get_live_state(self) -> LiveState: @@ -185,6 +223,9 @@ async def _run_once(self, uri: str) -> None: """Connect, subscribe, and receive updates until disconnect.""" logger.info("Connecting to %s", uri) async with websockets.connect(uri, ping_interval=None) as ws: + self._box_entry_times.clear() + self._box_entry_mono.clear() + self._was_in_box.clear() self._connected = True logger.info("Connected to scoreboard WS") await ws.send(REGISTER_MSG) @@ -203,6 +244,7 @@ async def _run_once(self, uri: str) -> None: try: if "state" in msg: self._apply_update(msg["state"]) + self._update_box_entry_times() elif "error" in msg: logger.warning("Scoreboard error: %s", msg["error"]) except Exception: diff --git a/models.py b/models.py index 1bec3c5..76cc217 100644 --- a/models.py +++ b/models.py @@ -7,6 +7,21 @@ class SkaterPosition(BaseModel): name: Optional[str] = None number: Optional[str] = None in_box: bool = False + box_entered_at_ms: Optional[int] = Field( + default=None, + description=( + "Unix epoch milliseconds when this skater entered the penalty box. " + "None when the skater is not in the box. " + "Overlay usage: elapsed_ms = Date.now() - box_entered_at_ms" + ), + ) + box_time_remaining_s: Optional[int] = Field( + default=None, + description=( + "Whole seconds remaining in the 30-second penalty box. " + "None when the skater is not in the box. 0 when time has expired." + ), + ) class TeamState(BaseModel): diff --git a/tests/test_client.py b/tests/test_client.py index 0a937a3..331dced 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -2,6 +2,8 @@ import asyncio import json +import time +from unittest.mock import patch import pytest import pytest_asyncio @@ -143,6 +145,133 @@ async def test_penalty_box_update(mock_server): await asyncio.sleep(0.05) +async def test_box_entered_at_ms_is_none_when_not_in_box(mock_server): + """box_entered_at_ms is None for skaters not in the penalty box.""" + client, task = await _connected_client(mock_server) + try: + state = client.get_live_state() + assert state.team1.blocker1.box_entered_at_ms is None + finally: + client.stop() + await asyncio.sleep(0.05) + + +async def test_box_entered_at_ms_set_on_entry(mock_server): + """box_entered_at_ms is set to a recent epoch-ms timestamp on box entry.""" + client, task = await _connected_client(mock_server) + try: + before_ms = int(time.time() * 1000) + await mock_server.push_update({ + "ScoreBoard.CurrentGame.Team(1).Position(Blocker1).PenaltyBox": True, + }) + await asyncio.sleep(0.1) + after_ms = int(time.time() * 1000) + + state = client.get_live_state() + assert state.team1.blocker1.in_box is True + ts = state.team1.blocker1.box_entered_at_ms + assert ts is not None + assert before_ms <= ts <= after_ms + finally: + client.stop() + await asyncio.sleep(0.05) + + +async def test_box_entered_at_ms_cleared_on_exit(mock_server): + """box_entered_at_ms returns to None when in_box transitions to False.""" + client, task = await _connected_client(mock_server) + try: + await mock_server.push_update({ + "ScoreBoard.CurrentGame.Team(1).Position(Blocker1).PenaltyBox": True, + }) + await asyncio.sleep(0.1) + assert client.get_live_state().team1.blocker1.box_entered_at_ms is not None + + await mock_server.push_update({ + "ScoreBoard.CurrentGame.Team(1).Position(Blocker1).PenaltyBox": False, + }) + await asyncio.sleep(0.1) + state = client.get_live_state() + assert state.team1.blocker1.in_box is False + assert state.team1.blocker1.box_entered_at_ms is None + finally: + client.stop() + await asyncio.sleep(0.05) + + +async def test_box_time_remaining_s_none_when_not_in_box(mock_server): + """box_time_remaining_s is None for skaters not in the penalty box.""" + client, task = await _connected_client(mock_server) + try: + state = client.get_live_state() + assert state.team1.blocker1.box_time_remaining_s is None + finally: + client.stop() + await asyncio.sleep(0.05) + + +async def test_box_time_remaining_s_counts_down(mock_server): + """box_time_remaining_s reflects correct remaining seconds after box entry.""" + client, task = await _connected_client(mock_server) + try: + await mock_server.push_update({ + "ScoreBoard.CurrentGame.Team(1).Position(Jammer).PenaltyBox": True, + }) + await asyncio.sleep(0.1) + state = client.get_live_state() + remaining = state.team1.jammer.box_time_remaining_s + assert remaining is not None + # Should be at most 30 and greater than 0 (entered less than a second ago) + assert 0 < remaining <= 30 + finally: + client.stop() + await asyncio.sleep(0.05) + + +async def test_box_time_remaining_s_zero_when_expired(mock_server): + """box_time_remaining_s is 0 when the 30-second window has already passed.""" + from client import PENALTY_BOX_DURATION_S + client, task = await _connected_client(mock_server) + try: + past_time = time.time() - (PENALTY_BOX_DURATION_S + 5) + past_mono = time.monotonic() - (PENALTY_BOX_DURATION_S + 5) + with patch("client.time") as mock_time: + mock_time.time.return_value = past_time + mock_time.monotonic.return_value = past_mono + await mock_server.push_update({ + "ScoreBoard.CurrentGame.Team(2).Position(Pivot).PenaltyBox": True, + }) + await asyncio.sleep(0.1) + state = client.get_live_state() + assert state.team2.pivot.box_time_remaining_s == 0 + finally: + client.stop() + await asyncio.sleep(0.05) + + +async def test_box_entered_at_ms_not_reset_on_repeated_true(mock_server): + """box_entered_at_ms is not updated if in_box is already True (no re-entry).""" + client, task = await _connected_client(mock_server) + try: + await mock_server.push_update({ + "ScoreBoard.CurrentGame.Team(1).Position(Blocker1).PenaltyBox": True, + }) + await asyncio.sleep(0.1) + first_ts = client.get_live_state().team1.blocker1.box_entered_at_ms + + # Another PenaltyBox=True (e.g. scoreboard resending state) + await mock_server.push_update({ + "ScoreBoard.CurrentGame.Team(1).Position(Blocker1).PenaltyBox": True, + }) + await asyncio.sleep(0.1) + second_ts = client.get_live_state().team1.blocker1.box_entered_at_ms + + assert first_ts == second_ts + finally: + client.stop() + await asyncio.sleep(0.05) + + async def test_state_update_reflects_new_score(mock_server): client, task = await _connected_client(mock_server) try: