Sea 178 test actions loading

.pre-commit-config.yaml

@@ -18,7 +18,7 @@ repos:
       - id: end-of-file-fixer
       - id: trailing-whitespace
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.15.0
+    rev: v3.15.1
     hooks:
       - id: pyupgrade
         args: ["--py38-plus"]
@@ -30,12 +30,12 @@ repos:
         types: [file, python]
         args: ["--profile", "black", "--filter-files", "--gitignore"]
   - repo: https://github.com/psf/black
-    rev: 23.12.1
+    rev: 24.2.0
     hooks:
       - id: black
         types: [file, python]
   - repo: https://github.com/igorshubovych/markdownlint-cli
-    rev: v0.38.0
+    rev: v0.39.0
     hooks:
       - id: markdownlint
         types: [file, markdown]

sample_debug.py

@@ -2,6 +2,7 @@
 This is a sample file showing how an action be created and called for debugging purposes
 using a mock signal analyzer.
 """
+
 import json
 
 from scos_actions.actions.acquire_single_freq_fft import SingleFrequencyFftAcquisition

scos_actions/__init__.py

@@ -1 +1 @@
-__version__ = "8.0.1"
+__version__ = "9.0.0"

scos_actions/actions/acquire_sea_data_product.py

@@ -99,6 +99,7 @@
 DURATION_MS = "duration_ms"
 NUM_SKIP = "nskip"
 PFP_FRAME_PERIOD_MS = "pfp_frame_period_ms"
+CAL_ADJUST = "calibration_adjust"
 
 # Constants
 DATA_TYPE = np.half
@@ -158,6 +159,7 @@ def __init__(
             - 10.0 * np.log10(sample_rate_Hz * fft_size)  # PSD scaling
             + 20.0 * np.log10(window_ecf)  # Window energy correction
         )
+        self.cal_adjust = True
 
     def run(self, iq: ray.ObjectRef) -> np.ndarray:
         """
@@ -630,14 +632,20 @@ def capture_iq(self, params: dict) -> dict:
         nskip = utils.get_parameter(NUM_SKIP, params)
         num_samples = int(params[SAMPLE_RATE] * duration_ms * 1e-3)
         # Collect IQ data
+        self.cal_adjust = True
+        for key, value in params.items():
+            logger.debug(f"param {key}={value}")
+        self.cal_adjust = utils.get_parameter(CAL_ADJUST, params)
+        logger.debug(f"cal_adjust={self.cal_adjust}")
         measurement_result = self.sensor.signal_analyzer.acquire_time_domain_samples(
-            num_samples, nskip
+            num_samples, nskip, cal_adjust=self.cal_adjust
         )
         # Store some metadata with the IQ
         measurement_result.update(params)
-        measurement_result[
-            "sensor_cal"
-        ] = self.sensor.signal_analyzer.sensor_calibration_data
+        if self.cal_adjust:
+            measurement_result["sensor_cal"] = (
+                self.sensor.signal_analyzer.sensor_calibration_data
+            )
         toc = perf_counter()
         logger.debug(
             f"IQ Capture ({duration_ms} ms @ {(params[FREQUENCY]/1e6):.1f} MHz) completed in {toc-tic:.2f} s."
@@ -786,7 +794,8 @@ def capture_diagnostics(
             "scos_sensor_version": SCOS_SENSOR_GIT_TAG,
             "scos_actions_version": SCOS_ACTIONS_VERSION,
             "scos_sigan_plugin": ntia_diagnostics.ScosPlugin(
-                name="scos_tekrsa", version=self.sensor.signal_analyzer.plugin_version
+                name=self.sensor.signal_analyzer.plugin_name,
+                version=self.sensor.signal_analyzer.plugin_version,
             ),
             "preselector_api_version": PRESELECTOR_API_VERSION,
             "sigan_firmware_version": self.sensor.signal_analyzer.firmware_version,
@@ -1125,19 +1134,20 @@ def create_capture_segment(
             datetime=measurement_result["capture_time"],
             duration=measurement_result[DURATION_MS],
             overload=measurement_result["overload"],
-            sensor_calibration=ntia_sensor.Calibration(
-                datetime=measurement_result["sensor_cal"]["datetime"],
-                gain=round(measurement_result["sensor_cal"]["gain"], 3),
-                noise_figure=round(measurement_result["sensor_cal"]["noise_figure"], 3),
-                temperature=round(measurement_result["sensor_cal"]["temperature"], 1),
-                reference=DATA_REFERENCE_POINT,
-            ),
             sigan_settings=ntia_sensor.SiganSettings(
                 reference_level=self.sensor.signal_analyzer.reference_level,
                 attenuation=self.sensor.signal_analyzer.attenuation,
                 preamp_enable=self.sensor.signal_analyzer.preamp_enable,
             ),
         )
+        if self.cal_adjust:
+            capture_segment.sensor_calibration = ntia_sensor.Calibration(
+                datetime=measurement_result["sensor_cal"]["datetime"],
+                gain=round(measurement_result["sensor_cal"]["gain"], 3),
+                noise_figure=round(measurement_result["sensor_cal"]["noise_figure"], 3),
+                temperature=round(measurement_result["sensor_cal"]["temperature"], 1),
+                reference=DATA_REFERENCE_POINT,
+            )
         self.sigmf_builder.add_capture(capture_segment)
 
     def get_sigmf_builder(

scos_actions/actions/acquire_single_freq_fft.py

@@ -178,9 +178,6 @@ def execute(self, schedule_entry: dict, task_id: int) -> dict:
         # Save measurement results
         measurement_result["data"] = m4s_result
         measurement_result.update(self.parameters)
-        measurement_result[
-            "calibration_datetime"
-        ] = self.sensor.signal_analyzer.sensor_calibration_data["datetime"]
         measurement_result["task_id"] = task_id
         measurement_result["classification"] = self.classification
 
@@ -191,7 +188,7 @@ def execute(self, schedule_entry: dict, task_id: int) -> dict:
             sample_start=0,
             start_time=measurement_result["capture_time"],
             center_frequency_Hz=self.frequency_Hz,
-            duration_ms=int(self.num_samples / sample_rate_Hz),
+            duration_ms=round((self.num_samples / sample_rate_Hz) * 1000),
             overload=measurement_result["overload"],
             sigan_settings=sigan_settings,
         )

scos_actions/actions/acquire_single_freq_tdomain_iq.py

@@ -89,9 +89,6 @@ def execute(self, schedule_entry: dict, task_id: int) -> dict:
         measurement_result.update(self.parameters)
         measurement_result["end_time"] = end_time
         measurement_result["task_id"] = task_id
-        measurement_result[
-            "calibration_datetime"
-        ] = self.sensor.signal_analyzer.sensor_calibration_data["datetime"]
         measurement_result["classification"] = self.classification
         sigan_settings = self.get_sigan_settings(measurement_result)
         logger.debug(f"sigan settings:{sigan_settings}")

scos_actions/actions/acquire_stepped_freq_tdomain_iq.py

@@ -130,9 +130,9 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
                     sensor_cal["compression_point"] = sensor_cal.pop(
                         "1db_compression_point"
                     )
-                    capture_segment.sensor_calibration = ntia_sensor.Calibration(
-                        **sensor_cal
-                    )
+                capture_segment.sensor_calibration = ntia_sensor.Calibration(
+                    **sensor_cal
+                )
             measurement_result["capture_segment"] = capture_segment
 
             self.create_metadata(measurement_result, recording_id)

scos_actions/actions/interfaces/action.py

@@ -61,15 +61,16 @@ def configure_sigan(self, params: dict):
 
     def configure_preselector(self, params: dict):
         preselector = self.sensor.preselector
-        if self.PRESELECTOR_PATH_KEY in params:
-            path = params[self.PRESELECTOR_PATH_KEY]
-            logger.debug(f"Setting preselector RF path: {path}")
-            preselector.set_state(path)
-        elif self.sensor.has_configurable_preselector:
-            # Require the RF path to be specified if the sensor has a preselector.
-            raise ParameterException(
-                f"No {self.PRESELECTOR_PATH_KEY} value specified in the YAML config."
-            )
+        if self.sensor.has_configurable_preselector:
+            if self.PRESELECTOR_PATH_KEY in params:
+                path = params[self.PRESELECTOR_PATH_KEY]
+                logger.debug(f"Setting preselector RF path: {path}")
+                preselector.set_state(path)
+            else:
+                # Require the RF path to be specified if the sensor has a preselector.
+                raise ParameterException(
+                    f"No {self.PRESELECTOR_PATH_KEY} value specified in the YAML config."
+                )
         else:
             # No preselector in use, so do not require an RF path
             pass

scos_actions/actions/interfaces/measurement_action.py

@@ -104,12 +104,6 @@ def create_metadata(
             self.sigmf_builder.set_classification(measurement_result["classification"])
         except KeyError:
             logger.warning(warning_str.format("classification"))
-        try:
-            self.sigmf_builder.set_last_calibration_time(
-                measurement_result["calibration_datetime"]
-            )
-        except KeyError:
-            logger.warning(warning_str.format("calibration_datetime"))
         try:
             cap = measurement_result["capture_segment"]
             logger.debug(f"Adding capture:{cap}")

scos_actions/actions/tests/test_sync_gps.py

@@ -30,9 +30,8 @@ def callback(sender, **kwargs):
 
     location_action_completed.connect(callback)
     action = test_actions["test_sync_gps"]
-    sensor = Sensor(
-        signal_analyzer=MockSignalAnalyzer(), capabilities={}, gps=MockGPS()
-    )
+    sigan = MockSignalAnalyzer()
+    sensor = Sensor(signal_analyzer=sigan, capabilities={}, gps=MockGPS(sigan))
     if sys.platform == "linux":
         action(sensor, SYNC_GPS, 1)
         assert _latitude

scos_actions/configs/actions/test_nasctn_sea_data_product.yml → scos_actions/configs/actions/test_SEA_CBRS_Measure_Baseline.yml

@@ -1,26 +1,22 @@
 nasctn_sea_data_product:
-  name: test_nasctn_sea_data_product
+  name: test_SEA_CBRS_Measure_Baseline
   rf_path: antenna
+  calibration_adjust: False
 # IIR filter settings
-  iir_apply: True
   iir_gpass_dB: 0.1  # Max passband ripple below unity gain
   iir_gstop_dB: 40  # Minimum stopband attenuation
   iir_pb_edge_Hz: 5e6  # Passband edge frequency
   iir_sb_edge_Hz: 5.008e6  # Stopband edge frequency
-# Mean/Max FFT settings
-  fft_size: 175
+# FFT settings
   nffts: 320e3
-  fft_window_type: flattop # See scipy.signal.get_window for supported input
 # PFP frame
   pfp_frame_period_ms: 10
 # APD downsampling settings
-  apd_bin_size_dB: 0.5  # Set to 0 or negative for no downsampling
-  apd_min_bin_dBm: -180
+  apd_bin_size_dB: 1.0  # Set to 0 or negative for no downsampling
   apd_max_bin_dBm: -30
+  apd_min_bin_dBm: -180
 # Time domain power statistics settings
   td_bin_size_ms: 10
-# Round all power results to X decimal places
-  round_to_places: 2
 # Sigan Settings
   preamp_enable: True
   reference_level: -25

scos_actions/discover/__init__.py

@@ -3,18 +3,10 @@
 from scos_actions.actions.monitor_sigan import MonitorSignalAnalyzer
 from scos_actions.actions.sync_gps import SyncGps
 from scos_actions.discover.yaml import load_from_yaml
-from scos_actions.settings import ACTION_DEFINITIONS_DIR
+from scos_actions.settings import ACTION_DEFINITIONS_DIR, SIGAN_CLASS, SIGAN_MODULE
 
-actions = {
-    "logger": Logger(),
-}
-test_actions = {
-    "test_sync_gps": SyncGps(parameters={"name": "test_sync_gps"}),
-    "test_monitor_sigan": MonitorSignalAnalyzer(
-        parameters={"name": "test_monitor_sigan"}
-    ),
-    "logger": Logger(),
-}
+actions = {"logger": Logger()}
+test_actions = {"logger": Logger()}
 
 
 def init(
@@ -33,4 +25,16 @@ def init(
 
 yaml_actions, yaml_test_actions = init()
 actions.update(yaml_actions)
-test_actions.update(yaml_test_actions)
+if (
+    SIGAN_MODULE == "scos_actions.hardware.mocks.mock_sigan"
+    and SIGAN_CLASS == "MockSignalAnalyzer"
+):
+    test_actions.update(
+        {
+            "test_sync_gps": SyncGps(parameters={"name": "test_sync_gps"}),
+            "test_monitor_sigan": MonitorSignalAnalyzer(
+                parameters={"name": "test_monitor_sigan"}
+            ),
+        }
+    )
+    test_actions.update(yaml_test_actions)

scos_actions/discover/tests/test_yaml.py

@@ -25,7 +25,7 @@
 """
 
 sigan = MockSignalAnalyzer()
-gps = MockGPS()
+gps = MockGPS(sigan)
 
 
 def test_load_from_yaml_existing():

scos_actions/hardware/mocks/mock_gps.py

@@ -7,6 +7,9 @@
 
 
 class MockGPS(GPSInterface):
+    def __init__(self, sigan):
+        self.sigan = sigan
+
     def get_location(timeout_s=1):
         logger.warning("Using mock GPS!")
         return 39.995118, -105.261572, 1651.0

scos_actions/hardware/mocks/mock_sigan.py

@@ -1,10 +1,12 @@
 """Mock a signal analyzer for testing."""
+
 import logging
 from collections import namedtuple
 from typing import Optional
 
 import numpy as np
 
+from scos_actions import __package__ as SCOS_ACTIONS_NAME
 from scos_actions import __version__ as SCOS_ACTIONS_VERSION
 from scos_actions.calibration.calibration import Calibration
 from scos_actions.hardware.sigan_iface import SignalAnalyzerInterface
@@ -64,6 +66,7 @@ def __init__(
         self._capture_time = None
         self._is_available = True
         self._plugin_version = SCOS_ACTIONS_VERSION
+        self._plugin_name = SCOS_ACTIONS_NAME
         self._firmware_version = "1.2.3"
         self._api_version = "v1.2.3"
 
@@ -84,6 +87,11 @@ def is_available(self):
     def plugin_version(self):
         return self._plugin_version
 
+    @property
+    def plugin_name(self) -> str:
+        """Returns the current package name of scos-actions."""
+        return self._plugin_name
+
     @property
     def firmware_version(self):
         return self._firmware_version
@@ -212,7 +220,7 @@ def update_calibration(self, params):
     def recompute_sensor_calibration_data(self, cal_args: list) -> None:
         if self.sensor_calibration is not None:
             self.sensor_calibration_data.update(
-                self._sensor_calibration.get_calibration_dict(cal_args)
+                self.sensor_calibration.get_calibration_dict(cal_args)
             )
         else:
             logger.warning("Sensor calibration does not exist.")

scos_actions/hardware/sigan_iface.py

@@ -56,6 +56,12 @@ def plugin_version(self) -> str:
         """Returns the version of the SCOS plugin defining this interface."""
         pass
 
+    @property
+    @abstractmethod
+    def plugin_name(self) -> str:
+        """Returns the name of the SCOS plugin defining this interface."""
+        pass
+
     @property
     def firmware_version(self) -> str:
         """Returns the version of the signal analyzer firmware."""

scos_actions/hardware/tests/test_sensor.py

@@ -7,9 +7,8 @@
 
 
 def test_sensor():
-    sensor = Sensor(
-        signal_analyzer=MockSignalAnalyzer(), capabilities={}, gps=MockGPS()
-    )
+    sigan = MockSignalAnalyzer()
+    sensor = Sensor(signal_analyzer=sigan, capabilities={}, gps=MockGPS(sigan))
     assert sensor is not None
     assert sensor.signal_analyzer is not None
     assert sensor.gps is not None
@@ -23,8 +22,9 @@ def test_set_get_preselector():
 
 
 def test_set_get_gps():
-    gps = MockGPS()
-    sensor = Sensor(signal_analyzer=MockSignalAnalyzer(), capabilities={})
+    sigan = MockSignalAnalyzer()
+    gps = MockGPS(sigan)
+    sensor = Sensor(signal_analyzer=sigan, capabilities={})
     sensor.gps = gps
     assert sensor.gps == gps
 

scos_actions/metadata/sigmf_builder.py

@@ -426,11 +426,6 @@ def add_annotation(self, start_index, length, annotation_md):
             start_index=start_index, length=length, metadata=annotation_md
         )
 
-    def set_last_calibration_time(self, last_cal_time):
-        self.sigmf_md.set_global_field(
-            "ntia-sensor:calibration_datetime", last_cal_time
-        )
-
     def add_to_global(self, key, value):
         self.sigmf_md.set_global_field(key, value)
 

scos_actions/settings.py

@@ -27,6 +27,10 @@
 logger.debug(f"scos-actions: RUNNING_TESTS:{RUNNING_TESTS}")
 FQDN = env("FQDN", None)
 logger.debug(f"scos-actions: FQDN:{FQDN}")
+SIGAN_MODULE = env.str("SIGAN_MODULE", default="scos_actions.hardware.mocks.mock_sigan")
+logger.debug(f"scos-actions: SIGAN_MODULE:{SIGAN_MODULE}")
+SIGAN_CLASS = env.str("SIGAN_CLASS", default="MockSignalAnalyzer")
+logger.debug(f"scos-actions: SIGAN_CLASS:{SIGAN_CLASS}")
 SIGAN_POWER_SWITCH = env("SIGAN_POWER_SWITCH", default=None)
 logger.debug(f"scos-actions: SIGAN_POWER_SWITCH:{SIGAN_POWER_SWITCH}")
 SIGAN_POWER_CYCLE_STATES = env("SIGAN_POWER_CYCLE_STATES", default=None)