feat(ground-suite): add replay observability console

ground-suite/lib/features/observability/presentation/observability_controller.dart

@@ -0,0 +1,206 @@
+import 'dart:async';
+
+import 'package:flutter_riverpod/flutter_riverpod.dart';
+
+import '../../../core/observations/observed_event.dart';
+import '../../../core/observations/vehicle_data_port.dart';
+import '../../../core/observations/vehicle_data_update.dart';
+import '../../../core/transport/replay_vehicle_data_port.dart';
+
+final vehicleDataPortProvider = Provider<VehicleDataPort>((ref) {
+  final port = const ReplayVehicleDataPort();
+  ref.onDispose(() => unawaited(port.close()));
+  return port;
+});
+
+final observabilityControllerProvider =
+    NotifierProvider<ObservabilityController, ObservabilityState>(
+      ObservabilityController.new,
+    );
+
+class ObservabilityController extends Notifier<ObservabilityState> {
+  static const _criticalAnnouncementWindow = Duration(seconds: 2);
+
+  StreamSubscription<VehicleDataUpdate>? _subscription;
+  DateTime? _lastCriticalAnnouncementAt;
+
+  @override
+  ObservabilityState build() {
+    final source = ref.read(vehicleDataPortProvider);
+    ref.onDispose(_cancelSubscription);
+    Future<void>.microtask(() => _observe(source));
+    return ObservabilityState.initial(source);
+  }
+
+  void _observe(VehicleDataPort source) {
+    _subscription = source.observe().listen(
+      _handleUpdate,
+      onError: (Object _, StackTrace _) {
+        state = state.copyWith(
+          status: ObservationStatus.disconnected,
+          accessibilityAnnouncement: 'Observation connection closed.',
+          replaceAccessibilityAnnouncement: true,
+        );
+      },
+      onDone: () {
+        state = state.copyWith(
+          status: source.isReplay
+              ? ObservationStatus.replayComplete
+              : ObservationStatus.disconnected,
+          accessibilityAnnouncement: source.isReplay
+              ? 'Replay complete.'
+              : 'Observation connection closed.',
+          replaceAccessibilityAnnouncement: true,
+        );
+      },
+    );
+  }
+
+  void _handleUpdate(VehicleDataUpdate update) {
+    switch (update) {
+      case ObservationReceived(:final event):
+        _record(event);
+      case ObservationDropped(:final kind, :final count):
+        state = state.copyWith(
+          status: ObservationStatus.receiving,
+          parserDropCount:
+              state.parserDropCount +
+              (kind == ObservationDropKind.parser ? count : 0),
+          semanticDropCount:
+              state.semanticDropCount +
+              (kind == ObservationDropKind.semantic ? count : 0),
+        );
+    }
+  }
+
+  void _record(ObservedEvent event) {
+    final announcement = _announcementFor(event);
+    state = state.copyWith(
+      status: ObservationStatus.receiving,
+      lastFrameAt: event.receivedAt,
+      events: [...state.events, event],
+      accessibilityAnnouncement: announcement.message,
+      replaceAccessibilityAnnouncement: true,
+      coalescedAlertCount:
+          state.coalescedAlertCount + (announcement.coalesced ? 1 : 0),
+    );
+  }
+
+  _AnnouncementDecision _announcementFor(ObservedEvent event) {
+    final candidate = switch (event) {
+      ErrorEvent(:final path) => 'Error received: $path',
+      ShutdownEvent(:final reason) =>
+        'Orderly shutdown observed: ${reason.label}',
+      _ => null,
+    };
+    if (candidate == null) {
+      return const _AnnouncementDecision();
+    }
+
+    final now = DateTime.now();
+    final previous = _lastCriticalAnnouncementAt;
+    if (previous != null &&
+        now.difference(previous) < _criticalAnnouncementWindow) {
+      return const _AnnouncementDecision(coalesced: true);
+    }
+    _lastCriticalAnnouncementAt = now;
+    return _AnnouncementDecision(message: candidate);
+  }
+
+  void _cancelSubscription() {
+    _subscription?.cancel();
+  }
+}
+
+enum ObservationStatus {
+  starting('Starting'),
+  receiving('Observing'),
+  replayComplete('Replay complete'),
+  disconnected('Disconnected');
+
+  const ObservationStatus(this.label);
+
+  final String label;
+}
+
+class ObservabilityState {
+  const ObservabilityState({
+    required this.sourceLabel,
+    required this.isReplay,
+    required this.status,
+    required this.events,
+    required this.parserDropCount,
+    required this.semanticDropCount,
+    required this.coalescedAlertCount,
+    this.lastFrameAt,
+    this.accessibilityAnnouncement,
+  });
+
+  factory ObservabilityState.initial(VehicleDataPort source) {
+    return ObservabilityState(
+      sourceLabel: source.label,
+      isReplay: source.isReplay,
+      status: ObservationStatus.starting,
+      events: const [],
+      parserDropCount: 0,
+      semanticDropCount: 0,
+      coalescedAlertCount: 0,
+    );
+  }
+
+  final String sourceLabel;
+  final bool isReplay;
+  final ObservationStatus status;
+  final List<ObservedEvent> events;
+  final int parserDropCount;
+  final int semanticDropCount;
+  final int coalescedAlertCount;
+  final DateTime? lastFrameAt;
+  final String? accessibilityAnnouncement;
+
+  HeartbeatEvent? get lastHeartbeat {
+    for (final event in events.reversed) {
+      if (event case HeartbeatEvent()) {
+        return event;
+      }
+    }
+    return null;
+  }
+
+  Set<TelemetryKind> get observedTelemetryKinds => {
+    for (final event in events)
+      if (event case TelemetryEvent(:final kind)) kind,
+  };
+
+  ObservabilityState copyWith({
+    ObservationStatus? status,
+    List<ObservedEvent>? events,
+    int? parserDropCount,
+    int? semanticDropCount,
+    int? coalescedAlertCount,
+    DateTime? lastFrameAt,
+    String? accessibilityAnnouncement,
+    bool replaceAccessibilityAnnouncement = false,
+  }) {
+    return ObservabilityState(
+      sourceLabel: sourceLabel,
+      isReplay: isReplay,
+      status: status ?? this.status,
+      events: events ?? this.events,
+      parserDropCount: parserDropCount ?? this.parserDropCount,
+      semanticDropCount: semanticDropCount ?? this.semanticDropCount,
+      coalescedAlertCount: coalescedAlertCount ?? this.coalescedAlertCount,
+      lastFrameAt: lastFrameAt ?? this.lastFrameAt,
+      accessibilityAnnouncement: replaceAccessibilityAnnouncement
+          ? accessibilityAnnouncement
+          : this.accessibilityAnnouncement,
+    );
+  }
+}
+
+class _AnnouncementDecision {
+  const _AnnouncementDecision({this.message, this.coalesced = false});
+
+  final String? message;
+  final bool coalesced;
+}