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((ref) { final port = const ReplayVehicleDataPort(); ref.onDispose(() => unawaited(port.close())); return port; }); final observabilityControllerProvider = NotifierProvider( ObservabilityController.new, ); class ObservabilityController extends Notifier { static const _criticalAnnouncementWindow = Duration(seconds: 2); StreamSubscription? _subscription; DateTime? _lastCriticalAnnouncementAt; @override ObservabilityState build() { final source = ref.read(vehicleDataPortProvider); ref.onDispose(_cancelSubscription); Future.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 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 get observedTelemetryKinds => { for (final event in events) if (event case TelemetryEvent(:final kind)) kind, }; ObservabilityState copyWith({ ObservationStatus? status, List? 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; }