diff --git a/scos_actions/actions/acquire_single_freq_fft.py b/scos_actions/actions/acquire_single_freq_fft.py index 51522d9f..0b470c82 100644 --- a/scos_actions/actions/acquire_single_freq_fft.py +++ b/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 diff --git a/scos_actions/actions/acquire_single_freq_tdomain_iq.py b/scos_actions/actions/acquire_single_freq_tdomain_iq.py index 2ed3ec2b..e60d50bc 100644 --- a/scos_actions/actions/acquire_single_freq_tdomain_iq.py +++ b/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 ) diff --git a/scos_actions/actions/acquire_stepped_freq_tdomain_iq.py b/scos_actions/actions/acquire_stepped_freq_tdomain_iq.py index 733f8802..7a131258 100644 --- a/scos_actions/actions/acquire_stepped_freq_tdomain_iq.py +++ b/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 diff --git a/scos_actions/actions/interfaces/measurement_action.py b/scos_actions/actions/interfaces/measurement_action.py index 671a582e..d5c3c42f 100644 --- a/scos_actions/actions/interfaces/measurement_action.py +++ b/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.""" diff --git a/scos_actions/metadata/annotation_segment.py b/scos_actions/metadata/annotation_segment.py new file mode 100644 index 00000000..233b122b --- /dev/null +++ b/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) diff --git a/scos_actions/metadata/annotations/__init__.py b/scos_actions/metadata/annotations/__init__.py index e69de29b..3d435c7a 100644 --- a/scos_actions/metadata/annotations/__init__.py +++ b/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 diff --git a/scos_actions/metadata/annotations/calibration_annotation.py b/scos_actions/metadata/annotations/calibration_annotation.py index aa34a2c3..7178c5a4 100644 --- a/scos_actions/metadata/annotations/calibration_annotation.py +++ b/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 diff --git a/scos_actions/metadata/annotations/fft_annotation.py b/scos_actions/metadata/annotations/fft_annotation.py deleted file mode 100644 index dbd86cc1..00000000 --- a/scos_actions/metadata/annotations/fft_annotation.py +++ /dev/null @@ -1,26 +0,0 @@ -from scos_actions.metadata.metadata import Metadata -from scos_actions.metadata.sigmf_builder import SigMFBuilder - - -class FrequencyDomainDetectionAnnotation(Metadata): - def __init__(self, detector, start, count): - super().__init__(start, count) - self.detector = detector - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result): - metadata = { - "ntia-core:annotation_type": "FrequencyDomainDetection", - "ntia-algorithm:number_of_samples_in_fft": measurement_result["fft_size"], - "ntia-algorithm:window": measurement_result["window"], - "ntia-algorithm:equivalent_noise_bandwidth": measurement_result["enbw"], - "ntia-algorithm:detector": "fft_" + self.detector, - "ntia-algorithm:number_of_ffts": measurement_result["nffts"], - "ntia-algorithm:units": "dBm", - "ntia-algorithm:reference": '"preselector input"', - "ntia-algorithm:frequency_start": measurement_result["frequency_start"], - "ntia-algorithm:frequency_stop": measurement_result["frequency_stop"], - "ntia-algorithm:frequency_step": measurement_result["frequency_step"], - } - sigmf_builder.add_annotation( - self.start, measurement_result["fft_size"], metadata - ) diff --git a/scos_actions/metadata/annotations/frequency_domain_detection.py b/scos_actions/metadata/annotations/frequency_domain_detection.py new file mode 100644 index 00000000..3a8d90c4 --- /dev/null +++ b/scos_actions/metadata/annotations/frequency_domain_detection.py @@ -0,0 +1,77 @@ +from dataclasses import dataclass +from typing import List, Optional + +from scos_actions.metadata.annotation_segment import AnnotationSegment + + +@dataclass +class FrequencyDomainDetection(AnnotationSegment): + """ + Interface for generating FrequencyDomainDetection annotation segments. + + Refer to the documentation of the ``ntia-algorithm`` extension of SigMF + for more information. + + The parameters ``detector``, ``number_of_ffts``, ``number_of_samples_in_fft``, + ``window``, and ``units`` are required. + + :param detector: Detector type, e.g. ``"fft_sample_iq"``, ``"fft_sample_power"``, + ``"fft_mean_power"``, etc. If the detector string does not start with "fft_" + already, "fft_" will be prepended to the input detector string. + :param number_of_ffts: Number of FFTs to be integrated over by detector. + :param number_of_samples_in_fft: Number of samples in FFT to calculate + ``delta_f = samplerate / number_of_samples_in_fft``. + :param window: Window type used in FFT, e.g. ``"blackman-harris"``, ``"flattop"``, + ``"hanning"``, ``"rectangular"``, etc. + :param equivalent_noise_bandwidth: Bandwidth of brickwall filter that has the + same integrated noise power as that of the actual filter. + :param units: Data units, e.g. ``"dBm"``, ``"watts"``, ``"volts"``. + :param reference: Data reference point, e.g. ``"signal analyzer input"``, + ``"preselector input"``, ``"antenna terminal"``. + :param frequency_start: Frequency (Hz) of first data point. + :param frequency_stop: Frequency (Hz) of last data point. + :param frequency_step: Frequency step size (Hz) between data points. + :param frequencies: A list of the frequencies (Hz) of the data points. + """ + + detector: Optional[str] = None + number_of_ffts: Optional[int] = None + number_of_samples_in_fft: Optional[int] = None + window: Optional[str] = None + equivalent_noise_bandwidth: Optional[float] = None + units: Optional[str] = None + reference: Optional[str] = None + frequency_start: Optional[float] = None + frequency_stop: Optional[float] = None + frequency_step: Optional[float] = None + frequencies: Optional[List[float]] = None + + def __post_init__(self): + super().__post_init__() + # Ensure required keys have been set + self.check_required(self.detector, "detector") + self.check_required(self.number_of_ffts, "number_of_ffts") + self.check_required(self.number_of_samples_in_fft, "number_of_samples_in_fft") + self.check_required(self.window, "window") + self.check_required(self.units, "units") + # Prepend "fft" to detector name if needed + if self.detector[:4] != "fft_": + self.detector = "fft_" + self.detector + # Define SigMF key names + self.sigmf_keys.update( + { + "detector": "ntia-algorithm:detector", + "number_of_ffts": "ntia-algorithm:number_of_ffts", + "number_of_samples_in_fft": "ntia-algorithm:number_of_samples_in_fft", + "window": "ntia-algorithm:window", + "equivalent_noise_bandwidth": "ntia-algorithm:equivalent_noise_bandwidth", + "units": "ntia-algorithm:units", + "reference": "ntia-algorithm:reference", + "frequency_start": "ntia-algorithm:frequency_start", + "frequency_stop": "ntia-algorithm:frequency_stop", + "frequency_step": "ntia-algorithm:frequency_step", + "frequencies": "ntia-algorithm:frequencies", + } + ) + # Create annotation segment + super().create_annotation_segment() diff --git a/scos_actions/metadata/annotations/sensor_annotation.py b/scos_actions/metadata/annotations/sensor_annotation.py index de38a8db..9454bac0 100644 --- a/scos_actions/metadata/annotations/sensor_annotation.py +++ b/scos_actions/metadata/annotations/sensor_annotation.py @@ -1,19 +1,46 @@ -from scos_actions.metadata.metadata import Metadata -from scos_actions.metadata.sigmf_builder import SigMFBuilder - - -class SensorAnnotation(Metadata): - def __init__(self, start, count): - super().__init__(start, count) - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result): - metadata = {"ntia-core:annotation_type": "SensorAnnotation"} - if "overload" in measurement_result: - metadata["ntia-sensor:overload"] = measurement_result["overload"] - if "gain" in measurement_result: - metadata["ntia-sensor:gain_setting_sigan"] = measurement_result["gain"] - if "attenuation" in measurement_result: - metadata["ntia-sensor:attenuation_setting_sigan"] = measurement_result[ - "attenuation" - ] - sigmf_builder.add_annotation(self.start, self.count, metadata) +from dataclasses import dataclass +from typing import Optional + +from scos_actions.metadata.annotation_segment import AnnotationSegment + + +@dataclass +class SensorAnnotation(AnnotationSegment): + """ + Interface for generating SensorAnnotation segments. + + All parameters are optional. Attenuation, gain, and overload + values can generally be found in the measurement_result + dictionary. + + Refer to the documentation of the ``ntia-sensor`` extension of + SigMF for more information. + + :param rf_path_index: Index of the RF Path. + :param overload: Indicator of sensor overload. + :param attenuation_setting_sigan: Attenuation setting of the signal + analyzer. + :param gain_setting_sigan: Gain setting of the signal analyzer. + :param gps_nmea: NMEA message from a GPS receiver. + """ + + rf_path_index: Optional[int] = None + overload: Optional[bool] = None + attenuation_setting_sigan: Optional[float] = None + gain_setting_sigan: Optional[float] = None + gps_nmea: Optional[str] = None + + def __post_init__(self): + super().__post_init__() + # Define SigMF key names + self.sigmf_keys.update( + { + "rf_path_index": "ntia-sensor:rf_path_index", + "overload": "ntia-sensor:overload", + "attenuation_setting_sigan": "ntia-sensor:attenuation_setting_sigan", + "gain_setting_sigan": "ntia-sensor:gain_setting_sigan", + "gps_nmea": "ntia-sensor:gps_nmea", + } + ) + # Create annotation segment + super().create_annotation_segment() diff --git a/scos_actions/metadata/annotations/time_domain_annotation.py b/scos_actions/metadata/annotations/time_domain_annotation.py deleted file mode 100644 index 0bbd7ad0..00000000 --- a/scos_actions/metadata/annotations/time_domain_annotation.py +++ /dev/null @@ -1,17 +0,0 @@ -from scos_actions.metadata.metadata import Metadata -from scos_actions.metadata.sigmf_builder import SigMFBuilder - - -class TimeDomainAnnotation(Metadata): - def __init__(self, start, count): - super().__init__(start, count) - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result): - time_domain_detection_md = { - "ntia-core:annotation_type": "TimeDomainDetection", - "ntia-algorithm:detector": "sample_iq", - "ntia-algorithm:number_of_samples": self.count, - "ntia-algorithm:units": "volts", - "ntia-algorithm:reference": "preselector input", - } - sigmf_builder.add_annotation(self.start, self.count, time_domain_detection_md) diff --git a/scos_actions/metadata/annotations/time_domain_detection.py b/scos_actions/metadata/annotations/time_domain_detection.py new file mode 100644 index 00000000..e0164c5a --- /dev/null +++ b/scos_actions/metadata/annotations/time_domain_detection.py @@ -0,0 +1,47 @@ +from dataclasses import dataclass +from typing import Optional + +from scos_actions.metadata.annotation_segment import AnnotationSegment + + +@dataclass +class TimeDomainDetection(AnnotationSegment): + """ + Interface for generating TimeDomainDetection annotation segments. + + Refer to the documentation of the ``ntia-algorithm`` extension of + SigMF for more information. + + The parameters ``detector``, ``number_of_samples``, and ``units`` are + required. + + :param detector: Detector type, e.g. ``"sample_power"``, ``"mean_power"``, + ``"max_power"``, etc. + :param number_of_samples: Number of samples integrated over by the detector. + :param units: Data units, e.g. ``"dBm"``, ``"watts"``, etc. + :param reference: Data reference point, e.g. ``"signal analyzer input"``, + ``"preselector input"``, ``"antenna terminal"``, etc. + """ + + detector: Optional[str] = None + number_of_samples: Optional[int] = None + units: Optional[str] = None + reference: Optional[str] = None + + def __post_init__(self): + super().__post_init__() + # Ensure required keys have been set + self.check_required(self.detector, "detector") + self.check_required(self.number_of_samples, "number_of_samples") + self.check_required(self.units, "units") + # Define SigMF key names + self.sigmf_keys.update( + { + "detector": "ntia-algorithm:detector", + "number_of_samples": "ntia-algorithm:number_of_samples", + "units": "ntia-algorithm:units", + "reference": "ntia-algorithm:reference", + } + ) + # Create annotation segment + super().create_annotation_segment() diff --git a/scos_actions/metadata/measurement_global.py b/scos_actions/metadata/measurement_global.py index 86950b91..a04cb2b0 100644 --- a/scos_actions/metadata/measurement_global.py +++ b/scos_actions/metadata/measurement_global.py @@ -1,36 +1,78 @@ -from scos_actions.metadata.metadata import Metadata +from dataclasses import dataclass +from datetime import datetime +from typing import Any, List, Optional + from scos_actions.metadata.sigmf_builder import SigMFBuilder -class MeasurementMetadata(Metadata): - def __init__(self): - super().__init__() - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): - freq_low = None - freq_high = None - - if "frequency_low" in measurement_result: - freq_low = measurement_result["frequency_low"] - elif "frequency" in measurement_result: - freq_low = measurement_result["frequency"] - freq_high = measurement_result["frequency"] - if "frequency_high" in measurement_result: - freq_high = measurement_result["frequency_high"] - - if freq_high is None: - raise Exception("frequency_high is a required measurement metadata value.") - if freq_low is None: - raise Exception("frequency_low is a required measurement metadata value.") - - sigmf_builder.add_to_global( - "ntia-core:measurement", - { - "time_start": measurement_result["start_time"], - "time_stop": measurement_result["end_time"], - "domain": measurement_result["domain"], - "measurement_type": measurement_result["measurement_type"], - "frequency_tuned_low": freq_low, - "frequency_tuned_high": freq_high, - }, +@dataclass +class MeasurementMetadata: + """ + Interface for generating SigMF ntia-core Measurement objects. + + The parameters ``domain``, ``measurement_type``, ``time_start``, + ``time_stop``, ``frequency_tuned_low``, ``frequency_tuned_high``, and + ``classification`` are required. Refer to the documentation for the + ``ntia-core`` extension of SigMF for more information. + + :param domain: Measurement domain, generally ``"time"`` or ``"frequency"``. + :param measurement_type: Method by which the signal analyzer acquires data: + ``"single-frequency"`` or ``"scan"``. + :param time_start: When the action began execution. + :param time_stop: When the action finished execution. + :param frequency_tuned_low: Lowest tuned frequency (Hz). + :param frequency_tuned_high: Highest tuned frequency (Hz). + :param frequency_tuned_step: Step between tuned frequencies of a ``"scan"`` + measurement. Either ``frequency_tuned_step`` or ``frequencies_tuned`` + SHOULD be included for ``"scan"`` measurements. + :param classification: The classification markings for the acquisition, e.g. + ``"UNCLASSIFIED"``, ``"CONTROLLED//FEDCON"``, ``"SECRET"``, etc. + """ + + domain: Optional[str] + measurement_type: Optional[str] + time_start: Optional[datetime] + time_stop: Optional[datetime] + frequency_tuned_low: Optional[float] + frequency_tuned_high: Optional[float] + frequency_tuned_step: Optional[float] = None + frequencies_tuned: Optional[List[float]] = None + classification: Optional[str] = None + + def __post_init__(self): + # Ensure required keys have been set + self.check_required(self.domain, "domain") + self.check_required(self.measurement_type, "measurement_type") + self.check_required(self.time_start, "time_start") + self.check_required(self.time_stop, "time_stop") + self.check_required(self.frequency_tuned_low, "frequency_tuned_low") + self.check_required(self.frequency_tuned_high, "frequency_tuned_high") + self.check_required(self.classification, "classification") + # Define SigMF key names + self.sigmf_keys = { + "domain": "domain", + "measurement_type": "measurement_type", + "time_start": "time_start", + "time_stop": "time_stop", + "frequency_tuned_low": "frequency_tuned_low", + "frequency_tuned_high": "frequency_tuned_high", + "frequencies_tuned": "frequencies_tuned", + "classification": "classification", + } + + def check_required(self, value: Any, keyname: str) -> None: + assert value is not None, ( + "Measurement metadata requires a value to be specified for " + keyname ) + + def create_metadata(self, sigmf_builder: SigMFBuilder): + segment = {} + meta_vars = vars(self) + for varname, value in meta_vars.items(): + if value is not None: + try: + sigmf_key = meta_vars["sigmf_keys"][varname] + segment[sigmf_key] = value + except KeyError: + pass + sigmf_builder.add_to_global("ntia-core:measurement", segment) diff --git a/scos_actions/metadata/metadata.py b/scos_actions/metadata/metadata.py deleted file mode 100644 index 36139334..00000000 --- a/scos_actions/metadata/metadata.py +++ /dev/null @@ -1,14 +0,0 @@ -from abc import ABC, abstractmethod - -from scos_actions.metadata.sigmf_builder import SigMFBuilder - - -class Metadata(ABC): - def __init__(self, start=None, count=None, recording=None): - self.start = start - self.count = count - self.recording = recording - - @abstractmethod - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): - pass diff --git a/scos_actions/metadata/metadata_exception.py b/scos_actions/metadata/metadata_exception.py new file mode 100644 index 00000000..052d400d --- /dev/null +++ b/scos_actions/metadata/metadata_exception.py @@ -0,0 +1,5 @@ +class MetadataException(Exception): + """Basic exception handling for metadata-related problems.""" + + def __init__(self, msg): + super().__init__(msg) diff --git a/scos_actions/metadata/sigmf_builder.py b/scos_actions/metadata/sigmf_builder.py index a36b69a4..9fad46b3 100644 --- a/scos_actions/metadata/sigmf_builder.py +++ b/scos_actions/metadata/sigmf_builder.py @@ -48,15 +48,57 @@ def reset(self): def metadata(self): return self.sigmf_md._metadata - def set_data_type(self, is_complex): - if is_complex: - self.sigmf_md.set_global_field( - "core:datatype", "cf32_le" - ) # 2x 32-bit float, Little Endian + def set_data_type( + self, + is_complex: bool, + sample_type: str = "floating-point", + bit_width: int = 32, + endianness: str = "little", + ): + """ + Set the global ``core:datatype`` field of a SigMF metadata file. + + Defaults to "cf32_le" for complex 32 bit little-endian floating point. + + :param is_complex: True if the data is complex, False if the data is real. + :param sample_type: The sample type, defaults to "floating-point" + :param bit_width: The bit-width of the data, defaults to 32 + :param endianness: Data endianness, defaults to "little". Provide an empty + string if the data is saved as bytes. + :raises ValueError: If ``bit_width`` is not an integer. + :raises ValueError: If ``sample_type`` is not one of: "floating-point", + "signed-integer", or "unsigned-integer". + :raises ValueError: If the endianness is not one of: "big", "little", or + "" (an empty string). + """ + if not isinstance(bit_width, int): + raise ValueError("Bit-width must be an integer.") + + dset_fmt = "c" if is_complex else "r" + + if sample_type == "floating-point": + dset_fmt += "f" + elif sample_type == "signed-integer": + dset_fmt += "i" + elif sample_type == "unsigned-integer": + dset_fmt += "u" else: - self.sigmf_md.set_global_field( - "core:datatype", "rf32_le" - ) # 32-bit float, Little Endian + raise ValueError( + 'Sample type must be one of: "floating-point", "signed-integer", or "unsigned-integer"' + ) + + dset_fmt += str(bit_width) + + if endianness == "little": + dset_fmt += "_le" + elif endianness == "big": + dset_fmt += "_be" + elif endianness != "": + raise ValueError( + 'Endianness must be either "big", "little", or "" (for saving bytes)' + ) + + self.sigmf_md.set_global_field("core:datatype", dset_fmt) def set_sample_rate(self, sample_rate): self.sigmf_md.set_global_field("core:sample_rate", sample_rate) @@ -160,6 +202,6 @@ def add_metadata_generator(self, key, generator): def remove_metadata_generator(self, key): self.metadata_generators.pop(key, "") - def build(self, measurement_result): + def build(self): for metadata_creator in self.metadata_generators.values(): - metadata_creator.create_metadata(self, measurement_result) + metadata_creator.create_metadata(self)