host.match_monitor.callbacks

  1"""Callback registry for post-match processing."""
  2
  3import base64
  4import json
  5import logging
  6import webbrowser
  7from abc import ABC, abstractmethod
  8from datetime import datetime
  9from pathlib import Path
 10from typing import List, Optional, TYPE_CHECKING
 11
 12from .analyzer import MatchStats
 13
 14if TYPE_CHECKING:
 15    from .match_data_client import MatchResult, EventStats
 16
 17logger = logging.getLogger("match_monitor")
 18
 19
 20class MatchCompleteCallback(ABC):
 21    """Base class for end-of-match processing callbacks."""
 22
 23    @abstractmethod
 24    def on_match_complete(self, match_dir: Path, metadata: dict,
 25                          stats: MatchStats,
 26                          match_result: Optional['MatchResult'] = None,
 27                          event_stats: Optional['EventStats'] = None) -> None:
 28        """Called after match log analysis completes.
 29
 30        Args:
 31            match_dir: Directory containing the match log files.
 32            metadata: Dict with event_name, match_type, match_number.
 33            stats: Extracted match statistics.
 34            match_result: Official match result from TBA/FRC Events API, or None.
 35            event_stats: OPR/rankings data for our team, or None.
 36        """
 37
 38
 39class CallbackRegistry:
 40    """Manages and runs end-of-match callbacks."""
 41
 42    def __init__(self) -> None:
 43        self._callbacks: List[MatchCompleteCallback] = []
 44
 45    def register(self, callback: MatchCompleteCallback) -> None:
 46        self._callbacks.append(callback)
 47        logger.info(f"Registered callback: {type(callback).__name__}")
 48
 49    def unregister(self, callback: MatchCompleteCallback) -> None:
 50        try:
 51            self._callbacks.remove(callback)
 52            logger.info(f"Unregistered callback: {type(callback).__name__}")
 53        except ValueError:
 54            pass
 55
 56    def run_all(self, match_dir: Path, metadata: dict,
 57                stats: MatchStats,
 58                match_result: Optional['MatchResult'] = None,
 59                event_stats: Optional['EventStats'] = None) -> None:
 60        for cb in self._callbacks:
 61            try:
 62                cb.on_match_complete(match_dir, metadata, stats,
 63                                     match_result, event_stats)
 64            except Exception:
 65                logger.exception(
 66                    f"Callback {type(cb).__name__} failed"
 67                )
 68
 69
 70class JsonSummaryCallback(MatchCompleteCallback):
 71    """Saves match_summary.json in the match directory."""
 72
 73    def on_match_complete(self, match_dir: Path, metadata: dict,
 74                          stats: MatchStats,
 75                          match_result: Optional['MatchResult'] = None,
 76                          event_stats: Optional['EventStats'] = None) -> None:
 77        summary = {
 78            'event_name': metadata.get('event_name', ''),
 79            'match_type': metadata.get('match_type', ''),
 80            'match_number': metadata.get('match_number', ''),
 81            'analyzed_at': datetime.now().isoformat(),
 82        }
 83        summary.update(stats.to_dict())
 84
 85        if match_result is not None:
 86            summary['official_match'] = match_result.to_dict()
 87        if event_stats is not None:
 88            summary['event_stats'] = event_stats.to_dict()
 89
 90        out_path = match_dir / 'match_summary.json'
 91        out_path.write_text(json.dumps(summary, indent=2) + '\n')
 92        logger.info(f"Saved {out_path}")
 93
 94
 95class HtmlPreviewCallback(MatchCompleteCallback):
 96    """Generates a local HTML preview page and opens it in the browser."""
 97
 98    def on_match_complete(self, match_dir: Path, metadata: dict,
 99                          stats: MatchStats,
100                          match_result: Optional['MatchResult'] = None,
101                          event_stats: Optional['EventStats'] = None) -> None:
102        # Generate voltage chart
103        chart_b64 = ''
104        if stats.voltage_timeseries:
105            try:
106                from .chart_generator import voltage_chart
107                chart_png = voltage_chart(
108                    stats.voltage_timeseries,
109                    brownout_times=stats.brownout_timestamps_sec,
110                    mode_transitions=stats.mode_transitions,
111                    match_duration=stats.match_duration_seconds,
112                )
113                if chart_png:
114                    chart_b64 = base64.b64encode(chart_png).decode('ascii')
115            except Exception:
116                logger.exception("Chart generation failed for HTML preview")
117
118        match_type = metadata.get('match_type', 'Match') or 'Match'
119        match_number = metadata.get('match_number', '?') or '?'
120        event_name = metadata.get('event_name', '') or ''
121        title = f"{match_type.capitalize()} {match_number}"
122        if event_name:
123            title = f"[{event_name.upper()}] {title}"
124
125        # Build stats rows
126        rows = []
127        if stats.avg_voltage is not None:
128            rows.append(('Battery Voltage',
129                         f"Start {stats.start_voltage}V · "
130                         f"Avg {stats.avg_voltage}V · "
131                         f"Min {stats.min_voltage}V · "
132                         f"Max {stats.max_voltage}V"))
133        rows.append(('Brownouts', f"{stats.brownout_count}"
134                      + (f" @ {', '.join(f'{t}s' for t in stats.brownout_timestamps_sec[:5])}"
135                         if stats.brownout_count else '')))
136        rows.append(('Disconnects', f"{stats.disconnect_count}"
137                      + (f" @ {', '.join(f'{t}s' for t in stats.disconnect_timestamps_sec[:5])}"
138                         if stats.disconnect_count else '')))
139        if stats.match_duration_seconds is not None:
140            rows.append(('Log Duration', f"{stats.match_duration_seconds:.1f}s"))
141        if stats.rail_voltages:
142            for rail, rv in stats.rail_voltages.items():
143                rows.append((f"Rail {rail}",
144                             f"Avg {rv['avg']}V · "
145                             f"Min {rv['min']}V · "
146                             f"Max {rv['max']}V"))
147        if stats.ds_log and stats.ds_log.avg_packet_loss_pct is not None:
148            rows.append(('Packet Loss', f"{stats.ds_log.avg_packet_loss_pct}%"))
149        if stats.ds_log and stats.ds_log.avg_trip_time_ms is not None:
150            rows.append(('Latency',
151                         f"avg {stats.ds_log.avg_trip_time_ms}ms / "
152                         f"max {stats.ds_log.max_trip_time_ms}ms"))
153
154        stats_html = '\n'.join(
155            f'<tr><td><b>{name}</b></td><td>{val}</td></tr>' for name, val in rows
156        )
157        chart_html = (f'<img src="data:image/png;base64,{chart_b64}" '
158                      f'style="max-width:100%">' if chart_b64 else
159                      '<p><i>No chart data</i></p>')
160
161        html = f"""<!DOCTYPE html>
162<html><head>
163<meta charset="utf-8">
164<title>{title} - Match Preview</title>
165<style>
166  body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
167         max-width: 900px; margin: 40px auto; padding: 0 20px;
168         background: #1a1a2e; color: #e0e0e0; }}
169  h1 {{ color: #3498db; }}
170  table {{ border-collapse: collapse; width: 100%; margin: 20px 0; }}
171  td {{ padding: 8px 12px; border-bottom: 1px solid #333; }}
172  tr:hover {{ background: #252545; }}
173  .chart {{ margin: 20px 0; text-align: center; }}
174  .timestamp {{ color: #666; font-size: 0.85em; }}
175</style>
176</head><body>
177<h1>{title}</h1>
178<p class="timestamp">Analyzed {datetime.now():%Y-%m-%d %H:%M:%S}</p>
179<div class="chart">{chart_html}</div>
180<h2>Match Statistics</h2>
181<table>{stats_html}</table>
182</body></html>"""
183
184        out_path = match_dir / 'match_preview.html'
185        out_path.write_text(html, encoding='utf-8')
186        logger.info(f"Saved HTML preview: {out_path}")
187        print(f"  Opening preview: {out_path}")
188        webbrowser.open(out_path.as_uri())