Merge pull request #106 from NTIA/empty_calibration_params

Move sigan agnostic default calibration file to scos-sensor and enable calibrations with default cal file.

scos_actions/__init__.py

@@ -1,2 +1 @@
 __version__ = "7.1.0"
-

scos_actions/actions/calibrate_y_factor.py

@@ -78,7 +78,7 @@
 
 from scos_actions import utils
 from scos_actions.actions.interfaces.action import Action
-from scos_actions.calibration import sensor_calibration
+from scos_actions.calibration import sensor_calibration, default_sensor_calibration
 from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.hardware.sigan_iface import SIGAN_SETTINGS_KEYS
 from scos_actions.settings import SENSOR_CALIBRATION_FILE
@@ -253,7 +253,7 @@ def calibrate(self, params):
         assert (
             sample_rate == noise_off_measurement_result["sample_rate"]
         ), "Sample rate mismatch"
-
+        sigan_params = {k: v for k, v in params.items() if k in SIGAN_SETTINGS_KEYS}
         # Apply IIR filtering to both captures if configured
         if self.iir_apply:
             # Estimate of IIR filter ENBW does NOT account for passband ripple in sensor transfer function!
@@ -262,9 +262,13 @@ def calibrate(self, params):
             noise_on_data = sosfilt(self.iir_sos, noise_on_measurement_result["data"])
             noise_off_data = sosfilt(self.iir_sos, noise_off_measurement_result["data"])
         else:
+            if default_sensor_calibration:
+                raise Exception(
+                    "Calibrations without IIR filter cannot be performed with default calibration."
+                )
+
             logger.debug("Skipping IIR filtering")
             # Get ENBW from sensor calibration
-            sigan_params = {k: v for k, v in params.items() if k in SIGAN_SETTINGS_KEYS}
             assert set(sensor_calibration.calibration_parameters) <= set(
                 sigan_params.keys()
             ), f"Action parameters do not include all required calibration parameters"
@@ -289,7 +293,7 @@ def calibrate(self, params):
 
         # Update sensor calibration with results
         sensor_calibration.update(
-            params,
+            sigan_params,
             utils.get_datetime_str_now(),
             gain,
             noise_figure,

scos_actions/calibration/__init__.py

@@ -1,13 +1,17 @@
 import logging
-from pathlib import Path
+from os import path
 
 from scos_actions.calibration.calibration import Calibration, load_from_json
-from scos_actions.settings import SENSOR_CALIBRATION_FILE, SIGAN_CALIBRATION_FILE
+from scos_actions.settings import (
+    DEFAULT_CALIBRATION_FILE,
+    SENSOR_CALIBRATION_FILE,
+    SIGAN_CALIBRATION_FILE,
+)
 
 logger = logging.getLogger(__name__)
 
 
-def get_sigan_calibration(sigan_cal_file: Path) -> Calibration:
+def get_sigan_calibration(sigan_cal_file: str) -> Calibration:
     """
     Load signal analyzer calibration data from file.
 
@@ -23,7 +27,7 @@ def get_sigan_calibration(sigan_cal_file: Path) -> Calibration:
     return sigan_cal
 
 
-def get_sensor_calibration(sensor_cal_file: Path) -> Calibration:
+def get_sensor_calibration(sensor_cal_file: str) -> Calibration:
     """
     Load sensor calibration data from file.
 
@@ -39,9 +43,48 @@ def get_sensor_calibration(sensor_cal_file: Path) -> Calibration:
     return sensor_cal
 
 
-logger.debug(f"Loading sensor cal file: {SENSOR_CALIBRATION_FILE}")
-sensor_calibration = get_sensor_calibration(SENSOR_CALIBRATION_FILE)
-logger.debug(f"Loading sigan cal file: {SIGAN_CALIBRATION_FILE}")
-sigan_calibration = get_sigan_calibration(SIGAN_CALIBRATION_FILE)
+def check_for_default_calibration(cal_file_path: str, cal_type: str) -> bool:
+    default_cal = False
+    if cal_file_path == DEFAULT_CALIBRATION_FILE:
+        default_cal = True
+        logger.warning(
+            f"***************LOADING DEFAULT {cal_type} CALIBRATION***************"
+        )
+    return default_cal
+
+
+sensor_calibration = None
+if SENSOR_CALIBRATION_FILE is None or SENSOR_CALIBRATION_FILE == "":
+    logger.warning(
+        "No sensor calibration file specified. Not loading calibration file."
+    )
+elif not path.exists(SENSOR_CALIBRATION_FILE):
+    logger.warning(
+        SENSOR_CALIBRATION_FILE
+        + " does not exist. Not loading sensor calibration file."
+    )
+else:
+    logger.debug(f"Loading sensor cal file: {SENSOR_CALIBRATION_FILE}")
+    default_sensor_calibration = check_for_default_calibration(
+        SENSOR_CALIBRATION_FILE, "Sensor"
+    )
+    sensor_calibration = get_sensor_calibration(SENSOR_CALIBRATION_FILE)
+
+sigan_calibration = None
+default_sensor_calibration = False
+default_sigan_calibration = False
+if SIGAN_CALIBRATION_FILE is None or SIGAN_CALIBRATION_FILE == "":
+    logger.warning("No sigan calibration file specified. Not loading calibration file.")
+elif not path.exists(SIGAN_CALIBRATION_FILE):
+    logger.warning(
+        SIGAN_CALIBRATION_FILE + " does not exist. Not loading sigan calibration file."
+    )
+else:
+    logger.debug(f"Loading sigan cal file: {SIGAN_CALIBRATION_FILE}")
+    default_sigan_calibration = check_for_default_calibration(
+        SIGAN_CALIBRATION_FILE, "Sigan"
+    )
+    sigan_calibration = get_sigan_calibration(SIGAN_CALIBRATION_FILE)
+
 if sensor_calibration:
     logger.debug(f"Last sensor cal: {sensor_calibration.last_calibration_datetime}")

scos_actions/calibration/calibration.py

@@ -38,14 +38,14 @@ def get_calibration_dict(self, cal_params: List[Union[float, int, bool]]) -> dic
             then the input to this method could be ``["15360000.0", "40"]``.
         :return: The calibration data corresponding to the input parameter values.
         """
-        # Check if the sample rate was calibrated
+
         cal_data = self.calibration_data
-        # raise Exception(self)
         for i, setting_value in enumerate(cal_params):
             setting = self.calibration_parameters[i]
             logger.debug(f"Looking up calibration for {setting} at {setting_value}")
             cal_data = filter_by_parameter(cal_data, setting_value)
         logger.debug(f"Got calibration data: {cal_data}")
+
         return cal_data
 
     def update(
@@ -77,9 +77,10 @@ def update(
         """
         cal_data = self.calibration_data
         self.last_calibration_datetime = calibration_datetime_str
-
+        if len(self.calibration_parameters) == 0:
+            self.calibration_parameters = list(params.keys())
         # Ensure all required calibration parameters were used
-        if not set(params.keys()) >= set(self.calibration_parameters):
+        elif not set(params.keys()) >= set(self.calibration_parameters):
             raise Exception(
                 "Not enough parameters specified to update calibration.\n"
                 + f"Required parameters are {self.calibration_parameters}"

scos_actions/calibration/tests/test_calibration.py

@@ -310,12 +310,6 @@ def test_update(self):
         assert cal_from_file.calibration_data["100.0"]["200.0"]["gain"] == 30.0
         assert cal_from_file.calibration_data["100.0"]["200.0"]["noise_figure"] == 5.0
 
-    def test_default_sensor_cal(self):
-        assert sensor_calibration is not None
-
-    def test_default_sigan_cal_location(self):
-        assert sigan_calibration is not None
-
     def test_filter_by_paramter_integer(self):
         calibrations = {"200.0": {"some_cal_data"}, 300.0: {"more cal data"}}
         filtered_data = filter_by_parameter(calibrations, 200)