host.match_monitor.ds_log_reader

  1"""
  2Locate, copy, and parse FRC Driver Station log files.
  3
  4DS Log files are stored in C:\\Users\\Public\\Documents\\FRC\\Log Files\\ with names:
  5  2024_01_15_15_23_45 FRC.dslog     — 50Hz binary data
  6  2024_01_15_15_23_45 FRC.dsevents  — text event log
  7
  8dsevents format (tab-separated fields per line):
  9  <seconds_from_log_start>\\t<message>\\t[detail]
 10
 11dslog binary format (v3, 50Hz):
 12  Header (16 bytes): version(4) + start_time_ms(8) + reserved(4)
 13  Each record (21 bytes):
 14    uint8  packet_loss_pct
 15    uint8  trip_time_ms
 16    uint16 battery_voltage  (big-endian, volts * 256.0)
 17    uint8  robot_voltage    (volts * 10.2, approx)
 18    uint8  status_flags     (bit0=estop, bit1=disabled, bit2=auto, bit3=enabled,
 19                             bit4=robot_code, bit5=ds_link, bit6=fms_link)
 20    uint8  can_utilization_pct
 21    uint8  signal_quality_pct
 22    uint8  robot_cpu_pct
 23    uint8  ds_cpu_pct
 24    uint8  ram_pct
 25    uint16 bandwidth_kbps   (big-endian)
 26    uint8  pdp_channel_0_current (x 8 ... varies by record type)
 27    -- remaining bytes vary --
 28"""
 29
 30import logging
 31import re
 32import shutil
 33import struct
 34from dataclasses import dataclass, field
 35from datetime import datetime, timedelta
 36from pathlib import Path
 37from typing import List, Optional
 38
 39logger = logging.getLogger("match_monitor")
 40
 41# Default DS log directory (Windows Driver Station)
 42DS_LOG_DIR = Path(r"C:\Users\Public\Documents\FRC\Log Files")
 43
 44# Filename timestamp pattern: "2024_01_15_15_23_45 FRC"
 45_DS_FNAME_RE = re.compile(
 46    r"(\d{4})_(\d{2})_(\d{2})_(\d{2})_(\d{2})_(\d{2})\s+FRC\.(dslog|dsevents)$",
 47    re.IGNORECASE,
 48)
 49
 50# How far before/after the match window to look for a DS log file (seconds)
 51_MATCH_WINDOW_SLOP = 300  # 5 minutes
 52
 53
 54@dataclass
 55class DsEvent:
 56    """A single entry from a .dsevents file."""
 57    offset_sec: float     # Seconds from log start
 58    message: str
 59    detail: str = ""
 60
 61
 62@dataclass
 63class DsRecord:
 64    """One 50Hz record from a .dslog file."""
 65    offset_sec: float
 66    packet_loss_pct: int
 67    trip_time_ms: int
 68    battery_voltage: float
 69    signal_quality_pct: int
 70    robot_cpu_pct: int
 71    ds_cpu_pct: int
 72    ram_pct: int
 73    can_utilization_pct: int
 74    bandwidth_kbps: int
 75    enabled: bool
 76    autonomous: bool
 77    fms_link: bool
 78    ds_link: bool
 79
 80
 81@dataclass
 82class DsLogData:
 83    """Parsed content from DS log files for one match."""
 84    log_start_wall: Optional[datetime] = None   # Wall clock of log file start
 85    events: List[DsEvent] = field(default_factory=list)
 86    records: List[DsRecord] = field(default_factory=list)
 87    copied_files: List[str] = field(default_factory=list)
 88
 89    # Derived summary (filled by summarize())
 90    avg_packet_loss_pct: Optional[float] = None
 91    max_trip_time_ms: Optional[int] = None
 92    avg_trip_time_ms: Optional[float] = None
 93    min_battery_v: Optional[float] = None
 94    avg_battery_v: Optional[float] = None
 95    peak_ds_cpu_pct: Optional[int] = None
 96    peak_robot_cpu_pct: Optional[int] = None
 97    peak_ram_pct: Optional[int] = None
 98    peak_bandwidth_kbps: Optional[int] = None
 99
100    def summarize(self) -> None:
101        """Compute derived summary fields from raw records."""
102        if not self.records:
103            return
104        self.avg_packet_loss_pct = round(
105            sum(r.packet_loss_pct for r in self.records) / len(self.records), 1)
106        self.max_trip_time_ms = max(r.trip_time_ms for r in self.records)
107        self.avg_trip_time_ms = round(
108            sum(r.trip_time_ms for r in self.records) / len(self.records), 1)
109        voltages = [r.battery_voltage for r in self.records if r.battery_voltage > 0]
110        if voltages:
111            self.min_battery_v = round(min(voltages), 3)
112            self.avg_battery_v = round(sum(voltages) / len(voltages), 3)
113        self.peak_ds_cpu_pct = max(r.ds_cpu_pct for r in self.records)
114        self.peak_robot_cpu_pct = max(r.robot_cpu_pct for r in self.records)
115        self.peak_ram_pct = max(r.ram_pct for r in self.records)
116        self.peak_bandwidth_kbps = max(r.bandwidth_kbps for r in self.records)
117
118    def to_dict(self) -> dict:
119        d = {
120            'log_start_wall': self.log_start_wall.isoformat() if self.log_start_wall else None,
121            'copied_files': self.copied_files,
122            'event_count': len(self.events),
123            'record_count': len(self.records),
124            'events': [
125                {'offset_sec': e.offset_sec, 'message': e.message, 'detail': e.detail}
126                for e in self.events
127            ],
128        }
129        if self.records:
130            d['summary'] = {
131                'avg_packet_loss_pct': self.avg_packet_loss_pct,
132                'avg_trip_time_ms': self.avg_trip_time_ms,
133                'max_trip_time_ms': self.max_trip_time_ms,
134                'min_battery_v': self.min_battery_v,
135                'avg_battery_v': self.avg_battery_v,
136                'peak_robot_cpu_pct': self.peak_robot_cpu_pct,
137                'peak_ds_cpu_pct': self.peak_ds_cpu_pct,
138                'peak_ram_pct': self.peak_ram_pct,
139                'peak_bandwidth_kbps': self.peak_bandwidth_kbps,
140            }
141        return d
142
143
144def _parse_filename_time(path: Path) -> Optional[datetime]:
145    """Extract datetime from a DS log filename, or None if unrecognized."""
146    m = _DS_FNAME_RE.match(path.name)
147    if not m:
148        return None
149    year, month, day, hour, minute, second = (int(m.group(i)) for i in range(1, 7))
150    try:
151        return datetime(year, month, day, hour, minute, second)
152    except ValueError:
153        return None
154
155
156def find_match_logs(match_time: datetime,
157                    ds_log_dir: Path = DS_LOG_DIR) -> List[Path]:
158    """
159    Find DS log files whose filename timestamp is within MATCH_WINDOW_SLOP
160    seconds of match_time.  Returns a list of .dslog / .dsevents paths.
161    """
162    if not ds_log_dir.is_dir():
163        return []
164
165    window = timedelta(seconds=_MATCH_WINDOW_SLOP)
166    results = []
167    for path in ds_log_dir.iterdir():
168        if path.suffix.lower() not in ('.dslog', '.dsevents'):
169            continue
170        file_time = _parse_filename_time(path)
171        if file_time is None:
172            continue
173        if abs(file_time - match_time) <= window:
174            results.append(path)
175
176    results.sort()
177    return results
178
179
180def copy_ds_logs(paths: List[Path], dest_dir: Path) -> List[str]:
181    """Copy DS log files into dest_dir. Returns list of copied filenames."""
182    copied = []
183    for src in paths:
184        dest = dest_dir / src.name
185        if dest.exists():
186            copied.append(src.name)
187            continue
188        try:
189            shutil.copy2(src, dest)
190            copied.append(src.name)
191            logger.info(f"Copied DS log: {src.name}")
192        except OSError:
193            logger.warning(f"Could not copy DS log: {src.name}")
194    return copied
195
196
197def parse_dsevents(path: Path) -> tuple[Optional[datetime], List[DsEvent]]:
198    """
199    Parse a .dsevents file.  Returns (log_start_wall, [DsEvent, ...]).
200
201    The file contains tab-separated lines:
202        <float_seconds>\\t<message>[\\t<detail>]
203    The log start wall time is inferred from the filename.
204    """
205    log_start = _parse_filename_time(path)
206    events = []
207    try:
208        for line in path.read_text(encoding='utf-8', errors='replace').splitlines():
209            line = line.strip()
210            if not line:
211                continue
212            parts = line.split('\t')
213            if len(parts) < 2:
214                continue
215            try:
216                offset = float(parts[0])
217            except ValueError:
218                continue
219            message = parts[1].strip()
220            detail = parts[2].strip() if len(parts) > 2 else ""
221            events.append(DsEvent(offset_sec=round(offset, 3),
222                                  message=message, detail=detail))
223    except OSError:
224        logger.warning(f"Could not read DS events file: {path.name}")
225    return log_start, events
226
227
228# dslog record: packet_loss(B) trip_time(B) battery_v(H) robot_v(B)
229# status(B) can(B) signal(B) robot_cpu(B) ds_cpu(B) ram(B) bandwidth(H)
230# = 13 bytes; remaining bytes in record vary (PDP dumps etc.)
231_RECORD_FMT = struct.Struct('>BBHBBBBBBBH')       # 12 bytes
232_RECORD_SIZE = 21  # total record size including PDP/trailing bytes
233
234
235def parse_dslog(path: Path) -> tuple[Optional[datetime], List[DsRecord]]:
236    """
237    Parse a .dslog binary file.  Returns (log_start_wall, [DsRecord, ...]).
238
239    Note: The binary format is community-documented and may vary slightly
240    between DS versions.  Fields beyond the core 12 bytes are skipped.
241    """
242    log_start = _parse_filename_time(path)
243    records = []
244    try:
245        data = path.read_bytes()
246    except OSError:
247        logger.warning(f"Could not read DS log file: {path.name}")
248        return log_start, records
249
250    if len(data) < 16:
251        return log_start, records
252
253    # Version check
254    version = struct.unpack_from('>I', data, 0)[0]
255    if version != 3:
256        logger.warning(f"Unsupported dslog version {version} in {path.name}")
257        return log_start, records
258
259    offset = 16  # skip header
260    record_index = 0
261    while offset + _RECORD_FMT.size <= len(data):
262        try:
263            (packet_loss, trip_time, batt_raw, robot_v_raw,
264             status, can_pct, signal_pct, robot_cpu, ds_cpu,
265             ram_pct, bw_kbps) = _RECORD_FMT.unpack_from(data, offset)
266
267            battery_v = batt_raw / 256.0
268            enabled = bool(status & 0x08)
269            autonomous = bool(status & 0x04)
270            ds_link = bool(status & 0x20)
271            fms_link = bool(status & 0x40)
272
273            records.append(DsRecord(
274                offset_sec=round(record_index / 50.0, 3),
275                packet_loss_pct=min(packet_loss, 100),
276                trip_time_ms=trip_time,
277                battery_voltage=round(battery_v, 3),
278                signal_quality_pct=signal_pct,
279                robot_cpu_pct=robot_cpu,
280                ds_cpu_pct=ds_cpu,
281                ram_pct=ram_pct,
282                can_utilization_pct=can_pct,
283                bandwidth_kbps=bw_kbps,
284                enabled=enabled,
285                autonomous=autonomous,
286                fms_link=fms_link,
287                ds_link=ds_link,
288            ))
289        except struct.error:
290            break
291
292        offset += _RECORD_SIZE
293        record_index += 1
294
295    logger.info(f"Parsed {len(records)} records from {path.name}")
296    return log_start, records
297
298
299def collect_for_match(match_time: datetime, dest_dir: Path,
300                      ds_log_dir: Path = DS_LOG_DIR) -> DsLogData:
301    """
302    Find DS logs matching match_time, copy them to dest_dir, parse them,
303    and return a populated DsLogData.
304    """
305    result = DsLogData()
306
307    paths = find_match_logs(match_time, ds_log_dir)
308    if not paths:
309        logger.info("No matching DS log files found")
310        return result
311
312    result.copied_files = copy_ds_logs(paths, dest_dir)
313
314    for path in paths:
315        suffix = path.suffix.lower()
316        local = dest_dir / path.name   # work from the copy
317
318        if suffix == '.dsevents':
319            wall_time, events = parse_dsevents(local)
320            if wall_time and result.log_start_wall is None:
321                result.log_start_wall = wall_time
322            result.events.extend(events)
323
324        elif suffix == '.dslog':
325            wall_time, records = parse_dslog(local)
326            if wall_time and result.log_start_wall is None:
327                result.log_start_wall = wall_time
328            result.records.extend(records)
329
330    result.summarize()
331    return result