Sea 178 test actions loading

scos_actions/__init__.py

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

scos_actions/actions/acquire_sea_data_product.py

@@ -103,6 +103,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
@@ -161,6 +162,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:
         """
@@ -643,9 +645,16 @@ def capture_iq(self, params: dict) -> dict:
         duration_ms = utils.get_parameter(DURATION_MS, params)
         nskip = utils.get_parameter(NUM_SKIP, params)
         num_samples = int(params[SAMPLE_RATE] * duration_ms * 1e-3)
+        if logger.getEffectiveLevel() == logging.DEBUG:
+            for key, value in params.items():
+                logger.debug(f"param {key}={value}")
+        self.cal_adjust = True
+        if CAL_ADJUST in params:
+            self.cal_adjust = utils.get_parameter(CAL_ADJUST, params)
+        logger.debug(f"cal_adjust={self.cal_adjust}")
         # Collect IQ data
         measurement_result = self.sensor.acquire_time_domain_samples(
-            num_samples, nskip, cal_params=params
+            num_samples, nskip, cal_params=params, cal_adjust=self.cal_adjust
         )
         # Store some metadata with the IQ
         measurement_result.update(params)
@@ -802,7 +811,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,
@@ -1143,7 +1153,14 @@ def create_capture_segment(
             datetime=measurement_result["capture_time"],
             duration=measurement_result[DURATION_MS],
             overload=measurement_result["overload"],
-            sensor_calibration=ntia_sensor.Calibration(
+            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=self.sensor.sensor_calibration_data["datetime"],
                 gain=round(measurement_result["applied_calibration"]["gain"], 3),
                 noise_figure=round(
@@ -1153,17 +1170,11 @@ def create_capture_segment(
                     self.sensor.sensor_calibration_data["temperature"], 1
                 ),
                 reference=measurement_result["reference"],
-            ),
-            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 "compression_point" in measurement_result["applied_calibration"]:
-            capture_segment.sensor_calibration.compression_point = measurement_result[
-                "applied_calibration"
-            ]["compression_point"]
+            )
+            if "compression_point" in measurement_result["applied_calibration"]:
+                capture_segment.sensor_calibration.compression_point = (
+                    measurement_result["applied_calibration"]["compression_point"]
+                )
         self.sigmf_builder.add_capture(capture_segment)
 
     def get_sigmf_builder(

scos_actions/actions/calibrate_y_factor.py

@@ -231,14 +231,12 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
             logger.debug(f"cal_params: {cal_params}")
             cal_data = dict()
             last_cal_datetime = get_datetime_str_now()
-            clock_rate_lookup_by_sample_rate = []
             self.sensor.sensor_calibration = SensorCalibration(
                 calibration_parameters=cal_params,
                 calibration_data=cal_data,
                 calibration_reference=onboard_cal_reference,
                 file_path=Path(env("ONBOARD_CALIBRATION_FILE")),
                 last_calibration_datetime=last_cal_datetime,
-                clock_rate_lookup_by_sample_rate=clock_rate_lookup_by_sample_rate,
                 sensor_uid=sensor_uid,
             )
         elif self.sensor.sensor_calibration.file_path == env(

scos_actions/actions/interfaces/action.py

@@ -59,15 +59,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

@@ -67,15 +67,16 @@ def get_calibration(self, measurement_result: dict) -> ntia_sensor.Calibration:
                 noise_figure=round(
                     measurement_result["applied_calibration"]["noise_figure"], 3
                 ),
-                temperature=round(
-                    self.sensor.sensor_calibration_data["temperature"], 1
-                ),
                 reference=measurement_result["reference"],
             )
             if "compression_point" in measurement_result["applied_calibration"]:
                 cal_meta.compression_point = measurement_result["applied_calibration"][
                     "compression_point"
                 ]
+            if "temperature" in self.sensor.sensor_calibration_data:
+                cal_meta.temperature = round(
+                    self.sensor.sensor_calibration_data["temperature"], 1
+                )
         return cal_meta
 
     def create_metadata(

scos_actions/actions/sync_gps.py

@@ -19,12 +19,12 @@ def __init__(self, parameters: dict):
     def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
         logger.debug("Syncing to GPS")
         self.sensor = sensor
-        dt = self.sensor.gps.get_gps_time()
+        dt = self.sensor.gps.get_gps_time(self.sensor)
         date_cmd = ["date", "-s", "{:}".format(dt.strftime("%Y/%m/%d %H:%M:%S"))]
         subprocess.check_output(date_cmd, shell=True)
         logger.info(f"Set system time to GPS time {dt.ctime()}")
 
-        location = sensor.gps.get_location()
+        location = sensor.gps.get_location(self.sensor)
         if location is None:
             raise RuntimeError("Unable to synchronize to GPS")
 

scos_actions/calibration/sensor_calibration.py

@@ -25,16 +25,8 @@ class provides an implementation for the update method to allow calibration
     """
 
     last_calibration_datetime: str
-    clock_rate_lookup_by_sample_rate: List[Dict[str, float]]
     sensor_uid: str
 
-    def get_clock_rate(self, sample_rate: Union[float, int]) -> Union[float, int]:
-        """Find the clock rate (Hz) using the given sample_rate (samples per second)"""
-        for mapping in self.clock_rate_lookup_by_sample_rate:
-            if mapping["sample_rate"] == sample_rate:
-                return mapping["clock_frequency"]
-        return sample_rate
-
     def update(
         self,
         params: dict,

scos_actions/calibration/tests/test_calibration.py

@@ -103,7 +103,6 @@ def test_to_and_from_json(self, tmp_path: Path):
             "testing",
             tmp_path / "testing.json",
             "dt_str",
-            [],
             "uid",
         )
         sensor_cal.to_json()

scos_actions/calibration/tests/test_sensor_calibration.py

@@ -83,17 +83,6 @@ def setup_calibration_file(self, tmp_path: Path):
         cal_data["sensor_uid"] = "SAMPLE_CALIBRATION"
         cal_data["calibration_reference"] = "TESTING"
 
-        # Add SR/CF lookup table
-        cal_data["clock_rate_lookup_by_sample_rate"] = []
-        for sr in self.sample_rates:
-            cr = sr
-            while cr <= 40e6:
-                cr *= 2
-            cr /= 2
-            cal_data["clock_rate_lookup_by_sample_rate"].append(
-                {"sample_rate": int(sr), "clock_frequency": int(cr)}
-            )
-
         # Create the JSON architecture for the calibration data
         cal_data["calibration_data"] = {}
         cal_data["calibration_parameters"] = ["sample_rate", "frequency", "gain"]
@@ -151,7 +140,6 @@ def test_sensor_calibration_dataclass_fields(self):
         # Note: does not check field order
         assert fields == {
             "last_calibration_datetime": str,
-            "clock_rate_lookup_by_sample_rate": List[Dict[str, float]],
             "sensor_uid": str,
         }
 
@@ -167,13 +155,6 @@ def test_field_validator(self):
                 [], {}, False, Path(""), datetime.datetime.now(), [], "uid"
             )
 
-    def test_get_clock_rate(self):
-        """Test the get_clock_rate method"""
-        # Test getting a clock rate by sample rate
-        assert self.sample_cal.get_clock_rate(10e6) == 40e6
-        # If there isn't an entry, the sample rate should be returned
-        assert self.sample_cal.get_clock_rate(-999) == -999
-
     def test_get_calibration_dict_exact_match_lookup(self):
         calibration_datetime = get_datetime_str_now()
         calibration_params = ["sample_rate", "frequency"]
@@ -187,7 +168,6 @@ def test_get_calibration_dict_exact_match_lookup(self):
             calibration_reference="testing",
             file_path=Path(""),
             last_calibration_datetime=calibration_datetime,
-            clock_rate_lookup_by_sample_rate=[],
             sensor_uid="TESTING",
         )
         cal_data = cal.get_calibration_dict({"sample_rate": 100.0, "frequency": 200.0})
@@ -206,7 +186,6 @@ def test_get_calibration_dict_within_range(self):
             calibration_reference="testing",
             file_path=Path("test_calibration.json"),
             last_calibration_datetime=calibration_datetime,
-            clock_rate_lookup_by_sample_rate=[],
             sensor_uid="TESTING",
         )
 
@@ -234,7 +213,6 @@ def test_update(self):
             calibration_reference="testing",
             file_path=test_cal_path,
             last_calibration_datetime=calibration_datetime,
-            clock_rate_lookup_by_sample_rate=[],
             sensor_uid="TESTING",
         )
         action_params = {"sample_rate": 100.0, "frequency": 200.0}

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/configs/actions/test_multi_frequency_iq_action.yml

@@ -17,3 +17,4 @@ stepped_frequency_time_domain_iq:
   nskip: 15.36e4
   calibration_adjust: False
   classification: UNCLASSIFIED
+  rf_path: antenna

scos_actions/configs/actions/test_single_frequency_iq_action.yml

@@ -7,3 +7,4 @@ single_frequency_time_domain_iq:
   nskip: 15.36e4
   calibration_adjust: False
   classification: UNCLASSIFIED
+  rf_path: antenna

scos_actions/configs/actions/test_single_frequency_m4s_action.yml

@@ -8,3 +8,4 @@ single_frequency_fft:
   nskip: 15.36e4
   calibration_adjust: False
   classification: UNCLASSIFIED
+  rf_path: antenna

scos_actions/configs/actions/test_survey_iq_action.yml

@@ -27,3 +27,4 @@ stepped_frequency_time_domain_iq:
     - 10000
   nskip: 15.36e4
   calibration_adjust: False
+  rf_path: antenna

scos_actions/discover/__init__.py

@@ -1,20 +1,11 @@
 from scos_actions.actions import action_classes
-from scos_actions.actions.logger import Logger
 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 = {}
+test_actions = {}
 
 
 def init(
@@ -31,6 +22,18 @@ def init(
     return yaml_actions, yaml_test_actions
 
 
-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"
+):
+    yaml_actions, yaml_test_actions = init()
+    actions.update(yaml_actions)
+    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/hardware/gps_iface.py

@@ -3,9 +3,11 @@
 
 class GPSInterface(ABC):
     @abstractmethod
-    def get_location(self, timeout_s=1):
+    def get_location(
+        self, sensor: "scos_actions.hardware.sensor.Sensor", timeout_s: float = 1
+    ):
         pass
 
     @abstractmethod
-    def get_gps_time(self):
+    def get_gps_time(self, sensor: "scos_actions.hardware.sensor.Sensor"):
         pass