add dummy sensor

reskin_sensor/__init__.py

@@ -1,2 +1,2 @@
-from .sensor import ReSkinBase
+from .sensor import ReSkinBase, ReSkinDummy
 from .sensor_proc import ReSkinProcess

reskin_sensor/sensor.py

@@ -45,11 +45,10 @@ def __init__(
         reskin_data_struct: bool = True,
     ) -> None:
         """Initializes a ReSkinBase object."""
-        super(ReSkinBase, self).__init__(port=port, baudrate=baudrate)
 
         self.num_mags = num_mags
-        self.port = port
-        self.baudrate = baudrate
+        self.port_name = port
+        self.baud_rate = baudrate
         self.burst_mode = burst_mode
         self.device_id = device_id
         self.reskin_data_struct = reskin_data_struct
@@ -61,6 +60,7 @@ def __init__(
         if temp_filtered:
             self._temp_mask[::4] = False
 
+        super(ReSkinBase, self).__init__(port=port, baudrate=baudrate)
         self._initialize()
 
     def _initialize(self):
@@ -149,3 +149,40 @@ def get_sample(self, num_samples=1):
             else:
                 # Need checks to timeout if required
                 pass
+
+
+class ReSkinDummy(ReSkinBase):
+    def __init__(
+        self,
+        num_mags: int = 1,
+        port: str = None,
+        baudrate: int = 115200,
+        burst_mode: bool = True,
+        device_id: int = -1,
+        temp_filtered: bool = False,
+        reskin_data_struct: bool = True,
+    ):
+
+        self.num_mags = num_mags
+        self.port_name = port
+        self.baud_rate = baudrate
+        self.burst_mode = burst_mode
+        self.device_id = device_id
+        self.reskin_data_struct = reskin_data_struct
+
+        self._msg_floats = 4 * num_mags
+        self._msg_length = 4 * self._msg_floats + 2
+
+        self._temp_mask = np.ones((self._msg_floats,), dtype=bool)
+        if temp_filtered:
+            self._temp_mask[::4] = False
+
+    def _initialize(self):
+        pass
+
+    def get_sample(self, num_samples=1):
+        collect_start = time.time()
+        data = np.random.uniform(-1., 1., size=(np.sum(self._temp_mask),))
+        acq_delay = time.time() - collect_start
+
+        return collect_start, acq_delay, data

reskin_sensor/sensor_proc.py

@@ -6,7 +6,7 @@
 import numpy as np
 import serial
 
-from .sensor import ReSkinBase, ReSkinData
+from .sensor import ReSkinBase, ReSkinData, ReSkinDummy
 
 
 class ReSkinProcess(Process):
@@ -16,7 +16,7 @@ class ReSkinProcess(Process):
     Attributes
     ----------
     num_mags: int
-            Number of magnetometers connected to the sensor
+        Number of magnetometers connected to the sensor
     port : str
         System port that the sensor is connected to
     baudrate: int
@@ -28,6 +28,9 @@ class ReSkinProcess(Process):
     temp_filtered: bool
         Flag indicating if temperature readings should be filtered from
         the output
+    allow_dummy_sensor: bool
+        Flag to instantiate a dummy sensor if a real sensor with the specified
+        configurations is unavailable
     chunk_size : int
         Quantum of data piped from buffer at one time.
 
@@ -56,6 +59,7 @@ def __init__(
         device_id: int = -1,
         temp_filtered: bool = False,
         reskin_data_struct: bool = True,
+        allow_dummy_sensor: bool = False,
         chunk_size: int = 10000,
     ):
         """Initializes a ReSkinProcess object."""
@@ -67,6 +71,7 @@ def __init__(
         self.device_id = device_id
         self.temp_filtered = temp_filtered
         self.reskin_data_struct = reskin_data_struct
+        self.allow_dummy_sensor = allow_dummy_sensor
 
         self._pipe_in, self._pipe_out = Pipe()
         self._sample_cnt = Value(ct.c_uint64)
@@ -229,10 +234,22 @@ def run(self):
             )
             # self.sensor._initialize()
             self.start_streaming()
-        except serial.serialutil.SerialException as e:
-            self._event_quit_request.set()
+        except (serial.serialutil.SerialException, AttributeError) as e:
             print("ERROR: ", e)
-            sys.exit(1)
+            if self.allow_dummy_sensor:
+                print("Using dummy sensor")
+                self.sensor = ReSkinDummy(
+                    num_mags=self.num_mags,
+                    port=self.port,
+                    baudrate=self.baudrate,
+                    burst_mode=self.burst_mode,
+                    device_id=self.device_id,
+                    temp_filtered=self.temp_filtered,
+                    reskin_data_struct=True,
+                )
+                self.start_streaming()
+            else:
+                sys.exit(-1)
 
         is_streaming = False
         while not self._event_quit_request.is_set():

tests/sensor_test.py

@@ -1,5 +1,7 @@
+import serial
+
 import argparse
-from reskin_sensor import ReSkinBase
+from reskin_sensor import ReSkinBase, ReSkinDummy
 
 if __name__ == "__main__":
     # fmt: off
@@ -12,13 +14,24 @@
     # fmt: on
     args = parser.parse_args()
 
-    test_sensor = ReSkinBase(
-        num_mags=args.num_mags,
-        port=args.port,
-        baudrate=args.baudrate,
-        burst_mode=True,
-        device_id=1,
-    )
+    try:
+        test_sensor = ReSkinBase(
+            num_mags=args.num_mags,
+            port=args.port,
+            baudrate=args.baudrate,
+            burst_mode=True,
+            device_id=1,
+        )
+    except serial.serialutil.SerialException as e:
+        print("ERROR: ", e)
+        print("Using dummy sensor")
+        test_sensor = ReSkinDummy(
+            num_mags=args.num_mags,
+            port=args.port,
+            baudrate=args.baudrate,
+            burst_mode=True,
+            device_id=1,
+        )
 
     # Get 5 samples from sensor
     test_samples = test_sensor.get_data(num_samples=5)