add formatting and linting

.github/workflows/lint.yml

@@ -0,0 +1,28 @@
+name: Linting and Formatting
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up OnAIR conda environment
+      uses: conda-incubator/setup-miniconda@v3
+      with:
+        activate-environment: onair
+        environment-file: environment.yml
+        python-version: ${{ matrix.python-version }}
+        auto-activate-base: false
+    - name: Lint Project
+      shell: bash -l {0}
+      run: pylint onair plugins test
+    - name: Check Formatting
+      shell: bash -l {0}
+      run: black --check .

.pylintrc

@@ -0,0 +1,3 @@
+[MAIN]
+
+exit-zero=True

conftest.py

@@ -13,32 +13,34 @@
 from unittest.mock import MagicMock
 import sys
 
+
 def pytest_addoption(parser):
-  parser.addoption("--conftest-seed", action="store", type=int, default=None)
+    parser.addoption("--conftest-seed", action="store", type=int, default=None)
+
 
 def pytest_configure(config):
-  seed = config.getoption("--conftest-seed")
-  if config.getoption("--conftest-seed") == None:
-    seed = int(time())
-  pytest.gen = random.Random(seed)
-  print(f"Using --conftest-seed={seed}")
-
-  # Mock simdkalman for kalman_plugin testing
-  simdkalman = MagicMock()
-  sys.modules['simdkalman'] = simdkalman
-
-  # Mock sbn_client for sbn_adapter testing
-  sc = MagicMock()
-  sys.modules['sbn_client'] = sc
-
-  # Mock message_headers for sbn_adapter testing
-  mh = MagicMock()
-  mh.sample_data_tlm_t = MagicMock()
-  mh.sample_data_tlm_t.__name__ = 'mock_sample_data_tlm_t'
-  mh.sample_data_power_t = MagicMock()
-  mh.sample_data_power_t.__name__ = 'mock_sample_data_power_t'
-  mh.sample_data_thermal_t = MagicMock()
-  mh.sample_data_thermal_t.__name__ = 'mock_sample_data_thermal_t'
-  mh.sample_data_gps_t = MagicMock()
-  mh.sample_data_gps_t.__name__ = 'mock_sample_data_gps_t'
-  sys.modules['message_headers'] = mh
+    seed = config.getoption("--conftest-seed")
+    if config.getoption("--conftest-seed") == None:
+        seed = int(time())
+    pytest.gen = random.Random(seed)
+    print(f"Using --conftest-seed={seed}")
+
+    # Mock simdkalman for kalman_plugin testing
+    simdkalman = MagicMock()
+    sys.modules["simdkalman"] = simdkalman
+
+    # Mock sbn_client for sbn_adapter testing
+    sc = MagicMock()
+    sys.modules["sbn_client"] = sc
+
+    # Mock message_headers for sbn_adapter testing
+    mh = MagicMock()
+    mh.sample_data_tlm_t = MagicMock()
+    mh.sample_data_tlm_t.__name__ = "mock_sample_data_tlm_t"
+    mh.sample_data_power_t = MagicMock()
+    mh.sample_data_power_t.__name__ = "mock_sample_data_power_t"
+    mh.sample_data_thermal_t = MagicMock()
+    mh.sample_data_thermal_t.__name__ = "mock_sample_data_thermal_t"
+    mh.sample_data_gps_t = MagicMock()
+    mh.sample_data_gps_t.__name__ = "mock_sample_data_gps_t"
+    sys.modules["message_headers"] = mh

driver.py

@@ -27,34 +27,53 @@ def main():
     Check the .ini file for the filenames used
     """
 
-    arg_parser = argparse.ArgumentParser(description='')
-    arg_parser.add_argument('configfile', nargs='?',
-                            default='./onair/config/default_config.ini',
-                            help='Config file to be used')
-    arg_parser.add_argument('--save', '-s', action='store_true',
-                            help='Should log files be saved?')
-    arg_parser.add_argument('--save_name', '--name', '-n',
-                            help='Name of saved log files')
-    arg_parser.add_argument('--mute', '-m', action='store_true',
-                            help='Mute all non-error output')
+    arg_parser = argparse.ArgumentParser(description="")
+    arg_parser.add_argument(
+        "configfile",
+        nargs="?",
+        default="./onair/config/default_config.ini",
+        help="Config file to be used",
+    )
+    arg_parser.add_argument(
+        "--save", "-s", action="store_true", help="Should log files be saved?"
+    )
+    arg_parser.add_argument(
+        "--save_name", "--name", "-n", help="Name of saved log files"
+    )
+    arg_parser.add_argument(
+        "--mute", "-m", action="store_true", help="Mute all non-error output"
+    )
 
     """
     Testing specific arguments
     """
-    arg_parser.add_argument('--test', '-t', action='store_true',
-                            help='Run tests')
-    arg_parser.add_argument('--verbose', '-v', action='count', default=0,
-                            help="Increase verbosity in tests")
-    arg_parser.add_argument('-k', action='store', dest='keyword', default="",
-                            metavar='EXPRESSION',
-                            help="Pass thru for pytest's -k option. Runs only"
-                                 " tests with names that match EXPRESSION.")
-    arg_parser.add_argument('--conftest-seed', action='store',
-                            type=int, default=None,
-                            help="Set the random seed for test values")
-    arg_parser.add_argument('--randomly-seed', action='store',
-                            type=int, default=None,
-                            help="Set the random seed for test run order")
+    arg_parser.add_argument("--test", "-t", action="store_true", help="Run tests")
+    arg_parser.add_argument(
+        "--verbose", "-v", action="count", default=0, help="Increase verbosity in tests"
+    )
+    arg_parser.add_argument(
+        "-k",
+        action="store",
+        dest="keyword",
+        default="",
+        metavar="EXPRESSION",
+        help="Pass thru for pytest's -k option. Runs only"
+        " tests with names that match EXPRESSION.",
+    )
+    arg_parser.add_argument(
+        "--conftest-seed",
+        action="store",
+        type=int,
+        default=None,
+        help="Set the random seed for test values",
+    )
+    arg_parser.add_argument(
+        "--randomly-seed",
+        action="store",
+        type=int,
+        default=None,
+        help="Set the random seed for test run order",
+    )
     args = arg_parser.parse_args()
 
     """
@@ -63,7 +82,8 @@ def main():
     """
     if args.test:
         import coverage
-        cov = coverage.Coverage(source=['onair', 'plugins'], branch=True)
+
+        cov = coverage.Coverage(source=["onair", "plugins"], branch=True)
         cov.start()
 
     """
@@ -80,10 +100,11 @@ def main():
     """ Runs all unit tests """
     if args.test:
         import pytest
+
         test_directory_name = "test"
         pytest_args = [test_directory_name]
 
-        pytest_args.extend(['-v'] * args.verbose)
+        pytest_args.extend(["-v"] * args.verbose)
         if args.conftest_seed:
             pytest_args.extend([f"--conftest-seed={args.conftest_seed}"])
         if args.randomly_seed:
@@ -95,7 +116,7 @@ def main():
         cov.save()
         cov.html_report()
     else:
-        setup_folders(os.environ['RESULTS_PATH'])
+        setup_folders(os.environ["RESULTS_PATH"])
         if args.save_name:
             save_name = args.save_name
         else:
@@ -108,26 +129,28 @@ def init_global_paths(test=False):
     """
     Initializes global paths, used throughout execution
     """
-    run_path = 'onair/src/test' if test else './'
-    results_path = 'onair/src/test/results' if test else 'results/'
+    run_path = "onair/src/test" if test else "./"
+    results_path = "onair/src/test/results" if test else "results/"
 
-    os.environ['BASE_PATH'] = os.path.dirname(os.path.realpath(__file__))
-    os.environ['RUN_PATH'] = os.path.join(os.path.dirname(
-        os.path.realpath(__file__)), run_path)
-    os.environ['RESULTS_PATH'] = os.path.join(os.path.dirname(
-        os.path.realpath(__file__)), results_path)
-    os.environ['SRC_ROOT_PATH'] = os.path.dirname(os.path.realpath(__file__))
+    os.environ["BASE_PATH"] = os.path.dirname(os.path.realpath(__file__))
+    os.environ["RUN_PATH"] = os.path.join(
+        os.path.dirname(os.path.realpath(__file__)), run_path
+    )
+    os.environ["RESULTS_PATH"] = os.path.join(
+        os.path.dirname(os.path.realpath(__file__)), results_path
+    )
+    os.environ["SRC_ROOT_PATH"] = os.path.dirname(os.path.realpath(__file__))
 
 
 def blockPrint():
-    """ Disable terminal output """
-    sys.stdout = open(os.devnull, 'w')
+    """Disable terminal output"""
+    sys.stdout = open(os.devnull, "w")
 
 
 def enablePrint():
-    """ Restore terminal output """
+    """Restore terminal output"""
     sys.stdout = sys.__stdout__
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

environment.yml

@@ -5,6 +5,8 @@ channels:
 dependencies:
   - python>=3.8,<3.13
   - numpy
+  - black
+  - pylint
   - coverage
   - pytest
   - pytest-mock

onair/data_handling/csv_parser.py

@@ -18,22 +18,23 @@
 from onair.src.util.print_io import *
 from onair.data_handling.parser_util import *
 
+
 class DataSource(OnAirDataSource):
 
     def process_data_file(self, data_file):
         self.sim_data = self.parse_csv_data(data_file)
         self.frame_index = 0
 
-##### INITIAL PROCESSING ####
+    ##### INITIAL PROCESSING ####
     def parse_csv_data(self, data_file):
-        #Read in the data set
+        # Read in the data set
 
         all_data = []
 
-        with open(data_file, 'r', newline='') as csv_file:
-            dataset = csv.reader(csv_file, delimiter=',')
+        with open(data_file, "r", newline="") as csv_file:
+            dataset = csv.reader(csv_file, delimiter=",")
 
-            #Initialize the entire data dictionary
+            # Initialize the entire data dictionary
             index = 0
             for row in dataset:
                 if index == 0:
@@ -49,10 +50,10 @@ def parse_csv_data(self, data_file):
     def parse_meta_data_file(self, meta_data_file, ss_breakdown):
         return extract_meta_data_handle_ss_breakdown(meta_data_file, ss_breakdown)
 
-##### GETTERS ##################################
+    ##### GETTERS ##################################
 
     def get_vehicle_metadata(self):
-        return self.all_headers, self.binning_configs['test_assignments']
+        return self.all_headers, self.binning_configs["test_assignments"]
 
     # Get the data at self.index and increment the index
     def get_next(self):

onair/data_handling/on_air_data_source.py

@@ -10,14 +10,16 @@
 from abc import ABC, abstractmethod
 from .parser_util import *
 
+
 class ConfigKeyError(KeyError):
     pass
 
+
 class OnAirDataSource(ABC):
-    def __init__(self, data_file, meta_file, ss_breakdown = False):
+    def __init__(self, data_file, meta_file, ss_breakdown=False):
         """An initial parsing needs to happen in order to use the parser classes
-            This means that, if you want to use this class to parse in real time,
-            it needs to at least have seen one sample of the anticipated format """
+        This means that, if you want to use this class to parse in real time,
+        it needs to at least have seen one sample of the anticipated format"""
 
         self.raw_data_file = data_file
         self.meta_data_file = meta_file
@@ -27,10 +29,12 @@ def __init__(self, data_file, meta_file, ss_breakdown = False):
         self.binning_configs = {}
 
         configs = self.parse_meta_data_file(self.meta_data_file, ss_breakdown)
-        self.binning_configs['subsystem_assignments'] = configs['subsystem_assignments']
-        self.binning_configs['test_assignments'] = configs['test_assignments']
-        self.binning_configs['description_assignments'] = configs['description_assignments']
-        self.all_headers = configs['data_labels']
+        self.binning_configs["subsystem_assignments"] = configs["subsystem_assignments"]
+        self.binning_configs["test_assignments"] = configs["test_assignments"]
+        self.binning_configs["description_assignments"] = configs[
+            "description_assignments"
+        ]
+        self.all_headers = configs["data_labels"]
 
         self.process_data_file(self.raw_data_file)
 

onair/data_handling/parser_util.py

@@ -11,28 +11,32 @@
 from .tlm_json_parser import parseTlmConfJson, str2lst
 import datetime
 
+
 def extract_meta_data_handle_ss_breakdown(meta_data_file, ss_breakdown):
     parsed_meta_data = extract_meta_data(meta_data_file)
     if ss_breakdown == False:
-        num_elements = len(parsed_meta_data['subsystem_assignments'])
-        parsed_meta_data['subsystem_assignments'] = [['MISSION'] for elem in range(num_elements)]
+        num_elements = len(parsed_meta_data["subsystem_assignments"])
+        parsed_meta_data["subsystem_assignments"] = [
+            ["MISSION"] for elem in range(num_elements)
+        ]
     return parsed_meta_data
 
+
 ## Method to extract configuration data and return 3 dictionaries
 def extract_meta_data(meta_data_file):
-    assert meta_data_file != ''
+    assert meta_data_file != ""
 
     configs = parseTlmConfJson(meta_data_file)
 
-    configs_len = len(configs['subsystem_assignments'])
+    configs_len = len(configs["subsystem_assignments"])
 
     for i in range(configs_len):
-        if configs['subsystem_assignments'][i] != 'NONE':
-            configs['subsystem_assignments'][i] = [configs['subsystem_assignments'][i]]
+        if configs["subsystem_assignments"][i] != "NONE":
+            configs["subsystem_assignments"][i] = [configs["subsystem_assignments"][i]]
         else:
-            configs['subsystem_assignments'][i] = []
+            configs["subsystem_assignments"][i] = []
 
-        test_assign = configs['test_assignments'][i]
+        test_assign = configs["test_assignments"][i]
 
         for j in range(len(test_assign)):
             if len(test_assign[j]) > 1:
@@ -42,6 +46,7 @@ def extract_meta_data(meta_data_file):
 
     return configs
 
+
 def floatify_input(_input, remove_str=False):
     floatified = []
     for i in _input:
@@ -60,12 +65,13 @@ def floatify_input(_input, remove_str=False):
                 continue
     return floatified
 
+
 def convert_str_to_timestamp(time_str):
     try:
-        t = datetime.datetime.strptime(time_str, '%Y-%j-%H:%M:%S.%f')
+        t = datetime.datetime.strptime(time_str, "%Y-%j-%H:%M:%S.%f")
         return t.timestamp()
     except:
-        min_sec = time_str.split(':')
+        min_sec = time_str.split(":")
         # Use 1 am on Jan 1st, 2000 as the date if only minutes and seconds are specified
         t = datetime.datetime(2000, 1, 1, 1, int(min_sec[0]), int(min_sec[1]), 0)
         return t.timestamp()