[draft] handle differential calibration correction

.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

@@ -23,14 +23,14 @@
 import logging
 import lzma
 import platform
+import ray
 import sys
 from enum import EnumMeta
 from time import perf_counter
 from typing import Tuple
 
 import numpy as np
 import psutil
-import ray
 from environs import Env
 from its_preselector import __version__ as PRESELECTOR_API_VERSION
 from scipy.signal import sos2tf, sosfilt
@@ -109,7 +109,6 @@
 FFT_WINDOW = get_fft_window(FFT_WINDOW_TYPE, FFT_SIZE)
 FFT_WINDOW_ECF = get_fft_window_correction(FFT_WINDOW, "energy")
 IMPEDANCE_OHMS = 50.0
-DATA_REFERENCE_POINT = "noise source output"
 NUM_ACTORS = 3  # Number of ray actors to initialize
 
 # Create power detectors
@@ -422,7 +421,7 @@ def __init__(self, params: dict, iir_sos: np.ndarray):
         ]
         del params
 
-    def run(self, iqdata: np.ndarray) -> list:
+    def run(self, iqdata: np.ndarray):
         """
         Filter the input IQ data and concurrently compute FFT, PVT, PFP, and APD results.
 
@@ -449,6 +448,7 @@ class NasctnSeaDataProduct(Action):
     def __init__(self, parameters: dict):
         super().__init__(parameters)
         # Assume preselector is present
+        self.total_channel_data_length = None
         rf_path_name = utils.get_parameter(RF_PATH, self.parameters)
         self.rf_path = {self.PRESELECTOR_PATH_KEY: rf_path_name}
 
@@ -506,6 +506,7 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
         action_start_tic = perf_counter()
         # Ray should have already been initialized within scos-sensor,
         # but check and initialize just in case.
+
         if not ray.is_initialized():
             logger.info("Initializing ray.")
             logger.info("Set RAY_INIT=true to avoid initializing within " + __name__)
@@ -525,8 +526,6 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
             self.iteration_params,
         )
         self.create_global_sensor_metadata(self.sensor)
-        self.create_global_data_product_metadata()
-
         # Initialize remote supervisor actors for IQ processing
         tic = perf_counter()
         # This uses iteration_params[0] because
@@ -538,10 +537,15 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
         logger.debug(f"Spawned {NUM_ACTORS} supervisor actors in {toc-tic:.2f} s")
 
         # Collect all IQ data and spawn data product computation processes
-        dp_procs, cpu_speed = [], []
+        dp_procs, cpu_speed, reference_points = [], [], []
         capture_tic = perf_counter()
+
         for i, parameters in enumerate(self.iteration_params):
             measurement_result = self.capture_iq(parameters)
+            if i == 0:
+                self.create_global_data_product_metadata(
+                    measurement_result["reference"]
+                )
             # Start data product processing but do not block next IQ capture
             tic = perf_counter()
 
@@ -552,16 +556,22 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
             toc = perf_counter()
             logger.debug(f"IQ data delivered for processing in {toc-tic:.2f} s")
             # Create capture segment with channel-specific metadata before sigan is reconfigured
-            tic = perf_counter()
             self.create_capture_segment(i, measurement_result)
-            toc = perf_counter()
-            logger.debug(f"Created capture metadata in {toc-tic:.2f} s")
+            # Query CPU speed for later averaging in diagnostics metadata
             cpu_speed.append(get_current_cpu_clock_speed())
+            # Append list of data reference points; later we require these to be identical
+            reference_points.append(measurement_result["reference"])
         capture_toc = perf_counter()
         logger.debug(
             f"Collected all IQ data and started all processing in {capture_toc-capture_tic:.2f} s"
         )
 
+        # Create data product metadata: requires all data reference points
+        # to be identical.
+        assert (
+            len(set(reference_points)) == 1
+        ), "Channel data were scaled to different reference points. Cannot build metadata."
+
         # Collect processed data product results
         all_data, max_max_ch_pwrs, med_mean_ch_pwrs, mean_ch_pwrs, median_ch_pwrs = (
             [],
@@ -630,14 +640,11 @@ 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
-        measurement_result = self.sensor.signal_analyzer.acquire_time_domain_samples(
-            num_samples, nskip
+        measurement_result = self.sensor.acquire_time_domain_samples(
+            num_samples, nskip, cal_params=params
         )
         # Store some metadata with the IQ
         measurement_result.update(params)
-        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."
@@ -978,7 +985,7 @@ def test_required_components(self):
             trigger_api_restart.send(sender=self.__class__)
         return None
 
-    def create_global_data_product_metadata(self) -> None:
+    def create_global_data_product_metadata(self, data_products_reference: str) -> None:
         p = self.parameters
         num_iq_samples = int(p[SAMPLE_RATE] * p[DURATION_MS] * 1e-3)
         iir_obj = ntia_algorithm.DigitalFilter(
@@ -1023,7 +1030,7 @@ def create_global_data_product_metadata(self) -> None:
             x_step=[p[SAMPLE_RATE] / FFT_SIZE],
             y_units="dBm/Hz",
             processing=[dft_obj.id],
-            reference=DATA_REFERENCE_POINT,
+            reference=data_products_reference,
             description=(
                 "Results of statistical detectors (max, mean, median, 25th_percentile, 75th_percentile, "
                 + "90th_percentile, 95th_percentile, 99th_percentile, 99.9th_percentile, 99.99th_percentile) "
@@ -1043,7 +1050,7 @@ def create_global_data_product_metadata(self) -> None:
             x_stop=[pvt_x_axis__s[-1]],
             x_step=[pvt_x_axis__s[1] - pvt_x_axis__s[0]],
             y_units="dBm",
-            reference=DATA_REFERENCE_POINT,
+            reference=data_products_reference,
             description=(
                 "Max- and mean-detected channel power vs. time, with "
                 + f"an integration time of {p[TD_BIN_SIZE_MS]} ms. "
@@ -1070,7 +1077,7 @@ def create_global_data_product_metadata(self) -> None:
             x_stop=[pfp_x_axis__s[-1]],
             x_step=[pfp_x_axis__s[1] - pfp_x_axis__s[0]],
             y_units="dBm",
-            reference=DATA_REFERENCE_POINT,
+            reference=data_products_reference,
             description=(
                 "Channelized periodic frame power statistics reported over"
                 + f" a {p[PFP_FRAME_PERIOD_MS]} ms frame period, with frame resolution"
@@ -1093,6 +1100,7 @@ def create_global_data_product_metadata(self) -> None:
             y_start=[apd_y_axis__dBm[0]],
             y_stop=[apd_y_axis__dBm[-1]],
             y_step=[apd_y_axis__dBm[1] - apd_y_axis__dBm[0]],
+            reference=data_products_reference,
             description=(
                 f"Estimate of the APD, using a {p[APD_BIN_SIZE_DB]} dB "
                 + "bin size for amplitude values. The data payload includes"
@@ -1111,6 +1119,7 @@ def create_global_data_product_metadata(self) -> None:
             + pfp_length * len(PFP_M3_DETECTOR) * 2
             + apd_graph.length
         )
+        logger.debug(f"Total channel length:{self.total_channel_data_length}")
 
     def create_capture_segment(
         self,
@@ -1126,11 +1135,15 @@ def create_capture_segment(
             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,
+                datetime=self.sensor.sensor_calibration_data["datetime"],
+                gain=round(measurement_result["applied_calibration"]["gain"], 3),
+                noise_figure=round(
+                    measurement_result["applied_calibration"]["noise_figure"], 3
+                ),
+                temperature=round(
+                    self.sensor.sensor_calibration_data["temperature"], 1
+                ),
+                reference=measurement_result["reference"],
             ),
             sigan_settings=ntia_sensor.SiganSettings(
                 reference_level=self.sensor.signal_analyzer.reference_level,

scos_actions/actions/acquire_single_freq_fft.py

@@ -91,7 +91,6 @@
 from numpy import float32, ndarray
 
 from scos_actions.actions.interfaces.measurement_action import MeasurementAction
-from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.metadata.structs import ntia_algorithm
 from scos_actions.signal_processing.fft import (
     get_fft,
@@ -153,10 +152,6 @@ def __init__(self, parameters: dict):
         self.classification = get_parameter(CLASSIFICATION, self.parameters)
         self.cal_adjust = get_parameter(CAL_ADJUST, self.parameters)
         assert isinstance(self.cal_adjust, bool)
-        if self.cal_adjust:
-            self.data_reference = "calibration terminal"
-        else:
-            self.data_reference = "signal analyzer input"
         # FFT setup
         self.fft_detector = create_statistical_detector(
             "M4sDetector", ["min", "max", "mean", "median", "sample"]
@@ -169,7 +164,7 @@ def __init__(self, parameters: dict):
     def execute(self, schedule_entry: dict, task_id: int) -> dict:
         # Acquire IQ data and generate M4S result
         measurement_result = self.acquire_data(
-            self.num_samples, self.nskip, self.cal_adjust
+            self.num_samples, self.nskip, self.cal_adjust, cal_params=self.parameters
         )
         # Actual sample rate may differ from configured value
         sample_rate_Hz = measurement_result["sample_rate"]
@@ -178,9 +173,9 @@ 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["calibration_datetime"] = (
+        #     self.sensor.sensor_calibration_data["datetime"]
+        # )
         measurement_result["task_id"] = task_id
         measurement_result["classification"] = self.classification
 
@@ -270,7 +265,7 @@ def create_metadata(self, measurement_result: dict, recording: int = None) -> No
             x_stop=[frequencies[-1]],
             x_step=[frequencies[1] - frequencies[0]],
             y_units="dBm",
-            reference=self.data_reference,
+            reference=measurement_result["reference"],
             description=(
                 "Results of min, max, mean, and median statistical detectors, "
                 + f"along with a random sampling, from a set of {self.nffts} "

scos_actions/actions/acquire_single_freq_tdomain_iq.py

@@ -37,7 +37,6 @@
 
 from scos_actions import utils
 from scos_actions.actions.interfaces.measurement_action import MeasurementAction
-from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.utils import get_parameter
 
 logger = logging.getLogger(__name__)
@@ -84,14 +83,16 @@ def execute(self, schedule_entry: dict, task_id: int) -> dict:
         # Use the sigan's actual reported instead of requested sample rate
         sample_rate = self.sensor.signal_analyzer.sample_rate
         num_samples = int(sample_rate * self.duration_ms * 1e-3)
-        measurement_result = self.acquire_data(num_samples, self.nskip, self.cal_adjust)
+        measurement_result = self.acquire_data(
+            num_samples, self.nskip, self.cal_adjust, cal_params=self.parameters
+        )
         end_time = utils.get_datetime_str_now()
         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["calibration_datetime"] = (
+        #     self.sensor.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

@@ -100,7 +100,9 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
             cal_adjust = get_parameter(CAL_ADJUST, measurement_params)
             sample_rate = self.sensor.signal_analyzer.sample_rate
             num_samples = int(sample_rate * duration_ms * 1e-3)
-            measurement_result = super().acquire_data(num_samples, nskip, cal_adjust)
+            measurement_result = super().acquire_data(
+                num_samples, nskip, cal_adjust, cal_params=measurement_params
+            )
             measurement_result.update(measurement_params)
             end_time = utils.get_datetime_str_now()
             measurement_result["end_time"] = end_time
@@ -117,22 +119,18 @@ def __call__(self, sensor: Sensor, schedule_entry: dict, task_id: int):
                 overload=measurement_result["overload"],
                 sigan_settings=sigan_settings,
             )
-            sigan_cal = self.sensor.signal_analyzer.sigan_calibration_data
-            sensor_cal = self.sensor.signal_analyzer.sensor_calibration_data
-            if sigan_cal is not None:
-                if "1db_compression_point" in sigan_cal:
-                    sigan_cal["compression_point"] = sigan_cal.pop(
-                        "1db_compression_point"
-                    )
-                capture_segment.sigan_calibration = ntia_sensor.Calibration(**sigan_cal)
+            sensor_cal = self.sensor.sensor_calibration_data
             if sensor_cal is not None:
                 if "1db_compression_point" in sensor_cal:
                     sensor_cal["compression_point"] = sensor_cal.pop(
                         "1db_compression_point"
                     )
-                    capture_segment.sensor_calibration = ntia_sensor.Calibration(
-                        **sensor_cal
-                    )
+                if "reference" not in sensor_cal:
+                    # If the calibration data already includes this, don't overwrite
+                    sensor_cal["reference"] = measurement_result["reference"]
+                capture_segment.sensor_calibration = ntia_sensor.Calibration(
+                    **sensor_cal
+                )
             measurement_result["capture_segment"] = capture_segment
 
             self.create_metadata(measurement_result, recording_id)