host.match_monitor.match_data_client

  1"""
  2Fetch official match data from The Blue Alliance (TBA) and FRC Events API.
  3
  4Configuration is read from a JSON file (default: match_monitor_config.json
  5next to the server log directory).  Create it with your API keys:
  6
  7    {
  8        "tba_api_key": "YOUR_TBA_KEY",
  9        "frc_events_username": "YOUR_USERNAME",
 10        "frc_events_auth_key": "YOUR_AUTH_KEY",
 11        "team_number": 3200
 12    }
 13
 14TBA keys: https://www.thebluealliance.com (Account > Read API Keys)
 15FRC Events keys: https://frc-events.firstinspires.org/services/api/register
 16"""
 17
 18import base64
 19import json
 20import logging
 21import urllib.error
 22import urllib.request
 23from dataclasses import dataclass, field
 24from pathlib import Path
 25from typing import Optional
 26
 27logger = logging.getLogger("match_monitor")
 28
 29_TBA_BASE = "https://www.thebluealliance.com/api/v3"
 30_FRC_BASE = "https://frc-api.firstinspires.org/v2.0"
 31_TIMEOUT = 10  # seconds
 32
 33
 34@dataclass
 35class AllianceResult:
 36    teams: list[str] = field(default_factory=list)   # e.g. ["frc3200", "frc1234", "frc5678"]
 37    score: Optional[int] = None
 38    won: bool = False
 39
 40
 41@dataclass
 42class MatchResult:
 43    """Official match result from TBA / FRC Events API."""
 44    match_key: str = ""           # e.g. "2026calas_qm12"
 45    event_key: str = ""           # e.g. "2026calas"
 46    match_number: int = 0
 47    tournament_level: str = ""    # "qm", "sf", "f"
 48    red: AllianceResult = field(default_factory=AllianceResult)
 49    blue: AllianceResult = field(default_factory=AllianceResult)
 50    score_breakdown: dict = field(default_factory=dict)
 51    our_alliance: str = ""        # "red" or "blue"
 52    our_score: Optional[int] = None
 53    opponent_score: Optional[int] = None
 54    result: str = ""              # "win", "loss", "tie", "unknown"
 55    video_url: Optional[str] = None
 56
 57    def to_dict(self) -> dict:
 58        return {
 59            'match_key': self.match_key,
 60            'event_key': self.event_key,
 61            'match_number': self.match_number,
 62            'tournament_level': self.tournament_level,
 63            'red': {'teams': self.red.teams, 'score': self.red.score},
 64            'blue': {'teams': self.blue.teams, 'score': self.blue.score},
 65            'score_breakdown': self.score_breakdown,
 66            'our_alliance': self.our_alliance,
 67            'our_score': self.our_score,
 68            'opponent_score': self.opponent_score,
 69            'result': self.result,
 70            'video_url': self.video_url,
 71        }
 72
 73
 74@dataclass
 75class EventStats:
 76    """OPR/rankings data for the team at this event."""
 77    opr: Optional[float] = None
 78    dpr: Optional[float] = None
 79    ccwm: Optional[float] = None
 80    rank: Optional[int] = None
 81    wins: Optional[int] = None
 82    losses: Optional[int] = None
 83    ties: Optional[int] = None
 84    ranking_points: Optional[float] = None
 85
 86    def to_dict(self) -> dict:
 87        return {k: v for k, v in vars(self).items() if v is not None}
 88
 89
 90def _load_config(config_path: Path) -> dict:
 91    """Load API key config from JSON file.  Returns empty dict if missing."""
 92    if config_path.exists():
 93        try:
 94            return json.loads(config_path.read_text())
 95        except Exception:
 96            logger.warning(f"Failed to read match data config: {config_path}")
 97    return {}
 98
 99
100def _tba_get(endpoint: str, api_key: str) -> Optional[dict | list]:
101    """Make a GET request to the TBA API.  Returns parsed JSON or None."""
102    url = f"{_TBA_BASE}/{endpoint.lstrip('/')}"
103    req = urllib.request.Request(url, headers={
104        'X-TBA-Auth-Key': api_key,
105        'User-Agent': 'FRC3200-MatchMonitor/1.0',
106    })
107    try:
108        with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
109            return json.loads(resp.read().decode())
110    except urllib.error.HTTPError as e:
111        logger.warning(f"TBA API error {e.code} for {endpoint}")
112    except urllib.error.URLError as e:
113        # Connection failure (no internet, DNS, timeout) — expected offline
114        logger.debug(f"TBA API unreachable ({e.reason})")
115    except Exception as e:
116        logger.debug(f"TBA API request failed for {endpoint}: {e}")
117    return None
118
119
120def _frc_get(endpoint: str, username: str, auth_key: str) -> Optional[dict | list]:
121    """Make a GET request to the FRC Events API.  Returns parsed JSON or None."""
122    url = f"{_FRC_BASE}/{endpoint.lstrip('/')}"
123    token = base64.b64encode(f"{username}:{auth_key}".encode()).decode()
124    req = urllib.request.Request(url, headers={
125        'Authorization': f'Basic {token}',
126        'Accept': 'application/json',
127    })
128    try:
129        with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
130            return json.loads(resp.read().decode())
131    except urllib.error.HTTPError as e:
132        logger.warning(f"FRC Events API error {e.code} for {endpoint}")
133    except urllib.error.URLError as e:
134        # Connection failure (no internet, DNS, timeout) — expected offline
135        logger.debug(f"FRC Events API unreachable ({e.reason})")
136    except Exception as e:
137        logger.debug(f"FRC Events API request failed for {endpoint}: {e}")
138    return None
139
140
141def _alliance_result_from_tba(raw: dict) -> AllianceResult:
142    return AllianceResult(
143        teams=raw.get('team_keys', []),
144        score=raw.get('score'),
145    )
146
147
148class MatchDataClient:
149    """Fetches match data from TBA and FRC Events API for a given team/event."""
150
151    def __init__(self, config_path: Path) -> None:
152        cfg = _load_config(config_path)
153        self._tba_key: str = cfg.get('tba_api_key', '')
154        self._frc_user: str = cfg.get('frc_events_username', '')
155        self._frc_auth: str = cfg.get('frc_events_auth_key', '')
156        self._team_num: int = int(cfg.get('team_number', 3200))
157        self._team_key = f"frc{self._team_num}"
158
159    @property
160    def configured(self) -> bool:
161        return bool(self._tba_key or (self._frc_user and self._frc_auth))
162
163    def fetch_match(self, event_key: str, match_number: int,
164                    level: str = 'qm') -> Optional[MatchResult]:
165        """
166        Fetch a single match result.
167        level: 'qm' (qualification), 'sf' (semifinal), 'f' (final)
168        """
169        match_key = f"{event_key}_{level}{match_number}"
170
171        if self._tba_key:
172            result = self._fetch_match_tba(match_key)
173            if result:
174                return result
175
176        if self._frc_user and self._frc_auth:
177            return self._fetch_match_frc(event_key, match_number, level)
178
179        return None
180
181    def _fetch_match_tba(self, match_key: str) -> Optional[MatchResult]:
182        data = _tba_get(f"/match/{match_key}", self._tba_key)
183        if not data or not isinstance(data, dict):
184            return None
185
186        alliances = data.get('alliances', {})
187        red = _alliance_result_from_tba(alliances.get('red', {}))
188        blue = _alliance_result_from_tba(alliances.get('blue', {}))
189
190        # Determine winner
191        winning_alliance = data.get('winning_alliance', '')
192        if winning_alliance == 'red':
193            red.won = True
194        elif winning_alliance == 'blue':
195            blue.won = True
196
197        # Find our alliance
198        our_alliance = ''
199        if self._team_key in red.teams:
200            our_alliance = 'red'
201        elif self._team_key in blue.teams:
202            our_alliance = 'blue'
203
204        our_score = (red.score if our_alliance == 'red' else
205                     blue.score if our_alliance == 'blue' else None)
206        opp_score = (blue.score if our_alliance == 'red' else
207                     red.score if our_alliance == 'blue' else None)
208
209        result_str = 'unknown'
210        if our_alliance:
211            if winning_alliance == our_alliance:
212                result_str = 'win'
213            elif winning_alliance and winning_alliance != our_alliance:
214                result_str = 'loss'
215            elif winning_alliance == '':
216                result_str = 'tie'
217
218        # Video URL (first available)
219        video_url = None
220        for vid in data.get('videos', []):
221            if vid.get('type') == 'youtube':
222                video_url = f"https://www.youtube.com/watch?v={vid['key']}"
223                break
224
225        # Score breakdown (game-specific)
226        breakdown = data.get('score_breakdown') or {}
227
228        return MatchResult(
229            match_key=data.get('key', ''),
230            event_key=data.get('event_key', ''),
231            match_number=data.get('match_number', 0),
232            tournament_level=data.get('comp_level', ''),
233            red=red,
234            blue=blue,
235            score_breakdown=breakdown,
236            our_alliance=our_alliance,
237            our_score=our_score,
238            opponent_score=opp_score,
239            result=result_str,
240            video_url=video_url,
241        )
242
243    def _fetch_match_frc(self, event_key: str, match_number: int,
244                          level: str) -> Optional[MatchResult]:
245        year = event_key[:4]
246        event_code = event_key[4:]
247        level_map = {'qm': 'Qualification', 'sf': 'Playoff', 'f': 'Playoff'}
248        tournament_level = level_map.get(level, 'Qualification')
249        data = _frc_get(
250            f"/{year}/matches/{event_code}?matchNumber={match_number}"
251            f"&tournamentLevel={tournament_level}",
252            self._frc_user, self._frc_auth,
253        )
254        if not data or not isinstance(data, dict):
255            return None
256
257        matches = data.get('Matches', [])
258        if not matches:
259            return None
260
261        m = matches[0]
262        red_teams = [f"frc{t['teamNumber']}" for t in m.get('teams', [])
263                     if t.get('station', '').startswith('Red')]
264        blue_teams = [f"frc{t['teamNumber']}" for t in m.get('teams', [])
265                      if t.get('station', '').startswith('Blue')]
266        red_score = m.get('scoreRedFinal')
267        blue_score = m.get('scoreBlueFinal')
268
269        our_alliance = ''
270        if self._team_key in red_teams:
271            our_alliance = 'red'
272        elif self._team_key in blue_teams:
273            our_alliance = 'blue'
274
275        our_score = red_score if our_alliance == 'red' else blue_score if our_alliance == 'blue' else None
276        opp_score = blue_score if our_alliance == 'red' else red_score if our_alliance == 'blue' else None
277
278        result_str = 'unknown'
279        if our_score is not None and opp_score is not None:
280            if our_score > opp_score:
281                result_str = 'win'
282            elif our_score < opp_score:
283                result_str = 'loss'
284            else:
285                result_str = 'tie'
286
287        return MatchResult(
288            match_key=f"{event_key}_{level}{match_number}",
289            event_key=event_key,
290            match_number=match_number,
291            tournament_level=level,
292            red=AllianceResult(teams=red_teams, score=red_score),
293            blue=AllianceResult(teams=blue_teams, score=blue_score),
294            our_alliance=our_alliance,
295            our_score=our_score,
296            opponent_score=opp_score,
297            result=result_str,
298        )
299
300    def fetch_event_stats(self, event_key: str) -> Optional[EventStats]:
301        """Fetch OPR, DPR, CCWM, and ranking for our team at this event."""
302        if not self._tba_key:
303            return None
304
305        stats = EventStats()
306
307        # OPR/DPR/CCWM
308        oprs = _tba_get(f"/event/{event_key}/oprs", self._tba_key)
309        if oprs and isinstance(oprs, dict):
310            stats.opr = oprs.get('oprs', {}).get(self._team_key)
311            stats.dpr = oprs.get('dprs', {}).get(self._team_key)
312            stats.ccwm = oprs.get('ccwms', {}).get(self._team_key)
313
314        # Rankings
315        rankings = _tba_get(f"/event/{event_key}/rankings", self._tba_key)
316        if rankings and isinstance(rankings, dict):
317            for entry in rankings.get('rankings', []):
318                if entry.get('team_key') == self._team_key:
319                    stats.rank = entry.get('rank')
320                    record = entry.get('record', {})
321                    stats.wins = record.get('wins')
322                    stats.losses = record.get('losses')
323                    stats.ties = record.get('ties')
324                    stats.ranking_points = entry.get('extra_stats', [None])[0]
325                    break
326
327        return stats if any(v is not None for v in vars(stats).values()) else None