Generalize and Improve Annotation Handling

scos_actions/actions/acquire_single_freq_fft.py

@@ -92,9 +92,7 @@
 
 from scos_actions.actions.interfaces.measurement_action import MeasurementAction
 from scos_actions.hardware import gps as mock_gps
-from scos_actions.metadata.annotations.fft_annotation import (
-    FrequencyDomainDetectionAnnotation,
-)
+from scos_actions.metadata.annotations import FrequencyDomainDetection
 from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder
 from scos_actions.signal_processing.fft import (
     get_fft,
@@ -190,6 +188,7 @@ def execute(self, schedule_entry, task_id) -> dict:
         measurement_result["measurement_type"] = MeasurementType.SINGLE_FREQUENCY.value
         measurement_result["sigan_cal"] = self.sigan.sigan_calibration_data
         measurement_result["sensor_cal"] = self.sigan.sensor_calibration_data
+        measurement_result["classification"] = "UNCLASSIFIED"
         return measurement_result
 
     def apply_m4s(self, measurement_result: dict) -> ndarray:
@@ -242,8 +241,19 @@ def description(self):
     def get_sigmf_builder(self, measurement_result) -> SigMFBuilder:
         sigmf_builder = super().get_sigmf_builder(measurement_result)
         for i, detector in enumerate(self.fft_detector):
-            fft_annotation = FrequencyDomainDetectionAnnotation(
-                detector.value, i * self.fft_size, self.fft_size
+            fft_annotation = FrequencyDomainDetection(
+                sample_start=i * self.fft_size,
+                sample_count=self.fft_size,
+                detector=detector.value,
+                number_of_ffts=self.nffts,
+                number_of_samples_in_fft=self.fft_size,
+                window=self.fft_window_type,
+                equivalent_noise_bandwidth=measurement_result["enbw"],
+                units="dBm",
+                reference="preselector input",
+                frequency_start=measurement_result["frequency_start"],
+                frequency_stop=measurement_result["frequency_stop"],
+                frequency_step=measurement_result["frequency_step"],
             )
             sigmf_builder.add_metadata_generator(
                 type(fft_annotation).__name__ + "_" + detector.value, fft_annotation

scos_actions/actions/acquire_single_freq_tdomain_iq.py

@@ -38,9 +38,7 @@
 from scos_actions import utils
 from scos_actions.actions.interfaces.measurement_action import MeasurementAction
 from scos_actions.hardware import gps as mock_gps
-from scos_actions.metadata.annotations.time_domain_annotation import (
-    TimeDomainAnnotation,
-)
+from scos_actions.metadata.annotations import TimeDomainDetection
 from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder
 from scos_actions.utils import get_parameter
 
@@ -99,11 +97,19 @@ def execute(self, schedule_entry, task_id) -> dict:
         measurement_result["description"] = self.description
         measurement_result["sigan_cal"] = self.sigan.sigan_calibration_data
         measurement_result["sensor_cal"] = self.sigan.sensor_calibration_data
+        measurement_result["classification"] = "UNCLASSIFIED"
         return measurement_result
 
     def get_sigmf_builder(self, measurement_result: dict) -> SigMFBuilder:
         sigmf_builder = super().get_sigmf_builder(measurement_result)
-        time_domain_annotation = TimeDomainAnnotation(0, self.received_samples)
+        time_domain_annotation = TimeDomainDetection(
+            sample_start=0,
+            sample_count=self.received_samples,
+            detector="sample_iq",
+            number_of_samples=self.received_samples,
+            units="volts",
+            reference="preselector input",
+        )
         sigmf_builder.add_metadata_generator(
             type(time_domain_annotation).__name__, time_domain_annotation
         )

scos_actions/actions/acquire_stepped_freq_tdomain_iq.py

@@ -115,6 +115,7 @@ def __call__(self, schedule_entry_json, task_id):
             measurement_result["name"] = self.name
             measurement_result["sigan_cal"] = self.sigan.sigan_calibration_data
             measurement_result["sensor_cal"] = self.sigan.sensor_calibration_data
+            measurement_result["classification"] = "UNCLASSIFIED"
             sigmf_builder = self.get_sigmf_builder(measurement_result)
             self.create_metadata(
                 sigmf_builder, schedule_entry_json, measurement_result, recording_id

scos_actions/actions/interfaces/measurement_action.py

@@ -4,10 +4,7 @@
 from scos_actions.actions.interfaces.action import Action
 from scos_actions.actions.interfaces.signals import measurement_action_completed
 from scos_actions.hardware import gps as mock_gps
-from scos_actions.metadata.annotations.calibration_annotation import (
-    CalibrationAnnotation,
-)
-from scos_actions.metadata.annotations.sensor_annotation import SensorAnnotation
+from scos_actions.metadata.annotations import CalibrationAnnotation, SensorAnnotation
 from scos_actions.metadata.measurement_global import MeasurementMetadata
 from scos_actions.metadata.sigmf_builder import SigMFBuilder
 
@@ -38,15 +35,50 @@ def __call__(self, schedule_entry, task_id):
     def get_sigmf_builder(self, measurement_result) -> SigMFBuilder:
         sigmf_builder = SigMFBuilder()
         self.received_samples = len(measurement_result["data"].flatten())
-        calibration_annotation = CalibrationAnnotation(0, self.received_samples)
+        calibration_annotation = CalibrationAnnotation(
+            sample_start=0,
+            sample_count=self.received_samples,
+            sigan_cal=measurement_result["sigan_cal"],
+            sensor_cal=measurement_result["sensor_cal"],
+        )
         sigmf_builder.add_metadata_generator(
             type(calibration_annotation).__name__, calibration_annotation
         )
-        measurement_metadata = MeasurementMetadata()
+        f_low, f_high = None, None
+        if "frequency_low" in measurement_result:
+            f_low = measurement_result["frequency_low"]
+        elif "frequency" in measurement_result:
+            f_low = measurement_result["frequency"]
+            f_high = measurement_result["frequency"]
+        if "frequency_high" in measurement_result:
+            f_high = measurement_result["frequency_high"]
+
+        measurement_metadata = MeasurementMetadata(
+            domain=measurement_result["domain"],
+            measurement_type=measurement_result["measurement_type"],
+            time_start=measurement_result["start_time"],
+            time_stop=measurement_result["end_time"],
+            frequency_tuned_low=f_low,
+            frequency_tuned_high=f_high,
+            classification=measurement_result["classification"],
+        )
         sigmf_builder.add_metadata_generator(
             type(measurement_metadata).__name__, measurement_metadata
         )
-        sensor_annotation = SensorAnnotation(0, self.received_samples)
+
+        sensor_annotation = SensorAnnotation(
+            sample_start=0,
+            sample_count=self.received_samples,
+            overload=measurement_result["overload"]
+            if "overload" in measurement_result
+            else None,
+            attenuation_setting_sigan=measurement_result["attenuation"]
+            if "attenuation" in measurement_result
+            else None,
+            gain_setting_sigan=measurement_result["gain"]
+            if "gain" in measurement_result
+            else None,
+        )
         sigmf_builder.add_metadata_generator(
             type(sensor_annotation).__name__, sensor_annotation
         )
@@ -63,7 +95,7 @@ def create_metadata(
             self.is_complex(),
         )
         sigmf_builder.add_sigmf_capture(sigmf_builder, measurement_result)
-        sigmf_builder.build(measurement_result)
+        sigmf_builder.build()
 
     def test_required_components(self):
         """Fail acquisition if a required component is not available."""

scos_actions/metadata/annotation_segment.py

@@ -0,0 +1,76 @@
+from abc import ABC
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from scos_actions.metadata.sigmf_builder import SigMFBuilder
+
+
+@dataclass
+class AnnotationSegment(ABC):
+    """
+    Interface for generating SigMF annotation segments.
+
+    The only required parameter is ``sample_start``. Refer to the SigMF
+    documentation for more information.
+
+    :param sample_start: The sample index at which the segment takes effect.
+    :param sample_count: The number of samples to which the segment applies.
+    :param generator: Human-readable name of the entity that created this annotation.
+    :param label: A short form human/machine-readable label for the annotation.
+    :param comment: A human-readable comment.
+    :param freq_lower_edge: The frequency (Hz) of the lower edge of the feature
+        described by this annotation.
+    :param freq_upper_edge: The frequency (Hz) of the upper edge of the feature
+        described by this annotation.
+    """
+
+    sample_start: Optional[int] = None
+    sample_count: Optional[int] = None
+    generator: Optional[str] = None
+    label: Optional[str] = None
+    comment: Optional[str] = None
+    freq_lower_edge: Optional[float] = None
+    freq_upper_edge: Optional[float] = None
+
+    def __post_init__(self):
+        # Initialization
+        self.annotation_type = self.__class__.__name__
+        self.segment = {"ntia-core:annotation_type": self.annotation_type}
+        self.required_err_msg = (
+            f"{self.annotation_type} segments require a value to be specified for "
+        )
+        # Ensure required keys have been set
+        self.check_required(self.sample_start, "sample_start")
+        if self.freq_lower_edge is not None or self.freq_upper_edge is not None:
+            err_msg = "Both freq_lower_edge and freq_upper_edge must be provided if one is provided."
+            assert (
+                self.freq_lower_edge is not None and self.freq_upper_edge is not None
+            ), err_msg
+        # Define SigMF key names
+        self.sigmf_keys = {
+            "sample_start": "core:sample_start",
+            "sample_count": "core:sample_count",
+            "generator": "core:generator",
+            "label": "core:label",
+            "comment": "core:comment",
+            "freq_lower_edge": "core:freq_lower_edge",
+            "freq_upper_edge": "core:freq_upper_edge",
+            "recording": "core:recording",
+        }
+
+    def check_required(self, value: Any, keyname: str) -> None:
+        assert value is not None, self.required_err_msg + keyname
+
+    def create_annotation_segment(self) -> None:
+        meta_vars = vars(self)
+        for varname, value in meta_vars.items():
+            if value is not None:
+                try:
+                    sigmf_key = meta_vars["sigmf_keys"][varname]
+                    self.segment[sigmf_key] = value
+                except KeyError:
+                    pass
+        return
+
+    def create_metadata(self, sigmf_builder: SigMFBuilder) -> None:
+        sigmf_builder.add_annotation(self.sample_start, self.sample_count, self.segment)

scos_actions/metadata/annotations/__init__.py

@@ -0,0 +1,8 @@
+from scos_actions.metadata.annotations.calibration_annotation import (
+    CalibrationAnnotation,
+)
+from scos_actions.metadata.annotations.frequency_domain_detection import (
+    FrequencyDomainDetection,
+)
+from scos_actions.metadata.annotations.sensor_annotation import SensorAnnotation
+from scos_actions.metadata.annotations.time_domain_detection import TimeDomainDetection

scos_actions/metadata/annotations/calibration_annotation.py

@@ -1,36 +1,100 @@
-from scos_actions.metadata.metadata import Metadata
-from scos_actions.metadata.sigmf_builder import SigMFBuilder
-
-
-class CalibrationAnnotation(Metadata):
-    def __init__(self, start, count):
-        super().__init__(start, count)
-
-    def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict):
-        sigan_cal = measurement_result["sigan_cal"]
-        sensor_cal = measurement_result["sensor_cal"]
-        annotation = self.create_calibration_annotation(sigan_cal, sensor_cal)
-        sigmf_builder.add_annotation(self.start, self.count, annotation)
-
-    def create_calibration_annotation(self, sigan_cal, sensor_cal):
-        """Create the SigMF calibration annotation."""
-        annotation_md = {
-            "ntia-core:annotation_type": "CalibrationAnnotation",
-            "ntia-sensor:gain_sigan": sigan_cal["gain_sigan"],
-            "ntia-sensor:gain_sensor": sensor_cal["gain_sensor"],
-            "ntia-sensor:noise_figure_sigan": sigan_cal["noise_figure_sigan"],
-            "ntia-sensor:1db_compression_point_sigan": sigan_cal[
-                "1db_compression_sigan"
-            ],
-            "ntia-sensor:enbw_sigan": sigan_cal["enbw_sigan"],
-            "ntia-sensor:gain_preselector": sensor_cal["gain_preselector"],
-            "ntia-sensor:noise_figure_sensor": sensor_cal["noise_figure_sensor"],
-            "ntia-sensor:1db_compression_point_sensor": sensor_cal[
-                "1db_compression_sensor"
-            ],
-            "ntia-sensor:enbw_sensor": sensor_cal["enbw_sensor"],
-        }
-        if "temperature" in sensor_cal:
-            annotation_md["ntia-sensor:temperature"] = sensor_cal["temperature"]
-
-        return annotation_md
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from scos_actions.metadata.annotation_segment import AnnotationSegment
+
+
+@dataclass
+class CalibrationAnnotation(AnnotationSegment):
+    """
+    Interface for generating CalibrationAnnotation segments.
+
+    Most values are read from sensor and sigan calibrations,
+    expected to exist in a ``measurement_result`` dictionary.
+    No parameters are required.
+
+    Refer to the documentation of the ``ntia-sensor`` extension of
+    SigMF for more information.
+
+    :param sigan_cal: Sigan calibration result, likely stored
+        in the ``measurement_result`` dictionary. This should contain
+        keys: ``gain_sigan``, ``noise_figure_sigan``, ``1db_compression_sigan``,
+        and ``enbw_sigan``. Any missing keys will be skipped.
+    :param sensor_cal: Sensor calibration result, likely stored
+        in the ``measurement_result`` dictionary. This should contain
+        keys: ``gain_preselector``, ``noise_figure_sensor``, ``1db_compression_sensor``,
+        ``enbw_sensor``, and ``gain_sensor``. Optionally, it can also include
+        a ``temperature`` key. Any missing keys will be skipped.
+    :param mean_noise_power_sensor: Mean noise power density of the sensor.
+    :param mean_noise_power_units: The units of ``mean_noise_power_sensor``.
+    :param mean_noise_power_reference: Reference point for ``mean_noise_power_sensor``,
+        e.g. ``"signal analyzer input"``, ``"preselector input"``, ``"antenna terminal"``.
+    """
+
+    sigan_cal: Optional[dict] = None
+    sensor_cal: Optional[dict] = None
+    mean_noise_power_sensor: Optional[float] = None
+    mean_noise_power_units: Optional[str] = None
+    mean_noise_power_reference: Optional[str] = None
+
+    def __post_init__(self):
+        super().__post_init__()
+        # Load values from sensor and sigan calibrations
+        if self.sigan_cal is not None:
+            self.gain_sigan = self.get_cal_value_if_exists(self.sigan_cal, "gain_sigan")
+            self.noise_figure_sigan = self.get_cal_value_if_exists(
+                self.sigan_cal, "noise_figure_sigan"
+            )
+            self.compression_point_sigan = self.get_cal_value_if_exists(
+                self.sigan_cal, "1db_compression_sigan"
+            )
+            self.enbw_sigan = self.get_cal_value_if_exists(self.sigan_cal, "enbw_sigan")
+        if self.sensor_cal is not None:
+            self.gain_preselector = self.get_cal_value_if_exists(
+                self.sensor_cal, "gain_preselector"
+            )
+            self.noise_figure_sensor = self.get_cal_value_if_exists(
+                self.sensor_cal, "noise_figure_sensor"
+            )
+            self.compression_point_sensor = self.get_cal_value_if_exists(
+                self.sensor_cal, "1db_compression_sensor"
+            )
+            self.enbw_sensor = self.get_cal_value_if_exists(
+                self.sensor_cal, "enbw_sensor"
+            )
+            self.temperature = self.get_cal_value_if_exists(
+                self.sensor_cal, "temperature"
+            )
+            # Additional key gain_sensor is not in SigMF ntia-sensor spec but is included
+            self.gain_sensor = self.get_cal_value_if_exists(
+                self.sensor_cal, "gain_sensor"
+            )
+        # Define SigMF key names
+        self.sigmf_keys.update(
+            {
+                "gain_sigan": "ntia-sensor:gain_sigan",
+                "noise_figure_sigan": "ntia-sensor:noise_figure_sigan",
+                "one_db_compression_point_sigan": "ntia-sensor:1db_compression_point_sigan",
+                "enbw_sigan": "ntia-sensor:enbw_sigan",
+                "gain_preselector": "ntia-sensor:gain_preselector",
+                "gain_sensor": "ntia-sensor:gain_sensor",  # This is not a valid ntia-sensor key
+                "noise_figure_sensor": "ntia-sensor:noise_figure_sensor",
+                "one_db_compression_point_sensor": "ntia-sensor:1db_compression_point_sensor",
+                "enbw_sensor": "ntia-sensor:enbw_sensor",
+                "mean_noise_power_sensor": "ntia-sensor:mean_noise_power_sensor",
+                "mean_noise_power_units": "ntia-sensor:mean_noise_power_units",
+                "mean_noise_power_reference": "ntia-sensor:mean_noise_power_reference",
+                "temperature": "ntia-sensor:temperature",
+            }
+        )
+        # Create annotation segment
+        super().create_annotation_segment()
+
+    @staticmethod
+    def get_cal_value_if_exists(cal_dict: dict, key_name: str) -> Any:
+        try:
+            value = cal_dict[key_name]
+            return value
+        except KeyError:
+            # create_annotation_segment skips None values
+            return None