identify sensor streams by wearable and sensor to avoid collisions

open_wearable/lib/models/sensor_streams.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:collection';
 
 import 'package:open_earable_flutter/open_earable_flutter.dart';
 
@@ -7,16 +8,38 @@ import 'package:open_earable_flutter/open_earable_flutter.dart';
 class SensorStreams {
   SensorStreams._();
 
-  static final Map<Sensor, Stream<SensorValue>> _sharedStreams = {};
+  static final Map<String, Map<Sensor, Stream<SensorValue>>>
+      _sharedStreamsByDevice = {};
+  static Map<Sensor, Stream<SensorValue>> _createIdentitySensorStreamMap() =>
+      LinkedHashMap<Sensor, Stream<SensorValue>>.identity();
 
-  static Stream<SensorValue> shared(Sensor sensor) {
-    return _sharedStreams.putIfAbsent(
+  static Stream<SensorValue> shared({
+    required Wearable wearable,
+    required Sensor sensor,
+  }) {
+    final deviceStreams = _sharedStreamsByDevice.putIfAbsent(
+      wearable.deviceId,
+      // Identity map avoids collisions when Sensor overrides ==/hashCode
+      // non-uniquely across different devices.
+      _createIdentitySensorStreamMap,
+    );
+    return deviceStreams.putIfAbsent(
       sensor,
       () => sensor.sensorStream.asBroadcastStream(),
     );
   }
 
-  static void clearForSensor(Sensor sensor) {
-    _sharedStreams.remove(sensor);
+  static void clearForSensor({
+    required Wearable wearable,
+    required Sensor sensor,
+  }) {
+    final deviceStreams = _sharedStreamsByDevice[wearable.deviceId];
+    if (deviceStreams == null) {
+      return;
+    }
+    deviceStreams.remove(sensor);
+    if (deviceStreams.isEmpty) {
+      _sharedStreamsByDevice.remove(wearable.deviceId);
+    }
   }
 }

open_wearable/lib/view_models/sensor_data_provider.dart

@@ -7,6 +7,7 @@ import 'package:open_earable_flutter/open_earable_flutter.dart';
 import 'package:open_wearable/models/sensor_streams.dart';
 
 class SensorDataProvider with ChangeNotifier {
+  final Wearable wearable;
   final Sensor sensor;
   final int timeWindow; // seconds
 
@@ -26,6 +27,7 @@ class SensorDataProvider with ChangeNotifier {
   DateTime? _lastSensorArrivalTime;
 
   SensorDataProvider({
+    required this.wearable,
     required this.sensor,
     this.timeWindow = 5,
   }) {
@@ -52,8 +54,10 @@ class SensorDataProvider with ChangeNotifier {
   }
 
   void _listenToStream() {
-    _sensorStreamSubscription =
-        SensorStreams.shared(sensor).listen((sensorValue) {
+    _sensorStreamSubscription = SensorStreams.shared(
+      wearable: wearable,
+      sensor: sensor,
+    ).listen((sensorValue) {
       sensorValues.add(sensorValue);
       _lastSensorTimestamp = sensorValue.timestamp;
       _lastSensorArrivalTime = DateTime.now();

open_wearable/lib/view_models/sensor_recorder_provider.dart

@@ -148,7 +148,10 @@ class SensorRecorderProvider with ChangeNotifier {
 
       File file = await recorder.start(
         filepath: filepath,
-        inputStream: SensorStreams.shared(sensor),
+        inputStream: SensorStreams.shared(
+          wearable: wearable,
+          sensor: sensor,
+        ),
       );
 
       logger.i(

open_wearable/lib/widgets/sensors/sensor_page.dart

@@ -189,7 +189,10 @@ class _SensorPageState extends State<SensorPage>
           in wearable.requireCapability<SensorManager>().sensors) {
         _sensorDataProviders.putIfAbsent(
           (wearable, sensor),
-          () => SensorDataProvider(sensor: sensor),
+          () => SensorDataProvider(
+            wearable: wearable,
+            sensor: sensor,
+          ),
         );
       }
     }

open_wearable/lib/widgets/sensors/values/sensor_values_page.dart

@@ -144,7 +144,10 @@ class _SensorValuesPageState extends State<SensorValuesPage>
           in wearable.requireCapability<SensorManager>().sensors) {
         _sensorDataProvider.putIfAbsent(
           (wearable, sensor),
-          () => SensorDataProvider(sensor: sensor),
+          () => SensorDataProvider(
+            wearable: wearable,
+            sensor: sensor,
+          ),
         );
       }
     }