Provide means to mock ADC

src/client/adc_config.py

@@ -4,6 +4,7 @@
 
 
 class AdcConfig(object):
+    mock_adc: bool
     bias_point_channel: Optional[int]
     coil_input_channel: int
     adc_bit_resolution: int
@@ -12,7 +13,8 @@ class AdcConfig(object):
     MOSI: int
     CS: int
 
-    def __init__(self, seismometer_id: str):
+    def __init__(self, seismometer_id: str, mock_adc: bool):
+        self.mock_adc = mock_adc
         if seismometer_id == SEISMOMETER_ID_LEHMAN:
             self.bias_point_channel = None
             self.coil_input_channel = 7

src/client/adc_wrapper.py

@@ -4,12 +4,22 @@
 from src.client.mcp3208 import MCP3208
 
 
+class MockAdc(object):
+
+    def read_adc(self, channel: int): return 1
+
+
 class AdcWrapper(object):
     config: AdcConfig
-    adc: Adafruit_MCP3008.MCP3008
+    adc: Adafruit_MCP3008.MCP3008 | MockAdc
 
     def __init__(self, config: AdcConfig):
         self.config = config
+
+        if config.mock_adc:
+            self.adc = MockAdc()
+            return
+
         if config.adc_bit_resolution == 12:
             self.adc = MCP3208(
                 clk=config.CLK,

src/client/create_web_api_socket.py

@@ -28,8 +28,8 @@ def on_close(ws: WebSocket, close_status_code: Any, close_msg: Any):
     return ws
 
 
-def start_websocket_and_logger(seismometer_id: str) -> None:
-    config = SeismometerConfig(seismometer_id)
+def start_websocket_and_logger(seismometer_id: str, mock_adc: bool) -> None:
+    config = SeismometerConfig(seismometer_id, mock_adc)
     print("starting", '\'' + seismometer_id + '\'')
 
     def on_websocket_open(ws):

src/client/data_uploader.py

@@ -68,7 +68,7 @@ def run(self):
                 if not self.data_box.is_full():
                     self.condition.wait()
 
-                values: NDArray = self.data_box.get_values()
+                values: NDArray[int] = self.data_box.get_values()
                 bias_point = self.data_box.bias_point
                 actual_sampling_rate = self.data_box.actual_sampling_rate
 

src/client/seism_logger.py

@@ -24,7 +24,7 @@ class SeismometerConfig(object):
     chunk_size: int
     adc_config: AdcConfig
 
-    def __init__(self, seismometer_id: str):
+    def __init__(self, seismometer_id: str, mock_adc: bool):
         self.sampling_rate = 750
         self.decimated_sampling_rate = 30
         self.scale_factor = 8
@@ -34,7 +34,7 @@ def __init__(self, seismometer_id: str):
         self.filter_cutoff_freq = 6
         self.use_rolling_avg = False
         self.chunk_size = self.sampling_rate * self.upload_interval
-        self.adc_config = AdcConfig(seismometer_id)
+        self.adc_config = AdcConfig(seismometer_id, mock_adc)
 
 
 class SeismLogger(object):

start_logger.py

@@ -12,7 +12,8 @@
         print("No AUTH_TOKEN defined as env var")
     else:
         seismometer_id = sys.argv[1]
+        mock_adc = len(sys.argv) > 2 and sys.argv[2] == '--mock-adc'
         if seismometer_id in SEISMOMETER_IDS:
-            start_websocket_and_logger(seismometer_id)
+            start_websocket_and_logger(seismometer_id, mock_adc)
         else:
             print("Invalid seismometer_id:", seismometer_id)