From 7c0454d925b9ee09990e1be598972c56dd8cb319 Mon Sep 17 00:00:00 2001 From: Justin Haze Date: Thu, 25 Apr 2024 13:37:45 -0600 Subject: [PATCH] Address feedback --- scos_actions/actions/logger.py | 34 ------------------------------- scos_actions/discover/__init__.py | 5 ++--- scos_actions/hardware/sensor.py | 13 ++++++------ scos_actions/settings.py | 14 ++++++++++--- 4 files changed, 19 insertions(+), 47 deletions(-) delete mode 100644 scos_actions/actions/logger.py diff --git a/scos_actions/actions/logger.py b/scos_actions/actions/logger.py deleted file mode 100644 index ae2c6e60..00000000 --- a/scos_actions/actions/logger.py +++ /dev/null @@ -1,34 +0,0 @@ -"""A simple example action that logs a message.""" - -import logging -from typing import Optional - -from scos_actions.actions.interfaces.action import Action -from scos_actions.hardware.sensor import Sensor - -logger = logging.getLogger(__name__) - -LOGLVL_INFO = 20 -LOGLVL_ERROR = 40 - - -class Logger(Action): - """Log the message "running test {name}/{tid}". - - This is useful for testing and debugging. - - `{name}` will be replaced with the parent schedule entry's name, and - `{tid}` will be replaced with the sequential task id. - - """ - - def __init__(self, loglvl=LOGLVL_INFO): - super().__init__(parameters={"name": "logger"}) - self.loglvl = loglvl - - def __call__(self, sensor: Optional[Sensor], schedule_entry: dict, task_id: int): - msg = "running test {name}/{tid}" - schedule_entry_name = schedule_entry["name"] - logger.log( - level=self.loglvl, msg=msg.format(name=schedule_entry_name, tid=task_id) - ) diff --git a/scos_actions/discover/__init__.py b/scos_actions/discover/__init__.py index a3430161..30e1cb29 100644 --- a/scos_actions/discover/__init__.py +++ b/scos_actions/discover/__init__.py @@ -1,12 +1,11 @@ from scos_actions.actions import action_classes -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.settings import ACTION_DEFINITIONS_DIR, SIGAN_CLASS, SIGAN_MODULE -actions = {"logger": Logger()} -test_actions = {"logger": Logger()} +actions = {} +test_actions = {} def init( diff --git a/scos_actions/hardware/sensor.py b/scos_actions/hardware/sensor.py index f4e1f6fb..6a9e9701 100644 --- a/scos_actions/hardware/sensor.py +++ b/scos_actions/hardware/sensor.py @@ -5,6 +5,7 @@ from typing import Any, Dict, List, Optional import numpy as np +from numpy.typing import ArrayLike from its_preselector.preselector import Preselector from its_preselector.web_relay import WebRelay @@ -246,16 +247,14 @@ def recompute_calibration_data(self, params: dict) -> None: if not recomputed: logger.warning("Failed to recompute calibration data") - def check_sensor_overload(self, data) -> bool: + def check_sensor_overload(self, data: ArrayLike) -> bool: """Check for sensor overload in the measurement data.""" - measured_data = data.astype(np.complex64) - - time_domain_avg_power = 10 * np.log10(np.mean(np.abs(measured_data) ** 2)) - time_domain_avg_power += ( - 10 * np.log10(1 / (2 * 50)) + 30 - ) # Convert log(V^2) to dBm # explicitly check is not None since 1db compression could be 0 if self.sensor_calibration_data["compression_point"] is not None: + time_domain_avg_power = 10 * np.log10(np.mean(np.abs(data) ** 2)) + time_domain_avg_power += ( + 10 * np.log10(1 / (2 * 50)) + 30 + ) # Convert log(V^2) to dBm return bool( time_domain_avg_power > self.sensor_calibration_data["compression_point"] diff --git a/scos_actions/settings.py b/scos_actions/settings.py index 158de256..eeca61ad 100644 --- a/scos_actions/settings.py +++ b/scos_actions/settings.py @@ -1,4 +1,6 @@ import logging +import sys +from os import path from pathlib import Path from environs import Env @@ -21,14 +23,20 @@ logger.debug(f"scos-actions: MOCK_SIGAN:{MOCK_SIGAN}") MOCK_SIGAN_RANDOM = env.bool("MOCK_SIGAN_RANDOM", default=False) logger.debug(f"scos-actions: MOCK_SIGAN_RANDOM:{MOCK_SIGAN_RANDOM}") -RUNNING_TESTS = env.bool("RUNNING_TESTS", False) +__cmd = path.split(sys.argv[0])[-1] +RUNNING_TESTS = env.bool("RUNNING_TESTS", "test" in __cmd) logger.debug(f"scos-actions: RUNNING_TESTS:{RUNNING_TESTS}") logger.debug(f"scos-actions: RUNNING_TESTS:{RUNNING_TESTS}") FQDN = env("FQDN", None) logger.debug(f"scos-actions: FQDN:{FQDN}") -SIGAN_MODULE = env.str("SIGAN_MODULE", default="scos_actions.hardware.mocks.mock_sigan") + +SIGAN_MODULE = env.str("SIGAN_MODULE", default=None) +if RUNNING_TESTS: + SIGAN_MODULE = "scos_actions.hardware.mocks.mock_sigan" logger.debug(f"scos-actions: SIGAN_MODULE:{SIGAN_MODULE}") -SIGAN_CLASS = env.str("SIGAN_CLASS", default="MockSignalAnalyzer") +SIGAN_CLASS = env.str("SIGAN_CLASS", default=None) +if RUNNING_TESTS: + SIGAN_CLASS = "MockSignalAnalyzer" logger.debug(f"scos-actions: SIGAN_CLASS:{SIGAN_CLASS}") SIGAN_POWER_SWITCH = env("SIGAN_POWER_SWITCH", default=None) logger.debug(f"scos-actions: SIGAN_POWER_SWITCH:{SIGAN_POWER_SWITCH}")