Merge pull request #33 from NTIA/blacken

Run pre-commit hooks on all files

.markdownlint.yaml

@@ -1,3 +1,3 @@
 default: true
 MD013:
-  line_length: 88
+  line_length: 88

.pre-commit-config.yaml

@@ -21,6 +21,7 @@ repos:
     rev: v2.34.0
     hooks:
       - id: pyupgrade
+        args: ["--py3-plus"]
   - repo: https://github.com/pycqa/isort
     rev: 5.10.1
     hooks:

scos_actions/__init__.py

@@ -1 +1 @@
-__version__ = "3.0.0"
+__version__ = "3.0.1"

scos_actions/actions/__init__.py

@@ -3,18 +3,11 @@
 from .acquire_stepped_freq_tdomain_iq import SteppedFrequencyTimeDomainIqAcquisition
 from .calibrate_y_factor import YFactorCalibration
 
-
-
-
-
 # Map a class name to an action class
 # The YAML loader can key an object with parameters on these class names
 action_classes = {
     "single_frequency_fft": SingleFrequencyFftAcquisition,
     "stepped_frequency_time_domain_iq": SteppedFrequencyTimeDomainIqAcquisition,
     "single_frequency_time_domain_iq": SingleFrequencyTimeDomainIqAcquisition,
-    "y_factor_cal": YFactorCalibration
+    "y_factor_cal": YFactorCalibration,
 }
-
-
-

scos_actions/actions/acquire_single_freq_fft.py

@@ -87,27 +87,33 @@
 """
 
 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,
     get_fft_frequencies,
     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,27 @@ 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["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:
@@ -202,7 +210,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

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:

scos_actions/actions/acquire_stepped_freq_tdomain_iq.py

@@ -43,10 +43,10 @@
 from scos_actions.actions.acquire_single_freq_tdomain_iq import (
     SingleFrequencyTimeDomainIqAcquisition,
 )
-from scos_actions.utils import get_parameter
 from scos_actions.actions.interfaces.signals import measurement_action_completed
-from scos_actions.metadata.sigmf_builder import Domain, MeasurementType
 from scos_actions.hardware import gps as mock_gps
+from scos_actions.metadata.sigmf_builder import Domain, MeasurementType
+from scos_actions.utils import get_parameter
 
 logger = logging.getLogger(__name__)
 
@@ -84,7 +84,7 @@ def __init__(self, parameters, sigan, gps=mock_gps):
 
         # Create iterable parameter set
         self.iterable_params = utils.get_iterable_parameters(parameters)
-        
+
         self.sigan = sigan  # make instance variable to allow mocking
         self.num_center_frequencies = num_center_frequencies
 
@@ -104,15 +104,17 @@ def __call__(self, schedule_entry_json, task_id):
             measurement_result = super().acquire_data(num_samples, nskip)
             measurement_result.update(measurement_params)
             end_time = utils.get_datetime_str_now()
-            measurement_result['start_time'] = start_time
-            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['description'] = self.description
-            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["start_time"] = start_time
+            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["description"] = self.description
+            measurement_result["name"] = self.name
+            measurement_result["sigan_cal"] = self.sigan.sigan_calibration_data
+            measurement_result["sensor_cal"] = self.sigan.sensor_calibration_data
             sigmf_builder = self.get_sigmf_builder(measurement_result)
             self.create_metadata(
                 sigmf_builder, schedule_entry_json, measurement_result, recording_id

scos_actions/actions/calibrate_y_factor.py

@@ -285,8 +285,8 @@ def calibrate(self, params):
             noise_off_data = noise_off_measurement_result["data"]
 
         # Get power values in time domain (division by 2 for RF/baseband conversion)
-        pwr_on_watts = calculate_power_watts(noise_on_data / 2.0)
-        pwr_off_watts = calculate_power_watts(noise_off_data / 2.0)
+        pwr_on_watts = calculate_power_watts(noise_on_data) / 2.0
+        pwr_off_watts = calculate_power_watts(noise_off_data) / 2.0
 
         # Y-Factor
         enr_linear = get_linear_enr(cal_source_idx)

scos_actions/actions/interfaces/action.py

@@ -2,11 +2,11 @@
 from abc import ABC, abstractmethod
 from copy import deepcopy
 
-from scos_actions.utils import get_parameter
-from scos_actions.hardware import gps as mock_gps
-from scos_actions.hardware import sigan as mock_sigan
 from scos_actions.capabilities import capabilities
+from scos_actions.hardware import gps as mock_gps
 from scos_actions.hardware import preselector
+from scos_actions.hardware import sigan as mock_sigan
+from scos_actions.utils import get_parameter
 
 logger = logging.getLogger(__name__)
 
@@ -29,13 +29,14 @@ class Action(ABC):
         added to the task result's detail field.
 
     """
-    PRESELECTOR_PATH_KEY='rf_path'
+
+    PRESELECTOR_PATH_KEY = "rf_path"
 
     def __init__(self, parameters, sigan=mock_sigan, gps=mock_gps):
         self.parameters = deepcopy(parameters)
         self.sigan = sigan
         self.gps = gps
-        self.sensor_definition = capabilities['sensor']
+        self.sensor_definition = capabilities["sensor"]
 
     def configure(self, measurement_params: dict):
         self.configure_sigan(measurement_params)
@@ -72,6 +73,3 @@ def name(self):
     @abstractmethod
     def __call__(self, schedule_entry, task_id):
         pass
-
-
-

scos_actions/actions/interfaces/measurement_action.py

@@ -1,13 +1,15 @@
 import logging
-
 from abc import abstractmethod
+
 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.actions.interfaces.signals import measurement_action_completed
-from scos_actions.metadata.annotations.calibration_annotation import CalibrationAnnotation
-from scos_actions.metadata.measurement_global import MeasurementMetadata
+from scos_actions.metadata.annotations.calibration_annotation import (
+    CalibrationAnnotation,
+)
 from scos_actions.metadata.annotations.sensor_annotation import SensorAnnotation
+from scos_actions.metadata.measurement_global import MeasurementMetadata
 from scos_actions.metadata.sigmf_builder import SigMFBuilder
 
 logger = logging.getLogger(__name__)
@@ -38,18 +40,28 @@ 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)
-        sigmf_builder.add_metadata_generator(type(calibration_annotation).__name__, calibration_annotation)
+        sigmf_builder.add_metadata_generator(
+            type(calibration_annotation).__name__, calibration_annotation
+        )
         measurement_metadata = MeasurementMetadata()
-        sigmf_builder.add_metadata_generator(type(measurement_metadata).__name__, measurement_metadata)
+        sigmf_builder.add_metadata_generator(
+            type(measurement_metadata).__name__, measurement_metadata
+        )
         sensor_annotation = SensorAnnotation(0, self.received_samples)
-        sigmf_builder.add_metadata_generator(type(sensor_annotation).__name__, sensor_annotation)
+        sigmf_builder.add_metadata_generator(
+            type(sensor_annotation).__name__, sensor_annotation
+        )
         return sigmf_builder
 
-    def create_metadata(self, sigmf_builder, schedule_entry, measurement_result, recording=None):
+    def create_metadata(
+        self, sigmf_builder, schedule_entry, measurement_result, recording=None
+    ):
         sigmf_builder.set_base_sigmf_global(
             schedule_entry,
             self.sensor_definition,
-            measurement_result, recording, self.is_complex()
+            measurement_result,
+            recording,
+            self.is_complex(),
         )
         sigmf_builder.add_sigmf_capture(sigmf_builder, measurement_result)
         sigmf_builder.build(measurement_result)
@@ -87,4 +99,4 @@ def is_complex(self) -> bool:
 
     @abstractmethod
     def execute(self, schedule_entry, task_id):
-        pass
+        pass

scos_actions/actions/logger.py

@@ -30,4 +30,3 @@ def __call__(self, schedule_entry, task_id):
         logger.log(
             level=self.loglvl, msg=msg.format(name=schedule_entry_name, tid=task_id)
         )
-

scos_actions/actions/monitor_sigan.py

@@ -13,7 +13,7 @@ class MonitorSignalAnalyzer(Action):
     """Monitor signal analyzer connection and restart container if unreachable."""
 
     def __init__(self, sigan, parameters={}, gps=mock_gps):
-        super().__init__(parameters=parameters,sigan=sigan, gps=gps)
+        super().__init__(parameters=parameters, sigan=sigan, gps=gps)
 
     def __call__(self, schedule_entry_json, task_id):
         logger.debug("Performing signal analyzer health check")
@@ -30,4 +30,4 @@ def __call__(self, schedule_entry_json, task_id):
             logger.info("signal analyzer healthy")
         else:
             logger.warning("signal analyzer unhealthy")
-            monitor_action_completed.send(sender=self.__class__, sigan_healthy=False)
+            monitor_action_completed.send(sender=self.__class__, sigan_healthy=False)

scos_actions/actions/preselector_control/enable_antenna.py

@@ -12,8 +12,8 @@ class EnableAntenna(Action):
     """Set the preselector to the antenna"""
 
     def __init__(self, sigan):
-        super().__init__(parameters={'name':'enable_antenna'}, sigan=sigan)
+        super().__init__(parameters={"name": "enable_antenna"}, sigan=sigan)
 
     def __call__(self, schedule_entry_json, task_id):
         logger.debug("Enabling antenna")
-        preselector.set_state('antenna')
+        preselector.set_state("antenna")

scos_actions/actions/preselector_control/enable_noise_diode_off.py

@@ -12,8 +12,8 @@ class EnableNoiseDiodeOff(Action):
     """Set the preselector to noise diode off"""
 
     def __init__(self, sigan):
-        super().__init__(parameters={'name': 'enable_noise_diode_off'}, sigan=sigan)
+        super().__init__(parameters={"name": "enable_noise_diode_off"}, sigan=sigan)
 
     def __call__(self, schedule_entry_json, task_id):
         logger.debug("Setting noise diode OFF")
-        preselector.set_state('noise_diode_off')
+        preselector.set_state("noise_diode_off")

scos_actions/actions/preselector_control/enable_noise_diode_on.py

@@ -12,9 +12,8 @@ class EnableNoiseDiodeOn(Action):
     """Set the preselector to noise diode on"""
 
     def __init__(self, sigan):
-        super().__init__(parameters={'name': 'enable_noise_diode_on'}, sigan=sigan)
+        super().__init__(parameters={"name": "enable_noise_diode_on"}, sigan=sigan)
 
     def __call__(self, schedule_entry_json, task_id):
         logger.debug("Setting noise diode ON")
-        preselector.set_state('noise_diode_on')
-
+        preselector.set_state("noise_diode_on")