Feat/player tracker

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: "<team_n>.<crg_pos>" -> epoch ms
+        self._box_entry_mono: Dict[str, float] = {}   # key: "<team_n>.<crg_pos>" -> monotonic s
+        self._was_in_box: Dict[str, bool] = {}        # key: "<team_n>.<crg_pos>" -> last known in_box
 
     @property
     def connected(self) -> bool:
@@ -152,21 +157,54 @@ 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})."
         flat_fields = {
             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:

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):

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: