host.match_monitor.analyzer

  1"""Parse .wpilog files and extract match statistics."""
  2
  3import logging
  4import mmap
  5import re
  6from dataclasses import dataclass, field, asdict
  7from datetime import datetime, timezone
  8from pathlib import Path
  9from typing import Optional, List, Dict, TYPE_CHECKING
 10
 11if TYPE_CHECKING:
 12    from .ds_log_reader import DsLogData
 13
 14from wpiutil.log import DataLogReader
 15
 16logger = logging.getLogger("match_monitor")
 17
 18# Patterns to match entry names (case-insensitive)
 19VOLTAGE_PATTERNS = [
 20    re.compile(r'battery\s*voltage', re.IGNORECASE),
 21]
 22# Rail voltage patterns — tracked separately as named rails
 23RAIL_VOLTAGE_PATTERNS = {
 24    '3_3v': re.compile(r'rail3v3/voltage', re.IGNORECASE),
 25    '5v': re.compile(r'rail5v/voltage', re.IGNORECASE),
 26    '6v': re.compile(r'rail6v/voltage', re.IGNORECASE),
 27}
 28BROWNOUT_PATTERNS = [
 29    re.compile(r'brownout', re.IGNORECASE),
 30    re.compile(r'browned?\s*out', re.IGNORECASE),
 31]
 32DS_CONNECTED_PATTERNS = [
 33    re.compile(r'DS:ds', re.IGNORECASE),
 34    re.compile(r'ds\s*attached', re.IGNORECASE),
 35    re.compile(r'ds\s*connected', re.IGNORECASE),
 36]
 37MODE_PATTERNS = {
 38    'enabled': re.compile(r'DS:enabled', re.IGNORECASE),
 39    'autonomous': re.compile(r'DS:autonomous', re.IGNORECASE),
 40    'test': re.compile(r'DS:test', re.IGNORECASE),
 41}
 42CONFIG_HASH_PATTERNS = [
 43    re.compile(r'controller.*hash', re.IGNORECASE),
 44    re.compile(r'config.*hash', re.IGNORECASE),
 45]
 46SYSTEM_TIME_PATTERNS = [
 47    re.compile(r'^systemTime$'),
 48    re.compile(r'system.*time', re.IGNORECASE),
 49]
 50
 51
 52@dataclass
 53class MatchStats:
 54    """Data class holding extracted match statistics."""
 55    brownout_count: int = 0
 56    brownout_timestamps_sec: List[float] = field(default_factory=list)
 57    disconnect_count: int = 0
 58    disconnect_timestamps_sec: List[float] = field(default_factory=list)
 59    start_voltage: Optional[float] = None
 60    end_voltage: Optional[float] = None
 61    avg_voltage: Optional[float] = None
 62    min_voltage: Optional[float] = None
 63    max_voltage: Optional[float] = None
 64    voltage_samples: int = 0
 65    match_duration_seconds: Optional[float] = None
 66    controller_config_hash: Optional[str] = None
 67    mode_transitions: List[dict] = field(default_factory=list)
 68    entry_names: List[str] = field(default_factory=list)
 69    # Voltage timeseries: [(time_sec_from_start, voltage), ...]
 70    voltage_timeseries: List[tuple] = field(default_factory=list)
 71    # Rail voltages: {rail_name: {avg, min, max, samples}}
 72    rail_voltages: Dict[str, dict] = field(default_factory=dict)
 73    # Wall clock time of the first wpilog record (derived from systemTime entry)
 74    wpilog_wall_start: Optional[datetime] = field(default=None, compare=False)
 75    # DS log data attached after collection (not part of wpilog analysis)
 76    ds_log: Optional['DsLogData'] = field(default=None, compare=False)
 77
 78    def to_dict(self) -> dict:
 79        """Convert to a JSON-serializable dictionary."""
 80        d = asdict(self)
 81        # Remove non-serializable / large fields handled separately
 82        d.pop('wpilog_wall_start', None)
 83        d.pop('ds_log', None)
 84        d.pop('voltage_timeseries', None)
 85        # Group voltage fields
 86        d['voltage'] = {
 87            'start': d.pop('start_voltage'),
 88            'end': d.pop('end_voltage'),
 89            'average': d.pop('avg_voltage'),
 90            'min': d.pop('min_voltage'),
 91            'max': d.pop('max_voltage'),
 92            'samples': d.pop('voltage_samples'),
 93        }
 94        if self.wpilog_wall_start:
 95            d['wpilog_wall_start'] = self.wpilog_wall_start.isoformat()
 96        if self.ds_log is not None:
 97            d['ds_log'] = self.ds_log.to_dict()
 98        return d
 99
100
101def _match_any(name: str, patterns: list) -> bool:
102    return any(p.search(name) for p in patterns)
103
104
105class WpilogAnalyzer:
106    """Parses .wpilog files to extract match statistics."""
107
108    def analyze(self, wpilog_path: Path) -> MatchStats:
109        """Analyze a single .wpilog file and return match statistics."""
110        stats = MatchStats()
111
112        with open(wpilog_path, 'rb') as f:
113            mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
114            reader = DataLogReader(mm)
115
116            if not reader.isValid():
117                logger.warning(f"Invalid wpilog file: {wpilog_path.name}")
118                return stats
119
120            # First pass: discover entry names and their IDs
121            entries: Dict[int, dict] = {}  # entry_id -> {name, type}
122            voltage_ids = set()
123            brownout_ids = set()
124            ds_connected_ids = set()
125            mode_ids: Dict[str, set] = {k: set() for k in MODE_PATTERNS}
126            config_hash_ids = set()
127            system_time_ids = set()
128            # Rail voltage entry IDs: rail_name -> set of entry IDs
129            rail_voltage_ids: Dict[str, set] = {k: set() for k in RAIL_VOLTAGE_PATTERNS}
130
131            for record in reader:
132                if record.isStart():
133                    data = record.getStartData()
134                    entries[data.entry] = {
135                        'name': data.name, 'type': data.type
136                    }
137                    stats.entry_names.append(data.name)
138
139                    if _match_any(data.name, VOLTAGE_PATTERNS):
140                        voltage_ids.add(data.entry)
141                    if _match_any(data.name, BROWNOUT_PATTERNS):
142                        brownout_ids.add(data.entry)
143                    if _match_any(data.name, DS_CONNECTED_PATTERNS):
144                        ds_connected_ids.add(data.entry)
145                    for mode_name, pattern in MODE_PATTERNS.items():
146                        if pattern.search(data.name):
147                            mode_ids[mode_name].add(data.entry)
148                    if _match_any(data.name, CONFIG_HASH_PATTERNS):
149                        config_hash_ids.add(data.entry)
150                    if _match_any(data.name, SYSTEM_TIME_PATTERNS):
151                        system_time_ids.add(data.entry)
152                    for rail_name, pattern in RAIL_VOLTAGE_PATTERNS.items():
153                        if pattern.search(data.name):
154                            rail_voltage_ids[rail_name].add(data.entry)
155
156            # Log discovered entries
157            logger.info(f"Analyzing {wpilog_path.name}: "
158                        f"{len(entries)} entries, "
159                        f"voltage={len(voltage_ids)}, "
160                        f"brownout={len(brownout_ids)}, "
161                        f"ds_connected={len(ds_connected_ids)}")
162
163            # Second pass: extract data
164            voltage_sum = 0.0
165            voltage_count = 0
166            first_timestamp = None
167            last_timestamp = None
168            prev_brownout = False
169            prev_ds_connected = True
170            mode_state: Dict[str, bool] = {}
171            # Rail voltage accumulators: {rail: [sum, count, min, max]}
172            rail_accum: Dict[str, list] = {k: [0.0, 0, None, None] for k in RAIL_VOLTAGE_PATTERNS}
173
174            mm.seek(0)
175            reader2 = DataLogReader(mm)
176
177            for record in reader2:
178                if record.isControl():
179                    continue
180
181                entry_info = entries.get(record.getEntry())
182                if entry_info is None:
183                    continue
184
185                ts_sec = record.getTimestamp() / 1_000_000.0
186                if first_timestamp is None:
187                    first_timestamp = ts_sec
188                last_timestamp = ts_sec
189
190                try:
191                    # Voltage (skip 0.0 — default NT value before first real reading)
192                    if record.getEntry() in voltage_ids:
193                        v = self._get_numeric(record, entry_info['type'])
194                        if v is not None and v > 0.0:
195                            if stats.start_voltage is None:
196                                stats.start_voltage = v
197                            stats.end_voltage = v
198                            voltage_sum += v
199                            voltage_count += 1
200                            if stats.min_voltage is None or v < stats.min_voltage:
201                                stats.min_voltage = v
202                            if stats.max_voltage is None or v > stats.max_voltage:
203                                stats.max_voltage = v
204                            stats.voltage_timeseries.append((
205                                round(ts_sec - (first_timestamp or 0), 3), v
206                            ))
207
208                    # Brownout
209                    if record.getEntry() in brownout_ids:
210                        val = record.getBoolean()
211                        if val and not prev_brownout:
212                            stats.brownout_count += 1
213                            stats.brownout_timestamps_sec.append(
214                                round(ts_sec - (first_timestamp or 0), 3)
215                            )
216                        prev_brownout = val
217
218                    # DS Connected
219                    if record.getEntry() in ds_connected_ids:
220                        val = record.getBoolean()
221                        if not val and prev_ds_connected:
222                            stats.disconnect_count += 1
223                            stats.disconnect_timestamps_sec.append(
224                                round(ts_sec - (first_timestamp or 0), 3)
225                            )
226                        prev_ds_connected = val
227
228                    # Mode transitions (record both True and False)
229                    for mode_name, ids in mode_ids.items():
230                        if record.getEntry() in ids:
231                            val = record.getBoolean()
232                            prev = mode_state.get(mode_name)
233                            if prev is None or val != prev:
234                                mode_state[mode_name] = val
235                                stats.mode_transitions.append({
236                                    'time_sec': round(
237                                        ts_sec - (first_timestamp or 0), 3
238                                    ),
239                                    'mode': mode_name,
240                                    'value': val,
241                                })
242
243                    # Rail voltages
244                    for rail_name, ids in rail_voltage_ids.items():
245                        if record.getEntry() in ids:
246                            v = self._get_numeric(record, entry_info['type'])
247                            if v is not None:
248                                acc = rail_accum[rail_name]
249                                acc[0] += v
250                                acc[1] += 1
251                                if acc[2] is None or v < acc[2]:
252                                    acc[2] = v
253                                if acc[3] is None or v > acc[3]:
254                                    acc[3] = v
255
256                    # Controller config hash
257                    if record.getEntry() in config_hash_ids:
258                        stats.controller_config_hash = record.getString()
259
260                    # System time — wall clock offset for DS log correlation
261                    if record.getEntry() in system_time_ids and stats.wpilog_wall_start is None:
262                        try:
263                            entry_type = entries[record.getEntry()]['type']
264                            if entry_type == 'int64':
265                                # Microseconds since Unix epoch
266                                wall_us = record.getInteger()
267                                wall_sec = wall_us / 1_000_000.0
268                                stats.wpilog_wall_start = datetime.fromtimestamp(
269                                    wall_sec, tz=timezone.utc)
270                            elif entry_type == 'string':
271                                raw = record.getString()
272                                for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S'):
273                                    try:
274                                        stats.wpilog_wall_start = datetime.strptime(
275                                            raw[:19], fmt).replace(tzinfo=timezone.utc)
276                                        break
277                                    except ValueError:
278                                        pass  # try next format
279                        except Exception as e:
280                            logger.debug("Could not parse system time entry: %s", e)
281
282                except Exception as e:
283                    logger.debug("Skipping malformed record: %s", e)
284
285            # Finalize
286            if voltage_count > 0:
287                stats.avg_voltage = round(voltage_sum / voltage_count, 3)
288                stats.voltage_samples = voltage_count
289                if stats.start_voltage is not None:
290                    stats.start_voltage = round(stats.start_voltage, 3)
291                if stats.end_voltage is not None:
292                    stats.end_voltage = round(stats.end_voltage, 3)
293                if stats.min_voltage is not None:
294                    stats.min_voltage = round(stats.min_voltage, 3)
295                if stats.max_voltage is not None:
296                    stats.max_voltage = round(stats.max_voltage, 3)
297
298            if first_timestamp is not None and last_timestamp is not None:
299                stats.match_duration_seconds = round(
300                    last_timestamp - first_timestamp, 3
301                )
302
303            # Finalize rail voltages
304            for rail_name, acc in rail_accum.items():
305                if acc[1] > 0:
306                    stats.rail_voltages[rail_name] = {
307                        'avg': round(acc[0] / acc[1], 3),
308                        'min': round(acc[2], 3),
309                        'max': round(acc[3], 3),
310                        'samples': acc[1],
311                    }
312
313            mm.close()
314
315        return stats
316
317    def analyze_directory(self, match_dir: Path) -> MatchStats:
318        """Analyze all .wpilog files in a directory and merge results."""
319        wpilog_files = sorted(match_dir.glob('*.wpilog'))
320        if not wpilog_files:
321            logger.info(f"No .wpilog files found in {match_dir}")
322            return MatchStats()
323
324        all_stats = []
325        for f in wpilog_files:
326            try:
327                s = self.analyze(f)
328                all_stats.append(s)
329            except Exception:
330                logger.exception(f"Failed to analyze {f.name}")
331
332        if not all_stats:
333            return MatchStats()
334        if len(all_stats) == 1:
335            return all_stats[0]
336        return self._merge(all_stats)
337
338    def _merge(self, stats_list: List[MatchStats]) -> MatchStats:
339        """Merge stats from multiple .wpilog files."""
340        merged = MatchStats()
341        total_voltage_sum = 0.0
342        total_voltage_count = 0
343
344        for s in stats_list:
345            merged.brownout_count += s.brownout_count
346            merged.brownout_timestamps_sec.extend(s.brownout_timestamps_sec)
347            merged.disconnect_count += s.disconnect_count
348            merged.disconnect_timestamps_sec.extend(s.disconnect_timestamps_sec)
349            merged.mode_transitions.extend(s.mode_transitions)
350            merged.entry_names.extend(s.entry_names)
351
352            if s.start_voltage is not None and merged.start_voltage is None:
353                merged.start_voltage = s.start_voltage
354            if s.end_voltage is not None:
355                merged.end_voltage = s.end_voltage
356            if s.min_voltage is not None:
357                if merged.min_voltage is None or s.min_voltage < merged.min_voltage:
358                    merged.min_voltage = s.min_voltage
359            if s.max_voltage is not None:
360                if merged.max_voltage is None or s.max_voltage > merged.max_voltage:
361                    merged.max_voltage = s.max_voltage
362            if s.voltage_samples > 0 and s.avg_voltage is not None:
363                total_voltage_sum += s.avg_voltage * s.voltage_samples
364                total_voltage_count += s.voltage_samples
365            if s.match_duration_seconds is not None:
366                if (merged.match_duration_seconds is None
367                        or s.match_duration_seconds > merged.match_duration_seconds):
368                    merged.match_duration_seconds = s.match_duration_seconds
369            if s.controller_config_hash is not None:
370                merged.controller_config_hash = s.controller_config_hash
371
372        if total_voltage_count > 0:
373            merged.avg_voltage = round(total_voltage_sum / total_voltage_count, 3)
374            merged.voltage_samples = total_voltage_count
375
376        merged.mode_transitions.sort(key=lambda t: t['time_sec'])
377        merged.entry_names = sorted(set(merged.entry_names))
378        return merged
379
380    @staticmethod
381    def _get_numeric(record, type_name: str):
382        if type_name == 'double':
383            return record.getDouble()
384        elif type_name == 'float':
385            return record.getFloat()
386        elif type_name == 'int64':
387            return float(record.getInteger())
388        return None