Fix circular import, add test survey action, add interpolated values to calibration files

.pre-commit-config.yaml

@@ -2,7 +2,7 @@ default_language_version:
   python: python3.8
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.1.0
+    rev: v4.3.0
     hooks:
       - id: check-ast
         types: [file, python]

README.md

@@ -306,7 +306,7 @@ Depending on the type of action, a signal should be sent upon action completion.
 enables SCOS Sensor to do something with the results of the action. This could range
 from storing measurement data to recycling a Docker container or to fixing an unhealthy
 connection to the signal analyzer. You can see the available signals in
-[`scos_actions/actions/interfaces/signals.py`](scos_actions/actions/interfaces/signals.py).
+[`scos_actions/signals.py`](scos_actions/signals.py).
 The following signals are currently offered:
 
 - `measurement_action_completed` - signal expects task_id, data, and metadata

sample_debug.py

@@ -5,8 +5,8 @@
 import json
 
 from scos_actions.actions.acquire_single_freq_fft import SingleFrequencyFftAcquisition
-from scos_actions.actions.interfaces.signals import measurement_action_completed
-from scos_actions.hardware import sigan
+from scos_actions.hardware.mocks.mock_sigan import MockSignalAnalyzer
+from scos_actions.signals import measurement_action_completed
 
 parameters = {
     "name": "test_single_frequency_m4s_action",
@@ -15,6 +15,8 @@
     "sample_rate": 15.36e6,
     "fft_size": 1024,
     "nffts": 300,
+    "nskip": 0,
+    "classification": "UNCLASSIFIED",
 }
 schedule_entry_json = {
     "name": "test_m4s_multi_1",
@@ -48,7 +50,9 @@ def callback(sender, **kwargs):
 
 measurement_action_completed.connect(callback)
 
-action = SingleFrequencyFftAcquisition(parameters=parameters, sigan=sigan)
+action = SingleFrequencyFftAcquisition(
+    parameters=parameters, sigan=MockSignalAnalyzer(randomize_values=True)
+)
 action(schedule_entry_json, 1)
 print("metadata:")
 print(json.dumps(_metadata, indent=4))

scos_actions/actions/acquire_single_freq_fft.py

@@ -91,7 +91,7 @@
 from numpy import float32, ndarray
 
 from scos_actions.actions.interfaces.measurement_action import MeasurementAction
-from scos_actions.hardware import gps as mock_gps
+from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.metadata.annotations import FrequencyDomainDetection
 from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder
 from scos_actions.signal_processing.fft import (
@@ -143,7 +143,9 @@ class SingleFrequencyFftAcquisition(MeasurementAction):
     :param sigan: Instance of SignalAnalyzerInterface.
     """
 
-    def __init__(self, parameters, sigan, gps=mock_gps):
+    def __init__(self, parameters, sigan, gps=None):
+        if gps is None:
+            gps = MockGPS()
         super().__init__(parameters, sigan, gps)
         # Pull parameters from action config
         self.fft_size = get_parameter(FFT_SIZE, self.parameters)

scos_actions/actions/acquire_single_freq_tdomain_iq.py

@@ -37,7 +37,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.hardware.mocks.mock_gps import MockGPS
 from scos_actions.metadata.annotations import TimeDomainDetection
 from scos_actions.metadata.sigmf_builder import Domain, MeasurementType, SigMFBuilder
 from scos_actions.utils import get_parameter
@@ -72,7 +72,9 @@ class SingleFrequencyTimeDomainIqAcquisition(MeasurementAction):
     :param sigan: instance of SignalAnalyzerInterface.
     """
 
-    def __init__(self, parameters, sigan, gps=mock_gps):
+    def __init__(self, parameters, sigan, gps=None):
+        if gps is None:
+            gps = MockGPS()
         super().__init__(parameters=parameters, sigan=sigan, gps=gps)
         # Pull parameters from action config
         self.nskip = get_parameter(NUM_SKIP, self.parameters)

scos_actions/actions/acquire_stepped_freq_tdomain_iq.py

@@ -46,9 +46,9 @@
     NUM_SKIP,
     SingleFrequencyTimeDomainIqAcquisition,
 )
-from scos_actions.actions.interfaces.signals import measurement_action_completed
-from scos_actions.hardware import gps as mock_gps
+from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.metadata.sigmf_builder import Domain, MeasurementType
+from scos_actions.signals import measurement_action_completed
 from scos_actions.utils import get_parameter
 
 logger = logging.getLogger(__name__)
@@ -75,7 +75,9 @@ class SteppedFrequencyTimeDomainIqAcquisition(SingleFrequencyTimeDomainIqAcquisi
     :param sigan: instance of SignalAnalyzerInterface
     """
 
-    def __init__(self, parameters, sigan, gps=mock_gps):
+    def __init__(self, parameters, sigan, gps=None):
+        if gps is None:
+            gps = MockGPS()
         super().__init__(parameters=parameters, sigan=sigan, gps=gps)
         num_center_frequencies = len(parameters[FREQUENCY])
 

scos_actions/actions/calibrate_y_factor.py

@@ -79,7 +79,7 @@
 from scos_actions import utils
 from scos_actions.actions.interfaces.action import Action
 from scos_actions.calibration import sensor_calibration
-from scos_actions.hardware import gps as mock_gps
+from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.settings import SENSOR_CALIBRATION_FILE
 from scos_actions.signal_processing.calibration import (
     get_linear_enr,
@@ -138,7 +138,9 @@ class YFactorCalibration(Action):
     :param sigan: instance of SignalAnalyzerInterface.
     """
 
-    def __init__(self, parameters, sigan, gps=mock_gps):
+    def __init__(self, parameters, sigan, gps=None):
+        if gps is None:
+            gps = MockGPS()
         logger.debug("Initializing calibration action")
         super().__init__(parameters, sigan, gps)
         self.sigan = sigan

scos_actions/actions/interfaces/action.py

@@ -3,11 +3,10 @@
 from copy import deepcopy
 
 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.mocks.mock_gps import MockGPS
 from scos_actions.utils import ParameterException, get_parameter
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -32,7 +31,9 @@ class Action(ABC):
 
     PRESELECTOR_PATH_KEY = "rf_path"
 
-    def __init__(self, parameters, sigan, gps=mock_gps):
+    def __init__(self, parameters, sigan, gps=None):
+        if gps is None:
+            gps = MockGPS()
         self.parameters = deepcopy(parameters)
         self.sigan = sigan
         self.gps = gps

scos_actions/actions/interfaces/measurement_action.py

@@ -2,11 +2,11 @@
 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.mocks.mock_gps import MockGPS
 from scos_actions.metadata.annotations import CalibrationAnnotation, SensorAnnotation
 from scos_actions.metadata.measurement_global import MeasurementMetadata
 from scos_actions.metadata.sigmf_builder import SigMFBuilder
+from scos_actions.signals import measurement_action_completed
 
 logger = logging.getLogger(__name__)
 
@@ -19,7 +19,9 @@ class MeasurementAction(Action):
 
     """
 
-    def __init__(self, parameters, sigan, gps=mock_gps):
+    def __init__(self, parameters, sigan, gps=None):
+        if gps is None:
+            gps = MockGPS()
         super().__init__(parameters, sigan, gps)
         self.received_samples = 0
 

scos_actions/actions/monitor_sigan.py

@@ -3,16 +3,18 @@
 import logging
 
 from scos_actions.actions.interfaces.action import Action
-from scos_actions.actions.interfaces.signals import monitor_action_completed
-from scos_actions.hardware import gps as mock_gps
+from scos_actions.hardware.mocks.mock_gps import MockGPS
+from scos_actions.signals import monitor_action_completed
 
 logger = logging.getLogger(__name__)
 
 
 class MonitorSignalAnalyzer(Action):
     """Monitor signal analyzer connection and restart container if unreachable."""
 
-    def __init__(self, sigan, parameters={"name": "monitor_sigan"}, gps=mock_gps):
+    def __init__(self, sigan, parameters={"name": "monitor_sigan"}, gps=None):
+        if gps is None:
+            gps = MockGPS()
         super().__init__(parameters=parameters, sigan=sigan, gps=gps)
 
     def __call__(self, schedule_entry_json, task_id):

scos_actions/actions/sync_gps.py

@@ -4,7 +4,7 @@
 import subprocess
 
 from scos_actions.actions.interfaces.action import Action
-from scos_actions.actions.interfaces.signals import location_action_completed
+from scos_actions.signals import location_action_completed
 
 logger = logging.getLogger(__name__)
 

scos_actions/actions/tests/test_acquire_single_freq_fft.py

@@ -1,7 +1,7 @@
-from scos_actions.actions.interfaces.signals import measurement_action_completed
 from scos_actions.actions.tests.utils import SENSOR_DEFINITION, check_metadata_fields
 from scos_actions.capabilities import capabilities
 from scos_actions.discover import test_actions as actions
+from scos_actions.signals import measurement_action_completed
 
 SINGLE_FREQUENCY_FFT_ACQUISITION = {
     "name": "test_acq",

scos_actions/actions/tests/test_monitor_radio.py

@@ -1,5 +1,5 @@
-from scos_actions.actions.interfaces.signals import monitor_action_completed
 from scos_actions.discover import test_actions as actions
+from scos_actions.signals import monitor_action_completed
 
 MONITOR_SIGAN_SCHEDULE = {
     "name": "test_monitor",

scos_actions/actions/tests/test_single_freq_tdomain_iq.py

@@ -1,8 +1,8 @@
 import pytest
 
-from scos_actions.actions.interfaces.signals import measurement_action_completed
 from scos_actions.actions.tests.utils import check_metadata_fields
 from scos_actions.discover import test_actions as actions
+from scos_actions.signals import measurement_action_completed
 
 SINGLE_TIMEDOMAIN_IQ_ACQUISITION = {
     "name": "test_acq",

scos_actions/actions/tests/test_stepped_freq_tdomain_iq.py

@@ -1,6 +1,5 @@
-from scos_actions.actions.interfaces.signals import measurement_action_completed
-from scos_actions.actions.tests.utils import SENSOR_DEFINITION, check_metadata_fields
 from scos_actions.discover import test_actions as actions
+from scos_actions.signals import measurement_action_completed
 
 SINGLE_TIMEDOMAIN_IQ_MULTI_RECORDING_ACQUISITION = {
     "name": "test_multirec_acq",

scos_actions/actions/tests/test_sync_gps.py

@@ -1,6 +1,5 @@
-from scos_actions.actions.interfaces.signals import location_action_completed
-from scos_actions.actions.tests.utils import SENSOR_DEFINITION
 from scos_actions.discover import test_actions
+from scos_actions.signals import location_action_completed
 
 SYNC_GPS = {
     "name": "sync_gps",

scos_actions/configs/actions/test_survey_iq_action.yml

@@ -0,0 +1,28 @@
+stepped_frequency_time_domain_iq:
+  name: test_survey_iq_action
+  classification: UNCLASSIFIED
+  frequency:
+    - 700.5e6
+    - 709e6
+    - 731.5e6
+    - 739e6
+    - 751e6
+    - 763e6
+    - 772e6
+    - 782e6
+    - 793e6
+    - 802e6
+  gain: 40
+  sample_rate: 15.36e6
+  duration_ms:
+    - 1000
+    - 1000
+    - 1000
+    - 1000
+    - 1000
+    - 1000
+    - 10000
+    - 1000
+    - 1000
+    - 10000
+  nskip: 15.36e4

scos_actions/discover/__init__.py

@@ -1,14 +1,15 @@
 from scos_actions.actions import action_classes
-from scos_actions.actions.interfaces.signals import register_component_with_status
 from scos_actions.actions.logger import Logger
 from scos_actions.actions.monitor_sigan import MonitorSignalAnalyzer
 from scos_actions.actions.sync_gps import SyncGps
 from scos_actions.discover.yaml import load_from_yaml
-from scos_actions.hardware import gps as mock_gps
+from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.hardware.mocks.mock_sigan import MockSignalAnalyzer
 from scos_actions.settings import ACTION_DEFINITIONS_DIR, MOCK_SIGAN
+from scos_actions.signals import register_component_with_status
 
 mock_sigan = MockSignalAnalyzer(randomize_values=True)
+mock_gps = MockGPS()
 if MOCK_SIGAN:
     register_component_with_status.send(mock_sigan.__class__, component=mock_sigan)
 actions = {"logger": Logger()}

scos_actions/discover/tests/test_yaml.py

@@ -5,7 +5,7 @@
 
 from scos_actions import actions
 from scos_actions.discover.yaml import load_from_yaml
-from scos_actions.hardware import gps
+from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.hardware.mocks.mock_sigan import MockSignalAnalyzer
 
 INVALID_YAML = b"""\
@@ -24,6 +24,7 @@
 """
 
 sigan = MockSignalAnalyzer()
+gps = MockGPS()
 
 
 def test_load_from_yaml_existing():

scos_actions/hardware/__init__.py

@@ -6,15 +6,14 @@
 from its_preselector.controlbyweb_web_relay import ControlByWebWebRelay
 
 from scos_actions import utils
-from scos_actions.actions.interfaces.signals import register_component_with_status
 from scos_actions.capabilities import capabilities
-from scos_actions.hardware.mocks.mock_gps import MockGPS
 from scos_actions.settings import (
     PRESELECTOR_CLASS,
     PRESELECTOR_CONFIG_FILE,
     PRESELECTOR_MODULE,
     SWITCH_CONFIGS_DIR,
 )
+from scos_actions.signals import register_component_with_status
 from scos_actions.status.status_registration_handler import status_registration_handler
 
 logger = logging.getLogger(__name__)
@@ -72,7 +71,6 @@ def load_preselector(preselector_config, module, preselector_class_name):
 
 register_component_with_status.connect(status_registration_handler)
 logger.info("Connected status registration handler")
-gps = MockGPS()
 preselector = load_preslector_from_file(PRESELECTOR_CONFIG_FILE)
 switches = load_switches(SWITCH_CONFIGS_DIR)
 logger.info("Loaded {switch_count} switches.".format(switch_count=(len(switches))))

scos_actions/hardware/mocks/mock_gps.py

@@ -2,10 +2,16 @@
 
 from scos_actions.hardware.gps_iface import GPSInterface
 
+import logging
+
+logger = logging.getLogger(__name__)
+
 
 class MockGPS(GPSInterface):
     def get_location(timeout_s=1):
+        logger.warning("Using mock GPS!")
         return 39.995118, -105.261572, 1651.0
 
     def get_gps_time(self):
+        logger.warning("Using mock GPS!")
         return datetime.now()

scos_actions/hardware/mocks/mock_sigan.py

@@ -81,6 +81,7 @@ def connect(self):
     def acquire_time_domain_samples(
         self, num_samples, num_samples_skip=0, retries=5, gain_adjust=True
     ):
+        logger.warning("Using mock signal analyzer!")
         self.sigan_overload = False
         self._capture_time = None
         self._num_samples_skip = num_samples_skip