From 6b99a151fb40651e448d4c99e56b7d6c93af523b Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Mon, 15 Aug 2022 16:43:03 -0600 Subject: [PATCH 01/25] Generalize time domain annotation --- .../annotations/time_domain_annotation.py | 28 +++++++++++++------ 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/scos_actions/metadata/annotations/time_domain_annotation.py b/scos_actions/metadata/annotations/time_domain_annotation.py index 3f2f57c1..892ccb24 100644 --- a/scos_actions/metadata/annotations/time_domain_annotation.py +++ b/scos_actions/metadata/annotations/time_domain_annotation.py @@ -1,17 +1,27 @@ 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) +class TimeDomainAnnotation(Metadata): + def __init__( + self, + start: int, + count: int, + detector: str = "sample_iq", + units: str = "volts", + reference: str = "preselector input", + ): + super().__init__(start, count) + self.detector = detector + self.units = units + self.reference = reference - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result): - time_domain_detection_md = { + def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): + metadata = { "ntia-core:annotation_type": "TimeDomainDetection", - "ntia-algorithm:detector": "sample_iq", + "ntia-algorithm:detector": self.detector, "ntia-algorithm:number_of_samples": self.count, - "ntia-algorithm:units": "volts", - "ntia-algorithm:reference": "preselector input", + "ntia-algorithm:units": self.units, + "ntia-algorithm:reference": self.reference, } - sigmf_builder.add_annotation(self.start, self.count, time_domain_detection_md) + sigmf_builder.add_annotation(self.start, self.count, metadata) From f582ce3074fe3bf9accb2ce189f487f88f4410ea Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Mon, 15 Aug 2022 16:57:10 -0600 Subject: [PATCH 02/25] Generalize FFT annotation --- .../actions/acquire_single_freq_fft.py | 63 ++++++++++++------- .../metadata/annotations/fft_annotation.py | 44 ++++++++----- 2 files changed, 69 insertions(+), 38 deletions(-) diff --git a/scos_actions/actions/acquire_single_freq_fft.py b/scos_actions/actions/acquire_single_freq_fft.py index d6299c56..0f1adc37 100644 --- a/scos_actions/actions/acquire_single_freq_fft.py +++ b/scos_actions/actions/acquire_single_freq_fft.py @@ -87,8 +87,16 @@ """ import logging + +from numpy import float32, ndarray + 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.fft_annotation import ( + FrequencyDomainDetectionAnnotation, +) +from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder from scos_actions.signal_processing.fft import ( get_fft, get_fft_enbw, @@ -96,18 +104,16 @@ get_fft_window, get_fft_window_correction, ) -from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder -from scos_actions.metadata.annotations.fft_annotation import FrequencyDomainDetectionAnnotation -from scos_actions.hardware import gps as mock_gps - -from scos_actions.utils import get_parameter from scos_actions.signal_processing.power_analysis import ( apply_power_detector, calculate_power_watts, create_power_detector, ) -from scos_actions.signal_processing.unit_conversion import convert_watts_to_dBm, convert_linear_to_dB -from numpy import float32, ndarray +from scos_actions.signal_processing.unit_conversion import ( + convert_linear_to_dB, + convert_watts_to_dBm, +) +from scos_actions.utils import get_parameter logger = logging.getLogger(__name__) @@ -164,25 +170,26 @@ def execute(self, schedule_entry, task_id) -> dict: m4s_result = self.apply_m4s(measurement_result) # Save measurement results - measurement_result['data'] = m4s_result - measurement_result['start_time'] = start_time - measurement_result['end_time'] = utils.get_datetime_str_now() - measurement_result['enbw'] = get_fft_enbw(self.fft_window, sample_rate_Hz) + measurement_result["data"] = m4s_result + measurement_result["start_time"] = start_time + measurement_result["end_time"] = utils.get_datetime_str_now() + measurement_result["enbw"] = get_fft_enbw(self.fft_window, sample_rate_Hz) frequencies = get_fft_frequencies( self.fft_size, sample_rate_Hz, self.frequency_Hz ) measurement_result.update(self.parameters) - measurement_result['description'] = self.description - measurement_result['domain'] = Domain.FREQUENCY.value - measurement_result['frequency_start'] = frequencies[0] - measurement_result['frequency_stop'] = frequencies[-1] - measurement_result['frequency_step'] = frequencies[1] - frequencies[0] - measurement_result['window'] = self.fft_window_type - measurement_result['calibration_datetime'] = self.sigan.sensor_calibration_data['calibration_datetime'] - measurement_result['task_id'] = task_id - 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["description"] = self.description + measurement_result["domain"] = Domain.FREQUENCY.value + measurement_result["frequency_start"] = frequencies[0] + measurement_result["frequency_stop"] = frequencies[-1] + measurement_result["frequency_step"] = frequencies[1] - frequencies[0] + measurement_result["calibration_datetime"] = self.sigan.sensor_calibration_data[ + "calibration_datetime" + ] + measurement_result["task_id"] = task_id + 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 return measurement_result def apply_m4s(self, measurement_result: dict) -> ndarray: @@ -202,7 +209,7 @@ def apply_m4s(self, measurement_result: dict) -> ndarray: # RF/Baseband power conversion (-3 dB) # FFT window amplitude correction m4s_result -= 3 - m4s_result += 2. * convert_linear_to_dB(self.fft_window_acf) + m4s_result += 2.0 * convert_linear_to_dB(self.fft_window_acf) return m4s_result @property @@ -236,7 +243,15 @@ 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 + start=i * self.fft_size, + count=self.fft_size, + fft_size=self.fft_size, + window=self.fft_window_type, + enbw=measurement_result["enbw"], + detector=detector.value, + nffts=self.nffts, + units="dBm", + reference="preselector input", ) sigmf_builder.add_metadata_generator( type(fft_annotation).__name__ + "_" + detector.value, fft_annotation diff --git a/scos_actions/metadata/annotations/fft_annotation.py b/scos_actions/metadata/annotations/fft_annotation.py index 85a3ba7a..938c88e4 100644 --- a/scos_actions/metadata/annotations/fft_annotation.py +++ b/scos_actions/metadata/annotations/fft_annotation.py @@ -3,23 +3,39 @@ class FrequencyDomainDetectionAnnotation(Metadata): - - def __init__(self, detector, start, count): + def __init__( + self, + start: int, + count: int, + fft_size: int, + window: str, + enbw: float, + detector: str, + nffts: int, + units: str, + reference: str, + ): super().__init__(start, count) + self.fft_size = fft_size + self.window = window + self.enbw = enbw self.detector = detector + self.nffts = nffts + self.units = units + self.reference = reference - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result): + def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): 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'], + "ntia-algorithm:number_of_samples_in_fft": self.fft_size, + "ntia-algorithm:window": self.window, + "ntia-algorithm:equivalent_noise_bandwidth": self.enbw, + "ntia-algorithm:detector": "fft_" + self.detector, + "ntia-algorithm:number_of_ffts": self.nffts, + "ntia-algorithm:units": self.units, + "ntia-algorithm:reference": self.reference, + "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) + sigmf_builder.add_annotation(self.start, self.fft_size, metadata) From 0e446be38f9aceba2415c9a37d5ee8750c19a245 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Mon, 15 Aug 2022 17:05:03 -0600 Subject: [PATCH 03/25] Remove dependence on measurement result --- scos_actions/metadata/annotations/fft_annotation.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/scos_actions/metadata/annotations/fft_annotation.py b/scos_actions/metadata/annotations/fft_annotation.py index 938c88e4..cbd1b949 100644 --- a/scos_actions/metadata/annotations/fft_annotation.py +++ b/scos_actions/metadata/annotations/fft_annotation.py @@ -14,6 +14,9 @@ def __init__( nffts: int, units: str, reference: str, + frequency_start: float, + frequency_stop: float, + frequency_step: float, ): super().__init__(start, count) self.fft_size = fft_size @@ -23,6 +26,9 @@ def __init__( self.nffts = nffts self.units = units self.reference = reference + self.frequency_start = frequency_start + self.frequency_stop = frequency_stop + self.frequency_step = frequency_step def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): metadata = { @@ -34,8 +40,8 @@ def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict) "ntia-algorithm:number_of_ffts": self.nffts, "ntia-algorithm:units": self.units, "ntia-algorithm:reference": self.reference, - "ntia-algorithm:frequency_start": measurement_result["frequency_start"], - "ntia-algorithm:frequency_stop": measurement_result["frequency_stop"], - "ntia-algorithm:frequency_step": measurement_result["frequency_step"], + "ntia-algorithm:frequency_start": self.frequency_start, + "ntia-algorithm:frequency_stop": self.frequency_stop, + "ntia-algorithm:frequency_step": self.frequency_step, } sigmf_builder.add_annotation(self.start, self.fft_size, metadata) From 95ee9338c3f094762990ef29ef208fedaf2e1041 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Mon, 15 Aug 2022 17:06:16 -0600 Subject: [PATCH 04/25] Update FFT action to new annotation parameters --- scos_actions/actions/acquire_single_freq_fft.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/scos_actions/actions/acquire_single_freq_fft.py b/scos_actions/actions/acquire_single_freq_fft.py index 0f1adc37..61cff7f7 100644 --- a/scos_actions/actions/acquire_single_freq_fft.py +++ b/scos_actions/actions/acquire_single_freq_fft.py @@ -252,6 +252,9 @@ def get_sigmf_builder(self, measurement_result) -> SigMFBuilder: nffts=self.nffts, 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 From ab9b90cab781fab86250019e6c6cd425f03b9b82 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Mon, 15 Aug 2022 17:08:18 -0600 Subject: [PATCH 05/25] Update FFT action to new annotation parameters --- .../actions/acquire_single_freq_tdomain_iq.py | 39 ++++++++++++------- .../annotations/time_domain_annotation.py | 6 +-- 2 files changed, 28 insertions(+), 17 deletions(-) diff --git a/scos_actions/actions/acquire_single_freq_tdomain_iq.py b/scos_actions/actions/acquire_single_freq_tdomain_iq.py index 3e3b89ee..bc27ad86 100644 --- a/scos_actions/actions/acquire_single_freq_tdomain_iq.py +++ b/scos_actions/actions/acquire_single_freq_tdomain_iq.py @@ -33,13 +33,16 @@ import logging +from numpy import complex64 + from scos_actions import utils -from scos_actions.utils import get_parameter from scos_actions.actions.interfaces.measurement_action import MeasurementAction -from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder from scos_actions.hardware import gps as mock_gps -from scos_actions.metadata.annotations.time_domain_annotation import TimeDomainAnnotation -from numpy import complex64 +from scos_actions.metadata.annotations.time_domain_annotation import ( + TimeDomainAnnotation, +) +from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder +from scos_actions.utils import get_parameter logger = logging.getLogger(__name__) @@ -83,22 +86,30 @@ def execute(self, schedule_entry, task_id) -> dict: sample_rate = self.sigan.sample_rate num_samples = int(sample_rate * self.duration_ms * 1e-3) measurement_result = self.acquire_data(num_samples, self.nskip) - measurement_result['start_time'] = start_time + measurement_result["start_time"] = start_time end_time = utils.get_datetime_str_now() measurement_result.update(self.parameters) - measurement_result['end_time'] = end_time - measurement_result['domain'] = Domain.TIME.value - measurement_result['measurement_type'] = MeasurementType.SINGLE_FREQUENCY.value - measurement_result['task_id'] = task_id - measurement_result['calibration_datetime'] = self.sigan.sensor_calibration_data['calibration_datetime'] - 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["end_time"] = end_time + measurement_result["domain"] = Domain.TIME.value + measurement_result["measurement_type"] = MeasurementType.SINGLE_FREQUENCY.value + measurement_result["task_id"] = task_id + measurement_result["calibration_datetime"] = self.sigan.sensor_calibration_data[ + "calibration_datetime" + ] + measurement_result["description"] = self.description + measurement_result["sigan_cal"] = self.sigan.sigan_calibration_data + measurement_result["sensor_cal"] = self.sigan.sensor_calibration_data 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 = TimeDomainAnnotation( + start=0, + count=self.received_samples, + detector="sample_iq", + units="volts", + reference="preselector input", + ) sigmf_builder.add_metadata_generator( type(time_domain_annotation).__name__, time_domain_annotation ) diff --git a/scos_actions/metadata/annotations/time_domain_annotation.py b/scos_actions/metadata/annotations/time_domain_annotation.py index 892ccb24..c77bf25e 100644 --- a/scos_actions/metadata/annotations/time_domain_annotation.py +++ b/scos_actions/metadata/annotations/time_domain_annotation.py @@ -7,9 +7,9 @@ def __init__( self, start: int, count: int, - detector: str = "sample_iq", - units: str = "volts", - reference: str = "preselector input", + detector: str, + units: str, + reference: str, ): super().__init__(start, count) self.detector = detector From 03cf8a353dc66f1f5abed41ddb8c3fc62c8e9efc Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Mon, 15 Aug 2022 17:12:31 -0600 Subject: [PATCH 06/25] Cleanup action diffs --- .../actions/acquire_single_freq_fft.py | 53 ++++++++----------- .../actions/acquire_single_freq_tdomain_iq.py | 35 ++++++------ 2 files changed, 40 insertions(+), 48 deletions(-) diff --git a/scos_actions/actions/acquire_single_freq_fft.py b/scos_actions/actions/acquire_single_freq_fft.py index 61cff7f7..7fcab54a 100644 --- a/scos_actions/actions/acquire_single_freq_fft.py +++ b/scos_actions/actions/acquire_single_freq_fft.py @@ -87,16 +87,8 @@ """ import logging - -from numpy import float32, ndarray - 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.fft_annotation import ( - FrequencyDomainDetectionAnnotation, -) -from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder from scos_actions.signal_processing.fft import ( get_fft, get_fft_enbw, @@ -104,16 +96,18 @@ get_fft_window, get_fft_window_correction, ) +from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder +from scos_actions.metadata.annotations.fft_annotation import FrequencyDomainDetectionAnnotation +from scos_actions.hardware import gps as mock_gps + +from scos_actions.utils import get_parameter from scos_actions.signal_processing.power_analysis import ( apply_power_detector, calculate_power_watts, create_power_detector, ) -from scos_actions.signal_processing.unit_conversion import ( - convert_linear_to_dB, - convert_watts_to_dBm, -) -from scos_actions.utils import get_parameter +from scos_actions.signal_processing.unit_conversion import convert_watts_to_dBm, convert_linear_to_dB +from numpy import float32, ndarray logger = logging.getLogger(__name__) @@ -170,26 +164,25 @@ def execute(self, schedule_entry, task_id) -> dict: m4s_result = self.apply_m4s(measurement_result) # Save measurement results - measurement_result["data"] = m4s_result - measurement_result["start_time"] = start_time - measurement_result["end_time"] = utils.get_datetime_str_now() - measurement_result["enbw"] = get_fft_enbw(self.fft_window, sample_rate_Hz) + measurement_result['data'] = m4s_result + measurement_result['start_time'] = start_time + measurement_result['end_time'] = utils.get_datetime_str_now() + measurement_result['enbw'] = get_fft_enbw(self.fft_window, sample_rate_Hz) frequencies = get_fft_frequencies( self.fft_size, sample_rate_Hz, self.frequency_Hz ) measurement_result.update(self.parameters) - measurement_result["description"] = self.description - measurement_result["domain"] = Domain.FREQUENCY.value - measurement_result["frequency_start"] = frequencies[0] - measurement_result["frequency_stop"] = frequencies[-1] - measurement_result["frequency_step"] = frequencies[1] - frequencies[0] - measurement_result["calibration_datetime"] = self.sigan.sensor_calibration_data[ - "calibration_datetime" - ] - measurement_result["task_id"] = task_id - 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['description'] = self.description + measurement_result['domain'] = Domain.FREQUENCY.value + measurement_result['frequency_start'] = frequencies[0] + measurement_result['frequency_stop'] = frequencies[-1] + measurement_result['frequency_step'] = frequencies[1] - frequencies[0] + measurement_result['window'] = self.fft_window_type + measurement_result['calibration_datetime'] = self.sigan.sensor_calibration_data['calibration_datetime'] + measurement_result['task_id'] = task_id + 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 return measurement_result def apply_m4s(self, measurement_result: dict) -> ndarray: @@ -209,7 +202,7 @@ def apply_m4s(self, measurement_result: dict) -> ndarray: # RF/Baseband power conversion (-3 dB) # FFT window amplitude correction m4s_result -= 3 - m4s_result += 2.0 * convert_linear_to_dB(self.fft_window_acf) + m4s_result += 2. * convert_linear_to_dB(self.fft_window_acf) return m4s_result @property diff --git a/scos_actions/actions/acquire_single_freq_tdomain_iq.py b/scos_actions/actions/acquire_single_freq_tdomain_iq.py index bc27ad86..93e118c0 100644 --- a/scos_actions/actions/acquire_single_freq_tdomain_iq.py +++ b/scos_actions/actions/acquire_single_freq_tdomain_iq.py @@ -33,16 +33,13 @@ import logging -from numpy import complex64 - from scos_actions import utils +from scos_actions.utils import get_parameter 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.sigmf_builder import Domain, MeasurementType, SigMFBuilder -from scos_actions.utils import get_parameter +from scos_actions.hardware import gps as mock_gps +from scos_actions.metadata.annotations.time_domain_annotation import TimeDomainAnnotation +from numpy import complex64 logger = logging.getLogger(__name__) @@ -86,19 +83,17 @@ def execute(self, schedule_entry, task_id) -> dict: sample_rate = self.sigan.sample_rate num_samples = int(sample_rate * self.duration_ms * 1e-3) measurement_result = self.acquire_data(num_samples, self.nskip) - measurement_result["start_time"] = start_time + measurement_result['start_time'] = start_time end_time = utils.get_datetime_str_now() measurement_result.update(self.parameters) - measurement_result["end_time"] = end_time - measurement_result["domain"] = Domain.TIME.value - measurement_result["measurement_type"] = MeasurementType.SINGLE_FREQUENCY.value - measurement_result["task_id"] = task_id - measurement_result["calibration_datetime"] = self.sigan.sensor_calibration_data[ - "calibration_datetime" - ] - 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['end_time'] = end_time + measurement_result['domain'] = Domain.TIME.value + measurement_result['measurement_type'] = MeasurementType.SINGLE_FREQUENCY.value + measurement_result['task_id'] = task_id + measurement_result['calibration_datetime'] = self.sigan.sensor_calibration_data['calibration_datetime'] + measurement_result['description'] = self.description + measurement_result['sigan_cal'] = self.sigan.sigan_calibration_data + measurement_result['sensor_cal'] = self.sigan.sensor_calibration_data return measurement_result def get_sigmf_builder(self, measurement_result: dict) -> SigMFBuilder: @@ -143,3 +138,7 @@ def transform_data(self, measurement_result): def is_complex(self) -> bool: return True + + + + From 5f248656a97b0841a347245777231e4f8c8e3998 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Tue, 16 Aug 2022 10:08:16 -0600 Subject: [PATCH 07/25] Make count and num_samps distinct --- .../actions/acquire_single_freq_tdomain_iq.py | 36 ++++++++++--------- .../annotations/time_domain_annotation.py | 4 ++- 2 files changed, 22 insertions(+), 18 deletions(-) diff --git a/scos_actions/actions/acquire_single_freq_tdomain_iq.py b/scos_actions/actions/acquire_single_freq_tdomain_iq.py index 93e118c0..d9eb9c95 100644 --- a/scos_actions/actions/acquire_single_freq_tdomain_iq.py +++ b/scos_actions/actions/acquire_single_freq_tdomain_iq.py @@ -33,13 +33,16 @@ import logging +from numpy import complex64 + from scos_actions import utils -from scos_actions.utils import get_parameter from scos_actions.actions.interfaces.measurement_action import MeasurementAction -from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder from scos_actions.hardware import gps as mock_gps -from scos_actions.metadata.annotations.time_domain_annotation import TimeDomainAnnotation -from numpy import complex64 +from scos_actions.metadata.annotations.time_domain_annotation import ( + TimeDomainAnnotation, +) +from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder +from scos_actions.utils import get_parameter logger = logging.getLogger(__name__) @@ -83,17 +86,19 @@ def execute(self, schedule_entry, task_id) -> dict: sample_rate = self.sigan.sample_rate num_samples = int(sample_rate * self.duration_ms * 1e-3) measurement_result = self.acquire_data(num_samples, self.nskip) - measurement_result['start_time'] = start_time + measurement_result["start_time"] = start_time end_time = utils.get_datetime_str_now() measurement_result.update(self.parameters) - measurement_result['end_time'] = end_time - measurement_result['domain'] = Domain.TIME.value - measurement_result['measurement_type'] = MeasurementType.SINGLE_FREQUENCY.value - measurement_result['task_id'] = task_id - measurement_result['calibration_datetime'] = self.sigan.sensor_calibration_data['calibration_datetime'] - 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["end_time"] = end_time + measurement_result["domain"] = Domain.TIME.value + measurement_result["measurement_type"] = MeasurementType.SINGLE_FREQUENCY.value + measurement_result["task_id"] = task_id + measurement_result["calibration_datetime"] = self.sigan.sensor_calibration_data[ + "calibration_datetime" + ] + measurement_result["description"] = self.description + measurement_result["sigan_cal"] = self.sigan.sigan_calibration_data + measurement_result["sensor_cal"] = self.sigan.sensor_calibration_data return measurement_result def get_sigmf_builder(self, measurement_result: dict) -> SigMFBuilder: @@ -102,6 +107,7 @@ def get_sigmf_builder(self, measurement_result: dict) -> SigMFBuilder: start=0, count=self.received_samples, detector="sample_iq", + num_samps=self.received_samples, units="volts", reference="preselector input", ) @@ -138,7 +144,3 @@ def transform_data(self, measurement_result): def is_complex(self) -> bool: return True - - - - diff --git a/scos_actions/metadata/annotations/time_domain_annotation.py b/scos_actions/metadata/annotations/time_domain_annotation.py index c77bf25e..a1dfde4b 100644 --- a/scos_actions/metadata/annotations/time_domain_annotation.py +++ b/scos_actions/metadata/annotations/time_domain_annotation.py @@ -8,11 +8,13 @@ def __init__( start: int, count: int, detector: str, + num_samps: int, units: str, reference: str, ): super().__init__(start, count) self.detector = detector + self.num_samps = num_samps self.units = units self.reference = reference @@ -20,7 +22,7 @@ def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict) metadata = { "ntia-core:annotation_type": "TimeDomainDetection", "ntia-algorithm:detector": self.detector, - "ntia-algorithm:number_of_samples": self.count, + "ntia-algorithm:number_of_samples": self.num_samps, "ntia-algorithm:units": self.units, "ntia-algorithm:reference": self.reference, } From 66a6cb821f475e313bf675b40ed346312eaf5ce4 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Wed, 17 Aug 2022 11:59:46 -0600 Subject: [PATCH 08/25] APD annotation progress --- .../probability_distribution_annotation.py | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 scos_actions/metadata/annotations/probability_distribution_annotation.py diff --git a/scos_actions/metadata/annotations/probability_distribution_annotation.py b/scos_actions/metadata/annotations/probability_distribution_annotation.py new file mode 100644 index 00000000..c26afc91 --- /dev/null +++ b/scos_actions/metadata/annotations/probability_distribution_annotation.py @@ -0,0 +1,34 @@ +from scos_actions.metadata.metadata import Metadata +from scos_actions.metadata.sigmf_builder import SigMFBuilder + + +class ProbabilityDistributionAnnotation(Metadata): + def __init__( + self, + start: int, + count: int, + function: str, + units: str, + probability_units: str, + ): + super().__init__(start, count) + self.detector = detector + self.num_samps = num_samps + self.units = units + self.reference = reference + + def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): + metadata = { + "ntia-core:annotation_type": "ProbabilityDistributionAnnotation", + "ntia-algorithm:function": self.function, + "ntia-algorithm:units": self.units, + "ntia-algorithm:probability_units": self.probability_units, + "ntia-algorithm:number_of_samples": self.num_samps, + "ntia-algorithm:reference": self.reference, + "ntia-algorithm:probability_start": self.probability_start, + "ntia-algorithm:probability_stop": self.probability_stop, + "ntia-algorithm:probabilities": self.probabilities, + "ntia-algorithm:downsampled": self.downsampled, + "ntia-algorithm:downsampling_method": self.downsampling_method, + } + sigmf_builder.add_annotation(self.start, self.count, metadata) From 66fdb3de5aa8b785ed52d60007f2484284fc5b1b Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Thu, 25 Aug 2022 12:56:23 -0600 Subject: [PATCH 09/25] Use dataclasses for annotation segments --- .../actions/acquire_single_freq_fft.py | 6 +- .../actions/acquire_single_freq_tdomain_iq.py | 12 +- scos_actions/metadata/annotation_segment.py | 76 ++++++++++++ scos_actions/metadata/annotations/__init__.py | 7 ++ .../metadata/annotations/fft_annotation.py | 47 -------- .../annotations/frequency_domain_detection.py | 77 +++++++++++++ .../probability_distribution_annotation.py | 109 +++++++++++++----- .../annotations/time_domain_annotation.py | 29 ----- .../annotations/time_domain_detection.py | 47 ++++++++ scos_actions/metadata/measurement_global.py | 5 +- scos_actions/metadata/metadata.py | 14 --- 11 files changed, 295 insertions(+), 134 deletions(-) create mode 100644 scos_actions/metadata/annotation_segment.py delete mode 100644 scos_actions/metadata/annotations/fft_annotation.py create mode 100644 scos_actions/metadata/annotations/frequency_domain_detection.py delete mode 100644 scos_actions/metadata/annotations/time_domain_annotation.py create mode 100644 scos_actions/metadata/annotations/time_domain_detection.py delete mode 100644 scos_actions/metadata/metadata.py diff --git a/scos_actions/actions/acquire_single_freq_fft.py b/scos_actions/actions/acquire_single_freq_fft.py index 4d03fdda..d605e444 100644 --- a/scos_actions/actions/acquire_single_freq_fft.py +++ b/scos_actions/actions/acquire_single_freq_fft.py @@ -93,9 +93,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.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, @@ -243,7 +241,7 @@ 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( + fft_annotation = FrequencyDomainDetection( start=i * self.fft_size, count=self.fft_size, fft_size=self.fft_size, diff --git a/scos_actions/actions/acquire_single_freq_tdomain_iq.py b/scos_actions/actions/acquire_single_freq_tdomain_iq.py index d9eb9c95..987243ee 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 @@ -103,11 +101,11 @@ def execute(self, schedule_entry, task_id) -> dict: def get_sigmf_builder(self, measurement_result: dict) -> SigMFBuilder: sigmf_builder = super().get_sigmf_builder(measurement_result) - time_domain_annotation = TimeDomainAnnotation( - start=0, - count=self.received_samples, + time_domain_annotation = TimeDomainDetection( + sample_start=0, + sample_count=self.received_samples, detector="sample_iq", - num_samps=self.received_samples, + number_of_samples=self.received_samples, units="volts", reference="preselector input", ) diff --git a/scos_actions/metadata/annotation_segment.py b/scos_actions/metadata/annotation_segment.py new file mode 100644 index 00000000..23a0fdd8 --- /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] = "SCOS Sensor" + 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..bda08f7d 100644 --- a/scos_actions/metadata/annotations/__init__.py +++ b/scos_actions/metadata/annotations/__init__.py @@ -0,0 +1,7 @@ +from scos_actions.metadata.annotations.frequency_domain_detection import ( + FrequencyDomainDetection, +) +from scos_actions.metadata.annotations.probability_distribution_annotation import ( + ProbabilityDistributionAnnotation, +) +from scos_actions.metadata.annotations.time_domain_detection import TimeDomainDetection diff --git a/scos_actions/metadata/annotations/fft_annotation.py b/scos_actions/metadata/annotations/fft_annotation.py deleted file mode 100644 index cbd1b949..00000000 --- a/scos_actions/metadata/annotations/fft_annotation.py +++ /dev/null @@ -1,47 +0,0 @@ -from scos_actions.metadata.metadata import Metadata -from scos_actions.metadata.sigmf_builder import SigMFBuilder - - -class FrequencyDomainDetectionAnnotation(Metadata): - def __init__( - self, - start: int, - count: int, - fft_size: int, - window: str, - enbw: float, - detector: str, - nffts: int, - units: str, - reference: str, - frequency_start: float, - frequency_stop: float, - frequency_step: float, - ): - super().__init__(start, count) - self.fft_size = fft_size - self.window = window - self.enbw = enbw - self.detector = detector - self.nffts = nffts - self.units = units - self.reference = reference - self.frequency_start = frequency_start - self.frequency_stop = frequency_stop - self.frequency_step = frequency_step - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): - metadata = { - "ntia-core:annotation_type": "FrequencyDomainDetection", - "ntia-algorithm:number_of_samples_in_fft": self.fft_size, - "ntia-algorithm:window": self.window, - "ntia-algorithm:equivalent_noise_bandwidth": self.enbw, - "ntia-algorithm:detector": "fft_" + self.detector, - "ntia-algorithm:number_of_ffts": self.nffts, - "ntia-algorithm:units": self.units, - "ntia-algorithm:reference": self.reference, - "ntia-algorithm:frequency_start": self.frequency_start, - "ntia-algorithm:frequency_stop": self.frequency_stop, - "ntia-algorithm:frequency_step": self.frequency_step, - } - sigmf_builder.add_annotation(self.start, self.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/probability_distribution_annotation.py b/scos_actions/metadata/annotations/probability_distribution_annotation.py index c26afc91..1e6c676a 100644 --- a/scos_actions/metadata/annotations/probability_distribution_annotation.py +++ b/scos_actions/metadata/annotations/probability_distribution_annotation.py @@ -1,34 +1,81 @@ -from scos_actions.metadata.metadata import Metadata +from dataclasses import dataclass +from typing import List, Optional + +from scos_actions.metadata.annotation_segment import AnnotationSegment from scos_actions.metadata.sigmf_builder import SigMFBuilder -class ProbabilityDistributionAnnotation(Metadata): - def __init__( - self, - start: int, - count: int, - function: str, - units: str, - probability_units: str, - ): - super().__init__(start, count) - self.detector = detector - self.num_samps = num_samps - self.units = units - self.reference = reference - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): - metadata = { - "ntia-core:annotation_type": "ProbabilityDistributionAnnotation", - "ntia-algorithm:function": self.function, - "ntia-algorithm:units": self.units, - "ntia-algorithm:probability_units": self.probability_units, - "ntia-algorithm:number_of_samples": self.num_samps, - "ntia-algorithm:reference": self.reference, - "ntia-algorithm:probability_start": self.probability_start, - "ntia-algorithm:probability_stop": self.probability_stop, - "ntia-algorithm:probabilities": self.probabilities, - "ntia-algorithm:downsampled": self.downsampled, - "ntia-algorithm:downsampling_method": self.downsampling_method, - } - sigmf_builder.add_annotation(self.start, self.count, metadata) +@dataclass +class ProbabilityDistributionAnnotation(AnnotationSegment): + """ + Interface for generating ProbabilityDistributionAnnotation segments. + + Refer to the documentation of the ``ntia-algorithm`` extension of + SigMF for more information. + + The parameters ``function``, ``units``, and ``probability_units`` + are required. + + :param function: The estimated probability distribution function, e.g. + ``"cumulative distribution"``, ``"probability density"``, + ``"amplitude probability distribution"``. + :param units: Data units, e.g. ``"dBm"``, ``"volts"``, ``"watts"``. + :param probability_units: Unit of the probability values, generally + either ``"dimensionless"`` or ``"percent"``. + :param number_of_samples: Number of samples used to estimate the + probability distribution function. In the case of a downsampled + result, this number may be larger than the length of the annotated + data. + :param reference: Data reference point, e.g. ``"signal analyzer input"``, + ``"preselector input"``, ``"antenna terminal"``. + :param probability_start: Probability of the first data point, in units + specified by ``probability_units``. + :param probability_stop: Probability of the last data point, in units + specified by ``probability_units``. + :param probability_step: Step size, in ``probability_units``, between + data points. This should only be used if the step size is constant + across all data points. + :param probabilities: A list of the probabilities for all data points. + This must be used if the probability step size is not constant. + :param downsampled: Whether or not the probability distribution data + has been downsampled. + :param downsampling_method: The method used for downsampling, e.g. + ``"uniform downsampling by a factor of 2"``, etc. + """ + + function: Optional[str] = None + units: Optional[str] = None + probability_units: Optional[str] = None + number_of_samples: Optional[int] = None + reference: Optional[str] = None + probability_start: Optional[float] = None + probability_stop: Optional[float] = None + probability_step: Optional[float] = None + probabilities: Optional[List[float]] = None + downsampled: Optional[bool] = None + downsampling_method: Optional[str] = None + + def __post_init__(self): + super().__post_init__() + # Ensure required keys have been set + self.check_required(self.function, "function") + self.check_required(self.units, "units") + self.check_required(self.probability_units, "probability_units") + # Define SigMF key names + self.sigmf_keys.update( + { + "function": "ntia-algorithm:function", + "units": "ntia-algorithm:units", + "probability_units": "ntia-algorithm:probability_units", + "number_of_samples": "ntia-algorithm:number_of_samples", + "reference": "ntia-algorithm:reference", + "probability_start": "ntia-algorithm:probability_start", + "probability_stop": "ntia-algorithm:probability_stop", + "probability_step": "ntia-algorithm:probability_step", + "probabilities": "ntia-algorithm:probabilities", + "downsampled": "ntia-algorithm:downsampled", + "downsampling_method": "ntia-algorithm:downsampling_method", + } + ) + # 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 a1dfde4b..00000000 --- a/scos_actions/metadata/annotations/time_domain_annotation.py +++ /dev/null @@ -1,29 +0,0 @@ -from scos_actions.metadata.metadata import Metadata -from scos_actions.metadata.sigmf_builder import SigMFBuilder - - -class TimeDomainAnnotation(Metadata): - def __init__( - self, - start: int, - count: int, - detector: str, - num_samps: int, - units: str, - reference: str, - ): - super().__init__(start, count) - self.detector = detector - self.num_samps = num_samps - self.units = units - self.reference = reference - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): - metadata = { - "ntia-core:annotation_type": "TimeDomainDetection", - "ntia-algorithm:detector": self.detector, - "ntia-algorithm:number_of_samples": self.num_samps, - "ntia-algorithm:units": self.units, - "ntia-algorithm:reference": self.reference, - } - sigmf_builder.add_annotation(self.start, self.count, metadata) 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..1b707d97 100644 --- a/scos_actions/metadata/measurement_global.py +++ b/scos_actions/metadata/measurement_global.py @@ -1,8 +1,9 @@ -from scos_actions.metadata.metadata import Metadata +from scos_actions.metadata.annotation_segment import AnnotationSegment from scos_actions.metadata.sigmf_builder import SigMFBuilder -class MeasurementMetadata(Metadata): +# TODO: Decouple this from AnnotationSegment +class MeasurementMetadata(AnnotationSegment): def __init__(self): super().__init__() 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 From f207bfbe91fa11c0a83886ed804a94f844f56762 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 09:55:00 -0600 Subject: [PATCH 10/25] Removed default param value --- scos_actions/metadata/annotation_segment.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scos_actions/metadata/annotation_segment.py b/scos_actions/metadata/annotation_segment.py index 23a0fdd8..233b122b 100644 --- a/scos_actions/metadata/annotation_segment.py +++ b/scos_actions/metadata/annotation_segment.py @@ -26,7 +26,7 @@ class AnnotationSegment(ABC): sample_start: Optional[int] = None sample_count: Optional[int] = None - generator: Optional[str] = "SCOS Sensor" + generator: Optional[str] = None label: Optional[str] = None comment: Optional[str] = None freq_lower_edge: Optional[float] = None From 6c15126ed1c092b7b61b5927b9b8b5cbbb0bc9fe Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 10:16:51 -0600 Subject: [PATCH 11/25] Decouple MeasurementMetadata from AnnotationSegment --- scos_actions/metadata/measurement_global.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/scos_actions/metadata/measurement_global.py b/scos_actions/metadata/measurement_global.py index 1b707d97..a15c7e38 100644 --- a/scos_actions/metadata/measurement_global.py +++ b/scos_actions/metadata/measurement_global.py @@ -1,11 +1,9 @@ -from scos_actions.metadata.annotation_segment import AnnotationSegment from scos_actions.metadata.sigmf_builder import SigMFBuilder -# TODO: Decouple this from AnnotationSegment -class MeasurementMetadata(AnnotationSegment): +class MeasurementMetadata: def __init__(self): - super().__init__() + pass def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): freq_low = None From 93acac1b5a9c7aea56dfba47efab3657ca5da5bf Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 10:20:33 -0600 Subject: [PATCH 12/25] Update FFT action for new annotation code --- scos_actions/actions/acquire_single_freq_fft.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/scos_actions/actions/acquire_single_freq_fft.py b/scos_actions/actions/acquire_single_freq_fft.py index d605e444..452d1c42 100644 --- a/scos_actions/actions/acquire_single_freq_fft.py +++ b/scos_actions/actions/acquire_single_freq_fft.py @@ -242,13 +242,13 @@ 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 = FrequencyDomainDetection( - start=i * self.fft_size, - count=self.fft_size, - fft_size=self.fft_size, - window=self.fft_window_type, - enbw=measurement_result["enbw"], + sample_start=i * self.fft_size, + sample_count=self.fft_size, detector=detector.value, - nffts=self.nffts, + 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"], From 8e20039dfb722bf31a7414d2541ee587c52d212c Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 11:50:01 -0600 Subject: [PATCH 13/25] Update CalibrationAnnotation to dataclass --- .../annotations/calibration_annotation.py | 112 ++++++++++++------ 1 file changed, 76 insertions(+), 36 deletions(-) diff --git a/scos_actions/metadata/annotations/calibration_annotation.py b/scos_actions/metadata/annotations/calibration_annotation.py index aa34a2c3..702c7b1c 100644 --- a/scos_actions/metadata/annotations/calibration_annotation.py +++ b/scos_actions/metadata/annotations/calibration_annotation.py @@ -1,36 +1,76 @@ -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 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. + The sensor and sigan calibration 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``. + :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. + :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 + self.gain_sigan = self.sigan_cal["gain_sigan"] + self.noise_figure_sigan = self.sigan_cal["noise_figure_sigan"] + self.compression_point_sigan = self.sigan_cal["1db_compression_sigan"] + self.enbw_sigan = self.sigan_cal["enbw_sigan"] + self.gain_preselector = self.sensor_cal["gain_preselector"] + self.noise_figure_sensor = self.sensor_cal["noise_figure_sensor"] + self.compression_point_sensor = self.sensor_cal["1db_compression_sensor"] + self.enbw_sensor = self.sensor_cal["enbw_sensor"] + if "temperature" in self.sensor_cal: + self.temperature = self.sensor_cal["temperature"] + else: + self.temperature = None + # Additional key gain_sensor is not in SigMF ntia-sensor spec but is included + self.gain_sensor = 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() From c72040bc5e40994700ee95660a92a4284f65ef07 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 12:05:17 -0600 Subject: [PATCH 14/25] Switch sensor annotation to dataclass --- .../metadata/annotations/sensor_annotation.py | 65 +++++++++++++------ 1 file changed, 46 insertions(+), 19 deletions(-) diff --git a/scos_actions/metadata/annotations/sensor_annotation.py b/scos_actions/metadata/annotations/sensor_annotation.py index de38a8db..f80cc409 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:attenutation_setting_sigan", + "gain_setting_sigan": "ntia-sensor:gain_setting_sigan", + "gps_nmea": "ntia-sensor:gps_nmea", + } + ) + # Create annotation segment + super().create_annotation_segment() From 7f3893e38220d531ceecf13bedf78b93e5c802c3 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 12:22:49 -0600 Subject: [PATCH 15/25] Update measurement_action for new annotations --- .../actions/interfaces/measurement_action.py | 28 +++++++++++++++---- 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/scos_actions/actions/interfaces/measurement_action.py b/scos_actions/actions/interfaces/measurement_action.py index 4e9fd620..ce7c2643 100644 --- a/scos_actions/actions/interfaces/measurement_action.py +++ b/scos_actions/actions/interfaces/measurement_action.py @@ -1,14 +1,12 @@ import logging from abc import abstractmethod +from typing import overload 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.hardware import sigan as mock_sigan -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 @@ -39,7 +37,12 @@ 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 ) @@ -47,7 +50,20 @@ def get_sigmf_builder(self, measurement_result) -> SigMFBuilder: 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 ) From 511d5054a9fc749a53055ab3cf4cb30f8c3de646 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 12:23:17 -0600 Subject: [PATCH 16/25] import all annotations to .annotations --- scos_actions/metadata/annotations/__init__.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/scos_actions/metadata/annotations/__init__.py b/scos_actions/metadata/annotations/__init__.py index bda08f7d..803f95af 100644 --- a/scos_actions/metadata/annotations/__init__.py +++ b/scos_actions/metadata/annotations/__init__.py @@ -1,7 +1,11 @@ +from scos_actions.metadata.annotations.calibration_annotation import ( + CalibrationAnnotation, +) from scos_actions.metadata.annotations.frequency_domain_detection import ( FrequencyDomainDetection, ) from scos_actions.metadata.annotations.probability_distribution_annotation import ( ProbabilityDistributionAnnotation, ) +from scos_actions.metadata.annotations.sensor_annotation import SensorAnnotation from scos_actions.metadata.annotations.time_domain_detection import TimeDomainDetection From 3c0fbf40b21ead56f45e384882eafea1233e19df Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 13:33:47 -0600 Subject: [PATCH 17/25] Update create_metadata --- .../actions/acquire_single_freq_fft.py | 1 + .../actions/acquire_single_freq_tdomain_iq.py | 1 + .../acquire_stepped_freq_tdomain_iq.py | 1 + .../actions/interfaces/measurement_action.py | 20 ++++- .../annotations/calibration_annotation.py | 4 +- scos_actions/metadata/measurement_global.py | 77 ++++++++++++------- scos_actions/metadata/sigmf_builder.py | 4 +- 7 files changed, 74 insertions(+), 34 deletions(-) diff --git a/scos_actions/actions/acquire_single_freq_fft.py b/scos_actions/actions/acquire_single_freq_fft.py index 452d1c42..073fad49 100644 --- a/scos_actions/actions/acquire_single_freq_fft.py +++ b/scos_actions/actions/acquire_single_freq_fft.py @@ -189,6 +189,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: diff --git a/scos_actions/actions/acquire_single_freq_tdomain_iq.py b/scos_actions/actions/acquire_single_freq_tdomain_iq.py index 987243ee..e60d50bc 100644 --- a/scos_actions/actions/acquire_single_freq_tdomain_iq.py +++ b/scos_actions/actions/acquire_single_freq_tdomain_iq.py @@ -97,6 +97,7 @@ 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: 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 ce7c2643..df5d3204 100644 --- a/scos_actions/actions/interfaces/measurement_action.py +++ b/scos_actions/actions/interfaces/measurement_action.py @@ -1,6 +1,5 @@ import logging from abc import abstractmethod -from typing import overload from scos_actions.actions.interfaces.action import Action from scos_actions.actions.interfaces.signals import measurement_action_completed @@ -46,7 +45,24 @@ def get_sigmf_builder(self, measurement_result) -> SigMFBuilder: 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 ) diff --git a/scos_actions/metadata/annotations/calibration_annotation.py b/scos_actions/metadata/annotations/calibration_annotation.py index 702c7b1c..d6203fdd 100644 --- a/scos_actions/metadata/annotations/calibration_annotation.py +++ b/scos_actions/metadata/annotations/calibration_annotation.py @@ -31,8 +31,8 @@ class CalibrationAnnotation(AnnotationSegment): e.g. ``"signal analyzer input"``, ``"preselector input"``, ``"antenna terminal"``. """ - sigan_cal = Optional[dict] = None - sensor_cal = Optional[dict] = None + 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 diff --git a/scos_actions/metadata/measurement_global.py b/scos_actions/metadata/measurement_global.py index a15c7e38..97ac6ab4 100644 --- a/scos_actions/metadata/measurement_global.py +++ b/scos_actions/metadata/measurement_global.py @@ -1,35 +1,56 @@ +from dataclasses import dataclass +from datetime import datetime +from typing import Any, List, Optional + from scos_actions.metadata.sigmf_builder import SigMFBuilder +@dataclass class MeasurementMetadata: - def __init__(self): - pass - - def create_metadata(self, sigmf_builder: SigMFBuilder, measurement_result: dict): - freq_low = None - freq_high = None + 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 - 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"] + 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", + } - 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, - }, + 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/sigmf_builder.py b/scos_actions/metadata/sigmf_builder.py index a36b69a4..ec8801e2 100644 --- a/scos_actions/metadata/sigmf_builder.py +++ b/scos_actions/metadata/sigmf_builder.py @@ -160,6 +160,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, measurement_result: dict): for metadata_creator in self.metadata_generators.values(): - metadata_creator.create_metadata(self, measurement_result) + metadata_creator.create_metadata(self) From f306070e6ca72dbc12334158a6525a771dacccb5 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 14:04:02 -0600 Subject: [PATCH 18/25] Added docstring for measurement metadata --- scos_actions/metadata/measurement_global.py | 22 +++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/scos_actions/metadata/measurement_global.py b/scos_actions/metadata/measurement_global.py index 97ac6ab4..a04cb2b0 100644 --- a/scos_actions/metadata/measurement_global.py +++ b/scos_actions/metadata/measurement_global.py @@ -7,6 +7,28 @@ @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] From 1f1802649a45772b35fe4cdad575eaf570ab6053 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 14:35:09 -0600 Subject: [PATCH 19/25] Remove unused parameter --- scos_actions/actions/interfaces/measurement_action.py | 2 +- scos_actions/metadata/sigmf_builder.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/scos_actions/actions/interfaces/measurement_action.py b/scos_actions/actions/interfaces/measurement_action.py index df5d3204..d641c245 100644 --- a/scos_actions/actions/interfaces/measurement_action.py +++ b/scos_actions/actions/interfaces/measurement_action.py @@ -96,7 +96,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/sigmf_builder.py b/scos_actions/metadata/sigmf_builder.py index ec8801e2..9e054da3 100644 --- a/scos_actions/metadata/sigmf_builder.py +++ b/scos_actions/metadata/sigmf_builder.py @@ -160,6 +160,6 @@ def add_metadata_generator(self, key, generator): def remove_metadata_generator(self, key): self.metadata_generators.pop(key, "") - def build(self, measurement_result: dict): + def build(self): for metadata_creator in self.metadata_generators.values(): metadata_creator.create_metadata(self) From 43eda9062ad2140b48df55a19c570f7ceb30c979 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Fri, 26 Aug 2022 14:37:27 -0600 Subject: [PATCH 20/25] Fixed typo --- scos_actions/metadata/annotations/sensor_annotation.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scos_actions/metadata/annotations/sensor_annotation.py b/scos_actions/metadata/annotations/sensor_annotation.py index f80cc409..9454bac0 100644 --- a/scos_actions/metadata/annotations/sensor_annotation.py +++ b/scos_actions/metadata/annotations/sensor_annotation.py @@ -37,7 +37,7 @@ def __post_init__(self): { "rf_path_index": "ntia-sensor:rf_path_index", "overload": "ntia-sensor:overload", - "attenuation_setting_sigan": "ntia-sensor:attenutation_setting_sigan", + "attenuation_setting_sigan": "ntia-sensor:attenuation_setting_sigan", "gain_setting_sigan": "ntia-sensor:gain_setting_sigan", "gps_nmea": "ntia-sensor:gps_nmea", } From 05dee3aadfe422eaae8b9176f7d47ad3531c2cbf Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Tue, 30 Aug 2022 11:48:57 -0600 Subject: [PATCH 21/25] Remove WIP ProbabilityDistributionAnnotation --- scos_actions/metadata/annotations/__init__.py | 3 - .../probability_distribution_annotation.py | 81 ------------------- 2 files changed, 84 deletions(-) delete mode 100644 scos_actions/metadata/annotations/probability_distribution_annotation.py diff --git a/scos_actions/metadata/annotations/__init__.py b/scos_actions/metadata/annotations/__init__.py index 803f95af..3d435c7a 100644 --- a/scos_actions/metadata/annotations/__init__.py +++ b/scos_actions/metadata/annotations/__init__.py @@ -4,8 +4,5 @@ from scos_actions.metadata.annotations.frequency_domain_detection import ( FrequencyDomainDetection, ) -from scos_actions.metadata.annotations.probability_distribution_annotation import ( - ProbabilityDistributionAnnotation, -) 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/probability_distribution_annotation.py b/scos_actions/metadata/annotations/probability_distribution_annotation.py deleted file mode 100644 index 1e6c676a..00000000 --- a/scos_actions/metadata/annotations/probability_distribution_annotation.py +++ /dev/null @@ -1,81 +0,0 @@ -from dataclasses import dataclass -from typing import List, Optional - -from scos_actions.metadata.annotation_segment import AnnotationSegment -from scos_actions.metadata.sigmf_builder import SigMFBuilder - - -@dataclass -class ProbabilityDistributionAnnotation(AnnotationSegment): - """ - Interface for generating ProbabilityDistributionAnnotation segments. - - Refer to the documentation of the ``ntia-algorithm`` extension of - SigMF for more information. - - The parameters ``function``, ``units``, and ``probability_units`` - are required. - - :param function: The estimated probability distribution function, e.g. - ``"cumulative distribution"``, ``"probability density"``, - ``"amplitude probability distribution"``. - :param units: Data units, e.g. ``"dBm"``, ``"volts"``, ``"watts"``. - :param probability_units: Unit of the probability values, generally - either ``"dimensionless"`` or ``"percent"``. - :param number_of_samples: Number of samples used to estimate the - probability distribution function. In the case of a downsampled - result, this number may be larger than the length of the annotated - data. - :param reference: Data reference point, e.g. ``"signal analyzer input"``, - ``"preselector input"``, ``"antenna terminal"``. - :param probability_start: Probability of the first data point, in units - specified by ``probability_units``. - :param probability_stop: Probability of the last data point, in units - specified by ``probability_units``. - :param probability_step: Step size, in ``probability_units``, between - data points. This should only be used if the step size is constant - across all data points. - :param probabilities: A list of the probabilities for all data points. - This must be used if the probability step size is not constant. - :param downsampled: Whether or not the probability distribution data - has been downsampled. - :param downsampling_method: The method used for downsampling, e.g. - ``"uniform downsampling by a factor of 2"``, etc. - """ - - function: Optional[str] = None - units: Optional[str] = None - probability_units: Optional[str] = None - number_of_samples: Optional[int] = None - reference: Optional[str] = None - probability_start: Optional[float] = None - probability_stop: Optional[float] = None - probability_step: Optional[float] = None - probabilities: Optional[List[float]] = None - downsampled: Optional[bool] = None - downsampling_method: Optional[str] = None - - def __post_init__(self): - super().__post_init__() - # Ensure required keys have been set - self.check_required(self.function, "function") - self.check_required(self.units, "units") - self.check_required(self.probability_units, "probability_units") - # Define SigMF key names - self.sigmf_keys.update( - { - "function": "ntia-algorithm:function", - "units": "ntia-algorithm:units", - "probability_units": "ntia-algorithm:probability_units", - "number_of_samples": "ntia-algorithm:number_of_samples", - "reference": "ntia-algorithm:reference", - "probability_start": "ntia-algorithm:probability_start", - "probability_stop": "ntia-algorithm:probability_stop", - "probability_step": "ntia-algorithm:probability_step", - "probabilities": "ntia-algorithm:probabilities", - "downsampled": "ntia-algorithm:downsampled", - "downsampling_method": "ntia-algorithm:downsampling_method", - } - ) - # Create annotation segment - super().create_annotation_segment() From 0350f0eda63a5e5cf0e4f0fff7a8c89f5d040633 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Thu, 1 Sep 2022 20:07:58 -0600 Subject: [PATCH 22/25] Make calibration keys optional as in spec --- .../annotations/calibration_annotation.py | 50 ++++++++++++------- 1 file changed, 33 insertions(+), 17 deletions(-) diff --git a/scos_actions/metadata/annotations/calibration_annotation.py b/scos_actions/metadata/annotations/calibration_annotation.py index d6203fdd..2b9f8d82 100644 --- a/scos_actions/metadata/annotations/calibration_annotation.py +++ b/scos_actions/metadata/annotations/calibration_annotation.py @@ -1,5 +1,5 @@ from dataclasses import dataclass -from typing import Optional +from typing import Any, Optional from scos_actions.metadata.annotation_segment import AnnotationSegment @@ -11,7 +11,7 @@ class CalibrationAnnotation(AnnotationSegment): Most values are read from sensor and sigan calibrations, expected to exist in a ``measurement_result`` dictionary. - The sensor and sigan calibration parameters are required. + No parameters are required. Refer to the documentation of the ``ntia-sensor`` extension of SigMF for more information. @@ -19,12 +19,12 @@ class CalibrationAnnotation(AnnotationSegment): :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``. + 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. + 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``, @@ -40,20 +40,27 @@ class CalibrationAnnotation(AnnotationSegment): def __post_init__(self): super().__post_init__() # Load values from sensor and sigan calibrations - self.gain_sigan = self.sigan_cal["gain_sigan"] - self.noise_figure_sigan = self.sigan_cal["noise_figure_sigan"] - self.compression_point_sigan = self.sigan_cal["1db_compression_sigan"] - self.enbw_sigan = self.sigan_cal["enbw_sigan"] - self.gain_preselector = self.sensor_cal["gain_preselector"] - self.noise_figure_sensor = self.sensor_cal["noise_figure_sensor"] - self.compression_point_sensor = self.sensor_cal["1db_compression_sensor"] - self.enbw_sensor = self.sensor_cal["enbw_sensor"] - if "temperature" in self.sensor_cal: - self.temperature = self.sensor_cal["temperature"] - else: - self.temperature = 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") + 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.sensor_cal["gain_sensor"] + self.gain_sensor = self.get_cal_value_if_exists(self.sensor_cal, "gain_sensor") # Define SigMF key names self.sigmf_keys.update( { @@ -74,3 +81,12 @@ def __post_init__(self): ) # 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 From d0a3fe001748de790a22806f6d8923f3cbcf3662 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Thu, 1 Sep 2022 20:32:12 -0600 Subject: [PATCH 23/25] Add metadata exception class --- scos_actions/metadata/metadata_exception.py | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 scos_actions/metadata/metadata_exception.py 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) From 05900e39dd4d3719ea94f96bc1237a0a6210b6ea Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Thu, 1 Sep 2022 20:59:46 -0600 Subject: [PATCH 24/25] Allow setting datatypes other than cf32_le --- scos_actions/metadata/sigmf_builder.py | 58 ++++++++++++++++++++++---- 1 file changed, 50 insertions(+), 8 deletions(-) diff --git a/scos_actions/metadata/sigmf_builder.py b/scos_actions/metadata/sigmf_builder.py index 9e054da3..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) From 00d93e1d7cfc3d9d62dae3ff2f8b342493d6b553 Mon Sep 17 00:00:00 2001 From: Anthony Romaniello Date: Thu, 1 Sep 2022 21:20:11 -0600 Subject: [PATCH 25/25] Skip attempting to load values if not provided --- .../annotations/calibration_annotation.py | 50 +++++++++++-------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/scos_actions/metadata/annotations/calibration_annotation.py b/scos_actions/metadata/annotations/calibration_annotation.py index 2b9f8d82..7178c5a4 100644 --- a/scos_actions/metadata/annotations/calibration_annotation.py +++ b/scos_actions/metadata/annotations/calibration_annotation.py @@ -40,27 +40,35 @@ class CalibrationAnnotation(AnnotationSegment): def __post_init__(self): super().__post_init__() # Load values from sensor and sigan calibrations - 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") - 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") + 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( {