Refactor/add type hints

pyproject.toml

@@ -30,6 +30,14 @@ python_version = "3.11"
 strict = false
 ignore_missing_imports = true
 show_error_codes = true
+# pyrpod/ and its subpackages are namespace packages (no __init__.py), so mypy
+# would otherwise derive two module names for the same file (e.g. "fs" from the
+# file path and "pyrpod.util.io.fs" from an import) and abort with
+# "Source file found twice under different module names". Anchoring the package
+# base at the repo root makes `python -m mypy pyrpod` resolve a single name per
+# file without adding __init__.py files or changing how imports are written.
+mypy_path = "."
+explicit_package_bases = true
 
 [tool.ruff.per-file-ignores]
 # Ignore some noisy rules in tests and generated files

pyrpod/config_test.py

@@ -2,9 +2,10 @@
 # Developing a better soltuion for handling RCS working groups is a top priority for PyRPOD.
 
 import configparser
+
 config = configparser.ConfigParser()
 
-config['thruster_groups'] = {
+_THRUSTER_GROUPS: dict[str, list[str]] = {
                   '+x': ['P1T2', 'P2T2', 'P3T2', 'P4T2', 'P5T2', 'P6T2', 'P7T2', 'P8T2'],
                   '-x': ['P1T1', 'P2T1', 'P3T1', 'P4T1', 'P5T1', 'P6T1', 'P7T1', 'P8T1'],
                   '+y': ['P1T3', 'P2T3', 'P3T4', 'P4T3', 'P5T3', 'P6T3', 'P7T4', 'P8T3'],
@@ -20,5 +21,11 @@
                   '-yaw': ['P2T3', 'P3T4', 'P5T4', 'P8T4']
                      }
 
+# ConfigParser.__setitem__ is typed for Mapping[str, str]; it forwards to
+# read_dict, which calls str() on every value, so the group lists are written
+# out as their repr. Hoisting the literal keeps that deliberate looseness
+# stated once instead of repeating an ignore on all twelve entries.
+config['thruster_groups'] = _THRUSTER_GROUPS  # type: ignore[assignment]
+
 with open('example.ini', 'w') as configfile:
   config.write(configfile)

pyrpod/logging_utils.py

@@ -48,7 +48,7 @@
 import time
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, TypeVar
 
 import numpy as np
 
@@ -106,7 +106,9 @@ class LoggingSettings:
 )
 
 
-def _read_logging_ini(case_dir: str):
+def _read_logging_ini(
+    case_dir: str,
+) -> tuple[configparser.SectionProxy | None, str]:
     """Return (``[logging]`` section proxy or None, ini path)."""
     ini_path = os.path.join(case_dir, "logging.ini")
     if not os.path.isfile(ini_path):
@@ -121,27 +123,34 @@ def _read_logging_ini(case_dir: str):
     return parser["logging"], ini_path
 
 
-def _apply_ini(settings: LoggingSettings, section) -> None:
+def _apply_ini(settings: LoggingSettings,
+               section: configparser.SectionProxy) -> None:
     for key in _INI_BOOL_KEYS:
         if key in section:
             try:
                 setattr(settings, key, section.getboolean(key))
             except ValueError:
                 pass
+    # SectionProxy.get()/getint() are typed as returning ``| None`` because
+    # configparser supports allow_no_value=True. _read_logging_ini builds a
+    # plain ConfigParser() (allow_no_value=False), so an option that passes the
+    # ``in section`` guard above always carries a value; the stub cannot say so.
     if "level" in section:
-        settings.level = section.get("level")
+        settings.level = section.get("level")  # type: ignore[assignment]
     if "format" in section:
-        settings.log_format = section.get("format")
+        settings.log_format = section.get("format")  # type: ignore[assignment]
     if "progress_every_n_firings" in section:
         try:
-            settings.progress_every_n_firings = section.getint(
-                "progress_every_n_firings"
+            settings.progress_every_n_firings = (
+                section.getint("progress_every_n_firings")  # type: ignore[assignment]
             )
         except ValueError:
             pass
 
 
-def _resolve_settings(case_dir: str, overrides: Dict[str, Any]):
+def _resolve_settings(
+    case_dir: str, overrides: Dict[str, Any]
+) -> tuple[LoggingSettings, bool, str]:
     """Resolve :class:`LoggingSettings` following the documented precedence."""
     settings = LoggingSettings()
 
@@ -188,7 +197,13 @@ def emit(self, record: logging.LogRecord) -> None:  # noqa: D401
             self.warning_count += 1
 
 
-def _mark_owned(handler: logging.Handler) -> logging.Handler:
+# Marking a handler returns the *same* handler, so the concrete subclass must
+# survive the call: configure_logging feeds the result straight into
+# LoggingSession(_counter=...), which is typed _LevelCounter.
+_HandlerT = TypeVar("_HandlerT", bound=logging.Handler)
+
+
+def _mark_owned(handler: _HandlerT) -> _HandlerT:
     setattr(handler, _OWNED_HANDLER_ATTR, True)
     return handler
 
@@ -204,7 +219,7 @@ def _remove_owned_handlers(logger: logging.Logger) -> None:
                 pass
 
 
-def _level_to_int(level) -> int:
+def _level_to_int(level: int | str) -> int:
     if isinstance(level, int):
         return level
     return getattr(logging, str(level).upper(), logging.INFO)

pyrpod/mdao/SweepConfig.py

@@ -11,9 +11,16 @@
   Provides methods to aid optimization studies over sweeps of RCS thruster configurations.
 '''
 
+from __future__ import annotations
+
+from collections.abc import Sequence
+
 import numpy as np
 import copy
 
+from pyrpod.vehicle.LogisticsModule import LogisticsModule
+from pyrpod.vehicle.VisitingVehicle import ThrusterConfig
+
 class SweepCoordinates:
 
     '''
@@ -34,7 +41,11 @@ class SweepCoordinates:
     read_swept_coords(swept_configs = [{}, {}, ...])
         Prints to terminal the position of all thrusters for all the swept cofigurations.
     '''
-    def move_ring(config, xf):
+    # NOTE: declared without `self`; it is reached as an unbound
+    # SweepCoordinates.move_ring(config, xf). Adding self would change the
+    # signature, so `config` is annotated in place as the first positional.
+    def move_ring(config: dict[str, ThrusterConfig],  # type: ignore[misc]
+                  xf: float) -> dict[str, ThrusterConfig]:
         '''
             Change the location of the ring of thrusters to the position xf.
 
@@ -59,7 +70,8 @@ def move_ring(config, xf):
 
         return new_config
 
-    def sweep_coords(self, config, x0, xf, dx):
+    def sweep_coords(self, config: dict[str, ThrusterConfig], x0: int,
+                     xf: int, dx: int) -> list[dict[str, ThrusterConfig]]:
         '''
             Given the range of coordinates and the step size, this method makes copies of the given configuration -
             each copy shifts the thrusters by dx and each copy is saved into an array.
@@ -99,7 +111,9 @@ def sweep_coords(self, config, x0, xf, dx):
 
         return configs_swept_coords
 
-    def read_swept_coords(self, swept_configs):
+    def read_swept_coords(
+        self, swept_configs: Sequence[dict[str, ThrusterConfig]]
+    ) -> None:
         '''
             Prints to terminal the position of all thrusters for all the swept cofigurations.
 
@@ -167,7 +181,8 @@ class SweepDecelAngles:
         Prints to terminal the DCM of all thrusters for all the swept cofigurations
     '''
 
-    def __init__(self, config, thruster_groups):
+    def __init__(self, config: dict[str, ThrusterConfig],
+                 thruster_groups: dict[str, list[str]]) -> None:
         '''
             Simple constructor.
             Standardizes each thruster to normalize how cant sweeps are performed.
@@ -201,7 +216,9 @@ def __init__(self, config, thruster_groups):
         #     standardized_thruster = self.standardize_thruster_normal(config[thruster])
         #     config[thruster] = standardized_thruster
 
-    def standardize_thruster_normal(self, thruster):
+    def standardize_thruster_normal(
+            self, thruster: ThrusterConfig
+    ) -> ThrusterConfig:
             '''
                 Method grabs a thruster's group and sets its DCM such that the thruster is pointed
                 directly opposite to the translational direction. 
@@ -228,7 +245,7 @@ def standardize_thruster_normal(self, thruster):
 
             return thruster
 
-    def set_lm(self, LogisticsModule):
+    def set_lm(self, LogisticsModule: LogisticsModule) -> None:
         """
             Simple setter method to set VV/LM used in analysis.
 
@@ -243,7 +260,7 @@ def set_lm(self, LogisticsModule):
         """
         self.lm = LogisticsModule
 
-    def categorize_rcs_groups(self):
+    def categorize_rcs_groups(self) -> None:
         '''
             Populates lists containing thrusters from several tgf groups and
             saves them as class members.
@@ -298,7 +315,7 @@ def categorize_rcs_groups(self):
 
         return
 
-    def calculate_DCM(self, cant, thruster):
+    def calculate_DCM(self, cant: float, thruster: str) -> list[list[float]]:
         '''
             Given the cant angle:
             calculate the DCM to angle that thruster "cant" deg about z axis.
@@ -356,7 +373,7 @@ def calculate_DCM(self, cant, thruster):
 
         return DCM.tolist()
 
-    def calculate_frame_rot(self, thruster_name):
+    def calculate_frame_rot(self, thruster_name: str) -> list[list[float]]:
         '''
             Given the thruster_name,
             determine the matrix by which to rotate the coordinate frame.
@@ -392,12 +409,17 @@ def calculate_frame_rot(self, thruster_name):
 
         return Tx.tolist()
 
-    def cant_decel_thrusters(self, cant):
+    def cant_decel_thrusters(
+        self, cant: float
+    ) -> dict[str, ThrusterConfig]:
         """
         """
         new_config = {}
 
-        Rz = self.calculate_DCM(cant)
+        # calculate_DCM takes (cant, thruster); this call omits the thruster
+        # and raises TypeError today. Flagged, not fixed -- supplying an
+        # argument would invent behavior (see deferred observations).
+        Rz = self.calculate_DCM(cant)  # type: ignore[call-arg]
 
         for thruster, thruster_info in self.config.items():
             new_thruster_info = thruster_info.copy()
@@ -411,7 +433,9 @@ def cant_decel_thrusters(self, cant):
 
         return new_config
 
-    def sweep_decel_thrusters_all(self, dcant):
+    def sweep_decel_thrusters_all(
+        self, dcant: int
+    ) -> list[dict[str, ThrusterConfig]]:
         '''
             Sweeps the given config by angle. Performed over min and max allowed. 
             All thrusters are canted simultaneously.
@@ -445,7 +469,11 @@ def sweep_decel_thrusters_all(self, dcant):
 
             # Rz = self.calculate_DCM(cant)
 
-            for thruster, thruster_info in config.items():
+            # `config` is undefined in this scope (flake8 also reports F821);
+            # this loop raises NameError. The block below it recomputes the
+            # swept configs from cant_decel_thrusters, so the method 'works'
+            # only because nothing reaches here first. Flagged, not fixed.
+            for thruster, thruster_info in config.items():  # type: ignore[name-defined]
                 new_thruster_info = thruster_info.copy()
 
                 for match in self.lm.rcs_groups['neg_x']:
@@ -470,7 +498,9 @@ def sweep_decel_thrusters_all(self, dcant):
 
         return configs_swept_angles
 
-    def one_cant_decel_thrusters_all(self, config, cant):
+    def one_cant_decel_thrusters_all(
+        self, config: dict[str, ThrusterConfig], cant: float
+    ) -> dict[str, ThrusterConfig]:
         '''
             Sweeps the given config by angle. Performed over min and max allowed. 
             All thrusters are canted simultaneously.
@@ -511,7 +541,9 @@ def one_cant_decel_thrusters_all(self, config, cant):
 
         return new_config
 
-    def read_swept_angles(self, swept_configs):
+    def read_swept_angles(
+        self, swept_configs: Sequence[dict[str, ThrusterConfig]]
+    ) -> None:
         '''
             Prints to terminal the DCM of all thrusters for all the swept cofigurations.