Type hinting and misc changes

src/swell/deployment/create_experiment.py

@@ -14,7 +14,6 @@
 import shutil
 import sys
 import yaml
-
 from typing import Union, Optional
 
 from swell.deployment.prepare_config_and_suite.prepare_config_and_suite import \
@@ -163,7 +162,7 @@ def create_experiment_directory(
     platform: str,
     override: str,
     advanced: bool,
-    slurm: str
+    slurm: Optional[str]
 ) -> None:
 
     # Create a logger
@@ -209,7 +208,7 @@ def create_experiment_directory(
     # Copy suite and platform files to experiment suite directory
     # -----------------------------------------------------------
     swell_suite_path = os.path.join(get_swell_path(), 'suites', suite)
-    copy_platform_files(exp_suite_path, platform)
+    copy_platform_files(logger, exp_suite_path, platform)
 
     if os.path.exists(os.path.join(swell_suite_path, 'eva')):
         copy_eva_files(swell_suite_path, exp_suite_path)

src/swell/deployment/prepare_config_and_suite/question_and_answer_cli.py

@@ -117,7 +117,7 @@ def make_boolean(
         quest: str,
         default: str,
         prompt: questionary.text
-    ) -> bool:
+    ) -> str:
 
         answer = prompt(quest, default=default, auto_enter=False).ask()
 

src/swell/tasks/base/task_base.py

@@ -47,7 +47,6 @@ def __init__(
         # Create message logger
         # ---------------------
         self.logger = Logger(task_name)
-        self.logger.info('Logger type', type(self.logger))
 
         # Write out the initialization info
         # ---------------------------------
@@ -160,7 +159,7 @@ def experiment_config_path(self) -> str:
 
     # ----------------------------------------------------------------------------------------------
 
-    def get_ensemble_packet(self) -> str:
+    def get_ensemble_packet(self) -> Optional[str]:
         return self.__ensemble_packet__
 
     # ----------------------------------------------------------------------------------------------
@@ -198,7 +197,7 @@ def cycle_dir(self) -> str:
 
     # ----------------------------------------------------------------------------------------------
 
-    def forecast_dir(self, paths: list = []) -> str:
+    def forecast_dir(self, paths: Union[str, list[str]] = []) -> Optional[str]:
 
         # Make sure forecast directory exists
         # -----------------------------------
@@ -222,7 +221,7 @@ def forecast_dir(self, paths: list = []) -> str:
 
     # ----------------------------------------------------------------------------------------------
 
-    def cycle_time_dto(self) -> object:
+    def cycle_time_dto(self) -> dt:
 
         return self.__datetime__.dto()
 
@@ -240,7 +239,7 @@ def first_cycle_time(self) -> str:
 
     # ----------------------------------------------------------------------------------------------
 
-    def first_cycle_time_dto(self) -> object:
+    def first_cycle_time_dto(self) -> dt:
 
         return self.__start_cycle_point__.dto()
 
@@ -295,7 +294,7 @@ def task_wrapper(
     task: str,
     config: str,
     datetime: Union[str, dt, None],
-    model: str,
+    model: Optional[str],
     ensemblePacket: Optional[str]
 ) -> None:
 

src/swell/tasks/get_observations.py

@@ -272,7 +272,7 @@ def execute(self) -> None:
 
     # ----------------------------------------------------------------------------------------------
 
-    def get_tlapse_files(self, observation_dict: dict) -> Union[None, object]:
+    def get_tlapse_files(self, observation_dict: dict) -> Union[None, int]:
 
         # Function to locate instances of tlapse in the obs operator config
 

src/swell/utilities/build.py

@@ -9,6 +9,7 @@
 
 import os
 import shutil
+from typing import Tuple
 
 from jedi_bundle.bin.jedi_bundle import get_default_config
 from jedi_bundle.config.config import check_platform
@@ -17,7 +18,7 @@
 # --------------------------------------------------------------------------------------------------
 
 
-def build_and_source_dirs(package_path: str) -> str:
+def build_and_source_dirs(package_path: str) -> Tuple[str, str]:
 
     # Make package directory
     # ----------------------

src/swell/utilities/data_assimilation_window_params.py

@@ -9,6 +9,7 @@
 
 import datetime
 import isodate
+from typing import Union
 
 from swell.utilities.datetime_util import datetime_formats
 from swell.utilities.logger import Logger
@@ -57,7 +58,7 @@ def __get_local_background_time__(
 
     # ----------------------------------------------------------------------------------------------
 
-    def window_begin(self, window_offset: str, dto: bool = False) -> str:
+    def window_begin(self, window_offset: str, dto: bool = False) -> Union[str, datetime.datetime]:
 
         window_begin_dto = self.__get_window_begin_dto__(window_offset)
 

src/swell/utilities/exceptions.py

@@ -6,16 +6,15 @@
 # -----------------------------------------------------------------------------
 
 from typing import Optional
-
-from swell.utilities.logger import Logger
+from logging import Logger as pyLogger
 
 
 class SWELLError(Exception):
 
     def __init__(
         self,
         message: str,
-        logger: Optional[Logger] = None
+        logger: Optional[pyLogger] = None
     ) -> None:
 
         self.message = message

src/swell/utilities/file_system_operations.py

@@ -51,7 +51,6 @@ def link_all_files_from_first_in_hierarchy_of_sources(
     """
 
     # First sweep to see if directories exist
-    logger.info('Source Paths', source_paths)
 
     found_paths = []
     for source_path in source_paths:
@@ -99,13 +98,17 @@ def check_if_files_exist_in_path(logger: Logger, path_to_files: str) -> bool:
     path_to_files: path to target directory
     """
 
+    files_exist = False
+
     if os.path.exists(path_to_files):
         if os.listdir(path_to_files):
             logger.info(f'Files found within {path_to_files}')
-            return True
+            files_exist = True
         else:
             logger.info(f'No files found within {path_to_files}')
-            return False
+
+    return files_exist
+
 
 # --------------------------------------------------------------------------------------------------
 

src/swell/utilities/filehandler.py

@@ -63,7 +63,7 @@
 import copy
 import datetime as dt
 from shutil import copyfile
-from typing import Union, Optional
+from typing import Union, Optional, Any
 
 from swell.utilities.exceptions import *
 
@@ -385,7 +385,7 @@ def list(self, force: bool = False) -> list:
 
 class FileCollection(object):
 
-    def __init__(self, config: list) -> None:
+    def __init__(self, config: dict[Any, Any]) -> None:
 
         self.config = copy.deepcopy(config)
 

src/swell/utilities/geos.py

@@ -14,7 +14,7 @@
 import netCDF4
 import os
 import re
-from typing import Tuple
+from typing import Tuple, Optional, Union
 
 from swell.utilities.shell_commands import run_subprocess
 from swell.utilities.datetime_util import datetime_formats
@@ -27,7 +27,7 @@ class Geos():
 
     # ----------------------------------------------------------------------------------------------
 
-    def __init__(self, logger: Logger, forecast_dir: str) -> None:
+    def __init__(self, logger: Logger, forecast_dir: Optional[str]) -> None:
 
         '''
         Intention with GEOS class is to not have any model dependent methods.
@@ -40,7 +40,11 @@ def __init__(self, logger: Logger, forecast_dir: str) -> None:
 
     # ----------------------------------------------------------------------------------------------
 
-    def adjacent_cycle(self, offset: str, return_date: bool = False) -> str:
+    def adjacent_cycle(
+        self,
+        offset: str,
+        return_date: bool = False
+    ) -> Union[str, datetime.datetime]:
 
         # Basename consists of swell datetime and model
         # ---------------------------------------------

src/swell/utilities/get_channels.py

@@ -11,7 +11,7 @@
 import os
 from datetime import datetime as dt
 from itertools import groupby
-from typing import Tuple
+from typing import Tuple, Optional
 
 from swell.utilities.logger import Logger
 
@@ -40,7 +40,7 @@ def process_channel_lists(channel_list: list) -> list:
 # --------------------------------------------------------------------------------------------------
 
 
-def create_range_string(avail_list: list) -> list:
+def create_range_string(avail_list: list) -> str:
     '''
         Function converts integer list into string of ranges
     '''
@@ -76,7 +76,7 @@ def get_channels(
     observation: str,
     dt_cycle_time: dt,
     logger: Logger
-) -> Tuple[str, list]:
+) -> Tuple[Optional[str], Optional[list[int]]]:
 
     '''
         Comparing available channels and active channels from the observing
@@ -119,7 +119,7 @@ def num_active_channels(
     path_to_observing_sys_yamls: str,
     observation: str,
     dt_cycle_time: dt
-) -> None:
+) -> Optional[int]:
 
     # Retrieve available and active channels from records yaml
     path_to_observing_sys_config = path_to_observing_sys_yamls + '/' + \

src/swell/utilities/gsi_record_parser.py

@@ -23,7 +23,7 @@ def __init__(self) -> None:
         self.sat = None
         self.instr = None
 
-    def get_channel_list(self, start) -> list:
+    def get_channel_list(self, start: int) -> list:
         channel_list = []
         rows = self.instr_df.loc[self.instr_df["start"] == start]
         for row_ch_list in rows["channels"].values:

src/swell/utilities/observing_system_records.py

@@ -20,7 +20,7 @@ def format_date(old_date: str) -> str:
 # --------------------------------------------------------------------------------------------------
 
 
-def read_sat_db(path_to_sat_db: str, column_names: list) -> pd.DataFrame:
+def read_sat_db(path_to_sat_db: str, column_names: list[str]) -> pd.DataFrame:
 
     '''
         Reading GSI observing system records row by row into

src/swell/utilities/render_jedi_interface_files.py

@@ -9,12 +9,12 @@
 
 import os
 import yaml
-from typing import Union, Optional
+from typing import Union, Optional, Any
 
 from swell.utilities.jinja2 import template_string_jinja2
 from swell.utilities.get_channels import get_channels
 from swell.utilities.logger import Logger
-
+from swell.utilities.datetime_util import Datetime
 
 # --------------------------------------------------------------------------------------------------
 
@@ -26,8 +26,8 @@ def __init__(
         logger: Logger,
         experiment_root: str,
         experiment_id: str,
-        cycle_dir: str,
-        cycle_time: Optional[str],
+        cycle_dir: Optional[str],
+        cycle_time: Union[Datetime, None],
         jedi_interface: Optional[str] = None
     ) -> None:
 
@@ -121,7 +121,7 @@ def __init__(
     # ----------------------------------------------------------------------------------------------
 
     # Function to add key to the template dictionary
-    def add_key(self, key: str, element: object) -> None:
+    def add_key(self, key: str, element: Any) -> None:
 
         # First assert that key is allowed
         self.logger.assert_abort(key in self.valid_template_keys, f'Trying to add key \'{key}\' ' +
@@ -134,7 +134,7 @@ def add_key(self, key: str, element: object) -> None:
     # ----------------------------------------------------------------------------------------------
 
     # Open the file at the provided path, use dictionary to complete templates and return dictionary
-    def __open_file_render_to_dict__(self, config_file: str) -> object:
+    def __open_file_render_to_dict__(self, config_file: str) -> dict[Any, Any]:
 
         # Check that config file exists
         self.logger.assert_abort(os.path.exists(config_file), f'In open_file_and_render failed ' +
@@ -165,7 +165,7 @@ def render_oops_file(self, config_name: str) -> dict:
     # ----------------------------------------------------------------------------------------------
 
     # Prepare path to interface model file and call rendering
-    def render_interface_model(self, config_name: str) -> dict:
+    def render_interface_model(self, config_name: str) -> dict[Any, Any]:
 
         # Assert that there is a jedi interface associated with the task
         self.logger.assert_abort(self.jedi_interface is not None, f'In order to render a ' +

src/swell/utilities/run_jedi_executables.py

@@ -11,6 +11,7 @@
 import os
 import netCDF4 as nc
 from typing import Optional
+import datetime
 
 from swell.utilities.shell_commands import run_track_log_subprocess
 from swell.utilities.logger import Logger
@@ -20,10 +21,10 @@
 
 
 def check_obs(
-    path_to_observing_sys_yamls: str,
+    path_to_observing_sys_yamls: Optional[str],
     observation: str,
     obs_dict: dict,
-    cycle_time: str
+    cycle_time: Optional[str]
 ) -> bool:
 
     use_observation = False
@@ -51,8 +52,8 @@ def jedi_dictionary_iterator(
     jedi_config_dict: dict,
     jedi_rendering: JediConfigRendering,
     window_type: Optional[str] = None,
-    obs: Optional[str] = None,
-    cycle_time: Optional[str] = None,
+    obs: Optional[list[str]] = None,
+    cycle_time: Optional[datetime.datetime] = None,
     jedi_forecast_model: Optional[str] = None
 ) -> None: