diff --git a/pyproject.toml b/pyproject.toml index 2ed903c..cda3298 100644 --- a/pyproject.toml +++ b/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 diff --git a/pyrpod/config_test.py b/pyrpod/config_test.py index 12de6ac..844ac70 100644 --- a/pyrpod/config_test.py +++ b/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) \ No newline at end of file diff --git a/pyrpod/logging_utils.py b/pyrpod/logging_utils.py index aa0864c..86eb58c 100644 --- a/pyrpod/logging_utils.py +++ b/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) diff --git a/pyrpod/mdao/SweepConfig.py b/pyrpod/mdao/SweepConfig.py index a4219ba..6f8c6b9 100644 --- a/pyrpod/mdao/SweepConfig.py +++ b/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. diff --git a/pyrpod/mdao/TradeStudy.py b/pyrpod/mdao/TradeStudy.py index 662d222..89890b3 100644 --- a/pyrpod/mdao/TradeStudy.py +++ b/pyrpod/mdao/TradeStudy.py @@ -1,7 +1,12 @@ +from __future__ import annotations + import os import csv import time import logging +from collections.abc import Mapping +from typing import Any + import numpy as np import pandas as pd import matplotlib.pyplot as plt @@ -11,17 +16,26 @@ from pyrpod.util.io.fs import ensure_dir import configparser +from pyrpod.vehicle.LogisticsModule import LogisticsModule +from pyrpod.vehicle.TargetVehicle import TargetVehicle + logger = logging.getLogger(__name__) class TradeStudy(): - def __init__(self, case_dir): + # Built by init_trade_study and read by every sweep; max_v0 is recorded + # by the sweeps and read back by print_mission_report. Declared bare so + # no class attribute is created at runtime. + rpod: Any + max_v0: Any + + def __init__(self, case_dir: str) -> None: self.case_dir = case_dir config = configparser.ConfigParser() config.read(self.case_dir + "config.ini") self.config = config - def init_trade_study(self, lm, tv): + def init_trade_study(self, lm: LogisticsModule, tv: TargetVehicle) -> None: """ Organizes data needed to kick off an RPOD trade study. @@ -35,11 +49,14 @@ def init_trade_study(self, lm, tv): jfh = JetFiringHistory.JetFiringHistory(case_dir) # Instantiate RPOD object. - rpod = PlumeStrikeEstimationStudy.RPOD(case_dir) + # The class in that module is PlumeStrikeEstimationStudy; there is no + # RPOD, so this raises AttributeError. Renaming it is a behavior + # change (see deferred observations), so it is flagged here. + rpod = PlumeStrikeEstimationStudy.RPOD(case_dir) # type: ignore[attr-defined] rpod.study_init(jfh, tv, lm) self.rpod = rpod - def init_trade_study_case(self): + def init_trade_study_case(self) -> None: """ Resets JFH data according to current case key. @@ -53,7 +70,7 @@ def init_trade_study_case(self): self.rpod.jfh = jfh - def print_mission_report(self): + def print_mission_report(self) -> None: case_key = self.rpod.get_case_key() @@ -91,7 +108,7 @@ def print_mission_report(self): max_cum_heat_load ]) - def graph_mission_report(self, report_results): + def graph_mission_report(self, report_results: pd.DataFrame) -> None: """ """ # Extract the first column (assuming it's the parameter you want to plot against) @@ -116,7 +133,7 @@ def graph_mission_report(self, report_results): plt.show() - def interpret_mission_report(self): + def interpret_mission_report(self) -> None: """ """ report_path = self.case_dir + 'results/MissionReport.csv' @@ -154,7 +171,9 @@ def interpret_mission_report(self): report_results['PlumeFailureMode'] = plume_failure_mode self.graph_mission_report(report_results) - def run_axial_overshoot_sweep(self, sweep_vars, lm, tv): + def run_axial_overshoot_sweep(self, sweep_vars: Mapping[str, Any], + lm: LogisticsModule, + tv: TargetVehicle) -> None: """ Simple variable sweep study that assesses RCS performance for a given set of axial overshoot velocity values. """ @@ -210,7 +229,9 @@ def run_axial_overshoot_sweep(self, sweep_vars, lm, tv): time.perf_counter() - study_start) self.interpret_mission_report() - def run_surface_cant_sweep(self, sweep_vars, lm, tv): + def run_surface_cant_sweep(self, sweep_vars: Mapping[str, Any], + lm: LogisticsModule, + tv: TargetVehicle) -> None: """ """ @@ -259,7 +280,9 @@ def run_surface_cant_sweep(self, sweep_vars, lm, tv): self.print_mission_report() self.interpret_mission_report() - def run_multi_var_sweep(self, sweep_vars, lm, tv): + def run_multi_var_sweep(self, sweep_vars: Mapping[str, Any], + lm: LogisticsModule, + tv: TargetVehicle) -> None: """ """ # Organize variables to sweet over. diff --git a/pyrpod/mission/MissionEnvironment.py b/pyrpod/mission/MissionEnvironment.py index 78f2c7c..8be62f8 100644 --- a/pyrpod/mission/MissionEnvironment.py +++ b/pyrpod/mission/MissionEnvironment.py @@ -11,7 +11,7 @@ @dataclass class MissionEnvironment: - def __init__(self, case_dir): + def __init__(self, case_dir: str) -> None: self.case_dir = case_dir config = configparser.ConfigParser() config.read(self.case_dir + "config.ini") @@ -46,7 +46,11 @@ def __init__(self, case_dir): mass_profile: List[float] = field(default_factory=list) # --- Accessors and Mutators --- - def get_current_state(self) -> Dict[str, np.ndarray]: + # The four kinematic fields default to None and nothing in this class + # populates them, so the returned mapping is Optional-valued. The previous + # Dict[str, np.ndarray] annotation was wrong on a freshly built + # environment, which is the only state PyRPOD ever puts one in today. + def get_current_state(self) -> Dict[str, Optional[np.ndarray]]: return { "v_current": self.v_current, "w_current": self.w_current, @@ -54,34 +58,45 @@ def get_current_state(self) -> Dict[str, np.ndarray]: "w_desired": self.w_desired, } - def update_state(self, v: np.ndarray, w: np.ndarray): + def update_state(self, v: np.ndarray, w: np.ndarray) -> None: self.v_current = v self.w_current = w - def set_desired_state(self, v_des: np.ndarray, w_des: np.ndarray): + def set_desired_state(self, v_des: np.ndarray, w_des: np.ndarray) -> None: self.v_desired = v_des self.w_desired = w_des def get_jfh_segment(self, t: float) -> Any: + # As the comment says, this assumes a query interface that + # JetFiringHistory does not implement; the accessor has never worked. + # Flagged rather than fixed (see the report's deferred observations). # Assumes jfh provides a query interface by time - return self.jfh.query(t) + return self.jfh.query(t) # type: ignore[attr-defined] def get_thruster_by_id(self, thruster_id: str) -> Any: - return self.thruster_data.get(thruster_id) - - def log_impingement(self, strike_data: Dict[str, Any]): + # thruster_data is never assigned: neither __init__ nor any dataclass + # field defines it, and the commented-out declaration above was never + # reinstated. Declaring it here is not an option -- this is a dataclass + # with fields that already carry defaults, so a bare annotation in the + # class body would add a field and break class creation. Flagged rather + # than fixed; see the report's deferred observations. + return self.thruster_data.get(thruster_id) # type: ignore[attr-defined] + + def log_impingement(self, strike_data: Dict[str, Any]) -> None: self.impingement_log.append(strike_data) - def log_burn(self, burn_data: Dict[str, Any]): + def log_burn(self, burn_data: Dict[str, Any]) -> None: self.burn_log.append(burn_data) - def log_delta_v(self, dv_data: Dict[str, Any]): + def log_delta_v(self, dv_data: Dict[str, Any]) -> None: self.delta_v_log.append(dv_data) - def update_mass_profile(self, mass: float): + def update_mass_profile(self, mass: float) -> None: self.mass_profile.append(mass) - def clone(self) -> "MissionContext": + # Was annotated "MissionContext", a class that does not exist anywhere in + # the repository; copy.deepcopy(self) returns this class. + def clone(self) -> "MissionEnvironment": # Deep copy method for simulation branching (if needed) import copy return copy.deepcopy(self) diff --git a/pyrpod/mission/MissionPlanner.py b/pyrpod/mission/MissionPlanner.py index 7d8fce5..325e6e9 100644 --- a/pyrpod/mission/MissionPlanner.py +++ b/pyrpod/mission/MissionPlanner.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pyrpod.mission.state_vectors import StateVectors from pyrpod.mission.six_dof_dynamics import SixDOFDynamics from pyrpod.mission.fuel_management import FuelManager @@ -7,10 +9,18 @@ from pyrpod.mission.orbital_transfer import OrbitalTransferEngine from pyrpod.mission.MissionEnvironment import MissionEnvironment +from pyrpod.rpod.JetFiringHistory import JetFiringHistory as JetFiringHistoryType +from pyrpod.vehicle.LogisticsModule import LogisticsModule as LogisticsModuleType + class MissionPlanner: - def __init__(self, environment: MissionEnvironment): + # Set as a CLASS attribute by PlumeStrikeEstimationStudy.calc_jfh_1d_approach + # (MissionPlanner.cant = np.radians(cant)) and read back from there. + # Declared bare so no attribute is created at import time. + cant: float + + def __init__(self, environment: MissionEnvironment) -> None: self.environment = environment # Initialize submodules with context if needed @@ -21,7 +31,7 @@ def __init__(self, environment: MissionEnvironment): self.dynamics = SixDOFDynamics() self.propellant = FuelManager(self.environment) - def set_lm(self, LogisticsModule): + def set_lm(self, LogisticsModule: LogisticsModuleType) -> None: """ Simple setter method to set VV/LM used in analysis. @@ -41,7 +51,7 @@ def set_lm(self, LogisticsModule): """ self.vv = LogisticsModule - def set_jfh(self, JetFiringHistory): + def set_jfh(self, JetFiringHistory: JetFiringHistoryType) -> None: """ Simple setter method to set JFH used in propellant usage calculations. @@ -56,17 +66,17 @@ def set_jfh(self, JetFiringHistory): """ self.jfh = JetFiringHistory - def load_components(self): + def load_components(self) -> None: self.flight_eval.load_plan() - def execute_mission(self): + def execute_mission(self) -> None: self.flight_eval.execute(self) - def analyze_maneuver(self): + def analyze_maneuver(self) -> None: dv, dw = self.state.get_deltas() self.dynamics.evaluate(dv, dw) - def summarize_results(self): + def summarize_results(self) -> None: self.post_processor.plot_burn_profiles() self.post_processor.plot_mass_usage() self.orbital_transfer.summarize() diff --git a/pyrpod/mission/SubModule.py b/pyrpod/mission/SubModule.py index 86b5ecd..3f007a6 100644 --- a/pyrpod/mission/SubModule.py +++ b/pyrpod/mission/SubModule.py @@ -1,5 +1,10 @@ +from __future__ import annotations + +from pyrpod.mission.MissionEnvironment import MissionEnvironment + + class SubModule: - def __init__(self, environment): + def __init__(self, environment: MissionEnvironment) -> None: self.environment = environment self.case_dir = environment.case_dir self.config = environment.config \ No newline at end of file diff --git a/pyrpod/mission/flight_eval.py b/pyrpod/mission/flight_eval.py index d875594..b160641 100644 --- a/pyrpod/mission/flight_eval.py +++ b/pyrpod/mission/flight_eval.py @@ -1,5 +1,11 @@ +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + import pandas as pd import numpy as np +from numpy.typing import NDArray from pyrpod.mission.SubModule import SubModule from pyrpod.mission.six_dof_dynamics import SixDOFDynamics @@ -8,16 +14,23 @@ class FlightEvaluator(SubModule, SixDOFDynamics): # self.maneuvers = [] - def load_plan(self): + # Set by read_flight_plan; None when the case configures no flight plan. + flight_plan: pd.DataFrame | None + + def load_plan(self) -> None: # Parse CSV or other source pass - def execute(self, mission_planner): + # mission_planner is the owning MissionPlanner. It is typed Any rather than + # MissionPlanner because MissionPlanner imports FlightEvaluator at module + # level, so naming the class here -- with or without an import -- is a + # circular import. The method is a stub, so nothing is lost today. + def execute(self, mission_planner: Any) -> None: # Loop over maneuvers and update planner pass - def read_flight_plan(self, vv): + def read_flight_plan(self, vv: Any) -> None: """ Reads in VV flight as specified using CSV format. @@ -49,7 +62,7 @@ def read_flight_plan(self, vv): return - def calc_flight_performance(self): + def calc_flight_performance(self) -> None: """ Calculates 6DOF performance for all firings specified in the flight plan. @@ -61,7 +74,7 @@ def calc_flight_performance(self): ------- None """ - for firing in self.flight_plan.iterrows(): + for firing in self.flight_plan.iterrows(): # type: ignore[union-attr] # Convert firing data to numpy arra for easier data manipulation. firing_array = np.array(firing[1]) @@ -89,7 +102,8 @@ def calc_flight_performance(self): return - def set_current_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): + def set_current_6dof_state(self, v: Sequence[float] | NDArray[Any] = [0, 0, 0], + w: Sequence[float] | NDArray[Any] = [0, 0, 0]) -> None: """ Sets current inertial state for the VV. Can be done manually or read from flight plan. @@ -110,7 +124,8 @@ def set_current_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): self.w_current = np.array(w) return - def set_desired_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): + def set_desired_6dof_state(self, v: Sequence[float] | NDArray[Any] = [0, 0, 0], + w: Sequence[float] | NDArray[Any] = [0, 0, 0]) -> None: """ Sets desired inertial state for the VV. Can be done manually or read from flight plan. @@ -131,13 +146,13 @@ def set_desired_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): self.w_desired = np.array(w) return - def get_deltas(self): + def get_deltas(self) -> tuple[NDArray[Any], NDArray[Any]]: return self.v_desired - self.v_current, self.w_desired - self.w_current # calc_trans_performance is inherited from SixDOFDynamics - def calc_6dof_performance(self): + def calc_6dof_performance(self) -> None: """ Wrapper method used to calculate performance for translation and rotational maneuvers. @@ -185,7 +200,11 @@ def calc_6dof_performance(self): # self.calc_rot_performance(motion) return - def calc_flight_performance(self): + # Byte-for-byte duplicate of the definition above; this one unconditionally + # shadows it at class-creation time. Removing the earlier copy is a code + # change, which #103 excludes, so the shadowing is flagged here instead + # (see the report's deferred observations). + def calc_flight_performance(self) -> None: # type: ignore[no-redef] """ Calculates 6DOF performance for all firings specified in the flight plan. @@ -197,7 +216,7 @@ def calc_flight_performance(self): ------- None """ - for firing in self.flight_plan.iterrows(): + for firing in self.flight_plan.iterrows(): # type: ignore[union-attr] # Convert firing data to numpy arra for easier data manipulation. firing_array = np.array(firing[1]) diff --git a/pyrpod/mission/fuel_management.py b/pyrpod/mission/fuel_management.py index e59fc18..5eb820d 100644 --- a/pyrpod/mission/fuel_management.py +++ b/pyrpod/mission/fuel_management.py @@ -1,4 +1,8 @@ +from __future__ import annotations + import logging +from collections.abc import Callable +from typing import Any from pyrpod.mission.SubModule import SubModule import numpy as np @@ -12,15 +16,27 @@ class FuelManager(SubModule): # self.isp = isp # self.mass = mass - def compute_burn_time(self, dv, thrust): + # Collaborator state and thruster-group helpers that this class reads but + # never defines. They come from the object that owns the fuel manager + # (see MissionPlanner/PlumeStrikeEstimationStudy and ThrusterGrouping); + # declared bare so no class attribute is created at runtime. + vv: Any + jfh: Any + cant: float + rotational_maneuvers: bool + calc_thrust_sum: Callable[[str], float] + calc_m_dot_sum: Callable[[str], float] + calc_v_e: Callable[[str], float] + + def compute_burn_time(self, dv: float, thrust: float) -> None: # Simplified rocket equation stub pass - def compute_delta_mass(self, dv): + def compute_delta_mass(self, dv: float) -> None: # Return delta-mass from ideal rocket equation pass - def calc_burn_time(self, dv, isp, T): + def calc_burn_time(self, dv: float, isp: float, T: float) -> float: """ Calculates burn time when given change in velocity (dv), specific impulse (isp), and thrust (T) @@ -48,7 +64,7 @@ def calc_burn_time(self, dv, isp, T): K=(isp*g_0*m_f*(np.exp(a) - 1)) return K / T - def calc_delta_mass(self, dv, isp): + def calc_delta_mass(self, dv: float, isp: float) -> float: """ Calculates propellant usage using expressions derived from the ideal rocket equation. @@ -72,7 +88,8 @@ def calc_delta_mass(self, dv, isp): self.vv.mass += dm return dm - def calc_delta_v(self, dt, v_e, m_dot, m_current): + def calc_delta_v(self, dt: float, v_e: float, m_dot: float, + m_current: float) -> float: """ Calculates change in velocity from the Tsiolkovsky equation given change in time (dt), exhaust velocity (v_e), and mass flow rate (m_dot). @@ -98,7 +115,7 @@ def calc_delta_v(self, dt, v_e, m_dot, m_current): dv = v_e*np.log(((m_dot*dt)/m_current)+1) return dv - def calc_delta_omega_rotation(self, dm, group): + def calc_delta_omega_rotation(self, dm: float, group: str) -> float: """ Calculates the angular velocity by discretizing the propellant expenditure and iteratively solving the rotational form of Newton's Second Law with updated moments of inertia. @@ -125,7 +142,8 @@ def calc_delta_omega_rotation(self, dm, group): return dw - def calc_delta_mass_rotation(self, dw, group, forward_propagation): + def calc_delta_mass_rotation(self, dw: float, group: str, + forward_propagation: bool) -> float: """ Calculates propellant usage for a rotational maneuver by discretizing the desired angular velocity and iteratively solving the rotational form of Newton's Second Law with updated moments of inertia. @@ -170,7 +188,8 @@ def calc_delta_mass_rotation(self, dw, group, forward_propagation): logger.error('ERROR: functionality not added for a roll rotation') return dm - def calc_delta_mass_v_e(self, dv, v_e, forward_propagation): + def calc_delta_mass_v_e(self, dv: float, v_e: float, + forward_propagation: bool) -> float: """ Calculates propellant usage using expressions derived from the ideal rocket equation. @@ -202,7 +221,7 @@ def calc_delta_mass_v_e(self, dv, v_e, forward_propagation): return dm - def calc_total_delta_mass(self): + def calc_total_delta_mass(self) -> None: """ Sums total propellant expenditure. Starts with calculating the propellant expenditure for the JFH twice (approach which back propagates with a starting mass of 14,000 kg, @@ -218,7 +237,7 @@ def calc_total_delta_mass(self): Total change in mass. """ # Initialization - self.dm_total = 0 + self.dm_total: float = 0 dm_jfh_total = 0 payload_mass = 5400 # Docking mass and post delivery mass @@ -233,7 +252,7 @@ def calc_total_delta_mass(self): if self.jfh.JFH != None and len(self.jfh.JFH) > 0: # Read the JFH and add propellant expended for each firing to a sum for f in range(len(self.jfh.JFH)): - dm = 0 + dm: float = 0 # Backpropagate with a vv.mass of 14,000 kg to find the vv.mass pre-approach if m == 0: forward_propagation = False @@ -372,7 +391,7 @@ def calc_total_delta_mass(self): # print('-------- dm / 10 is', (dm / 10)) # Disregarding minimum duty cycle, discretize the dm discretizing_resolution = 0.0001 # kg / s - dw_sum_rot = 0 # rad / s + dw_sum_rot: float = 0 # rad / s num_iters = round(((dm / 10) / np.cos(self.cant)) / (discretizing_resolution)) for i in range(num_iters): dw_rot = self.calc_delta_omega_rotation(discretizing_resolution, 'pos_pitch') @@ -380,7 +399,7 @@ def calc_total_delta_mass(self): # print('dw_sum_rot is', dw_sum_rot) discretizing_resolution = 0.0001 # rad / s - dm_sum_rot = 0 + dm_sum_rot: float = 0 # Currently the rotation calculation is hardcoded for the pitch maneuver, which is why two is subtracted num_iters = round(dw_sum_rot / discretizing_resolution) diff --git a/pyrpod/mission/orbital_transfer.py b/pyrpod/mission/orbital_transfer.py index 782160c..061790f 100644 --- a/pyrpod/mission/orbital_transfer.py +++ b/pyrpod/mission/orbital_transfer.py @@ -1,3 +1,7 @@ +from __future__ import annotations + +from typing import Any + from pyrpod.orbital import HohmannTransfer from astropy import units as u import logging @@ -5,21 +9,27 @@ logger = logging.getLogger(__name__) class OrbitalTransferEngine: - def __init__(self): - self.transfers = [] + def __init__(self) -> None: + self.transfers: list[Any] = [] - def add_hohmann_transfer(self, h1, h2): + def add_hohmann_transfer(self, h1: float, h2: float) -> None: # Add a Hohmann transfer case pass - def summarize(self): + def summarize(self) -> None: # Print or return summary pass - def init_hohmann_transfers(self): - self.hohmann_transfers = [] + def init_hohmann_transfers(self) -> None: + self.hohmann_transfers: list[dict[str, Any]] = [] - def add_hohmann_transfer(self, h1_km: float, h2_km: float, leg_id: str = None): + # The stub above is dead: this definition unconditionally shadows it at + # class-creation time, and it is the one every caller reaches. Removing + # the stub is a code change, which #103 excludes, so the shadowing is + # flagged here instead (see the report's deferred observations). + def add_hohmann_transfer( # type: ignore[no-redef] + self, h1_km: float, h2_km: float, leg_id: str | None = None + ) -> None: """ Computes and stores the delta-v and time of flight for a Hohmann transfer leg. @@ -36,7 +46,7 @@ def add_hohmann_transfer(self, h1_km: float, h2_km: float, leg_id: str = None): result['leg_id'] = leg_id self.hohmann_transfers.append(result) - def summarize_hohmann_transfers(self): + def summarize_hohmann_transfers(self) -> None: """ Prints a summary of all stored Hohmann transfers. """ diff --git a/pyrpod/mission/post_processing.py b/pyrpod/mission/post_processing.py index 6e037ca..b5cb4a4 100644 --- a/pyrpod/mission/post_processing.py +++ b/pyrpod/mission/post_processing.py @@ -1,4 +1,10 @@ +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + import numpy as np +import pandas as pd import matplotlib.pyplot as plt import logging @@ -6,17 +12,27 @@ class PostProcessor: - def __init__(self): + # Collaborator state and helpers the plotting methods read but this class + # never defines; they come from the object that owns the post-processor + # (see MissionPlanner, FuelManager and SixDOFDynamics). Declared bare so no + # class attribute is created at runtime. + vv: Any + flight_plan: pd.DataFrame + calc_burn_time: Callable[[float, float, float], float] + calc_delta_mass: Callable[[float, float], float] + calc_trans_performance: Callable[[str, Any], tuple[float, float, float]] + + def __init__(self) -> None: pass - def plot_burn_profiles(self): + def plot_burn_profiles(self) -> None: # Stub plot for burn times pass - def plot_mass_usage(self): + def plot_mass_usage(self) -> None: # Stub plot for fuel usage pass - def plot_burn_time(self, dv): + def plot_burn_time(self, dv: float) -> None: """ Plots burn time for a given dv and isp value. Varies thrust according to user inputs. @@ -41,7 +57,11 @@ def plot_burn_time(self, dv): """ isp_vals = [50, 200, 300, 400, 500] thrust_range = np.linspace(50, 600, 5000) - burn_time = [] + # These plotting accumulators start as a Python list and are then + # rebound to the NumPy array built from them. A list|NDArray union does + # not survive mypy's loop binder (it widens back at the loop edge and + # rejects .append), so the rebound locals in this module stay dynamic. + burn_time: Any = [] isp = 200 for thrust in thrust_range: @@ -60,7 +80,7 @@ def plot_burn_time(self, dv): # plt.yscale("log") fig.savefig("test.png") - def plot_burn_time_contour(self, dv): + def plot_burn_time_contour(self, dv: float) -> None: """ Plots burn time for a given dv by varrying thrust values. Graph is contoured using ISP values. @@ -82,7 +102,7 @@ def plot_burn_time_contour(self, dv): fig, ax = plt.subplots() for isp in isp_vals: - burn_time = [] + burn_time: Any = [] for thrust in thrust_range: burn_time.append(abs(self.calc_burn_time(dv, isp, thrust))) burn_time = np.array(burn_time) / (3600*24) @@ -97,7 +117,7 @@ def plot_burn_time_contour(self, dv): fig.savefig("test.png") return - def plot_burn_time_flight_plan(self): + def plot_burn_time_flight_plan(self) -> None: """ Plots burn time for all dv maneuvers in the specified flight plan. @@ -121,7 +141,7 @@ def plot_burn_time_flight_plan(self): # print(type(v[1])) dv = v[1][1] logger.debug('Processing dv entry in flight plan') - burn_time = [] + burn_time: Any = [] for thrust in thrust_range: burn_time.append(abs(self.calc_burn_time(dv, isp, thrust))) burn_time = np.array(burn_time) / (3600*24) @@ -137,7 +157,7 @@ def plot_burn_time_flight_plan(self): return - def plot_delta_mass(self, dv): + def plot_delta_mass(self, dv: float) -> None: """ Plots propellant usage for a given dv requirements by varying ISP according to user inputs. @@ -152,7 +172,7 @@ def plot_delta_mass(self, dv): Does the method need to return a status message? or pass similar data? """ isp_range = np.linspace(100, 600, 5000) - delta_mass = [] + delta_mass: Any = [] for isp in isp_range: delta_mass.append(abs(self.calc_delta_mass(dv, isp))) @@ -186,7 +206,7 @@ def plot_delta_mass(self, dv): # plt.yscale("log") fig.savefig("test.png") - def plot_delta_mass_contour(self): + def plot_delta_mass_contour(self) -> None: """ Co-Plots propellant usage for all dv maneuvers in the specified flight plan. @@ -204,8 +224,8 @@ def plot_delta_mass_contour(self): #creat plotting object. fig, ax = plt.subplots() - delta_mass_min = 10e9 - delta_mass_max = 0 + delta_mass_min: float = 10e9 + delta_mass_max: float = 0 # Step through all planned firings in the flight plan for firing in self.flight_plan.iterrows(): @@ -214,7 +234,7 @@ def plot_delta_mass_contour(self): # Calculate change in mass for a given range of ISP values. isp_range = np.linspace(50, 400, 5000) - delta_mass = [] + delta_mass: Any = [] for isp in isp_range: delta_mass.append(abs(self.calc_delta_mass(dv, isp))) @@ -259,7 +279,7 @@ def plot_delta_mass_contour(self): fig.savefig("test.png") return - def plot_thrust_envelope(self): + def plot_thrust_envelope(self) -> None: """ Plots operational envelope relating burn time to thrust required for all firings in the flight plan. diff --git a/pyrpod/mission/six_dof_dynamics.py b/pyrpod/mission/six_dof_dynamics.py index c863926..77da14b 100644 --- a/pyrpod/mission/six_dof_dynamics.py +++ b/pyrpod/mission/six_dof_dynamics.py @@ -1,21 +1,32 @@ +from __future__ import annotations + +from typing import Any + + class SixDOFDynamics: # def __init__(self): # self.thruster_model = thruster_model - def evaluate_translational(self, dv): + # Supplied by whichever object mixes this class in (FlightEvaluator sets + # it in read_flight_plan); SixDOFDynamics never constructs it itself. + vv: Any + + def evaluate_translational(self, dv: Any) -> None: # Placeholder logic for translational maneuver pass - def evaluate_rotational(self, dw): + def evaluate_rotational(self, dw: Any) -> None: # Placeholder logic for rotational maneuver pass - def evaluate(self, dv, dw): + def evaluate(self, dv: Any, dw: Any) -> None: # Combined 6DOF evaluation logic self.evaluate_translational(dv) self.evaluate_rotational(dw) - def calc_trans_performance(self, motion, dv): + def calc_trans_performance( + self, motion: str, dv: float + ) -> tuple[float, float, float] | None: """ Calculates RCS performance according to thruster working groups for a direction of motion. @@ -48,7 +59,11 @@ def calc_trans_performance(self, motion, dv): # print(self.vv) if self.vv.rcs_groups == None: # print("WARNING: Thruster Grouping File not Set") - return + # mypy asks for an explicit `return None` whenever the declared + # return type is not plain None, even when None is part of the + # union. A bare return and `return None` compile identically, so + # the source is left as-is rather than edited for the checker. + return # type: ignore[return-value] n_thrusters = len(self.vv.rcs_groups[motion]) total_thrust = n_thrusters * self.vv.thrust diff --git a/pyrpod/mission/state_vectors.py b/pyrpod/mission/state_vectors.py index 24f1b64..b175a8e 100644 --- a/pyrpod/mission/state_vectors.py +++ b/pyrpod/mission/state_vectors.py @@ -1,13 +1,21 @@ +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + import numpy as np +from numpy.typing import NDArray class StateVectors: - def __init__(self, v=None, w=None): + def __init__(self, v: Sequence[float] | None = None, + w: Sequence[float] | None = None) -> None: self.v_current = np.array(v) if v else np.zeros(3) self.w_current = np.array(w) if w else np.zeros(3) self.v_desired = np.zeros(3) self.w_desired = np.zeros(3) - def set_current_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): + def set_current_6dof_state(self, v: Sequence[float] = [0, 0, 0], + w: Sequence[float] = [0, 0, 0]) -> None: """ Sets current inertial state for the VV. Can be done manually or read from flight plan. @@ -28,7 +36,8 @@ def set_current_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): self.w_current = np.array(w) return - def set_desired_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): + def set_desired_6dof_state(self, v: Sequence[float] = [0, 0, 0], + w: Sequence[float] = [0, 0, 0]) -> None: """ Sets desired inertial state for the VV. Can be done manually or read from flight plan. @@ -49,5 +58,5 @@ def set_desired_6dof_state(self, v = [0, 0, 0], w = [0,0,0]): self.w_desired = np.array(w) return - def get_deltas(self): + def get_deltas(self) -> tuple[NDArray[Any], NDArray[Any]]: return self.v_desired - self.v_current, self.w_desired - self.w_current \ No newline at end of file diff --git a/pyrpod/mission/thruster_grouping.py b/pyrpod/mission/thruster_grouping.py index a6cfc90..1485871 100644 --- a/pyrpod/mission/thruster_grouping.py +++ b/pyrpod/mission/thruster_grouping.py @@ -1,20 +1,32 @@ +from __future__ import annotations + +from typing import Any + + class ThrusterGrouping: - def __init__(self, rcs_config): + # Collaborator state this class reads but never sets: the visiting vehicle + # and the deceleration cant angle are expected to be supplied by the owning + # study object. Declared (bare, so no class attribute is created) to keep + # the aggregation signatures below meaningful. + vv: Any + cant: float + + def __init__(self, rcs_config: dict[str, list[str]]) -> None: self.rcs_groups = rcs_config - def sum_thrust(self, group): + def sum_thrust(self, group: str) -> None: # Aggregate thrust from a group pass - def sum_mass_flow(self, group): + def sum_mass_flow(self, group: str) -> None: # Aggregate m_dot from a group pass - def mean_exhaust_velocity(self, group): + def mean_exhaust_velocity(self, group: str) -> None: # Compute effective exhaust velocity pass - def calc_thrust_sum(self, group): + def calc_thrust_sum(self, group: str) -> float: """ Calculates thrust sum for a thruster group. @@ -41,12 +53,16 @@ def calc_thrust_sum(self, group): for thruster_name in self.vv.rcs_groups[group]: thruster_type = self.vv.thruster_data[thruster_name]['type'][0] # print('thruster_type is', thruster_type) - thrust = np.cos(self.cant) * self.vv.thruster_metrics[thruster_type]['F'] + # numpy is not imported in this module, so this branch raises + # NameError today. Adding the import would turn a crash into a + # working calculation, which is a behavior change #103 must not + # make; flagged here instead (see deferred observations). + thrust = np.cos(self.cant) * self.vv.thruster_metrics[thruster_type]['F'] # type: ignore[name-defined] # print('thrust is', thrust) thrust_sum += thrust return thrust_sum - def calc_m_dot_sum(self, group): + def calc_m_dot_sum(self, group: str) -> float: """ Calculates mass flow rate sum for a thruster group. @@ -66,7 +82,7 @@ def calc_m_dot_sum(self, group): m_dot_sum += m_dot return m_dot_sum - def calc_v_e(self, group): + def calc_v_e(self, group: str) -> float: """ Calculates mean exhaust velocity for a thruster group. @@ -79,8 +95,8 @@ def calc_v_e(self, group): ------- Exhaust velocity. """ - thrust_sum = 0 - m_dot_sum = 0 + thrust_sum: float = 0 + m_dot_sum: float = 0 thrust_sum = self.calc_thrust_sum(group) m_dot_sum = self.calc_m_dot_sum(group) v_e = thrust_sum / m_dot_sum diff --git a/pyrpod/orbital/HohmannTransfer.py b/pyrpod/orbital/HohmannTransfer.py index 417496e..8ffba1d 100644 --- a/pyrpod/orbital/HohmannTransfer.py +++ b/pyrpod/orbital/HohmannTransfer.py @@ -1,9 +1,17 @@ +from __future__ import annotations + +from typing import Any + from astropy import units as u from astropy.constants import G, R_earth, M_earth import numpy as np class HohmannTransfer: - def __init__(self, r1_km: float, r2_km: float, leg_id: str = ""): + # leg_id was annotated ``str = ""``, but OrbitalTransferEngine passes None + # for an unnamed leg and the ``or`` below is what supplies the fallback + # label, so None has always been accepted at runtime. + def __init__(self, r1_km: float, r2_km: float, + leg_id: str | None = "") -> None: self.leg_id = leg_id or f"{r1_km} km → {r2_km} km" self.r1_km = r1_km self.r2_km = r2_km @@ -20,7 +28,7 @@ def __init__(self, r1_km: float, r2_km: float, leg_id: str = ""): self._compute_transfer() - def _compute_transfer(self): + def _compute_transfer(self) -> dict[str, Any]: # Circular speeds self.v1_circ = np.sqrt(self.mu_earth / self.r1) self.v2_circ = np.sqrt(self.mu_earth / self.r2) @@ -42,7 +50,7 @@ def _compute_transfer(self): return self.summary(verbose=False) - def summary(self, verbose=True) -> dict: + def summary(self, verbose: bool = True) -> dict[str, Any]: if verbose: print(f"\n=== Hohmann Transfer: {self.leg_id} ===") print(f"Initial altitude: {self.r1_km} km → Radius: {self.r1.to(u.km):.2f}") diff --git a/pyrpod/plume/CaiImpingement2016.py b/pyrpod/plume/CaiImpingement2016.py index 3a6a6dc..2d7b39f 100644 --- a/pyrpod/plume/CaiImpingement2016.py +++ b/pyrpod/plume/CaiImpingement2016.py @@ -111,7 +111,12 @@ dimensionless and independent of n_0. """ +from __future__ import annotations + +from typing import Any, TypeAlias + import numpy as np +from numpy.typing import ArrayLike, NDArray from scipy.special import erf from pyrpod.plume.RarefiedPlumeGasKinetics import ( @@ -121,6 +126,11 @@ get_Q_full, ) +# Local shorthands: every routine here returns float64 arrays, or a dict of +# them keyed by coefficient name. +FloatArray: TypeAlias = "NDArray[np.float64]" +CoefficientField: TypeAlias = "dict[str, NDArray[np.float64]]" + #: Gauss-Legendre order per axis for the exit-disk integrals. The integrand #: is analytic on the compact disk (denominators bounded below by X^2 > 0), #: so convergence is geometric; 48 nodes reach ~1e-12 for every geometry in @@ -128,7 +138,9 @@ DEFAULT_ORDER = 48 -def _scaled_A_factors(a, S_0): +def _scaled_A_factors( + a: FloatArray, S_0: float +) -> tuple[FloatArray, FloatArray, FloatArray]: ''' Appendix-A factors A_1, A_2, A_3 evaluated at a and pre-multiplied by the solutions' e^(-S_0^2) prefactor, combined overflow-safely @@ -156,7 +168,9 @@ def _scaled_A_factors(a, S_0): return A1, A2, A3 -def plate_point_coords(s, tau, alpha_0, L): +def plate_point_coords( + s: ArrayLike, tau: ArrayLike, alpha_0: float, L: float +) -> tuple[FloatArray, FloatArray, FloatArray]: ''' Global coordinates of plate points from local plate coordinates (see module docstring for the convention). @@ -183,8 +197,10 @@ def plate_point_coords(s, tau, alpha_0, L): return X, Y, Z -def surface_coefficients(X, Y, Z, S_0, alpha_0, eps, R_0, - order=DEFAULT_ORDER, chunk=256): +def surface_coefficients(X: ArrayLike, Y: ArrayLike, Z: ArrayLike, + S_0: float, alpha_0: float, eps: float, + R_0: float, order: int = DEFAULT_ORDER, + chunk: int = 256) -> CoefficientField: ''' Exact surface coefficients of Cai 2016 Eqs. 9-14 at global plate points (X, Y, Z), vectorized with tensor-product Gauss-Legendre @@ -296,8 +312,10 @@ def surface_coefficients(X, Y, Z, S_0, alpha_0, eps, R_0, 'Cp_s': Cp_s.reshape(shape), 'nw': nw.reshape(shape)} -def surface_coefficients_plate(s, tau, S_0, alpha_0, eps, R_0, L, - order=DEFAULT_ORDER): +def surface_coefficients_plate(s: ArrayLike, tau: ArrayLike, S_0: float, + alpha_0: float, eps: float, R_0: float, + L: float, + order: int = DEFAULT_ORDER) -> CoefficientField: ''' Convenience wrapper of surface_coefficients over local plate coordinates (s, tau); see plate_point_coords for the mapping. @@ -311,8 +329,10 @@ def surface_coefficients_plate(s, tau, S_0, alpha_0, eps, R_0, L, return surface_coefficients(X, Y, Z, S_0, alpha_0, eps, R_0, order=order) -def averaged_coefficients(S_0, alpha_0, eps, R_0, L, W_0, H_0, - n_gl=64, order=DEFAULT_ORDER): +def averaged_coefficients(S_0: float, alpha_0: float, eps: float, + R_0: float, L: float, W_0: float, H_0: float, + n_gl: int = 64, + order: int = DEFAULT_ORDER) -> dict[str, float]: ''' Plate-averaged properties of Eq. 15 by Gauss-Legendre quadrature over the plate: CP, CF1, CF2, CQ, the moment coefficient @@ -360,7 +380,9 @@ def averaged_coefficients(S_0, alpha_0, eps, R_0, L, W_0, H_0, # Section 3: 2D slot jet impinging on an inclined planar plate (Eqs. 1-8) # --------------------------------------------------------------------------- -def _scaled_planar_factors(a, S): +def _scaled_planar_factors( + a: FloatArray, S: float +) -> tuple[FloatArray, FloatArray, FloatArray]: ''' e^(-S^2)-scaled polar-velocity wedge moments e^(a^2) * I_k(a) of a drifting Maxwellian in 2D, k = 1, 2, 3, where @@ -381,7 +403,8 @@ def _scaled_planar_factors(a, S): return E1, A0, A1 -def _scaled_G_factor(a, S): +def _scaled_G_factor(a: NDArray[np.float64], + S: float) -> NDArray[np.float64]: ''' e^(-S^2)-scaled energy-flux integrand of Eq. 4, G(a) = (sqrt(pi)/2)(2 + 7a^2 + 2a^4) e^(a^2) [1 + erf(a)] + 3a + a^3. @@ -393,14 +416,17 @@ def _scaled_G_factor(a, S): + (3 * a + a ** 3) * np.exp(-S ** 2)) -def planar_plate_point_coords(s, alpha_0, L): +def planar_plate_point_coords( + s: ArrayLike, alpha_0: float, L: float +) -> tuple[NDArray[np.float64], NDArray[np.float64]]: '''2D plate point (X, Y) = (L + s cos(alpha_0), s sin(alpha_0)).''' s = np.asarray(s, dtype=float) return L + s * np.cos(alpha_0), s * np.sin(alpha_0) -def planar_surface_coefficients(s, S_0, alpha_0, eps, H, L, - order=DEFAULT_ORDER): +def planar_surface_coefficients(s: ArrayLike, S_0: float, alpha_0: float, + eps: float, H: float, L: float, + order: int = DEFAULT_ORDER) -> CoefficientField: ''' Exact 2D surface coefficients of Cai 2016 Eqs. 2-4 and 8 at plate positions s (distance from the plate center along the plate). @@ -442,7 +468,7 @@ def planar_surface_coefficients(s, S_0, alpha_0, eps, H, L, I_q = np.sum(w * Gs * mu_p, axis=1) I_n = np.sum(w * A0s * mu_p, axis=1) - def expand(vals): + def expand(vals: NDArray[Any]) -> NDArray[np.float64]: full = np.zeros(X.size) full[ahead] = vals return full.reshape(shape) @@ -456,7 +482,8 @@ def expand(vals): 'Cf_d': Cf_d, 'Cq_d': Cq_d, 'Cp_s': 2.0 * Cp_jet, 'nw': nw} -def _mirror_about_plate_2d(p, alpha_0, L): +def _mirror_about_plate_2d(p: ArrayLike, alpha_0: float, + L: float) -> NDArray[np.float64]: '''Mirror 2D point(s) about the plate line through (L, 0) at alpha_0.''' n_hat = np.array([-np.sin(alpha_0), np.cos(alpha_0)]) p = np.asarray(p, dtype=float) @@ -464,8 +491,11 @@ def _mirror_about_plate_2d(p, alpha_0, L): return p - 2.0 * d[..., None] * n_hat -def planar_flowfield(X, Y, S_0, alpha_0, eps, H, L, W, plate='diffuse', - order=DEFAULT_ORDER, nw_grid=1024): +def planar_flowfield(X: ArrayLike, Y: ArrayLike, S_0: float, + alpha_0: float, eps: float, H: float, L: float, + W: float, plate: str | None = 'diffuse', + order: int = DEFAULT_ORDER, + nw_grid: int = 1024) -> CoefficientField: ''' Combined 2D flowfield moments at points (X, Y) for the Section-3 problem: the free slot jet plus either the diffuse-plate wall @@ -500,8 +530,10 @@ def planar_flowfield(X, Y, S_0, alpha_0, eps, H, L, W, plate='diffuse', MY = np.zeros(Xa.size) M2 = np.zeros(Xa.size) - def add_population(th_lo, span, drift_angle, S, beta_ratio, - n_ref=None): + def add_population(th_lo: NDArray[np.float64], + span: NDArray[np.float64], drift_angle: float, + S: float, beta_ratio: float, + n_ref: NDArray[np.float64] | None = None) -> None: '''Wedge population moments; n_ref = None means constant n_0.''' nonlocal N, MX, MY, M2 theta = th_lo + span * 0.5 * (nodes[None, :] + 1.0) @@ -572,7 +604,7 @@ def add_population(th_lo, span, drift_angle, S, beta_ratio, Uy = MY / N T = (2.0 / 3.0) * (M2 / N - Ux ** 2 - Uy ** 2) - def expand(vals): + def expand(vals: NDArray[Any]) -> NDArray[np.float64]: full = np.full(Xf.size, np.nan) full[valid] = vals return full.reshape(shape) @@ -585,8 +617,19 @@ def expand(vals): # Section 4 flowfield pressure in the Y = 0 plane (Figs. 15-16) # --------------------------------------------------------------------------- -def _plate_emission_moments(Xa, Za, Px, Py, Pz, nw, wA, alpha_0, eps, - chunk=128): +def _plate_emission_moments( + Xa: NDArray[np.float64], + Za: NDArray[np.float64], + Px: NDArray[np.float64], + Py: NDArray[np.float64], + Pz: NDArray[np.float64], + nw: NDArray[np.float64], + wA: NDArray[np.float64], + alpha_0: float, + eps: float, + chunk: int = 128, +) -> tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64], + NDArray[np.float64], NDArray[np.float64]]: ''' Raw-moment contributions of the diffuse-wall emission at field points (Xa, 0, Za): solid-angle integrals over plate nodes @@ -625,8 +668,11 @@ def _plate_emission_moments(Xa, Za, Px, Py, Pz, nw, wA, alpha_0, eps, return dN, dMx, dMy, dMz, dM2 -def _jet_moments_3d(x_loc, rho_loc, S_0, R_0, order=DEFAULT_ORDER, - chunk=256): +def _jet_moments_3d( + x_loc: ArrayLike, rho_loc: ArrayLike, S_0: float, R_0: float, + order: int = DEFAULT_ORDER, chunk: int = 256, +) -> tuple[NDArray[np.float64], NDArray[np.float64], NDArray[np.float64], + NDArray[np.float64]]: ''' Raw moments (n/n_0, U*sqrt(beta_0), W*sqrt(beta_0), M2 = n beta_0^2 / n_0) of the free round jet at local @@ -673,9 +719,13 @@ def _jet_moments_3d(x_loc, rho_loc, S_0, R_0, order=DEFAULT_ORDER, return n, U, W, M2 -def flowfield_pressure_plane(X, Z, S_0, alpha_0, eps, R_0, L, W_0, H_0, - plate='diffuse', order=DEFAULT_ORDER, - plate_order=48, chunk=128): +def flowfield_pressure_plane(X: ArrayLike, Z: ArrayLike, S_0: float, + alpha_0: float, eps: float, R_0: float, + L: float, W_0: float, H_0: float, + plate: str | None = 'diffuse', + order: int = DEFAULT_ORDER, + plate_order: int = 48, + chunk: int = 128) -> CoefficientField: ''' Static-pressure field p/p_0 = (n/n_0)(T/T_0) in the vertical Y = 0 plane for the Section-4 3D impingement problem (Figs. 15-16), @@ -761,7 +811,7 @@ def flowfield_pressure_plane(X, Z, S_0, alpha_0, eps, R_0, L, W_0, H_0, T = (2.0 / 3.0) * (M2 / N - Vx ** 2 - Vy ** 2 - Vz ** 2) p = N * T - def expand(vals): + def expand(vals: NDArray[Any]) -> NDArray[np.float64]: full = np.full(Xf.size, np.nan) full[valid] = vals return full.reshape(shape) @@ -769,8 +819,10 @@ def expand(vals): return {'n': expand(N), 'T': expand(T), 'p': expand(p)} -def run_sanity_checks(S_0=2.0, alpha_0=np.deg2rad(60.0), eps=1.5, - R_0=0.5, L=4.0, verbose=True): +def run_sanity_checks(S_0: float = 2.0, + alpha_0: float = np.deg2rad(60.0), eps: float = 1.5, + R_0: float = 0.5, L: float = 4.0, + verbose: bool = True) -> dict[str, float]: ''' Built-in verification of the implementation: @@ -812,9 +864,11 @@ def run_sanity_checks(S_0=2.0, alpha_0=np.deg2rad(60.0), eps=1.5, sin_a, cos_a = np.sin(alpha_0), np.cos(alpha_0) for s0, t0 in [(0.0, 0.0), (1.5, -2.0), (-2.5, 3.0)]: X, Y, Z = plate_point_coords(s0, t0, alpha_0, L) - X, Y, Z = float(X), float(Y), float(Z) + # Deliberate 0-d array -> scalar narrowing: the spot check feeds these + # to scipy.integrate.dblquad, which wants plain floats. + X, Y, Z = float(X), float(Y), float(Z) # type: ignore[assignment] - def flux_integrand(r, th): + def flux_integrand(r: float, th: float) -> float: dy = Y - r * np.cos(th) dz = Z - r * np.sin(th) Q = np.sqrt((X ** 2 + dy ** 2 + dz ** 2) / X ** 2) @@ -848,8 +902,11 @@ def flux_integrand(r, th): return results -def run_planar_sanity_checks(S_0=2.0, alpha_0=np.deg2rad(60.0), eps=1.5, - H=0.5, L=4.0, W=5.0, verbose=True): +def run_planar_sanity_checks(S_0: float = 2.0, + alpha_0: float = np.deg2rad(60.0), + eps: float = 1.5, H: float = 0.5, + L: float = 4.0, W: float = 5.0, + verbose: bool = True) -> dict[str, float]: ''' Verification of the Section-3 (2D planar) implementation: @@ -888,7 +945,7 @@ def run_planar_sanity_checks(S_0=2.0, alpha_0=np.deg2rad(60.0), eps=1.5, for s0 in (0.0, -2.0, 3.0): X, Y = planar_plate_point_coords(s0, alpha_0, L) - def flux_integrand(theta): + def flux_integrand(theta: float) -> float: a = S_0 * np.cos(theta) _, A0s, _ = _scaled_planar_factors(np.asarray(a), S_0) return float(A0s) * max(np.sin(alpha_0 - theta), 0.0) @@ -935,9 +992,12 @@ def flux_integrand(theta): return results -def run_flowfield3d_sanity_checks(S_0=2.0, alpha_0=np.deg2rad(60.0), - eps=1.5, R_0=0.5, L=4.0, W_0=4.0, - H_0=4.0, verbose=True): +def run_flowfield3d_sanity_checks(S_0: float = 2.0, + alpha_0: float = np.deg2rad(60.0), + eps: float = 1.5, R_0: float = 0.5, + L: float = 4.0, W_0: float = 4.0, + H_0: float = 4.0, + verbose: bool = True) -> dict[str, float]: ''' Verification of the Section-4 flowfield-pressure implementation (Figs. 15-16): @@ -960,8 +1020,10 @@ def run_flowfield3d_sanity_checks(S_0=2.0, alpha_0=np.deg2rad(60.0), from scipy import integrate results = {} - kw = dict(S_0=S_0, alpha_0=alpha_0, eps=eps, R_0=R_0, L=L, - W_0=W_0, H_0=H_0) + # Heterogeneous keyword bundle forwarded with ** into + # flowfield_pressure_plane, whose parameters are not all float. + kw: dict[str, Any] = dict(S_0=S_0, alpha_0=alpha_0, eps=eps, R_0=R_0, L=L, + W_0=W_0, H_0=H_0) # 1. uniform-emitter sum rule (synthetic plate, alpha_0 = 90 deg so # the plate normal is -x and "on-axis height" is along x) @@ -987,7 +1049,7 @@ def run_flowfield3d_sanity_checks(S_0=2.0, alpha_0=np.deg2rad(60.0), emission_gl = float(both['n'][0] - jet['n'][0]) n_hat3 = np.array([-np.sin(alpha_0), 0.0, np.cos(alpha_0)]) - def integrand(tau, s): + def integrand(tau: float, s: float) -> float: px, py, pz = plate_point_coords(s, tau, alpha_0, L) nw = float(surface_coefficients( np.array([px]), np.array([py]), np.array([pz]), diff --git a/pyrpod/plume/IsentropicExpansion.py b/pyrpod/plume/IsentropicExpansion.py index c6f7299..28c6e42 100644 --- a/pyrpod/plume/IsentropicExpansion.py +++ b/pyrpod/plume/IsentropicExpansion.py @@ -1,37 +1,43 @@ #spherically symmetric isentropic expanison into a vacuum +from __future__ import annotations + import matplotlib.pyplot as plt import numpy as np from math import sqrt class IsentropicExpansion: - def calculate_temp(self, M, gamma, T_star): + def calculate_temp(self, M: float, gamma: float, T_star: float) -> float: T = (gamma + 1) * T_star / (2 + (gamma - 1) * M**2) return T - def calculate_temp_ratio(self, M, gamma): + def calculate_temp_ratio(self, M: float, gamma: float) -> float: T_ratio = (gamma + 1) / (2 + (gamma - 1) * M**2) return T_ratio - def calculate_radius(self, M, gamma, r_star): + def calculate_radius(self, M: float, gamma: float, + r_star: float) -> float: x = ((2 + (gamma - 1) * M**2) / (gamma + 1))**((gamma + 1) / (gamma - 1)) r = r_star * sqrt((1/M) * sqrt(x)) return r - def calculate_radius_ratio(self, M, gamma): + def calculate_radius_ratio(self, M: float, gamma: float) -> float: x = ((2 + (gamma - 1) * M**2) / (gamma + 1))**((gamma + 1) / (gamma - 1)) r_ratio = sqrt((1/M) * sqrt(x)) return r_ratio - def calculate_number_density(self, M, gamma, n_star): + def calculate_number_density(self, M: float, gamma: float, + n_star: float) -> float: n = n_star * ((gamma + 1) / (2 + (gamma - 1)* M**2)) ** (1 / (gamma - 1)) return n - def calculate_number_density_ratio(self, M, gamma): + def calculate_number_density_ratio(self, M: float, + gamma: float) -> float: n_ratio = ((gamma + 1) / (2 + (gamma - 1)* M**2)) ** (1 / (gamma - 1)) return n_ratio - def plot_temp_vs_radius(self, M1, M2, gamma, T_star, r_star): + def plot_temp_vs_radius(self, M1: float, M2: float, gamma: float, + T_star: float, r_star: float) -> None: temps = [] radii = [] @@ -53,7 +59,8 @@ def plot_temp_vs_radius(self, M1, M2, gamma, T_star, r_star): plt.grid(True) plt.show() - def plot_temp_ratios_vs_radius(self, M1, M2, gamma, r_star): + def plot_temp_ratios_vs_radius(self, M1: float, M2: float, gamma: float, + r_star: float) -> None: temp_ratios = [] radius_ratios = [] @@ -68,7 +75,13 @@ def plot_temp_ratios_vs_radius(self, M1, M2, gamma, r_star): M += 0.5 - plt.plot(radius_ratios * r_star, temp_ratios, marker='o') + # radius_ratios is a Python list, so `list * float` raises TypeError: + # this method (and its sibling below) cannot run as written. Both are + # only referenced from commented-out lines in + # tests/plume/plume_verification_test_01.py. Flagged, not fixed -- + # scaling the ratios is a numerical change, out of scope for #103. + plt.plot(radius_ratios * r_star, # type: ignore[operator] + temp_ratios, marker='o') plt.yscale('log') plt.title("Temperature vs Radius") plt.xlabel("Radial distance, r (m)") @@ -76,7 +89,9 @@ def plot_temp_ratios_vs_radius(self, M1, M2, gamma, r_star): plt.grid(True) plt.show() - def plot_number_density_ratios_vs_radius(self, M1, M2, gamma, r_star): + def plot_number_density_ratios_vs_radius(self, M1: float, M2: float, + gamma: float, + r_star: float) -> None: n_ratios = [] radius_ratios = [] @@ -91,7 +106,8 @@ def plot_number_density_ratios_vs_radius(self, M1, M2, gamma, r_star): M += 0.5 - plt.plot(radius_ratios * r_star, n_ratios, marker='o') + # Same pre-existing `list * float` TypeError as above. + plt.plot(radius_ratios * r_star, n_ratios, marker='o') # type: ignore[operator] plt.yscale('log') plt.title("Number Density vs Radius") plt.xlabel("Radial distance, r (m)") diff --git a/pyrpod/plume/PlumeStrikeCalculator.py b/pyrpod/plume/PlumeStrikeCalculator.py index 5ecbef2..fb06572 100644 --- a/pyrpod/plume/PlumeStrikeCalculator.py +++ b/pyrpod/plume/PlumeStrikeCalculator.py @@ -35,11 +35,12 @@ import os import time from concurrent.futures import ProcessPoolExecutor -from typing import Any, Dict, List, Optional, Sequence, Tuple +from typing import Any, Dict, List, Literal, Optional, Sequence, Tuple, overload import numpy as np from pyrpod.plume.RarefiedPlumeGasKinetics import ( AVOGADROS_NUMBER, + Scalar, SimplifiedGasKinetics, get_maxwellian_heat_transfer, get_maxwellian_pressure, @@ -47,8 +48,9 @@ ) -def _surface_loads_with_incidence(simple_plume: SimplifiedGasKinetics, - incidence: float): +def _surface_loads_with_incidence( + simple_plume: SimplifiedGasKinetics, incidence: Scalar +) -> Tuple[float, float, float]: """Maxwellian surface loads for a struck face at the true incidence angle. simple_plume carries the plume-field state at the face's position (its @@ -216,7 +218,11 @@ def _compute_plume_strikes_core( T_w = plume_params['surface_temp'] sigma = plume_params['sigma'] t_type = thruster_data[thruster_id]['type'][0] - metrics = thruster_metrics[t_type] + # thruster_metrics is Optional because compute_plume_strikes passes + # getattr(vv, 'thruster_metrics', None); it is only ever read on + # this branch, which use_kinetics already gates, but that pairing + # is a caller invariant the signature cannot express. + metrics = thruster_metrics[t_type] # type: ignore[index] # True incidence angle between the local (radial) flow direction # -unit_distance and the face unit normal; fed to the wall # formulas in place of the positional theta (see module header). @@ -408,7 +414,9 @@ def _parallel_worker_init( _WORKER_STATE['plume_params'] = plume_params -def _parallel_worker_compute(task) -> Any: +def _parallel_worker_compute( + task: Tuple[int, Dict[str, Any]], +) -> Tuple[int, Dict[str, np.ndarray], Dict[str, Any]]: """Compute strikes for one firing inside a worker process. task is (firing_index, jfh_step); returns @@ -436,6 +444,36 @@ def _parallel_worker_compute(task) -> Any: return firing_index, result, meta +# The return shape is selected entirely by return_meta, so overloads let +# callers keep the precise type instead of unpacking a union at every use. +@overload +def run_parallel_plume_strikes( + jfh_steps: Sequence[Dict[str, Any]], + face_centroids: np.ndarray, + target_unit_normals: np.ndarray, + thruster_data: Dict[str, Any], + thruster_metrics: Optional[Dict[str, Any]], + plume_params: Dict[str, Any], + workers: int, + return_meta: Literal[False] = ..., +) -> List[Dict[str, np.ndarray]]: + ... + + +@overload +def run_parallel_plume_strikes( + jfh_steps: Sequence[Dict[str, Any]], + face_centroids: np.ndarray, + target_unit_normals: np.ndarray, + thruster_data: Dict[str, Any], + thruster_metrics: Optional[Dict[str, Any]], + plume_params: Dict[str, Any], + workers: int, + return_meta: Literal[True], +) -> Tuple[List[Dict[str, np.ndarray]], List[Dict[str, Any]]]: + ... + + def run_parallel_plume_strikes( jfh_steps: Sequence[Dict[str, Any]], face_centroids: np.ndarray, @@ -445,6 +483,9 @@ def run_parallel_plume_strikes( plume_params: Dict[str, Any], workers: int, return_meta: bool = False, +) -> ( + List[Dict[str, np.ndarray]] + | Tuple[List[Dict[str, np.ndarray]], List[Dict[str, Any]]] ): """Compute per-firing strike results across processes, one firing per task. @@ -484,9 +525,12 @@ def run_parallel_plume_strikes( firing_index, result, meta = future.result() results[firing_index] = result metas[firing_index] = meta + # results/metas are pre-allocated with None so they can be filled in + # firing order, but every slot is assigned from a future above before + # returning, so no None survives into the public return type. if return_meta: - return results, metas - return results + return results, metas # type: ignore[return-value] + return results # type: ignore[return-value] def accumulate_cumulative( diff --git a/pyrpod/plume/RarefiedPlumeGasKinetics.py b/pyrpod/plume/RarefiedPlumeGasKinetics.py index 47be556..c8f9ae5 100644 --- a/pyrpod/plume/RarefiedPlumeGasKinetics.py +++ b/pyrpod/plume/RarefiedPlumeGasKinetics.py @@ -34,18 +34,51 @@ ' = simplified analytical results """ +from __future__ import annotations + import warnings +from collections.abc import Mapping +from typing import Any, TypeAlias, overload import numpy as np +from numpy.typing import NDArray from scipy import integrate from scipy.special import erf +# The closed-form field expressions below are written entirely with NumPy +# ufuncs, so each one accepts a scalar or an array of any shape and returns the +# matching kind; their docstrings already document this as "float or ndarray", +# and the quadrature in CollisionlessGasKinetics._compute_field_integrals (and +# tests/plume/plume_figure_utils.py) evaluates them on meshgrids. +FloatOrArray: TypeAlias = "float | NDArray[np.float64]" + +# Field-point coordinates arrive from NumPy geometry in the impingement +# pipeline (np.linalg.norm for the distance, 3.14 - np.arccos(...) for theta), +# which produces np.floating rather than a plain float. +Scalar: TypeAlias = "float | np.floating[Any]" + +# The K/M/N special factors are overloaded on Q rather than just returning the +# union: SimplifiedGasKinetics feeds them a scalar Q' and stores the result in +# K_simple/M_simple/N_simple, which the Eq. 14-17 field ratios then divide as +# scalars, so collapsing scalar-in to scalar-out is what keeps those return +# types honest. + #define constants AVOGADROS_NUMBER = 6.0221e23 GAS_CONSTANT = 8.314 -def get_K_factor(Q, S_0): +@overload +def get_K_factor(Q: float, S_0: float) -> float: + ... + + +@overload +def get_K_factor(Q: NDArray[np.float64], S_0: float) -> NDArray[np.float64]: + ... + + +def get_K_factor(Q: FloatOrArray, S_0: float) -> FloatOrArray: ''' Scaled special factor exp(-S_0^2) * K [Cai & Wang 2012, Eq. 10]. @@ -74,7 +107,17 @@ def get_K_factor(Q, S_0): return Q * (term1 + term2 * erf_term) -def get_M_factor(Q, S_0): +@overload +def get_M_factor(Q: float, S_0: float) -> float: + ... + + +@overload +def get_M_factor(Q: NDArray[np.float64], S_0: float) -> NDArray[np.float64]: + ... + + +def get_M_factor(Q: FloatOrArray, S_0: float) -> FloatOrArray: ''' Scaled special factor exp(-S_0^2) * M [Cai & Wang 2012, Eq. 11]. @@ -98,7 +141,17 @@ def get_M_factor(Q, S_0): return Q ** 2 * (term1 + term2 * erf_term) -def get_N_factor(Q, S_0): +@overload +def get_N_factor(Q: float, S_0: float) -> float: + ... + + +@overload +def get_N_factor(Q: NDArray[np.float64], S_0: float) -> NDArray[np.float64]: + ... + + +def get_N_factor(Q: FloatOrArray, S_0: float) -> FloatOrArray: ''' Scaled special factor exp(-S_0^2) * N [Cai & Wang 2012, Eq. 12]. @@ -123,7 +176,19 @@ def get_N_factor(Q, S_0): return term1 + term2 * term3 * erf_term -def get_Q_full(r, epsilon, X, Z): +@overload +def get_Q_full(r: float, epsilon: float, X: float, Z: float) -> float: + ... + + +@overload +def get_Q_full(r: FloatOrArray, epsilon: FloatOrArray, X: FloatOrArray, + Z: FloatOrArray) -> FloatOrArray: + ... + + +def get_Q_full(r: FloatOrArray, epsilon: FloatOrArray, X: FloatOrArray, + Z: FloatOrArray) -> FloatOrArray: ''' Full special factor Q [Cai & Wang 2012, Eq. 9] in closed form. @@ -162,7 +227,7 @@ def get_Q_full(r, epsilon, X, Z): return X ** 2 / (X ** 2 + Z ** 2 - 2 * Z * r * np.sin(epsilon) + r ** 2) -def get_far_field_velocity_normalized(S_0): +def get_far_field_velocity_normalized(S_0: float) -> float: ''' Far-field centerline velocity asymptote, lim U_1 * sqrt(beta_0) as X -> infinity [Cai & Wang 2012, Eqs. 22 and 24]. A function of @@ -190,7 +255,7 @@ def get_far_field_velocity_normalized(S_0): return S_0 + (inv_E + sqpi * S_0) / (S_0 * inv_E + (0.5 + S_0 ** 2) * sqpi) -def get_far_field_temp_ratio(S_0): +def get_far_field_temp_ratio(S_0: float) -> float: ''' Far-field centerline temperature asymptote, lim T_1/T_0 as X -> infinity [Cai & Wang 2012, Eqs. 23-24]. A function of the @@ -221,7 +286,9 @@ def get_far_field_temp_ratio(S_0): return -(2 / 3) * G ** 2 + num / den -def get_maxwellian_pressure(rho_inf, U, S, sigma, theta, T, T_w): +def get_maxwellian_pressure(rho_inf: float, U: float, S: float, + sigma: float, theta: Scalar, T: float, + T_w: float) -> float: ''' Rarefied Gas Dynamics - Shen - eq. 4.19 Gas-surface interaction model for pressure based on Maxwell's model. @@ -262,7 +329,8 @@ def get_maxwellian_pressure(rho_inf, U, S, sigma, theta, T, T_w): p *= (rho_inf * U ** 2) / (2 * S ** 2) return p -def get_maxwellian_shear_pressure(rho_inf, U, S, sigma, theta): +def get_maxwellian_shear_pressure(rho_inf: float, U: float, S: float, + sigma: float, theta: Scalar) -> float: ''' Rarefied Gas Dynamics - Shen - eq. 4.20 Gas-surface interaction model for shear pressure based on Maxwell's model. @@ -294,7 +362,9 @@ def get_maxwellian_shear_pressure(rho_inf, U, S, sigma, theta): tau *= -(sigma * rho_inf * np.sin(theta) * U ** 2) / (2 * np.sqrt(np.pi) * S) return tau -def get_maxwellian_heat_transfer(rho_inf, S, sigma, theta, T, T_r, R, gamma): +def get_maxwellian_heat_transfer(rho_inf: float, S: float, sigma: float, + theta: Scalar, T: float, T_r: float, + R: float, gamma: float) -> float: ''' Rarefied Gas Dynamics - Shen - eq. 4.45' Gas-surface interaction model for heat transfer based on Maxwell's model. @@ -384,7 +454,8 @@ class Simons: Number density from continuity equation with constant mass flux across different spherical surfaces. Return the ratio of number density at an analyzed point outside of the exit vs at the exit. ''' - def __init__(self, gamma, R, T_c, P_c, R_0, r, kappa=None): + def __init__(self, gamma: float, R: float, T_c: float, P_c: float, + R_0: float, r: float, kappa: float | None = None) -> None: ''' Simple constructor, saves parameters to self. @@ -428,7 +499,7 @@ def __init__(self, gamma, R, T_c, P_c, R_0, r, kappa=None): self.set_normalization_constant() - def get_nozzle_throat_density(self): + def get_nozzle_throat_density(self) -> float: ''' Returns density at the throat, from chamber pressure/temp, and specific gas constant. Assumes isentropic flow from chamber to throat. @@ -450,7 +521,7 @@ def get_nozzle_throat_density(self): rho_throat = P_throat / (self.R * T_throat) return rho_throat - def get_limiting_turn_angle(self): + def get_limiting_turn_angle(self) -> float: ''' From Lumpkin 1999. Solve for limiting turn angle from specific heat ratio. @@ -465,7 +536,7 @@ def get_limiting_turn_angle(self): theta_max = (np.pi / 2) * (np.sqrt((self.gamma + 1) / (self.gamma - 1)) - 1) return theta_max - def get_plume_angular_density_decay_function(self, theta): + def get_plume_angular_density_decay_function(self, theta: float) -> float: ''' Solve for the density decay function at a given off-centerline angle [Cai & Wang 2012, Eq. 26], f(theta) = @@ -493,7 +564,7 @@ def get_plume_angular_density_decay_function(self, theta): f = (np.cos((np.pi / 2) * (theta / theta_max))) ** self.kappa return f - def set_normalization_constant(self): + def set_normalization_constant(self) -> None: ''' From Lumpkin 1999. Setter for normalization constant. Numerically integrates sin(theta) * cos^kappa(pi*theta/(2*theta_max)) @@ -512,14 +583,14 @@ def set_normalization_constant(self): theta_max = self.get_limiting_turn_angle() kappa = self.kappa - def integrand(theta): + def integrand(theta: float) -> float: return np.sin(theta) * np.cos((np.pi / 2) * (theta / theta_max)) ** kappa integral, _ = integrate.quad(integrand, 0, theta_max) self.A = 0.5 * np.sqrt((self.gamma - 1) / (self.gamma + 1)) / integral return - def get_sonic_velocity(self): + def get_sonic_velocity(self) -> float: ''' Method that returns the sonic velocity of a flow given specific heat ratio, specific gas constant, and chamber temperature. @@ -537,7 +608,7 @@ def get_sonic_velocity(self): sonic_velocity = np.sqrt(self.gamma * self.R * T_throat) return sonic_velocity - def get_limiting_velocity(self): + def get_limiting_velocity(self) -> float: ''' Calculates the limiting velocity of the plume. This is based on the specific heat ratio and the sonic velocity. @@ -565,7 +636,7 @@ def get_static_pressure(self, rho_ratio): return P_static ''' - def get_num_density_ratio(self, theta): + def get_num_density_ratio(self, theta: float) -> float: ''' Number density from continuity equation with constant mass flux across different spherical surfaces [Cai & Wang 2012, Eq. 25]. @@ -597,7 +668,8 @@ def get_num_density_ratio(self, theta): n_ratio = rho_ratio return n_ratio - def get_num_density_ratio_exit(self, theta, exit_mach): + def get_num_density_ratio_exit(self, theta: float, + exit_mach: float) -> float: ''' Exit-referenced number density ratio n/n_0. @@ -726,7 +798,9 @@ class SimplifiedGasKinetics: get_heat_flux() ''' #maybe group the special factors into one method and return an array of them? - def __init__(self, distance, theta, thruster_characteristics, T_w, sigma): + def __init__(self, distance: Scalar, theta: Scalar, + thruster_characteristics: Mapping[str, Any], T_w: float, + sigma: float) -> None: ''' Simple constructor. Can be reworked to save constants for a plume. ''' @@ -752,7 +826,9 @@ def __init__(self, distance, theta, thruster_characteristics, T_w, sigma): return - def set_thruster_characteristics(self, thruster_characteristics): + def set_thruster_characteristics( + self, thruster_characteristics: Mapping[str, Any] + ) -> None: ''' Setter for thruster-specific characteristics. @@ -780,7 +856,7 @@ def set_thruster_characteristics(self, thruster_characteristics): return - def get_beta(self, T): + def get_beta(self, T: float) -> float: ''' Solves for beta at a given temperature. @@ -797,7 +873,7 @@ def get_beta(self, T): beta = 1 / np.sqrt(2 * self.R * T) return beta - def get_speed_ratio(self, U, beta): + def get_speed_ratio(self, U: float, beta: float) -> float: ''' Method to solve for the speed ratio of the flow at a specified flow. @@ -817,7 +893,7 @@ def get_speed_ratio(self, U, beta): S = U * beta return S - def set_Q_simple(self): + def set_Q_simple(self) -> None: ''' Setter for Q in its simplified form. Returns Q'. @@ -834,7 +910,7 @@ def set_Q_simple(self): return - def set_K_simple(self): + def set_K_simple(self) -> None: ''' Setter for simplified special factor K [Cai & Wang 2012, Eq. 10] with Q substituted by Q'. Stored scaled by exp(-S_0^2) for @@ -854,7 +930,7 @@ def set_K_simple(self): return - def set_M_simple(self): + def set_M_simple(self) -> None: ''' Setter for simplified special factor M [Cai & Wang 2012, Eq. 11] with Q substituted by Q'. Stored scaled by exp(-S_0^2) for @@ -872,7 +948,7 @@ def set_M_simple(self): return - def set_N_simple(self): + def set_N_simple(self) -> None: ''' Setter for simplified special factor N [Cai & Wang 2012, Eq. 12] with Q substituted by Q'. Stored scaled by exp(-S_0^2) for @@ -890,7 +966,7 @@ def set_N_simple(self): return - def get_num_density_ratio(self): + def get_num_density_ratio(self) -> float: ''' Method to calculate the number denisty at a point (X, 0, Z) outside of the nozzle. This density is normalized over the number density at the nozzle exit. @@ -909,7 +985,7 @@ def get_num_density_ratio(self): num_density_ratio = (self.K_simple / (2 * np.sqrt(np.pi)) * (self.R_0 / self.X) ** 2) return num_density_ratio - def get_U_normalized(self): + def get_U_normalized(self) -> float: ''' Method to calculate the macroscopic x-component of velocity at a point (X, 0, Z) outside of the nozzle. This velocity component is normalized with the parameter beta at the exit. @@ -928,7 +1004,7 @@ def get_U_normalized(self): U_normalized = self.M_simple / self.K_simple return U_normalized - def get_W_normalized(self): + def get_W_normalized(self) -> float: ''' Method to calculate the macroscopic z-component of velocity at a point (X, 0, Z) outside of the nozzle. This velocity component is normalized with the parameter beta at the exit. @@ -947,7 +1023,7 @@ def get_W_normalized(self): W_normalized = (self.M_simple / self.K_simple) * (self.Z / self.X) return W_normalized - def get_temp_ratio(self): + def get_temp_ratio(self) -> float: ''' Method to calculate the temperature at a point (X, 0, Z) outside of the nozzle. This temperature is normalized over the temperature at the nozzle exit. @@ -966,7 +1042,7 @@ def get_temp_ratio(self): T_ratio += (4 * self.N_simple / (3 * self.K_simple)) return T_ratio - def get_num_density_centerline(self): + def get_num_density_centerline(self) -> float: ''' Method to calculate the number denisty at a point (X, 0, 0) outside of the nozzle. This density is normalized over the number density at the nozzle exit. @@ -988,7 +1064,7 @@ def get_num_density_centerline(self): return n_ratio - def get_velocity_centerline(self): + def get_velocity_centerline(self) -> float: ''' Method to calculate the macroscopic velocity at a point (X, 0, 0) outside of the nozzle. This velocity is normalized with the parameter beta at the exit. @@ -1013,7 +1089,7 @@ def get_velocity_centerline(self): U_ratio = 1 / (2 * n_ratio) * ((p2 ** 2 * np.exp(- self.S_0 ** 2) / np.sqrt(np.pi)) + (self.S_0 * (1 + erf(self.S_0))) - (np.exp(- p2 ** 2 * self.S_0 ** 2) * p1 ** 3 * self.S_0 * (1 + erf(p1 * self.S_0)))) return U_ratio - def get_temp_centerline(self): + def get_temp_centerline(self) -> float: ''' Method to calculate the temperature at a point (X, 0, 0) outside of the nozzle. This temperature is normalized over the temperature at the nozzle exit. @@ -1040,7 +1116,7 @@ def get_temp_centerline(self): n_ratio = self.get_num_density_centerline() U1 = self.get_velocity_centerline() - def integrand(r): + def integrand(r: float) -> float: Q = get_Q_full(r, 0.0, self.X, 0.0) return get_N_factor(Q, self.S_0) * r @@ -1048,7 +1124,7 @@ def integrand(r): temp_ratio = 4 / (3 * n_ratio * np.sqrt(np.pi) * self.X ** 2) * integral - (U1 ** 2 / (3/2)) return temp_ratio - def get_pressure(self): + def get_pressure(self) -> float: ''' Method to call gas-surface interaction model. Passes thruster characteristics and normalized plume parameters to the Maxwell model to solve for pressre. @@ -1088,7 +1164,7 @@ def get_pressure(self): pressure = get_maxwellian_pressure(rho_inf, U, S, self.sigma, self.theta, T, self.T_w) return pressure - def get_shear_pressure(self): + def get_shear_pressure(self) -> float: ''' Method to call gas-surface interaction model. Passes thruster characteristics and normalized plume parameters to the Maxwell model to solve for shear pressre. @@ -1128,7 +1204,7 @@ def get_shear_pressure(self): shear_pressure = get_maxwellian_shear_pressure(rho_inf, U, S, self.sigma, self.theta) return shear_pressure - def get_heat_flux(self): + def get_heat_flux(self) -> float: ''' Method to call gas-surface interaction model. Passes thruster characteristics and normalized plume parameters to the Maxwell model to solve for heat flux. @@ -1238,7 +1314,9 @@ class CollisionlessGasKinetics(SimplifiedGasKinetics): QUAD_ORDERS = (40, 80, 160) QUAD_RTOL = 1e-9 - def __init__(self, distance, theta, thruster_characteristics, T_w, sigma): + def __init__(self, distance: Scalar, theta: Scalar, + thruster_characteristics: Mapping[str, Any], T_w: float, + sigma: float) -> None: ''' Mirrors SimplifiedGasKinetics(distance, theta, thruster_characteristics, T_w, sigma) exactly; X = d*cos(theta) @@ -1250,7 +1328,9 @@ def __init__(self, distance, theta, thruster_characteristics, T_w, sigma): return - def _compute_field_integrals(self, order): + def _compute_field_integrals( + self, order: int + ) -> tuple[float, float, float, float]: ''' Evaluate the four exit-disk integrals of Eqs. 5-8 with a tensor-product Gauss-Legendre rule of the given order per @@ -1291,7 +1371,7 @@ def _compute_field_integrals(self, order): I_N = np.sum(W2D * R * N) return I_K, I_M, I_W, I_N - def set_field_integrals(self): + def set_field_integrals(self) -> None: ''' Setter for the field integrals I_K, I_M, I_W, I_N of Eqs. 5-8, with quadrature-order doubling until two @@ -1326,7 +1406,11 @@ def set_field_integrals(self): return - def _integrals_converged(self, previous, current): + def _integrals_converged( + self, + previous: tuple[float, float, float, float], + current: tuple[float, float, float, float], + ) -> bool: ''' Convergence test between two quadrature orders. I_K, I_M and I_N are strictly positive, so a relative test applies; I_W @@ -1355,7 +1439,7 @@ def _integrals_converged(self, previous, current): and abs(I_N1 - I_N0) <= rtol * abs(I_N1) and abs(I_W1 - I_W0) <= rtol * scale_W) - def get_num_density_ratio(self): + def get_num_density_ratio(self) -> float: ''' Number density at (X, 0, Z) normalized by the exit number density, n_1/n_0 [Cai & Wang 2012, Eq. 5]. The exp(-S_0^2) @@ -1372,7 +1456,7 @@ def get_num_density_ratio(self): ''' return self.I_K / (np.pi ** 1.5 * self.X ** 2) - def get_U_normalized(self): + def get_U_normalized(self) -> float: ''' Macroscopic x-velocity at (X, 0, Z) normalized by sqrt(beta_0), i.e. U_1 * sqrt(beta_0) [Cai & Wang 2012, @@ -1390,7 +1474,7 @@ def get_U_normalized(self): ''' return self.I_M / self.I_K - def get_W_normalized(self): + def get_W_normalized(self) -> float: ''' Macroscopic z-velocity at (X, 0, Z) normalized by sqrt(beta_0), i.e. W_1 * sqrt(beta_0) [Cai & Wang 2012, @@ -1412,7 +1496,7 @@ def get_W_normalized(self): ''' return self.I_W / (self.X * self.I_K) - def get_Vr_normalized(self): + def get_Vr_normalized(self) -> float: ''' Radial (spherical, from the nozzle exit center) velocity component in the Y = 0 plane, normalized by sqrt(beta_0): @@ -1432,7 +1516,7 @@ def get_Vr_normalized(self): W = self.get_W_normalized() return (self.X * U + self.Z * W) / np.sqrt(self.X ** 2 + self.Z ** 2) - def get_temp_ratio(self): + def get_temp_ratio(self) -> float: ''' Temperature at (X, 0, Z) normalized by the exit temperature, T_1/T_0 [Cai & Wang 2012, Eq. 8]. With beta_0 = 1/(2*R*T_0), @@ -1452,7 +1536,7 @@ def get_temp_ratio(self): W = self.get_W_normalized() return -(2 / 3) * (U ** 2 + W ** 2) + (4 / 3) * self.I_N / self.I_K - def get_pressure_ratio(self): + def get_pressure_ratio(self) -> float: ''' Flowfield static pressure at (X, 0, Z) normalized by the exit static pressure: p_1/p_0 = (n_1/n_0) * (T_1/T_0) from the diff --git a/pyrpod/rpod/JetFiringHistory.py b/pyrpod/rpod/JetFiringHistory.py index b369737..f9f85e0 100644 --- a/pyrpod/rpod/JetFiringHistory.py +++ b/pyrpod/rpod/JetFiringHistory.py @@ -1,6 +1,11 @@ +from __future__ import annotations + import configparser import logging import time +from collections.abc import Sequence +from typing import Any + import numpy as np import sympy as sp @@ -11,7 +16,7 @@ logger = logging.getLogger(__name__) -def make_norm(vector_value_function): +def make_norm(vector_value_function: Sequence[Any]) -> Any: """Calculate vector norm/magnitude using the Pythagoream Theorem.""" return sp.sqrt(sp.Pow(vector_value_function[0],2) + sp.Pow(vector_value_function[1],2)) @@ -47,7 +52,20 @@ class JetFiringHistory: STL file rotations. Data is then saved to a text file. """ - def __init__(self, case_dir): + # Declared bare (no class attribute is created) and annotated to the + # class contract: a parsed firing list. read_jfh writes None as a + # "nothing loaded" sentinel on two failure paths, but only + # FuelManager.calc_total_delta_mass ever checks for it -- every other + # consumer, including PlumeStrikeEstimationStudy throughout, indexes + # JFH unguarded. Widening to list|None would relocate one report onto + # roughly forty unguarded call sites without fixing any of them. + # Element type is dict[str, Any] rather than a TypedDict because the + # records exist in two shapes: read_jfh parses the time fields as + # strings with nested-list dcm/xyz, while tests and edit_1d_JFH + # synthesize steps with NumPy arrays under the same keys. + JFH: list[dict[str, Any]] + + def __init__(self, case_dir: str) -> None: """ Constructor simply sets case directory and parses the appopriate configuration file. @@ -68,7 +86,7 @@ def __init__(self, case_dir): config.read(self.case_dir + "config.ini") self.config = config - def read_jfh(self): + def read_jfh(self) -> None: """ Method responsible for reading and parsing through JFH data. @@ -92,7 +110,7 @@ def read_jfh(self): except KeyError: logger.debug("No [jfh] jfh configured for case %s; JFH not loaded.", self.case_dir) - self.JFH = None + self.JFH = None # type: ignore[assignment] # sentinel return logger.debug("Resolving JFH asset %r for case %s", jfh_name, self.case_dir) path_to_jfh = resolve_asset_path(self.case_dir, 'jfh', jfh_name) @@ -106,7 +124,7 @@ def read_jfh(self): except IndexError: logger.error("Supplied JFH file is empty (no header/firing " "count): %s", path_to_jfh) - self.JFH = None + self.JFH = None # type: ignore[assignment] # sentinel return # Throw away second line @@ -130,7 +148,7 @@ def read_jfh(self): curr_row[-1] = curr_row[-1].split('\n')[0] # print(curr_row) # Save all information in current row to a dictionary. - time_step = {} + time_step: dict[str, Any] = {} # Save time data. time_step['nt'] = curr_row.pop(0) @@ -193,7 +211,7 @@ def read_jfh(self): return - def graph_param_curve(self, t, r_of_t): + def graph_param_curve(self, t: Any, r_of_t: Sequence[Any]) -> None: ''' Used to quickly prototype and visualize a proposed approach path. Calculates the unit tangent vector at a given timestep and rotates the STL file accordingly. Data is plotted using matlab @@ -306,7 +324,9 @@ def graph_param_curve(self, t, r_of_t): plt.close() - def print_JFH_param_curve(self, jfh_path, t, r_of_t, align = False): + def print_JFH_param_curve(self, jfh_path: str, t: Any, + r_of_t: Sequence[Any], + align: bool = False) -> None: ''' Used to produce JFH data for a proposed approach path. Calculates the unit tangent vector at a given timestep and DCMs for STL file rotations. Data is then saved to a text file. diff --git a/pyrpod/rpod/PlumeStrikeEstimationStudy.py b/pyrpod/rpod/PlumeStrikeEstimationStudy.py index afe11ba..edb7998 100644 --- a/pyrpod/rpod/PlumeStrikeEstimationStudy.py +++ b/pyrpod/rpod/PlumeStrikeEstimationStudy.py @@ -1,4 +1,10 @@ +from __future__ import annotations + +from collections.abc import Callable, Sequence +from typing import Any + import numpy as np +from numpy.typing import NDArray import os import math import logging @@ -41,10 +47,19 @@ run_parallel_plume_strikes, ) +# Aliased because study_init's parameter names shadow the class names. +from pyrpod.rpod.JetFiringHistory import ( + JetFiringHistory as JetFiringHistoryType, +) +from pyrpod.vehicle.TargetVehicle import ( + TargetVehicle as TargetVehicleType, +) + logger = logging.getLogger(__name__) -def _log_jfh_generation_complete(jfh_path, n_firings, gen_start): +def _log_jfh_generation_complete(jfh_path: str, n_firings: int, + gen_start: float) -> None: """Log completion of a JFH-generation step (path, size, count, runtime). Never raises: JFH generation is a scientific write, so an observability @@ -126,7 +141,24 @@ class PlumeStrikeEstimationStudy (MissionPlanner): """ # def __init__(self): # print("Initialized Approach Visualizer") - def study_init(self, JetFiringHistory, Target, Vehicle): + # Collaborator state and thruster-group / fuel helpers that this + # class calls on self but neither it nor MissionPlanner defines. + # Declared bare, so no class attribute is created at runtime. + jfh: JetFiringHistoryType + target: TargetVehicleType + vv: LogisticsModule + viz: PlumeStudyExport + case_key: str + calc_m_dot_sum: Callable[[str], float] + calc_v_e: Callable[[str], float] + calc_delta_v: Callable[[float, float, float, float], float] + calc_delta_mass_v_e: Callable[[float, float, bool], float] + # Sweep index set externally by mdao.TradeStudy before visualize_sweep. + count: int + + def study_init(self, JetFiringHistory: JetFiringHistoryType, + Target: TargetVehicleType, + Vehicle: LogisticsModule) -> None: """ Designates assets for RPOD analysis. @@ -152,7 +184,7 @@ def study_init(self, JetFiringHistory, Target, Vehicle): # visualization/export helper self.viz = PlumeStudyExport(self.environment) - def graph_init_config(self): + def graph_init_config(self) -> None: """ Creates visualization data for initiial configuration of RPOD analysis. @@ -196,7 +228,7 @@ def graph_init_config(self): plt.show() - def graph_jfh_thruster_check(self): + def graph_jfh_thruster_check(self) -> None: """ Creates visualization data for initiial configuration of RPOD analysis. @@ -285,11 +317,16 @@ def graph_jfh_thruster_check(self): elif firing < 100: index = '0' + str(firing) else: - index = str(i) + # Pre-existing bug: `i` is not defined in this scope, so a + # 100th-or-later firing raises NameError. Flagged, not + # fixed (also reported by flake8 as F821). + index = str(i) # type: ignore[name-defined] # delegate figure saving to export helper self.viz.save_figure(figure, os.path.join(self.environment.case_dir, 'img', 'frame' + str(index) + '.png')) - def graph_clusters(self, firing, vv_orientation): + def graph_clusters( + self, firing: int, vv_orientation: NDArray[np.float64] + ) -> mesh.Mesh | None: """ Creates visualization data for the cluster. Parameters @@ -343,7 +380,7 @@ def graph_clusters(self, firing, vv_orientation): return None return compose_meshes(cluster_meshes) - def graph_jfh(self, trade_study = False): + def graph_jfh(self, trade_study: bool = False) -> None: """ Creates visualization data for the trajectory of the proposed RPOD analysis. @@ -438,7 +475,9 @@ def graph_jfh(self, trade_study = False): if active_cones is not None: components.append(active_cones) if self.vv.use_clusters == True: - components.append(active_clusters) + # graph_clusters returns None only when no clusters are + # configured, which use_clusters already excludes. + components.append(active_clusters) # type: ignore[arg-type] VVmesh = compose_meshes(components) if trade_study == False: @@ -466,7 +505,7 @@ def graph_jfh(self, trade_study = False): "failures=%d directory=%s", viz_files_written, viz_failures, jfh_export_dir) - def visualize_sweep(self, config_iter): + def visualize_sweep(self, config_iter: int) -> None: """ Creates visualization data for the trajectory of the proposed RPOD analysis. @@ -554,7 +593,8 @@ def visualize_sweep(self, config_iter): if active_cones is not None: components.append(active_cones) if self.vv.use_clusters == True: - components.append(active_clusters) + # See graph_jfh: None only without configured clusters. + components.append(active_clusters) # type: ignore[arg-type] VVmesh = compose_meshes(components) if self.count > 0: @@ -630,7 +670,7 @@ def visualize_sweep(self, config_iter): # return param_queue, param_window_sum # Helper functions for jfh_plume_strikes - def create_results_dir(self): + def create_results_dir(self) -> None: """ Creates a results directory and sub-directories if they don't already exist. """ @@ -638,7 +678,13 @@ def create_results_dir(self): for sub_dir in sub_dirs: ensure_dir(os.path.join(self.environment.case_dir, sub_dir)) - def set_strike_fields(self, target): + def set_strike_fields( + self, target: mesh.Mesh + ) -> ( + tuple[NDArray[np.float64], NDArray[np.float64], + NDArray[np.float64], NDArray[np.float64]] + | NDArray[np.float64] + ): # Initiate array containing cummulative strikes. cum_strikes = np.zeros(len(target.vectors)) @@ -658,7 +704,9 @@ def set_strike_fields(self, target): return cum_strikes - def extract_firing_data(self, firing): + def extract_firing_data( + self, firing: int + ) -> tuple[list[int], Sequence[float], NDArray[np.float64]]: # Save active thrusters for current firing. thrusters = self.jfh.JFH[firing]['thrusters'] # print("thrusters", thrusters) @@ -670,7 +718,14 @@ def extract_firing_data(self, firing): return thrusters, vv_pos, vv_orientation - def set_plume_strike_fields(self, target): + def set_plume_strike_fields( + self, target: mesh.Mesh + ) -> ( + tuple[NDArray[np.float64], NDArray[np.float64], + NDArray[np.float64], NDArray[np.float64], + NDArray[np.float64]] + | NDArray[np.float64] + ): # reset strikes for each firing strikes = np.zeros(len(target.vectors)) @@ -688,7 +743,11 @@ def set_plume_strike_fields(self, target): else: return strikes - def set_plume_transformations(self, thruster_id, vv_orientation, vv_pos): + def set_plume_transformations( + self, thruster_id: str, vv_orientation: NDArray[np.float64], + vv_pos: Sequence[float] | NDArray[np.float64], + ) -> tuple[NDArray[np.float64], NDArray[np.float64], + NDArray[np.float64]]: # Load data to calculate plume transformations # First, according to DCM and exit vector using current thruster id in TCD @@ -711,7 +770,9 @@ def set_plume_transformations(self, thruster_id, vv_orientation, vv_pos): return plume_normal, thruster_pos, thruster_orientation - def set_face_centroid(self, face): + def set_face_centroid( + self, face: NDArray[np.float64] + ) -> NDArray[np.float64]: # Calculate centroid for face x = np.array(face[0]).mean() @@ -722,7 +783,10 @@ def set_face_centroid(self, face): return centroid - def set_face_distance(self, thruster_pos, centroid): + def set_face_distance( + self, thruster_pos: NDArray[np.float64], + centroid: NDArray[np.float64], + ) -> tuple[NDArray[np.float64], Any, NDArray[np.float64]]: # Calculate distance vector between face centroid and thruster exit. distance = thruster_pos - centroid # print('distance vector', distance) @@ -733,7 +797,9 @@ def set_face_distance(self, thruster_pos, centroid): return distance, norm_distance, unit_distance - def _resolve_parallel_options(self, parallel, workers, n_firings): + def _resolve_parallel_options( + self, parallel: bool | None, workers: int | None, n_firings: int + ) -> tuple[bool, int]: """ Resolves parallel execution settings for jfh_plume_strikes(). @@ -785,7 +851,7 @@ def _resolve_parallel_options(self, parallel, workers, n_firings): return False, 1 return True, workers - def _validate_plume_strike_inputs(self): + def _validate_plume_strike_inputs(self) -> bool: """Fail-fast validation of the inputs a plume-strike run requires. Logs an ERROR naming the case, the offending section/key, and any @@ -802,7 +868,7 @@ def _validate_plume_strike_inputs(self): config = self.environment.config case_dir = self.environment.case_dir - def require_option(section, key): + def require_option(section: str, key: str) -> str: if not config.has_option(section, key): logger.error("Missing required config [%s] %s (case %s)", section, key, case_dir) @@ -873,7 +939,9 @@ def require_option(section, key): f"{case_dir}); call set_thruster_metrics().") return kinetics_on - def jfh_plume_strikes(self, parallel=None, workers=None): + def jfh_plume_strikes( + self, parallel: bool | None = None, workers: int | None = None + ) -> dict[str, dict[str, NDArray[np.float64]]]: """ Calculates number of plume strikes according to data provided for RPOD analysis. Method assumes that study assets are correctly configured. @@ -930,7 +998,7 @@ def jfh_plume_strikes(self, parallel=None, workers=None): else: cum_strikes = self.set_strike_fields(target) - firing_data = {} + firing_data: dict[str, dict[str, NDArray[np.float64]]] = {} n_firings = len(self.jfh.JFH) @@ -1031,10 +1099,13 @@ def jfh_plume_strikes(self, parallel=None, workers=None): } if kinetics_on: - pressures = result.get("pressures") - shear_stresses = result.get("shear_stress") - heat_flux_rate = result.get("heat_flux_rate") - heat_flux_load = result.get("heat_flux_load") + # dict.get() reports these as Optional; the enclosing + # kinetics_on branch is what guarantees the keys exist, an + # invariant the result dict's type cannot carry. + pressures: Any = result.get("pressures") + shear_stresses: Any = result.get("shear_stress") + heat_flux_rate: Any = result.get("heat_flux_rate") + heat_flux_load: Any = result.get("heat_flux_load") max_pressures = np.maximum(max_pressures, pressures) max_shears = np.maximum(max_shears, shear_stresses) @@ -1208,14 +1279,19 @@ def jfh_plume_strikes(self, parallel=None, workers=None): return firing_data - def calc_time_multiplier(self, v_ida, v_o, r_o): + def calc_time_multiplier(self, v_ida: float, v_o: float, + r_o: float) -> float: # Determine thruster configuration characterstics. # The JFH only contains firings done by the neg_x group m_dot_sum = self.calc_m_dot_sum('neg_x') # print('m_dot_sum is', m_dot_sum) - MIB = self.vv.thruster_metrics[self.vv.thruster_data[self.vv.rcs_groups['neg_x'][0]]['type'][0]]['MIB'] + # thruster_metrics is Optional (None when the case has no [tcd] tdf); + # this legacy path has never guarded it. + MIB = self.vv.thruster_metrics[ # type: ignore[index] + self.vv.thruster_data[self.vv.rcs_groups['neg_x'][0]]['type'][0]]['MIB'] # print('MIB is', MIB) - F_thruster = self.vv.thruster_metrics[self.vv.thruster_data[self.vv.rcs_groups['neg_x'][0]]['type'][0]]['F'] + F_thruster = self.vv.thruster_metrics[ # type: ignore[index] + self.vv.thruster_data[self.vv.rcs_groups['neg_x'][0]]['type'][0]]['F'] F = F_thruster * np.cos(self.vv.decel_cant) n_thrusters = len(self.vv.rcs_groups['neg_x']) F = F * n_thrusters @@ -1249,7 +1325,7 @@ def calc_time_multiplier(self, v_ida, v_o, r_o): # Initializing empty tracking lists dx = [0] t = [0] - dv = [0] + dv: list[float] = [0] # Initializing inertial state dxdt = [v_o] @@ -1332,17 +1408,19 @@ def calc_time_multiplier(self, v_ida, v_o, r_o): # print(one_d_results) - def print_jfh_1d_approach_n_fire(self, v_ida, v_o, r_o, n_firings, trade_study = False): + def print_jfh_1d_approach_n_fire(self, v_ida: float, v_o: float, + r_o: float, n_firings: int, + trade_study: bool = False) -> None: # Delegate to approach_maneuvers.compute_1d_approach and rpod.io.write_jfh tm = self.calc_time_multiplier(v_ida, v_o, r_o) inputs = ApproachInputs(v_ida=float(v_ida), v_o=float(v_o), r_o=float(r_o), group='neg_x') # Adapter for grouping methods if not explicitly available as a module class _GroupingAdapter: - def __init__(self, outer): + def __init__(self, outer: 'PlumeStrikeEstimationStudy') -> None: self._outer = outer - def calc_m_dot_sum(self, group): + def calc_m_dot_sum(self, group: str) -> float: return self._outer.calc_m_dot_sum(group) - def calc_v_e(self, group): + def calc_v_e(self, group: str) -> float: return self._outer.calc_v_e(group) results = compute_1d_approach( @@ -1373,7 +1451,8 @@ def calc_v_e(self, group): _log_jfh_generation_complete(jfh_path, len(t_values), gen_start) - def print_jfh_1d_approach(self, v_ida, v_o, r_o, trade_study = False): + def print_jfh_1d_approach(self, v_ida: float, v_o: float, r_o: float, + trade_study: bool = False) -> None: """ Method creates JFH data for axial approach using simpified physics calculations. @@ -1401,11 +1480,11 @@ def print_jfh_1d_approach(self, v_ida, v_o, r_o, trade_study = False): # Delegate to approach_maneuvers with a fixed multiplier similar to legacy inputs = ApproachInputs(v_ida=float(v_ida), v_o=float(v_o), r_o=float(r_o), group='neg_x') class _GroupingAdapter: - def __init__(self, outer): + def __init__(self, outer: 'PlumeStrikeEstimationStudy') -> None: self._outer = outer - def calc_m_dot_sum(self, group): + def calc_m_dot_sum(self, group: str) -> float: return self._outer.calc_m_dot_sum(group) - def calc_v_e(self, group): + def calc_v_e(self, group: str) -> float: return self._outer.calc_v_e(group) results = compute_1d_approach( @@ -1434,7 +1513,9 @@ def calc_v_e(self, group): _log_jfh_generation_complete(jfh_path, len(t_values), gen_start) return - def edit_1d_JFH(self, t_values, r, rot): + def edit_1d_JFH(self, t_values: NDArray[np.float64], + r: Sequence[NDArray[np.float64]], + rot: NDArray[np.float64]) -> None: """ Helper function to RPOD.calc_jfh_1d_approach() that is responsible for modifying the JFH attribute in memory with the values calculated. @@ -1503,7 +1584,8 @@ def edit_1d_JFH(self, t_values, r, rot): # print('self.jfh.JFH is', self.jfh.JFH) # print('len(self.jfh.JFH) is', len(self.jfh.JFH)) - def calc_jfh_1d_approach(self, v_ida, v_o, cant): + def calc_jfh_1d_approach(self, v_ida: float, v_o: float, + cant: float) -> None: """ The x-position represents the distance required to reach a velocity of zero. @@ -1535,11 +1617,11 @@ def calc_jfh_1d_approach(self, v_ida, v_o, cant): inputs = ApproachInputs(v_ida=float(v_ida), v_o=float(v_o), r_o=0.0, group='neg_x') class _GroupingAdapter: - def __init__(self, outer): + def __init__(self, outer: 'PlumeStrikeEstimationStudy') -> None: self._outer = outer - def calc_m_dot_sum(self, group): + def calc_m_dot_sum(self, group: str) -> float: return self._outer.calc_m_dot_sum(group) - def calc_v_e(self, group): + def calc_v_e(self, group: str) -> float: return self._outer.calc_v_e(group) results = compute_1d_approach( @@ -1556,15 +1638,18 @@ def calc_v_e(self, group): rot = results["rot"] self.edit_1d_JFH(t_values, r, rot) - def get_case_key(self): + def get_case_key(self) -> str: return self.case_key - def set_case_key(self, v0_iter, cant_iter): + def set_case_key(self, v0_iter: Any, cant_iter: Any) -> None: self.case_key = 'vo_' + str(v0_iter) + '_cant_' + str(cant_iter) return - def make_test_jfh(): + # Declared without self and never called; invoking it on an instance + # would raise TypeError. Adding self would change the signature, + # which #103 excludes, so the defect is flagged here instead. + def make_test_jfh() -> None: # type: ignore[misc] return \ No newline at end of file diff --git a/pyrpod/rpod/PlumeStudyExport.py b/pyrpod/rpod/PlumeStudyExport.py index 65cc10e..bd5f6c2 100644 --- a/pyrpod/rpod/PlumeStudyExport.py +++ b/pyrpod/rpod/PlumeStudyExport.py @@ -4,8 +4,10 @@ can delegate those responsibilities and remain focused on simulation logic. """ +from __future__ import annotations + import os -from typing import Optional +from typing import Any, Optional import matplotlib.pyplot as plt @@ -17,7 +19,7 @@ class PlumeStudyExport: (matching the existing usage in the codebase). """ - def __init__(self, environment): + def __init__(self, environment: Any) -> None: self.environment = environment def _ensure_parent(self, path: str) -> None: @@ -25,7 +27,7 @@ def _ensure_parent(self, path: str) -> None: if parent: os.makedirs(parent, exist_ok=True) - def export_firing(self, vv_mesh, path_to_stl: str) -> None: + def export_firing(self, vv_mesh: Any, path_to_stl: str) -> None: """Save the provided mesh to disk, ensuring directory exists. Keeps behavior identical to previous calls to ``mesh.save(path)`` diff --git a/pyrpod/rpod/io.py b/pyrpod/rpod/io.py index 7ecb456..dddefa8 100644 --- a/pyrpod/rpod/io.py +++ b/pyrpod/rpod/io.py @@ -11,7 +11,10 @@ from __future__ import annotations import os -from typing import Any, Sequence +from typing import Any, Sequence, Union + +import numpy as np +from numpy.typing import NDArray from pyrpod.util.io.file_print import print_JFH, print_1d_JFH from pyrpod.util.io.fs import ensure_dir @@ -28,7 +31,14 @@ def save_mesh_to_stl(mesh_obj: Any, path: str) -> None: mesh_obj.save(path) -def write_jfh(t_values, r, rot, path: str, mode: str = "1d") -> None: +# Same bound as pyrpod.util.io.file_print: callers pass either Python lists +# of per-axis rows or NumPy arrays of the same shape. +_Times = Union[Sequence[float], NDArray[np.float64]] +_Rows = Union[Sequence[Any], NDArray[Any]] + + +def write_jfh(t_values: _Times, r: _Rows, rot: _Rows, path: str, + mode: str = "1d") -> None: """Write a JFH file. mode="1d" uses print_1d_JFH formatting; mode="generic" uses print_JFH. diff --git a/pyrpod/util/io/file_print.py b/pyrpod/util/io/file_print.py index a419d2f..8a6d29f 100644 --- a/pyrpod/util/io/file_print.py +++ b/pyrpod/util/io/file_print.py @@ -3,9 +3,24 @@ Alternatively these could be moved to JetFiringHistory.py """ +from __future__ import annotations + import sys +from collections.abc import Sequence +from typing import Any + +import numpy as np +from numpy.typing import NDArray + +# Callers pass either Python lists of per-axis rows (JetFiringHistory builds +# ``r = [x, y, z]`` and a list of rotation matrices) or NumPy arrays of the +# same shape, so the accurate bound is "indexable rows" rather than one +# concrete container type. +_Times = Sequence[float] | NDArray[np.float64] +_Rows = Sequence[Any] | NDArray[Any] + -def print_JFH(t_values, r, rot, file_name): +def print_JFH(t_values: _Times, r: _Rows, rot: _Rows, file_name: str) -> None: ''' Helper function to RPOD.print_jfh_param_curve() that is responsible for printing the calculated JFH data to a text file. @@ -84,7 +99,7 @@ def print_JFH(t_values, r, rot, file_name): # print() sys.stdout = orig_stdout -def print_test_JFH(t_values, r, rot, file_name): +def print_test_JFH(t_values: _Times, r: _Rows, rot: _Rows, file_name: str) -> None: ''' Helper function to RPOD.print_jfh_param_curve() that is responsible for @@ -166,7 +181,7 @@ def print_test_JFH(t_values, r, rot, file_name): # print() sys.stdout = orig_stdout -def print_1d_JFH(t_values, r, rot, file_name): +def print_1d_JFH(t_values: _Times, r: _Rows, rot: _Rows, file_name: str) -> None: ''' Helper function to RPOD.print_jfh_param_curve() that is responsible for printing the calculated JFH data to a text file. diff --git a/pyrpod/util/io/fs.py b/pyrpod/util/io/fs.py index 52302c8..ec4dafe 100644 --- a/pyrpod/util/io/fs.py +++ b/pyrpod/util/io/fs.py @@ -8,8 +8,9 @@ log = logging.getLogger(__name__) -def resolve_asset_path(case_dir, subdir, filename, shared_subdir=None, *, - required=True): +def resolve_asset_path(case_dir: str, subdir: str, filename: str, + shared_subdir: str | None = None, *, + required: bool = True) -> str: """ Resolve the path to a shared data asset (STL, JFH, TCD, flight plan, ...). @@ -71,7 +72,7 @@ def resolve_asset_path(case_dir, subdir, filename, shared_subdir=None, *, f"{local_path!r} then shared {shared_path!r}") return local_path -def ensure_dir(path): +def ensure_dir(path: str) -> None: """ Ensure that a directory exists. If it does not exist, create it. @@ -97,7 +98,7 @@ def ensure_dir(path): except Exception: pass -def ensure_parent_dir(file_path): +def ensure_parent_dir(file_path: str) -> None: """ Ensure that the parent directory of a file exists. If it does not exist, create it. diff --git a/pyrpod/util/math/transform.py b/pyrpod/util/math/transform.py index 29a5553..6378e9e 100644 --- a/pyrpod/util/math/transform.py +++ b/pyrpod/util/math/transform.py @@ -1,7 +1,10 @@ import numpy as np +from numpy.typing import ArrayLike, NDArray -def rotation_matrix_from_vectors(vec1: np.ndarray, vec2: np.ndarray) -> np.ndarray: +def rotation_matrix_from_vectors( + vec1: ArrayLike, vec2: ArrayLike +) -> NDArray[np.float64]: """ Compute the 3x3 rotation matrix that rotates vec1 to align with vec2 using the Rodrigues' rotation formula. diff --git a/pyrpod/util/stl/stl.py b/pyrpod/util/stl/stl.py index 30bf09e..aa7a533 100644 --- a/pyrpod/util/stl/stl.py +++ b/pyrpod/util/stl/stl.py @@ -1,7 +1,10 @@ from stl import mesh import os +from collections.abc import Iterable, Mapping, Sequence from pathlib import Path +from typing import Any import numpy as np +from numpy.typing import NDArray from pyevtk.hl import unstructuredGridToVTK from pyevtk.vtk import VtkTriangle from pyrpod.util.io.fs import ensure_parent_dir @@ -10,7 +13,7 @@ import json -def load_stl(file_path): +def load_stl(file_path: str) -> mesh.Mesh: """ Load an STL file and return a mesh object. @@ -28,7 +31,12 @@ def load_stl(file_path): raise FileNotFoundError(f"STL file not found: {file_path}") return mesh.Mesh.from_file(file_path) -def transform_mesh(mesh_obj, rotation_matrix=None, translation_vector=None, scale_factor=None): +def transform_mesh( + mesh_obj: mesh.Mesh, + rotation_matrix: NDArray[np.float64] | None = None, + translation_vector: Sequence[float] | NDArray[np.float64] | None = None, + scale_factor: float | None = None, +) -> mesh.Mesh: """ Apply transformations to a mesh object. @@ -57,7 +65,12 @@ def transform_mesh(mesh_obj, rotation_matrix=None, translation_vector=None, scal return mesh_obj -def transform_mesh_from_file(input_file, output_file, scale, translate): +def transform_mesh_from_file( + input_file: str, + output_file: str, + scale: float, + translate: Sequence[float] | NDArray[np.float64], +) -> None: """ Transform an STL mesh with scaling and translation and save it to a file. @@ -83,7 +96,7 @@ def transform_mesh_from_file(input_file, output_file, scale, translate): print(f"Mesh saved to {output_file}") -def compose_meshes(meshes): +def compose_meshes(meshes: Iterable[mesh.Mesh]) -> mesh.Mesh: """Concatenate an ordered collection of numpy-stl meshes into a single mesh. This is the canonical home for generic mesh composition used by the RPOD @@ -123,7 +136,13 @@ def compose_meshes(meshes): return mesh.Mesh(np.concatenate([m.data for m in meshes])) -def convert_stl_to_vtk(surface, out_path, *, filename=None, cellData=None): +def convert_stl_to_vtk( + surface: mesh.Mesh | str | Path, + out_path: str | os.PathLike[str], + *, + filename: str | None = None, + cellData: Mapping[str, NDArray[Any]] | None = None, +) -> None: """ Convert an STL mesh (or path to an STL) to a VTK unstructured grid file. @@ -190,7 +209,12 @@ def convert_stl_to_vtk(surface, out_path, *, filename=None, cellData=None): from stl import mesh import numpy as np -def _transform_stl_file_cli(input_file, output_file, scale, translate): +def _transform_stl_file_cli( + input_file: str, + output_file: str, + scale: float, + translate: Sequence[float] | NDArray[np.float64], +) -> None: """Scale + translate an STL file in place from the command line. This is the file-based command-line helper invoked from ``__main__``. It diff --git a/pyrpod/vehicle/LogisticsModule.py b/pyrpod/vehicle/LogisticsModule.py index e5fd9c0..5a8ee49 100644 --- a/pyrpod/vehicle/LogisticsModule.py +++ b/pyrpod/vehicle/LogisticsModule.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pyrpod.vehicle.VisitingVehicle import VisitingVehicle from stl import mesh import numpy as np @@ -6,6 +8,8 @@ import os import configparser import logging +from collections.abc import Sequence +from typing import Any from pyrpod.util.io.fs import resolve_asset_path logger = logging.getLogger(__name__) @@ -68,8 +72,16 @@ class LogisticsModule(VisitingVehicle): Plots all thruster working groups in the RCS configuration. """ + # Declared (without a value, so no class attribute is created) to stop the + # ``= None`` sentinel in assign_thruster_groups from becoming the permanent + # inferred type. Annotated to the class contract -- a populated grouping + # dict -- because that None is a dead-end error sentinel that no consumer + # in this class (or anywhere else in pyrpod) checks for; see the ignore on + # the assignment itself. + rcs_groups: dict[str, list[str]] + # TODO: write a custom COM of calculator for comparing RCS configurations (method) - def __init__(self, case_dir): + def __init__(self, case_dir: str) -> None: """ Class responsible for handling visiting vehicle data. @@ -94,7 +106,8 @@ def __init__(self, case_dir): # Delegate initialization to parent to avoid duplicating config reading super().__init__(case_dir) - def set_inertial_props(self, mass, height, radius): + def set_inertial_props(self, mass: float, height: float, + radius: float) -> None: """ Simple constructor used to establish LM inertial properties. @@ -132,7 +145,7 @@ def set_inertial_props(self, mass, height, radius): return - def add_thruster_performance(self, thrust_val, isp): + def add_thruster_performance(self, thrust_val: float, isp: float) -> None: """ WIP. Assigns thruster performance characteristics using thruster ID specified in TCD file.""" # TODO: re-write method to read in data from CSV file. Do docstring after. # 1. mass, 2. chamber temp, 3. chamber pressure 4. velocity 5. impulse bit @@ -141,7 +154,7 @@ def add_thruster_performance(self, thrust_val, isp): self.isp = isp return - def calc_thruster_performance(self): + def calc_thruster_performance(self) -> list[dict[str, Any]]: """ Calculates performance of each thruster fired individually. @@ -188,7 +201,7 @@ def calc_thruster_performance(self): return thruster_performance_data - def rcs_group_str_to_list(self, working_group): + def rcs_group_str_to_list(self, working_group: str) -> list[str]: """ Helper method needed convert configuration data into a list. @@ -220,12 +233,12 @@ def rcs_group_str_to_list(self, working_group): return group_list - def print_rcs_groups(self): + def print_rcs_groups(self) -> None: """Simple method to format printing of RCS groups""" for group in self.rcs_groups: logger.info("%s %s", group, self.rcs_groups[group]) - def assign_thrusters(self, group): + def assign_thrusters(self, group: str) -> None: """ Assigns RCS thrusters to a specificed working group @@ -250,7 +263,7 @@ def assign_thrusters(self, group): self.rcs_groups[group].append(thruster) # print(self.rcs_groups) - def assign_thruster_groups(self): + def assign_thruster_groups(self) -> None: """Wrapper method for grouping RCS thrusters according to provided configuration data.""" # Read in grouping configuration file. @@ -259,7 +272,13 @@ def assign_thruster_groups(self): config.read(resolve_asset_path(self.case_dir, 'tcd', self.config['tcd']['tgf'])) except KeyError: # print("WARNING: Thruster Grouping File Not Set") - self.rcs_groups = None + # Pre-existing "no thruster grouping file" sentinel. Nothing reads + # it back as None -- print_rcs_groups, plot_thruster_group and + # calc_overshoot_v_range all index rcs_groups unguarded -- so + # widening the attribute to dict|None would only relocate this one + # report to those five call sites. Guarding them is a behavior + # change and out of scope for #103. + self.rcs_groups = None # type: ignore[assignment] return @@ -285,7 +304,9 @@ def assign_thruster_groups(self): decel_thruster_name = next(iter(self.rcs_groups['neg_x'])) self.decel_cant = self.get_thruster_cant(decel_thruster_name) - def plot_active_thrusters(self, active_thrusters, working_group, normals): + def plot_active_thrusters(self, active_thrusters: mesh.Mesh, + working_group: str, + normals: Sequence[Sequence[Sequence[float]]]) -> None: """ Plots active thrusters for a specified working group. @@ -339,7 +360,7 @@ def plot_active_thrusters(self, active_thrusters, working_group, normals): # Save to file plt.savefig('img/frame' + str(working_group) + '.png') - def plot_thruster_group(self, working_group): + def plot_thruster_group(self, working_group: str) -> None: """ Wrapper method to plot active thrusters in a given working group. Name is confusing need to revise. @@ -371,11 +392,17 @@ def plot_thruster_group(self, working_group): if not os.path.isdir('stl/groups/'): os.system('mkdir stl/groups') - active_thrusters.save('stl/groups/' + working_group + '.stl') + # active_thrusters is still None when the working group is empty; the + # legacy code has always assumed at least one thruster per group and + # would raise AttributeError otherwise. Adding a guard would change + # runtime behavior, so the two uses below are flagged, not fixed. + active_thrusters.save( # type: ignore[union-attr] + 'stl/groups/' + working_group + '.stl') - self.plot_active_thrusters(active_thrusters, working_group, normals) + self.plot_active_thrusters( + active_thrusters, working_group, normals) # type: ignore[arg-type] - def check_thruster_groups(self): + def check_thruster_groups(self) -> None: """ Plots all thruster working groups in the RCS configuration. @@ -389,14 +416,20 @@ def check_thruster_groups(self): # print() return - def calc_overshoot_v_range(self, v_ida, r_o): + def calc_overshoot_v_range(self, v_ida: float, + r_o: float) -> list[np.floating[Any]]: mass = self.mass # print(len(self.rcs_groups['neg_x'])) F_decel = 0 for thruster in self.rcs_groups['neg_x']: thruster_type = self.thruster_data[thruster]['type'][0] - thruster_metrics = self.thruster_metrics[thruster_type] + # vv.thruster_metrics is a real Optional -- set_thruster_metrics + # leaves it None when the case has no [tcd] tdf, and callers such + # as PlumeStrikeEstimationStudy check for that. This path never + # did, and adding a check here would change runtime behavior. + thruster_metrics = self.thruster_metrics[ # type: ignore[index] + thruster_type] cant = self.decel_cant @@ -412,6 +445,7 @@ def calc_overshoot_v_range(self, v_ida, r_o): logger.debug("vo_range: %s", vo_range) return vo_range - def debug_decel_calc_example(self, F_decel, v_o, vo_range): + def debug_decel_calc_example(self, F_decel: float, v_o: float, + vo_range: Sequence[float]) -> None: # Replaces ad-hoc prints with a single debug helper (optional usage) logger.debug("F_decel=%s, v_o=%s, vo_range=%s", F_decel, v_o, vo_range) \ No newline at end of file diff --git a/pyrpod/vehicle/TargetVehicle.py b/pyrpod/vehicle/TargetVehicle.py index 6c3bc37..62263ea 100644 --- a/pyrpod/vehicle/TargetVehicle.py +++ b/pyrpod/vehicle/TargetVehicle.py @@ -34,7 +34,7 @@ class TargetVehicle(Vehicle): """ - def set_stl(self): + def set_stl(self) -> None: """ Reads in Vehicle surface mesh from STL file. @@ -61,7 +61,7 @@ def set_stl(self): log_array_summary(logger, "target_mesh_vectors", self.mesh.vectors) return - def set_stl_elements(self): + def set_stl_elements(self) -> None: """ place holder method now. A strech goal could be to load in a multi surface stl file which accounts for @@ -71,8 +71,8 @@ def set_stl_elements(self): print('') return - def set_v_ida(self, v_ida): + def set_v_ida(self, v_ida: float) -> None: self.v_ida = v_ida - def set_r_o(self, r_o): + def set_r_o(self, r_o: float) -> None: self.r_o = r_o \ No newline at end of file diff --git a/pyrpod/vehicle/Vehicle.py b/pyrpod/vehicle/Vehicle.py index 431a6d2..0b7060e 100644 --- a/pyrpod/vehicle/Vehicle.py +++ b/pyrpod/vehicle/Vehicle.py @@ -1,8 +1,13 @@ +from __future__ import annotations + from stl import mesh import numpy as np +from numpy.typing import NDArray import os import configparser import logging +from collections.abc import Mapping +from typing import Any from pyevtk.vtk import VtkTriangle, VtkQuad from pyrpod.util.stl.stl import convert_stl_to_vtk @@ -30,13 +35,13 @@ class Vehicle: convert_stl_to_vtk(cellData, mesh) Converts STL mesh to a VTK file and attaches surface properties supplied in cellData. """ - def __init__(self, case_dir): + def __init__(self, case_dir: str) -> None: self.case_dir = case_dir config = configparser.ConfigParser() config.read(self.case_dir + "config.ini") self.config = config - def set_stl(self): + def set_stl(self) -> None: """ Reads in Vehicle surface mesh from STL file. @@ -60,7 +65,12 @@ def set_stl(self): len(self.mesh.vectors)) return - def convert_stl_to_vtk_strikes(self, path_to_vtk, cellData, mesh): + def convert_stl_to_vtk_strikes( + self, + path_to_vtk: str | None, + cellData: Mapping[str, NDArray[Any]] | None, + mesh: mesh.Mesh | None, + ) -> None: """ Converts STL mesh to a VTK file and attaches surface properties supplied in cellData. @@ -86,7 +96,7 @@ def convert_stl_to_vtk_strikes(self, path_to_vtk, cellData, mesh): return - def convert_stl_to_vtk(self): + def convert_stl_to_vtk(self) -> None: """ Converts STL mesh to a VTK file and attaches surface properties supplied in cellData. diff --git a/pyrpod/vehicle/VisitingVehicle.py b/pyrpod/vehicle/VisitingVehicle.py index 08ebb67..1b6ebbb 100644 --- a/pyrpod/vehicle/VisitingVehicle.py +++ b/pyrpod/vehicle/VisitingVehicle.py @@ -1,42 +1,70 @@ +from __future__ import annotations + import pandas as pd from stl import mesh from mpl_toolkits import mplot3d from matplotlib import pyplot as plt import numpy as np +from numpy.typing import ArrayLike, NDArray import math import os import re import logging +from collections.abc import Iterable, Sequence +from typing import Any, TypedDict from pyrpod.vehicle.Vehicle import Vehicle -from pyrpod.mdao import SweepConfig from pyrpod.logging_utils import log_asset, log_array_summary from pyrpod.util.io.fs import resolve_asset_path logger = logging.getLogger(__name__) + +# One parsed entry of the thruster configuration file (TCF), as produced by +# process_thruster_def() and stored in VisitingVehicle.thruster_data. Declared +# total=False purely so the parser can populate it key by key; every key is +# always present on a fully parsed entry. 'name' and 'type' are one-element +# lists and 'exit' is a one-element list of [x, y, z]; that shape is load +# bearing (callers index ['name'][0], ['exit'][0]) and is not changed here. +# 'dcm' is nested lists as parsed, but SweepConfig's cant sweeps replace it +# with an ndarray in place, so both forms are accepted. +class ThrusterConfig(TypedDict, total=False): + name: list[str] + type: list[str] + exit: list[list[float]] + dcm: list[list[float]] | NDArray[np.float64] + + +# One parsed entry of the cluster configuration file (CCF). Same shape as +# ThrusterConfig minus 'type'. +class ClusterConfig(TypedDict, total=False): + name: list[str] + exit: list[list[float]] + dcm: list[list[float]] | NDArray[np.float64] + + # Adapted from # https://stackoverflow.com/questions/54616049/converting-a-rotation-matrix-to-euler-angles-and-back-special-case -def rot2eul(R): +def rot2eul(R: NDArray[np.float64]) -> NDArray[np.float64]: beta = -np.arcsin(R[2][0]) alpha = np.arctan2(R[2][1]/np.cos(beta),R[2][2]/np.cos(beta)) gamma = np.arctan2(R[1][0]/np.cos(beta),R[0][0]/np.cos(beta)) return np.array((alpha, beta, gamma)) # Helper functions for constructer. -def process_coordinates(str_coord): +def process_coordinates(str_coord: str) -> list[float]: # Split str at spaces str_list = str_coord.split(' ') # Return as list of floats return [float(x) for x in str_list] # Process definition of an individual thruster. -def process_thruster_def(str_thruster): +def process_thruster_def(str_thruster: str) -> ThrusterConfig: columns = ['name', 'type', 'exit', 'dcm'] # thruster = pd.DataFrame(columns = columns) # print(thruster.dtypes) - thruster = {} + thruster: ThrusterConfig = {} # Remove new line char (last char) and split at any space char. str_list = str_thruster[:-1].split(' ') # str_list = str_thruster.split(' ') @@ -65,7 +93,9 @@ def process_thruster_def(str_thruster): return thruster # Wrapper function -def process_str_thrusters(str_thrusters): +def process_str_thrusters( + str_thrusters: Iterable[str], +) -> dict[str, ThrusterConfig]: # dcm = direction cosine matrix columns = ['name', 'type', 'exit', 'dcm'] thrusters_data = {} @@ -79,9 +109,9 @@ def process_str_thrusters(str_thrusters): return thrusters_data # Process definition of an individual cluster. -def process_cluster_def(str_cluster): +def process_cluster_def(str_cluster: str) -> ClusterConfig: columns = ['name', 'exit', 'dcm'] - cluster = {} + cluster: ClusterConfig = {} # Remove new line char (last char) and split at any space char. str_list = str_cluster[:-1].split(' ') # Save name of cluster @@ -103,7 +133,9 @@ def process_cluster_def(str_cluster): return cluster # Wrapper function -def process_str_clusters(str_clusters): +def process_str_clusters( + str_clusters: Iterable[str], +) -> dict[str, ClusterConfig]: # dcm = direction cosine matrix columns = ['name', 'exit', 'dcm'] clusters_data = {} @@ -186,7 +218,15 @@ class VisitingVehicle(Vehicle): Plots visiting vehicle and all thrusters in RCS configuration. """ - def print_info(self): + # Declared (without a value, so no class attribute is created) because the + # first assignment mypy sees is ``= None`` in set_thruster_config, which + # would otherwise pin the attribute to None and reject the cached mapping. + _thruster_id_map: dict[str, str] | None + # pandas' to_dict(orient='records') yields Hashable-keyed dicts, so the + # per-thruster metric records stay Any-valued at that boundary. + thruster_metrics: dict[str, Any] | None + + def print_info(self) -> None: """ Simple method to format printing of vehicle info. @@ -205,7 +245,7 @@ def print_info(self): logger.info('number of dual jet interactions: %s', self.jet_interactions) return - def set_stl(self): + def set_stl(self) -> None: """ Reads in Vehicle surface mesh from STL file. @@ -228,7 +268,7 @@ def set_stl(self): log_array_summary(logger, "vv_mesh_vectors", self.mesh.vectors) return - def get_thruster_cant(self, thruster_name): + def get_thruster_cant(self, thruster_name: str) -> float: """ Finds the cant angle defined as angle from the LM surface tangent. Takes the thruster's DCM, undoes the frame transformation @@ -278,7 +318,9 @@ def get_thruster_cant(self, thruster_name): return cant - def set_thruster_config(self, thruster_data=None): + def set_thruster_config( + self, thruster_data: dict[str, ThrusterConfig] | None = None + ) -> None: """ Reads the thruster configuration file from the config.ini for the Visiting Vehicle and saves it as class members. @@ -348,7 +390,7 @@ def set_thruster_config(self, thruster_data=None): return - def change_cluster_config(self, x): + def change_cluster_config(self, x: float | NDArray[np.float64]) -> None: """ Alters cluster configuration data using OpenMDAO inputs. @@ -367,7 +409,7 @@ def change_cluster_config(self, x): # print('cluster is', cluster) self.cluster_data[cluster]["exit"][0][0] = float(x) - def set_cluster_config(self): + def set_cluster_config(self) -> None: """ Read in cluster configuration data from the provided file path. Gathers cluster configuration data for the Visiting Vehicle from a .dat file @@ -404,7 +446,7 @@ def set_cluster_config(self): return - def set_thruster_metrics(self): + def set_thruster_metrics(self) -> None: """ Reads the csv thruster data file to gather thruster-specific performance parameters for the configuration and saves it in a list of dictionaries. These dictionaries are then saved into each thruster in the configuration. @@ -459,7 +501,7 @@ def set_thruster_metrics(self): return - def initiate_plume_mesh(self): + def initiate_plume_mesh(self) -> mesh.Mesh: """ Helper method that reads in surface mesh for plume clone. @@ -486,7 +528,7 @@ def initiate_plume_mesh(self): # single-digit clusters while additionally supporting multi-digit ones. _CLUSTER_ID_RE = re.compile(r'^(P\d+)') - def _cluster_id_for_thruster(self, thruster_id): + def _cluster_id_for_thruster(self, thruster_id: str) -> str: """Resolve the cluster a thruster belongs to from its id. The cluster association is encoded in the thruster naming convention: @@ -517,8 +559,10 @@ def _cluster_id_for_thruster(self, thruster_id): f"(available: {sorted(self.cluster_data)}).") return cluster_id - def transform_plume_mesh(self, thruster_id, plumeMesh, - vv_orientation=None, vv_position=None): + def transform_plume_mesh(self, thruster_id: str, plumeMesh: mesh.Mesh, + vv_orientation: ArrayLike | None = None, + vv_position: Sequence[float] | NDArray[np.float64] + | None = None) -> mesh.Mesh: """Place a plume mesh for a thruster, mutating it in place. Canonical owner of plume-placement geometry. Applies the legacy @@ -616,7 +660,7 @@ def transform_plume_mesh(self, thruster_id, plumeMesh, plumeMesh.translate(self.thruster_data[thruster_id]['exit'][0]) return plumeMesh - def get_thruster_id_map(self): + def get_thruster_id_map(self) -> dict[str, str]: """Return the cached JFH-index -> canonical-thruster-id mapping. The JFH references thrusters by 1-based numeric index into the thruster @@ -637,7 +681,7 @@ def get_thruster_id_map(self): self._thruster_id_map = cached return cached - def get_thruster_id(self, jfh_index): + def get_thruster_id(self, jfh_index: int | str) -> str: """Return the canonical thruster id for a 1-based JFH thruster index. Parameters @@ -664,7 +708,7 @@ def get_thruster_id(self, jfh_index): f"JFH references thruster index {jfh_index} outside the " f"configured range 1..{len(mapping)}.") from None - def initiate_plume_normal(self, thruster_id): + def initiate_plume_normal(self, thruster_id: str) -> list[list[float]]: """ Collects plume normal vectors data for visualization. @@ -704,7 +748,8 @@ def initiate_plume_normal(self, thruster_id): return [X,Y,Z,U,V,W] - def plot_vv_and_thruster(self, plumeMesh, thruster_id, normal, i): + def plot_vv_and_thruster(self, plumeMesh: mesh.Mesh, thruster_id: str, + normal: Sequence[Sequence[float]], i: int) -> int: """ Plots Visiting Vehicle and plume cone for provided thruster id. @@ -778,7 +823,7 @@ def plot_vv_and_thruster(self, plumeMesh, thruster_id, normal, i): plt.savefig('img/frame' + str(index) + '.png') return i + 1 - def check_thruster_configuration(self): + def check_thruster_configuration(self) -> None: """ Plots visiting vehicle and all thrusters in RCS configuration.