fix more telemetry bugs, separate classes

paulocode · May 4, 2024 · 6c080c7 · 6c080c7
1 parent c1054d4
commit 6c080c7
Show file tree

Hide file tree

Showing 8 changed files with 316 additions and 294 deletions.
diff --git a/lib/providers/telemetry/telemetry_latest_streamer.dart b/lib/providers/telemetry/telemetry_latest_streamer.dart
@@ -0,0 +1,65 @@
+import 'package:riverpod_annotation/riverpod_annotation.dart';
+
+import '../../models/telemetry_state.dart';
+import '../../models/timed_telemetry.dart';
+import '../../repository/telemetry_repository.dart';
+import '../radio_config/radio_config_service.dart';
+import '../repository/telemetry_repository.dart';
+import 'telemetry_receiver.dart';
+
+part 'telemetry_latest_streamer.g.dart';
+
+@Riverpod(keepAlive: true)
+class TelemetryLatestStreamer extends _$TelemetryLatestStreamer {
+  late TelemetryRepository _telemetryRepository;
+  late int _myNodeNum;
+  late int _nodeNum;
+
+  @override
+  TelemetryState build({required int nodeNum}) {
+    final sub = ref
+        .watch(telemetryReceiverProvider)
+        .addTelemetryListener(nodeNum: nodeNum, listener: _processTelemetry);
+    _myNodeNum = ref
+        .watch(radioConfigServiceProvider.select((value) => value.myNodeNum));
+    _telemetryRepository = ref.watch(telemetryRepositoryProvider);
+    _nodeNum = nodeNum;
+    ref.onDispose(sub.cancel);
+    _loadFirstTelemetry();
+    return const TelemetryState();
+  }
+
+  void _processTelemetry(TimedTelemetry timedTelemetry) {
+    final telemetry = timedTelemetry.telemetry;
+    final environmentMetrics = telemetry.environmentMetrics;
+    if (environmentMetrics.hasTemperature()) {
+      state = state.copyWith(temp: telemetry.environmentMetrics.temperature);
+    }
+    if (environmentMetrics.hasRelativeHumidity()) {
+      state = state.copyWith(
+        relativeHumidity: telemetry.environmentMetrics.relativeHumidity,
+      );
+    }
+    if (environmentMetrics.hasBarometricPressure()) {
+      state = state.copyWith(
+        barometricPressure: telemetry.environmentMetrics.barometricPressure,
+      );
+    }
+    if (environmentMetrics.hasGasResistance()) {
+      state = state.copyWith(
+        gasResistance: telemetry.environmentMetrics.gasResistance,
+      );
+    }
+  }
+
+  Future<void> _loadFirstTelemetry() async {
+    final telemetry = await _telemetryRepository.getOne(
+      index: 0,
+      fromNode: _nodeNum,
+      owner: _myNodeNum,
+    );
+    if (telemetry != null) {
+      _processTelemetry(telemetry);
+    }
+  }
+}
diff --git a/lib/providers/telemetry/telemetry_latest_streamer.g.dart b/lib/providers/telemetry/telemetry_latest_streamer.g.dart
diff --git a/lib/providers/telemetry/telemetry_receiver.dart b/lib/providers/telemetry/telemetry_receiver.dart
@@ -1,15 +1,6 @@
-import 'dart:async';
-
-import 'package:logger/logger.dart';
 import 'package:riverpod_annotation/riverpod_annotation.dart';
 
-import '../../models/telemetry_state.dart';
-import '../../models/timed_telemetry.dart';
-import '../../protobufs/generated/meshtastic/mesh.pb.dart';
-import '../../protobufs/generated/meshtastic/portnums.pb.dart';
-import '../../protobufs/generated/meshtastic/telemetry.pb.dart';
-import '../../repository/telemetry_repository.dart';
-import '../../services/interfaces/radio_reader.dart';
+import '../../services/telemetry/telemetry_receiver.dart';
 import '../radio_config/radio_config_service.dart';
 import '../radio_reader.dart';
 import '../repository/telemetry_repository.dart';
@@ -25,120 +16,3 @@ TelemetryReceiver telemetryReceiver(TelemetryReceiverRef ref) {
     onDispose: ref.onDispose,
   );
 }
-
-class TelemetryReceiver {
-  TelemetryReceiver({
-    required int myNodeNum,
-    required TelemetryRepository telemetryRepository,
-    required RadioReader radioReader,
-    required void Function(void Function() cb) onDispose,
-  })  : _myNodeNum = myNodeNum,
-        _telemetryRepository = telemetryRepository,
-        _radioReader = radioReader,
-        _onDispose = onDispose {
-    final sub = _radioReader.onPacketReceived().listen(_processPacket);
-    _streamController = StreamController<MeshPacket>.broadcast();
-    _onDispose(sub.cancel);
-    _onDispose(_streamController.close);
-  }
-
-  final int _myNodeNum;
-  final TelemetryRepository _telemetryRepository;
-  final void Function(void Function() cb) _onDispose;
-  final RadioReader _radioReader;
-  late StreamController<MeshPacket> _streamController;
-  final _logger = Logger();
-
-  void _processPacket(FromRadio event) {
-    final packet = event.packet;
-    final decoded = packet.decoded;
-    if (decoded.portnum != PortNum.TELEMETRY_APP) {
-      return;
-    }
-    final telemetry = Telemetry.fromBuffer(decoded.payload);
-    if (!telemetry.hasEnvironmentMetrics()) {
-      // TODO: handle deviceMetrics
-      return;
-    }
-
-    _logger.i(telemetry.toString());
-    _streamController.add(packet);
-    _telemetryRepository.add(
-      timedTelemetry:
-          TimedTelemetry(timeReceived: DateTime.now(), telemetry: telemetry),
-      fromNode: packet.from,
-      owner: _myNodeNum,
-    );
-  }
-
-  StreamSubscription<MeshPacket> addTelemetryListener({
-    required int nodeNum,
-    required void Function(TimedTelemetry) listener,
-  }) {
-    return _streamController.stream
-        .where((event) => event.from == nodeNum)
-        .listen((event) {
-      listener(
-        TimedTelemetry(
-          timeReceived: DateTime.now(),
-          telemetry: Telemetry.fromBuffer(event.decoded.payload),
-        ),
-      );
-    });
-  }
-}
-
-@Riverpod(keepAlive: true)
-class TelemetryLatestStreamer extends _$TelemetryLatestStreamer {
-  late TelemetryRepository _telemetryRepository;
-  late int _myNodeNum;
-  late int _nodeNum;
-
-  @override
-  TelemetryState build({required int nodeNum}) {
-    final sub = ref
-        .watch(telemetryReceiverProvider)
-        .addTelemetryListener(nodeNum: nodeNum, listener: _processTelemetry);
-    _myNodeNum = ref
-        .watch(radioConfigServiceProvider.select((value) => value.myNodeNum));
-    _telemetryRepository = ref.watch(telemetryRepositoryProvider);
-    _nodeNum = nodeNum;
-    ref.onDispose(sub.cancel);
-    _loadFirstTelemetry();
-    return const TelemetryState();
-  }
-
-  void _processTelemetry(TimedTelemetry timedTelemetry) {
-    final telemetry = timedTelemetry.telemetry;
-    final environmentMetrics = telemetry.environmentMetrics;
-    if (environmentMetrics.hasTemperature()) {
-      state = state.copyWith(temp: telemetry.environmentMetrics.temperature);
-    }
-    if (environmentMetrics.hasRelativeHumidity()) {
-      state = state.copyWith(
-        relativeHumidity: telemetry.environmentMetrics.relativeHumidity,
-      );
-    }
-    if (environmentMetrics.hasBarometricPressure()) {
-      state = state.copyWith(
-        barometricPressure: telemetry.environmentMetrics.barometricPressure,
-      );
-    }
-    if (environmentMetrics.hasGasResistance()) {
-      state = state.copyWith(
-        gasResistance: telemetry.environmentMetrics.gasResistance,
-      );
-    }
-  }
-
-  Future<void> _loadFirstTelemetry() async {
-    final telemetry = await _telemetryRepository.getOne(
-      index: 0,
-      fromNode: _nodeNum,
-      owner: _myNodeNum,
-    );
-    if (telemetry != null) {
-      _processTelemetry(telemetry);
-    }
-  }
-}