host.match_monitor.chart_generator

  1"""Generate match analysis charts as PNG images."""
  2
  3import io
  4import logging
  5from typing import List, Tuple, Dict
  6
  7logger = logging.getLogger("match_monitor")
  8
  9try:
 10    import matplotlib
 11    matplotlib.use('Agg')  # non-interactive backend
 12    import matplotlib.pyplot as plt
 13    from matplotlib.patches import Patch
 14    _HAS_MATPLOTLIB = True
 15except ImportError:
 16    _HAS_MATPLOTLIB = False
 17
 18# Mode display config: (color, label) — None color means no band drawn
 19_MODE_COLORS = {
 20    'disabled':   (None, 'Disabled'),
 21    'autonomous': ('#3498db', 'Auto'),
 22    'teleop':     ('#2ecc71', 'Teleop'),
 23    'test':       ('#f39c12', 'Test'),
 24}
 25
 26
 27def _build_mode_spans(mode_transitions: List[dict],
 28                      duration: float) -> List[Tuple[float, float, str]]:
 29    """Convert mode transitions into (start, end, mode_name) spans.
 30
 31    Mode transitions contain entries like:
 32        {'time_sec': 5.0, 'mode': 'enabled', 'value': True}
 33        {'time_sec': 5.0, 'mode': 'autonomous', 'value': True}
 34        {'time_sec': 20.0, 'mode': 'autonomous', 'value': False}
 35        {'time_sec': 20.0, 'mode': 'enabled', 'value': False}
 36
 37    Returns spans like: [(0, 5, 'disabled'), (5, 20, 'autonomous'), (20, 150, 'disabled')]
 38    """
 39    if not mode_transitions:
 40        return [(0, duration, 'disabled')]
 41
 42    # Collect state changes at each timestamp, preserving order within a timestamp
 43    events: List[Tuple[float, str, bool]] = []
 44    for t in mode_transitions:
 45        ts = t['time_sec']
 46        mode = t['mode']
 47        # Support both old format (no 'value' key, implied True) and new format
 48        val = t.get('value', True)
 49        if mode in ('enabled', 'autonomous', 'test'):
 50            events.append((ts, mode, val))
 51
 52    # Sort by timestamp (stable sort preserves order within same timestamp)
 53    events.sort(key=lambda e: e[0])
 54
 55    # Build spans by replaying events in order
 56    spans = []
 57    prev_time = 0.0
 58    prev_mode = 'disabled'
 59    state: Dict[str, bool] = {'enabled': False, 'autonomous': False, 'test': False}
 60
 61    for ts, mode, val in events:
 62        state[mode] = val
 63
 64        # Determine current mode from state
 65        if not state['enabled']:
 66            cur_mode = 'disabled'
 67        elif state['autonomous']:
 68            cur_mode = 'autonomous'
 69        elif state['test']:
 70            cur_mode = 'test'
 71        else:
 72            cur_mode = 'teleop'
 73
 74        # Only emit a span when mode actually changes
 75        if cur_mode != prev_mode:
 76            if ts > prev_time:
 77                spans.append((prev_time, ts, prev_mode))
 78            prev_mode = cur_mode
 79            prev_time = ts
 80
 81    # Final span to end of match
 82    if prev_time < duration:
 83        spans.append((prev_time, duration, prev_mode))
 84
 85    return spans
 86
 87
 88def _find_last_match_block(spans: List[Tuple[float, float, str]],
 89                           gap_tolerance: float = 5.0,
 90                           ) -> Tuple[float, float]:
 91    """Find the last contiguous block of non-disabled modes.
 92
 93    Walks backwards from the end of the spans list to find the last active
 94    block. Disabled gaps shorter than gap_tolerance are bridged (e.g.
 95    the brief disabled blip between auto and teleop in an FRC match).
 96
 97    Returns:
 98        (t_zero, match_end) where t_zero is the start of the first
 99        auto/teleop/test span in the block, and match_end is the end
100        of the last active span.  Falls back to (0, last_span_end) if
101        no active spans exist.
102    """
103    if not spans:
104        return (0.0, 0.0)
105
106    # Walk backwards: find last non-disabled span
107    i = len(spans) - 1
108    while i >= 0 and spans[i][2] == 'disabled':
109        i -= 1
110
111    if i < 0:
112        # All disabled
113        return (0.0, spans[-1][1])
114
115    last_block_end = spans[i][1]
116    last_block_start = spans[i][0]
117
118    # Walk backwards: bridge small disabled gaps
119    i -= 1
120    while i >= 0:
121        start, end, mode = spans[i]
122        if mode == 'disabled':
123            # Check the disabled span's own duration
124            gap_duration = end - start
125            if gap_duration > gap_tolerance:
126                break  # large gap — previous enable cycle
127        else:
128            # Active span — extend block start
129            last_block_start = start
130        i -= 1
131
132    # T=0 at first auto/teleop/test in this block, else block start
133    t_zero = last_block_start
134    for start, _end, mode in spans:
135        if start >= last_block_start and mode in ('autonomous', 'teleop', 'test'):
136            t_zero = start
137            break
138
139    return (t_zero, last_block_end)
140
141
142def voltage_chart(timeseries: List[Tuple[float, float]],
143                  brownout_times: List[float] = None,
144                  mode_transitions: List[dict] = None,
145                  match_duration: float = None) -> bytes:
146    """Generate a battery voltage chart with mode bands and return PNG bytes.
147
148    Args:
149        timeseries: List of (time_sec, voltage) tuples.
150        brownout_times: Optional list of brownout timestamps (seconds from start).
151        mode_transitions: Optional list of {'time_sec': float, 'mode': str} dicts.
152        match_duration: Total match duration in seconds.
153
154    Returns:
155        PNG image bytes, or empty bytes if matplotlib is unavailable or data is empty.
156    """
157    if not _HAS_MATPLOTLIB:
158        logger.debug("matplotlib not installed — skipping voltage chart")
159        return b''
160    if not timeseries:
161        return b''
162
163    duration = match_duration or (max(t for t, _ in timeseries))
164
165    _PRE_MATCH_BUFFER = 10.0   # seconds before match start to show
166    _POST_MATCH_BUFFER = 10.0  # seconds after match end to show
167
168    # Debug: dump mode transitions to help diagnose clipping issues
169    if mode_transitions:
170        has_false = any(not t.get('value', True) for t in mode_transitions)
171        logger.info(f"Chart: {len(mode_transitions)} mode transitions, "
172                    f"has_false_values={has_false}, duration={duration:.1f}")
173        for t in mode_transitions[:30]:
174            logger.info(f"  transition: {t}")
175
176    # Find the last contiguous "match" block
177    t_zero = 0.0
178    match_end = duration
179    if mode_transitions:
180        spans = _build_mode_spans(mode_transitions, duration)
181        t_zero, match_end = _find_last_match_block(spans)
182
183        logger.info(f"Chart: spans={len(spans)}, t_zero={t_zero:.1f}, "
184                    f"match_end={match_end:.1f}")
185        for s in spans:
186            logger.debug(f"  span: {s[0]:.1f}-{s[1]:.1f} {s[2]}")
187
188    clip_start = max(0.0, t_zero - _PRE_MATCH_BUFFER)
189    clip_end = min(duration, match_end + _POST_MATCH_BUFFER)
190
191    # Shift timeseries relative to t_zero, clip to window
192    shifted_ts = [(t - t_zero, v) for t, v in timeseries
193                  if clip_start <= t <= clip_end]
194    if not shifted_ts:
195        return b''
196    times = [t for t, _ in shifted_ts]
197    volts = [v for _, v in shifted_ts]
198    x_min = clip_start - t_zero
199    x_max = clip_end - t_zero
200
201    fig, ax = plt.subplots(figsize=(8, 3), dpi=120)
202
203    # Mode background bands
204    seen_modes = set()
205    if mode_transitions:
206        spans = _build_mode_spans(mode_transitions, duration)
207        for start, end, mode in spans:
208            # Shift and clip spans to visible window
209            s = max(start - t_zero, x_min)
210            e = min(end - t_zero, x_max)
211            if e <= s:
212                continue
213            color, _ = _MODE_COLORS.get(mode, ('#cccccc', mode))
214            if color is not None:
215                ax.axvspan(s, e, alpha=0.15, color=color, linewidth=0)
216                seen_modes.add(mode)
217
218    # Voltage trace
219    ax.plot(times, volts, color='#3498db', linewidth=1.0, label='Battery')
220
221    # Reference lines
222    ax.axhline(y=12.0, color='#2ecc71', linewidth=1.0, linestyle='--',
223               alpha=0.7, label='12V nominal')
224    ax.axhline(y=7.0, color='#e74c3c', linewidth=1.0, linestyle='--',
225               alpha=0.7, label='7V brownout')
226
227    # Brownout markers
228    if brownout_times:
229        for bt in brownout_times:
230            bt_shifted = bt - t_zero
231            if bt_shifted >= clip_start - t_zero:
232                ax.axvline(x=bt_shifted, color='#e74c3c', linewidth=0.8,
233                           linestyle=':', alpha=0.5)
234
235    ax.set_xlabel('Match Time (s)', fontsize=9)
236    ax.set_ylabel('Voltage (V)', fontsize=9)
237    ax.set_title('Battery Voltage', fontsize=10, fontweight='bold')
238    ax.set_xlim(left=x_min, right=x_max)
239    ax.set_ylim(bottom=0, top=14)
240    ax.grid(True, alpha=0.3)
241    ax.tick_params(labelsize=8)
242
243    # Build legend: voltage items on the left, mode patches on the right
244    voltage_legend = ax.legend(loc='lower left', fontsize=7, framealpha=0.8)
245    ax.add_artist(voltage_legend)
246
247    if seen_modes:
248        mode_patches = [
249            Patch(facecolor=_MODE_COLORS[m][0], alpha=0.3, label=_MODE_COLORS[m][1])
250            for m in ('disabled', 'autonomous', 'teleop', 'test')
251            if m in seen_modes
252        ]
253        ax.legend(handles=mode_patches, loc='lower right', fontsize=7,
254                  framealpha=0.8, title='Mode', title_fontsize=7)
255
256    fig.tight_layout()
257
258    buf = io.BytesIO()
259    fig.savefig(buf, format='png', bbox_inches='tight')
260    plt.close(fig)
261    buf.seek(0)
262    return buf.read()