diff --git a/python/ngen_cal/src/ngen/cal/__main__.py b/python/ngen_cal/src/ngen/cal/__main__.py index 00e58fd1..1d369f3c 100644 --- a/python/ngen_cal/src/ngen/cal/__main__.py +++ b/python/ngen_cal/src/ngen/cal/__main__.py @@ -20,7 +20,7 @@ def _loaded_plugins(pm: PluginManager) -> str: from .utils import type_as_import_string - plugins: List[str] = [] + plugins: list[str] = [] for (name, plugin) in pm.list_name_plugin(): if not name: continue @@ -34,7 +34,7 @@ def _loaded_plugins(pm: PluginManager) -> str: return f"Plugins Loaded: {', '.join(plugins)}" -def main(general: General, model_conf: "Mapping[str, Any]"): +def main(general: General, model_conf: Mapping[str, Any]): #seed the random number generators if requested if general.random_seed is not None: import random @@ -72,7 +72,7 @@ def main(general: General, model_conf: "Mapping[str, Any]"): print("Restart not supported for PSO search, starting at 0") func = pso_search - print("Starting Iteration: {}".format(start_iteration)) + print(f"Starting Iteration: {start_iteration}") # print("Starting Best param: {}".format(meta.best_params)) # print("Starting Best score: {}".format(meta.best_score)) print("Starting calibration loop") diff --git a/python/ngen_cal/src/ngen/cal/_hookspec.py b/python/ngen_cal/src/ngen/cal/_hookspec.py index 5b5f6ee6..c046dbc6 100644 --- a/python/ngen_cal/src/ngen/cal/_hookspec.py +++ b/python/ngen_cal/src/ngen/cal/_hookspec.py @@ -10,7 +10,6 @@ from ngen.cal.configuration import General from ngen.cal.meta import JobMeta from pandas import Series - from pathlib import Path hookspec = pluggy.HookspecMarker(PROJECT_SLUG) diff --git a/python/ngen_cal/src/ngen/cal/agent.py b/python/ngen_cal/src/ngen/cal/agent.py index ab0da88d..4f124c8d 100644 --- a/python/ngen_cal/src/ngen/cal/agent.py +++ b/python/ngen_cal/src/ngen/cal/agent.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from abc import ABC, abstractmethod from ngen.cal.meta import JobMeta from ngen.cal.configuration import Model @@ -5,7 +7,7 @@ from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: - from typing import Sequence, Optional, Mapping, Any + from typing import Sequence, Mapping, Any from pandas import DataFrame from pathlib import Path from ngen.cal.calibratable import Adjustable @@ -13,7 +15,7 @@ class BaseAgent(ABC): @property - def adjustables(self) -> 'Sequence[Adjustable]': + def adjustables(self) -> Sequence[Adjustable]: return self.model.adjustables def restart(self) -> int: @@ -29,15 +31,15 @@ def restart(self) -> int: @property @abstractmethod - def model(self) -> 'Model': + def model(self) -> Model: pass @property @abstractmethod - def job(self) -> 'JobMeta': + def job(self) -> JobMeta: pass - def update_config(self, i: int, params: 'DataFrame', id: str): + def update_config(self, i: int, params: DataFrame, id: str): """ For a given calibration iteration, i, update the input files/configuration to prepare for that iterations calibration run. @@ -57,7 +59,7 @@ def best_params(self) -> str: class Agent(BaseAgent): - def __init__(self, model_conf, workdir: 'Path', log: bool=False, restart: bool=False, parameters: 'Optional[Mapping[str, Any]]' = {}): + def __init__(self, model_conf, workdir: Path, log: bool=False, restart: bool=False, parameters: Mapping[str, Any] | None = {}): self._workdir = workdir self._job = None if restart: @@ -86,11 +88,11 @@ def __init__(self, model_conf, workdir: 'Path', log: bool=False, restart: bool=F self._params = parameters @property - def parameters(self) -> 'Mapping[str, Any]': + def parameters(self) -> Mapping[str, Any]: return self._params @property - def workdir(self) -> 'Path': + def workdir(self) -> Path: return self._workdir @property @@ -106,9 +108,9 @@ def cmd(self) -> str: """ Proxy method to build command from contained model binary and args """ - return "{} {}".format(self.model.get_binary(), self.model.get_args()) + return f"{self.model.get_binary()} {self.model.get_args()}" - def duplicate(self) -> 'Agent': + def duplicate(self) -> Agent: #serialize a copy of the model #FIXME ??? if you do self.model.resolve_paths() here, the duplicated agent #doesn't have fully qualified paths...but if you do it in constructor, it works fine... diff --git a/python/ngen_cal/src/ngen/cal/calibratable.py b/python/ngen_cal/src/ngen/cal/calibratable.py index ce3f2f2b..745fa647 100644 --- a/python/ngen_cal/src/ngen/cal/calibratable.py +++ b/python/ngen_cal/src/ngen/cal/calibratable.py @@ -2,14 +2,14 @@ from abc import ABC, abstractmethod from pandas import Series, read_parquet # type: ignore -from typing import Optional, TYPE_CHECKING +from typing import TYPE_CHECKING from pathlib import Path if TYPE_CHECKING: from pandas import DataFrame, Series from pathlib import Path from datetime import datetime - from typing import Tuple, Callable, Optional + from typing import Callable from ngen.cal.model import EvaluationOptions from ngen.cal.meta import JobMeta @@ -18,11 +18,11 @@ class Adjustable(ABC): An Adjustable interface defning required properties for adjusting an object's state """ - def __init__(self, df: Optional['DataFrame'] = None): + def __init__(self, df: DataFrame | None = None): self._df = df @property - def df(self) -> 'DataFrame': + def df(self) -> DataFrame: """ A dataframe of the objects parameter values to calculate indexed relative to the variables being calibrated. The columns of the dataframe will be appended to with each search iterations @@ -46,14 +46,14 @@ def id(self) -> str: pass @property - def variables(self) -> 'Series': + def variables(self) -> Series: """ Index series of variables """ return Series(self.df.index.values) @property - def bounds(self) -> 'Tuple[Series, Series]': + def bounds(self) -> tuple[Series, Series]: """The bounds of each parameter that is adjustable Returns: @@ -77,11 +77,11 @@ def update_params(self, iteration: int) -> None: pass @property - def check_point_file(self) -> 'Path': + def check_point_file(self) -> Path: """ Filename checkpoint files are saved to """ - return Path('{}_parameter_df_state.parquet'.format(self.id)) + return Path(f'{self.id}_parameter_df_state.parquet') def check_point(self, iteration: int, info: JobMeta) -> None: """ @@ -90,7 +90,7 @@ def check_point(self, iteration: int, info: JobMeta) -> None: path = info.workdir self.df.to_parquet(path/self.check_point_file) - def load_df(self, path: 'Path') -> None: + def load_df(self, path: Path) -> None: """ Load saved calibration information """ @@ -104,9 +104,9 @@ class Evaluatable(ABC): An Evaluatable interface defining required properties for a evaluating and object's state """ - eval_params: 'EvaluationOptions' + eval_params: EvaluationOptions - def __init__(self, eval_params: 'EvaluationOptions', **kwargs): + def __init__(self, eval_params: EvaluationOptions, **kwargs): """ Args: eval_params (EvaluationOptions): The options configuring this evaluatable @@ -115,7 +115,7 @@ def __init__(self, eval_params: 'EvaluationOptions', **kwargs): @property @abstractmethod - def output(self) -> 'DataFrame': + def output(self) -> DataFrame: """ The output data for the calibrated object Calibration re-reads the output each call, as the output for given calibration is expected to change @@ -125,7 +125,7 @@ def output(self) -> 'DataFrame': @property @abstractmethod - def observed(self) -> 'DataFrame': + def observed(self) -> DataFrame: """ The observed data for this calibratable. This should be rather static, and can be set at initialization then accessed via the property @@ -134,7 +134,7 @@ def observed(self) -> 'DataFrame': @property @abstractmethod - def evaluation_range(self) -> 'Optional[Tuple[datetime, datetime]]': + def evaluation_range(self) -> tuple[datetime, datetime] | None: """ The datetime range to evaluate the model results at. This should be a tuple in the form of (start_time, end_time). @@ -142,7 +142,7 @@ def evaluation_range(self) -> 'Optional[Tuple[datetime, datetime]]': pass @property - def objective(self, *args, **kwargs) -> 'Callable': + def objective(self, *args, **kwargs) -> Callable: """ The objective function to compute cost values with. @@ -191,5 +191,5 @@ class Calibratable(Adjustable, Evaluatable): """ A Calibratable interface defining required properties for a calibratable object """ - def __init__(self, df: Optional['DataFrame'] = None): + def __init__(self, df: DataFrame | None = None): Adjustable.__init__(self, df) diff --git a/python/ngen_cal/src/ngen/cal/calibration_cathment.py b/python/ngen_cal/src/ngen/cal/calibration_cathment.py index f15067bb..c99d7a29 100644 --- a/python/ngen_cal/src/ngen/cal/calibration_cathment.py +++ b/python/ngen_cal/src/ngen/cal/calibration_cathment.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pandas import DataFrame, read_csv # type: ignore import shutil from typing import TYPE_CHECKING @@ -7,7 +9,6 @@ from pathlib import Path from geopandas import GeoSeries from datetime import datetime - from typing import Tuple, Optional from .model import EvaluationOptions from hypy.catchment import FormulatableCatchment # type: ignore @@ -21,7 +22,7 @@ class AdjustableCatchment(FormulatableCatchment, Adjustable): parameteters used by the catchment. """ - def __init__(self, workdir: 'Path', id: str, nexus, params: dict = {}): + def __init__(self, workdir: Path, id: str, nexus, params: dict = {}): """Create an adjustable catchment and initialize its parameter space Args: @@ -34,7 +35,7 @@ def __init__(self, workdir: 'Path', id: str, nexus, params: dict = {}): FormulatableCatchment.__init__(self=self, catchment_id=id, params=params, outflow=nexus) Adjustable.__init__(self=self, df=DataFrame(params).rename(columns={'init': '0'})) #FIXME paramterize - self._output_file = workdir/'{}.csv'.format(self.id) + self._output_file = workdir/f'{self.id}.csv' self._workdir = workdir def save_output(self, i) -> None: @@ -43,7 +44,7 @@ def save_output(self, i) -> None: """ #FIXME ensure _output_file exists #FIXME re-enable this once more complete - shutil.move(self._output_file, '{}_last'.format(self._output_file)) + shutil.move(self._output_file, f'{self._output_file}_last') #update handled in meta, TODO remove this method??? def update_params(self, iteration: int) -> None: @@ -54,7 +55,7 @@ class EvaluatableCatchment(Evaluatable): A catchment which is "observable" which means model output can be evaluated against these observations for this catchment. """ - def __init__(self, nexus: Nexus, start_time: str, end_time: str, fabric: "GeoSeries", output_var: str, eval_params: 'EvaluationOptions'): + def __init__(self, nexus: Nexus, start_time: str, end_time: str, fabric: GeoSeries, output_var: str, eval_params: EvaluationOptions): """Initialize the evaluatable catchment Args: @@ -81,11 +82,11 @@ def __init__(self, nexus: Nexus, start_time: str, end_time: str, fabric: "GeoSer self._eval_range = self.eval_params._eval_range @property - def evaluation_range(self) -> 'Optional[Tuple[datetime, datetime]]': + def evaluation_range(self) -> tuple[datetime, datetime] | None: return self._eval_range @property - def output(self) -> 'DataFrame': + def output(self) -> DataFrame: """ The model output hydrograph for this catchment This re-reads the output file each call, as the output for given calibration catchment changes @@ -112,13 +113,13 @@ def output(self, df): self._output = df @property - def observed(self) -> 'DataFrame': + def observed(self) -> DataFrame: """ The observed hydrograph for this catchment FIXME move output/observed to calibratable? """ hydrograph = self._observed if hydrograph is None: - raise(RuntimeError("Error reading observation for {}".format(self._id))) + raise(RuntimeError(f"Error reading observation for {self._id}")) return hydrograph @observed.setter @@ -129,7 +130,7 @@ class CalibrationCatchment(AdjustableCatchment, EvaluatableCatchment): """ A Calibratable interface defining required properties for a calibratable object """ - def __init__(self, workdir: str, id: str, nexus: Nexus, start_time: str, end_time: str, fabric: "GeoSeries", output_var: str, eval_params: 'EvaluationOptions', params: dict = {}): + def __init__(self, workdir: str, id: str, nexus: Nexus, start_time: str, end_time: str, fabric: GeoSeries, output_var: str, eval_params: EvaluationOptions, params: dict = {}): EvaluatableCatchment.__init__(self, nexus, start_time, end_time, fabric, output_var, eval_params) AdjustableCatchment.__init__(self, workdir, id, nexus, params) diff --git a/python/ngen_cal/src/ngen/cal/calibration_set.py b/python/ngen_cal/src/ngen/cal/calibration_set.py index c750a88f..13a0a082 100644 --- a/python/ngen_cal/src/ngen/cal/calibration_set.py +++ b/python/ngen_cal/src/ngen/cal/calibration_set.py @@ -1,21 +1,16 @@ from __future__ import annotations from pandas import DataFrame# type: ignore -import shutil from typing import TYPE_CHECKING, Sequence -import pandas as pd if TYPE_CHECKING: from pandas import DataFrame from pathlib import Path from datetime import datetime - from typing import Tuple, Optional from ngen.cal._hookspec import ModelHooks from ngen.cal.model import EvaluationOptions from ngen.cal.meta import JobMeta -import os from pathlib import Path -import warnings from hypy.nexus import Nexus from .calibratable import Adjustable, Evaluatable @@ -25,7 +20,7 @@ class CalibrationSet(Evaluatable): A HY_Features based catchment with additional calibration information/functionality """ - def __init__(self, adjustables: Sequence[Adjustable], eval_nexus: Nexus, hooks: ModelHooks, start_time: str, end_time: str, eval_params: 'EvaluationOptions'): + def __init__(self, adjustables: Sequence[Adjustable], eval_nexus: Nexus, hooks: ModelHooks, start_time: str, end_time: str, eval_params: EvaluationOptions): """ """ @@ -46,7 +41,7 @@ def __init__(self, adjustables: Sequence[Adjustable], eval_nexus: Nexus, hooks: self._eval_range = self.eval_params._eval_range @property - def evaluation_range(self) -> 'Optional[Tuple[datetime, datetime]]': + def evaluation_range(self) -> tuple[datetime, datetime] | None: return self._eval_range @property @@ -54,7 +49,7 @@ def adjustables(self): return self._adjustables @property - def output(self) -> 'DataFrame': + def output(self) -> DataFrame: """ The model output hydrograph for this catchment This re-reads the output file each call, as the output for given calibration catchment changes @@ -77,13 +72,13 @@ def output(self, df): self._output = df @property - def observed(self) -> 'DataFrame': + def observed(self) -> DataFrame: """ The observed hydrograph for this catchment FIXME move output/observed to calibratable? """ hydrograph = self._observed if hydrograph is None: - raise(RuntimeError("Error reading observation for {}".format(self._id))) + raise(RuntimeError(f"Error reading observation for {self._id}")) return hydrograph @observed.setter @@ -112,7 +107,7 @@ class UniformCalibrationSet(CalibrationSet, Adjustable): A HY_Features based catchment with additional calibration information/functionality """ - def __init__(self, eval_nexus: Nexus, hooks: ModelHooks, start_time: str, end_time: str, eval_params: 'EvaluationOptions', params: dict = {}): + def __init__(self, eval_nexus: Nexus, hooks: ModelHooks, start_time: str, end_time: str, eval_params: EvaluationOptions, params: dict = {}): """ """ @@ -137,8 +132,8 @@ def update_params(self, iteration: int) -> None: #Override this file name @property - def check_point_file(self) -> 'Path': - return Path('{}_parameter_df_state.parquet'.format(self._eval_nexus.id)) + def check_point_file(self) -> Path: + return Path(f'{self._eval_nexus.id}_parameter_df_state.parquet') def restart(self): try: diff --git a/python/ngen_cal/src/ngen/cal/meta.py b/python/ngen_cal/src/ngen/cal/meta.py index 871df1ab..8eb26bc9 100644 --- a/python/ngen_cal/src/ngen/cal/meta.py +++ b/python/ngen_cal/src/ngen/cal/meta.py @@ -1,4 +1,5 @@ -import pandas as pd # type: ignore +from __future__ import annotations + from pathlib import Path from tempfile import mkdtemp from datetime import datetime @@ -31,17 +32,17 @@ def __init__(self, name: str, parent_workdir: Path, workdir: Path=None, log=Fals self._log_file = self._workdir/Path(name+".log") @property - def workdir(self) -> 'Path': + def workdir(self) -> Path: return self._workdir @workdir.setter - def workdir(self, path: 'Path') -> None: + def workdir(self, path: Path) -> None: self._workdir = path if self._log_file is not None: self._log_file = self._workdir/Path(self._log_file.name) @property - def log_file(self) -> 'Path': + def log_file(self) -> Path: """ Path to the job's log file, or None. """ diff --git a/python/ngen_cal/src/ngen/cal/model.py b/python/ngen_cal/src/ngen/cal/model.py index 00a31b2a..006b4c51 100644 --- a/python/ngen_cal/src/ngen/cal/model.py +++ b/python/ngen_cal/src/ngen/cal/model.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pydantic import BaseModel, DirectoryPath, conint, PyObject, validator, Field, root_validator from typing import Any, cast, Callable, Dict, List, Optional, Tuple, Union from types import ModuleType, FunctionType @@ -114,14 +116,14 @@ def update(self, i: int, score: float, log: bool) -> None: def write_objective_log_file(self, i, score): with open(self.objective_log_file, 'a+') as log_file: - log_file.write('{}, '.format(i)) - log_file.write('{}\n'.format(score)) + log_file.write(f'{i}, ') + log_file.write(f'{score}\n') def write_param_log_file(self, i): with open(self.param_log_file, 'w+') as log_file: - log_file.write('{}\n'.format(i)) - log_file.write('{}\n'.format(self.best_params)) - log_file.write('{}\n'.format(self.best_score)) + log_file.write(f'{i}\n') + log_file.write(f'{self.best_params}\n') + log_file.write(f'{self.best_score}\n') @property def best_score(self) -> float: @@ -181,7 +183,7 @@ def validate_objective(cls, value): return value def read_param_log_file(self): - with open(self.param_log_file, 'r') as log_file: + with open(self.param_log_file) as log_file: iteration = int(log_file.readline()) best_params = int(log_file.readline()) best_score = float(log_file.readline()) diff --git a/python/ngen_cal/src/ngen/cal/ngen.py b/python/ngen_cal/src/ngen/cal/ngen.py index e78a312a..661d92c9 100644 --- a/python/ngen_cal/src/ngen/cal/ngen.py +++ b/python/ngen_cal/src/ngen/cal/ngen.py @@ -1,5 +1,7 @@ +from __future__ import annotations + from pydantic import FilePath, root_validator, BaseModel, Field -from typing import Optional, Sequence, Dict, Mapping, Union +from typing import Optional, Sequence, Mapping, Union try: #to get literal in python 3.7, it was added to typing in 3.8 from typing import Literal except ImportError: @@ -22,7 +24,7 @@ from .parameter import Parameter, Parameters from .calibration_cathment import CalibrationCatchment, AdjustableCatchment from .calibration_set import CalibrationSet, UniformCalibrationSet -from .ngen_hooks.ngen_output import TrouteOutput, NgenSaveOutput +from .ngen_hooks.ngen_output import TrouteOutput #HyFeatures components from hypy.hydrolocation import NWISLocation from hypy.nexus import Nexus @@ -94,7 +96,7 @@ class NgenBase(ModelExec): args: Optional[str] #private, not validated - _catchments: Sequence['CalibrationCatchment'] = [] + _catchments: Sequence[CalibrationCatchment] = [] _catchment_hydro_fabric: gpd.GeoDataFrame _nexus_hydro_fabric: gpd.GeoDataFrame _flowpath_hydro_fabric: gpd.GeoDataFrame @@ -214,7 +216,7 @@ def config_file(self) -> Path: return self.realization @property - def adjustables(self) -> Sequence['CalibrationCatchment']: + def adjustables(self) -> Sequence[CalibrationCatchment]: """A list of Catchments for calibration These catchments hold information about the parameters/calibration data for that catchment @@ -225,7 +227,7 @@ def adjustables(self) -> Sequence['CalibrationCatchment']: return self._catchments @root_validator - def set_defaults(cls, values: Dict): + def set_defaults(cls, values: dict): """Compose default values This validator will set/adjust the following data values for the class @@ -253,9 +255,9 @@ def set_defaults(cls, values: Dict): custom_args = False if args is None: if hydrofabric is not None: - args = '{} "all" {} "all" {}'.format(hydrofabric.resolve(), hydrofabric.resolve(), realization.name) + args = f'{hydrofabric.resolve()} "all" {hydrofabric.resolve()} "all" {realization.name}' else: - args = '{} "all" {} "all" {}'.format(catchments.resolve(), nexus.resolve(), realization.name) + args = f'{catchments.resolve()} "all" {nexus.resolve()} "all" {realization.name}' values['args'] = args else: custom_args = True @@ -348,7 +350,7 @@ def _verify_ngen_realization(realization: Optional[NgenRealization]) -> None: "ngen realization `output_root` field is not supported by ngen.cal. will be removed in future; see https://github.com/NOAA-OWP/ngen-cal/issues/150" ) - def update_config(self, i: int, params: 'pd.DataFrame', id: str = None, path=Path("./")): + def update_config(self, i: int, params: pd.DataFrame, id: str = None, path=Path("./")): """_summary_ Args: @@ -401,11 +403,11 @@ def __init__(self, **kwargs): try: nwis = self._x_walk[id] except KeyError: - raise(RuntimeError("Cannot establish mapping of catchment {} to nwis location in cross walk".format(id))) + raise(RuntimeError(f"Cannot establish mapping of catchment {id} to nwis location in cross walk")) try: nexus_data = self._nexus_hydro_fabric.loc[fabric['toid']] except KeyError: - raise(RuntimeError("No suitable nexus found for catchment {}".format(id))) + raise(RuntimeError(f"No suitable nexus found for catchment {id}")) #establish the hydro location for the observation nexus associated with this catchment location = NWISLocation(nwis, nexus_data.name, nexus_data.geometry) @@ -419,7 +421,7 @@ def __init__(self, **kwargs): eval_params.id = id self._catchments.append(CalibrationCatchment(self.workdir, id, nexus, start_t, end_t, fabric, output_var, eval_params, params)) - def update_config(self, i: int, params: 'pd.DataFrame', id: str, **kwargs): + def update_config(self, i: int, params: pd.DataFrame, id: str, **kwargs): """_summary_ Args: @@ -482,7 +484,7 @@ def __init__(self, **kwargs): try: nexus_data = self._nexus_hydro_fabric.loc[fabric['toid']] except KeyError: - raise(RuntimeError("No suitable nexus found for catchment {}".format(id))) + raise(RuntimeError(f"No suitable nexus found for catchment {id}")) nwis = None try: nwis = self._x_walk.loc[id.replace('cat', 'wb')] @@ -608,7 +610,7 @@ def restart(self) -> int: def type(self): return self.__root__.type - def resolve_paths(self, relative_to: Optional[Path]=None): + def resolve_paths(self, relative_to: Path | None=None): """resolve any possible relative paths in the realization """ if self.__root__.ngen_realization != None: diff --git a/python/ngen_cal/src/ngen/cal/ngen_hooks/ngen_output.py b/python/ngen_cal/src/ngen/cal/ngen_hooks/ngen_output.py index 03f553bb..a0362f8d 100644 --- a/python/ngen_cal/src/ngen/cal/ngen_hooks/ngen_output.py +++ b/python/ngen_cal/src/ngen/cal/ngen_hooks/ngen_output.py @@ -61,7 +61,7 @@ def get_output(self, id: str) -> Series: output.name="sim_flow" return output except FileNotFoundError: - print("{} not found. Current working directory is {}".format(self._output_file, Path.cwd())) + print(f"{self._output_file} not found. Current working directory is {Path.cwd()}") print("Setting output to None") return None except Exception as e: diff --git a/python/ngen_cal/src/ngen/cal/objectives.py b/python/ngen_cal/src/ngen/cal/objectives.py index 573df159..b84b38a0 100644 --- a/python/ngen_cal/src/ngen/cal/objectives.py +++ b/python/ngen_cal/src/ngen/cal/objectives.py @@ -1,5 +1,4 @@ #!/usr/bin/env python -import pandas as pd # type: ignore from hydrotools.metrics.metrics import * weights = [0.4, 0.2, 0.4] diff --git a/python/ngen_cal/src/ngen/cal/parameter.py b/python/ngen_cal/src/ngen/cal/parameter.py index 4c6fefa9..8f83dc7e 100644 --- a/python/ngen_cal/src/ngen/cal/parameter.py +++ b/python/ngen_cal/src/ngen/cal/parameter.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pydantic import BaseModel, Field from typing import Sequence diff --git a/python/ngen_cal/src/ngen/cal/plot.py b/python/ngen_cal/src/ngen/cal/plot.py index d059e448..3256e4b2 100644 --- a/python/ngen_cal/src/ngen/cal/plot.py +++ b/python/ngen_cal/src/ngen/cal/plot.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import geopandas as gpd import pandas as pd import json @@ -12,7 +14,7 @@ if TYPE_CHECKING: from pathlib import Path -def plot_objective(objective_log_file: 'Path'): +def plot_objective(objective_log_file: Path): """ Plot the objective funtion """ @@ -39,7 +41,7 @@ def plot_stuff(workdir, catchment_data, nexus_data, cross_walk, config_file): start_t = data['time']['start_time'] end_t = data['time']['end_time'] except KeyError as e: - raise(RuntimeError("Invalid time configuration: {} key missing from {}".format(e.args[0], config_file))) + raise(RuntimeError(f"Invalid time configuration: {e.args[0]} key missing from {config_file}")) #Setup each calibration catchment for id, catchment in data['catchments'].items(): @@ -52,11 +54,11 @@ def plot_stuff(workdir, catchment_data, nexus_data, cross_walk, config_file): try: nwis = x_walk[id]['site_no'] except KeyError: - raise(RuntimeError("Cannot establish mapping of catchment {} to nwis location in cross walk".format(id))) + raise(RuntimeError(f"Cannot establish mapping of catchment {id} to nwis location in cross walk")) try: nexus_data = nexus_hydro_fabric.loc[fabric['toID']] except KeyError: - raise(RuntimeError("No suitable nexus found for catchment {}".format(id))) + raise(RuntimeError(f"No suitable nexus found for catchment {id}")) #establish the hydro location for the observation nexus associated with this catchment location = NWISLocation(nwis, nexus_data.name, nexus_data.geometry) @@ -82,15 +84,15 @@ def plot_obs(id, catchment_data, nexus_data, cross_walk): try: fabric = catchment_hydro_fabric.loc[id] except KeyError: - raise(RuntimeError("No data for id {}".format(id))) + raise(RuntimeError(f"No data for id {id}")) try: nwis = x_walk[id]['site_no'] except KeyError: - raise(RuntimeError("Cannot establish mapping of catchment {} to nwis location in cross walk".format(id))) + raise(RuntimeError(f"Cannot establish mapping of catchment {id} to nwis location in cross walk")) try: nexus_data = nexus_hydro_fabric.loc[fabric['toID']] except KeyError: - raise(RuntimeError("No suitable nexus found for catchment {}".format(id))) + raise(RuntimeError(f"No suitable nexus found for catchment {id}")) #establish the hydro location for the observation nexus associated with this catchment location = NWISLocation(nwis, nexus_data.name, nexus_data.geometry) @@ -103,9 +105,9 @@ def plot_obs(id, catchment_data, nexus_data, cross_walk): obs = obs * 0.028316847 #convert to m^3/s obs.rename('obs_flow', inplace=True) plt.figure() - obs.plot(title='Observation at USGS {}'.format(nwis)) + obs.plot(title=f'Observation at USGS {nwis}') -def plot_output(output_file: 'Path'): +def plot_output(output_file: Path): #output = pd.read_csv(output_file, usecols=["Time", "Flow"], parse_dates=['Time'], index_col='Time') #output.rename(columns={'Flow':'sim_flow'}, inplace=True) output = pd.read_csv(output_file, parse_dates=['Time'], index_col='Time') @@ -116,7 +118,7 @@ def plot_output(output_file: 'Path'): plt.figure() output['Flow'].plot(title='simulated flow') -def plot_parameter_space(path: 'Path'): +def plot_parameter_space(path: Path): params = pd.read_parquet(path) params.drop(columns=['min', 'max', 'sigma'], inplace=True) params.set_index('param', inplace=True) diff --git a/python/ngen_cal/src/ngen/cal/search.py b/python/ngen_cal/src/ngen/cal/search.py index 022dc22d..74fef2dc 100644 --- a/python/ngen_cal/src/ngen/cal/search.py +++ b/python/ngen_cal/src/ngen/cal/search.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import subprocess import pandas as pd # type: ignore from math import log @@ -9,7 +11,6 @@ if TYPE_CHECKING: from ngen.cal import Adjustable, Evaluatable from ngen.cal.agent import Agent - from typing import Tuple, Optional from datetime import datetime @@ -21,7 +22,7 @@ """ __iteration_counter = 0 -def _objective_func(simulated_hydrograph, observed_hydrograph, objective, eval_range: 'Optional[Tuple[datetime, datetime]]' = None): +def _objective_func(simulated_hydrograph, observed_hydrograph, objective, eval_range: tuple[datetime, datetime] | None = None): df = pd.merge(simulated_hydrograph, observed_hydrograph, left_index=True, right_index=True) if df.empty: print("WARNING: Cannot compute objective function, do time indicies align?") @@ -31,7 +32,7 @@ def _objective_func(simulated_hydrograph, observed_hydrograph, objective, eval_r #Evaluate custom objective function providing simulated, observed series return objective(df['obs_flow'], df['sim_flow']) -def _execute(meta: 'Agent'): +def _execute(meta: Agent): """ Execute a model run defined by the calibration meta cmd """ @@ -41,7 +42,7 @@ def _execute(meta: 'Agent'): with open(meta.job.log_file, 'a+') as log_file: subprocess.check_call(meta.cmd, stdout=log_file, stderr=log_file, shell=True, cwd=meta.job.workdir) -def _evaluate(i: int, calibration_object: 'Evaluatable', info=False) -> float: +def _evaluate(i: int, calibration_object: Evaluatable, info=False) -> float: """ Performs the evaluation logic of a calibration step """ @@ -51,25 +52,25 @@ def _evaluate(i: int, calibration_object: 'Evaluatable', info=False) -> float: #update meta info based on latest score and write some log files calibration_object.update(i, score, log=True) if info: - print("Current score {}\nBest score {}".format(score, calibration_object.best_score)) - print("Best parameters at iteration {}".format(calibration_object.best_params)) + print(f"Current score {score}\nBest score {calibration_object.best_score}") + print(f"Best parameters at iteration {calibration_object.best_params}") return score -def dds_update(iteration: int, inclusion_probability: float, calibration_object: 'Adjustable', agent: 'Agent'): +def dds_update(iteration: int, inclusion_probability: float, calibration_object: Adjustable, agent: Agent): """_summary_ Args: iteration (int): _description_ """ - print( "inclusion probability: {}".format(inclusion_probability) ) + print( f"inclusion probability: {inclusion_probability}" ) #select a random subset of variables to modify #TODO convince myself that grabbing a random selction of P fraction of items #is the same as selecting item with probability P neighborhood = calibration_object.variables.sample(frac=inclusion_probability) if neighborhood.empty: neighborhood = calibration_object.variables.sample(n=1) - print( "neighborhood: {}".format(neighborhood) ) + print( f"neighborhood: {neighborhood}" ) #Copy the best parameter values so far into the next iterations parameter list calibration_object.df[str(iteration)] = calibration_object.df[agent.best_params] #print( data.calibration_df ) @@ -100,7 +101,7 @@ def dds_update(iteration: int, inclusion_probability: float, calibration_object: agent.update_config(iteration, calibration_object.df[[str(iteration), 'param', 'model']], calibration_object.id) -def dds(start_iteration: int, iterations: int, calibration_object: 'Evaluatable', agent: 'Agent'): +def dds(start_iteration: int, iterations: int, calibration_object: Evaluatable, agent: Agent): """ """ if iterations < 2: @@ -120,7 +121,7 @@ def dds(start_iteration: int, iterations: int, calibration_object: 'Evaluatable if calibration_object.output is None: #We are starting a new calibration and do not have an initial output state to evaluate, compute it #Need initial states (iteration 0) to start DDS loop - print("Running {} to produce initial simulation".format(agent.cmd)) + print(f"Running {agent.cmd} to produce initial simulation") agent.update_config(start_iteration, calibration_object.df[[str(start_iteration), 'param', 'model']], calibration_object.id) _execute(agent) with pushd(agent.job.workdir): @@ -133,13 +134,13 @@ def dds(start_iteration: int, iterations: int, calibration_object: 'Evaluatable inclusion_probability = 1 - log(i)/log(iterations) dds_update(i, inclusion_probability, calibration_object, agent) #Run cmd Again... - print("Running {} for iteration {}".format(agent.cmd, i)) + print(f"Running {agent.cmd} for iteration {i}") _execute(agent) with pushd(agent.job.workdir): _evaluate(i, calibration_object, info=True) calibration_object.check_point(i, agent.job) -def dds_set(start_iteration: int, iterations: int, agent: 'Agent'): +def dds_set(start_iteration: int, iterations: int, agent: Agent): """ DDS search that applies to a set of calibration objects. @@ -171,7 +172,7 @@ def dds_set(start_iteration: int, iterations: int, agent: 'Agent'): if calibration_set.output is None: #We are starting a new calibration and do not have an initial output state to evaluate, compute it #Need initial states (iteration 0) to start DDS loop - print("Running {} to produce initial simulation".format(agent.cmd)) + print(f"Running {agent.cmd} to produce initial simulation") _execute(agent) with pushd(agent.job.workdir): _evaluate(0, calibration_set, info=True) @@ -184,7 +185,7 @@ def dds_set(start_iteration: int, iterations: int, agent: 'Agent'): for calibration_object in calibration_set.adjustables: dds_update(i, inclusion_probability, calibration_object, agent) #Run cmd Again... - print("Running {} for iteration {}".format(agent.cmd, i)) + print(f"Running {agent.cmd} for iteration {i}") _execute(agent) with pushd(agent.job.workdir): _evaluate(i, calibration_set, info=True) @@ -206,7 +207,7 @@ def compute(calibration_object, iteration, input) -> float: #cost = _objective_func(calibration_object.output, calibration_object.observed, calibration_object.objective, calibration_object.evaluation_range) return cost -def cost_func( calibration_object: 'Adjustable', agents: 'Agent', pool, params): +def cost_func( calibration_object: Adjustable, agents: Agent, pool, params): """_summary_ Args: @@ -244,7 +245,7 @@ def pso_search(start_iteration: int, iterations: int, agent): #TODO run first iteration? num_particles = agent.parameters.get('particles', 4) pool_size = agent.parameters.get("pool", 1) - print("Running PSO with {} particles using {} processes".format(num_particles, pool_size)) + print(f"Running PSO with {num_particles} particles using {pool_size} processes") #TODO warn about potential loss of data when particles > pool _pool = pool.Pool(pool_size) agents = [agent] + [ agent.duplicate() for i in range(num_particles-1) ] @@ -277,6 +278,5 @@ def pso_search(start_iteration: int, iterations: int, agent): cost, pos = optimizer.optimize(cf, iters=iterations, n_processes=None) calibration_object.df.loc[:,'global_best'] = pos calibration_object.check_point(iterations, agent.job) - print("Best params with cost {}:".format(cost)) + print(f"Best params with cost {cost}:") print(calibration_object.df[['param','global_best']].set_index('param')) - \ No newline at end of file diff --git a/python/ngen_cal/src/ngen/cal/strategy.py b/python/ngen_cal/src/ngen/cal/strategy.py index 7a6897c2..607a11a0 100644 --- a/python/ngen_cal/src/ngen/cal/strategy.py +++ b/python/ngen_cal/src/ngen/cal/strategy.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pydantic import BaseModel from typing import Optional, Mapping, Any try: #to get literal in python 3.7, it was added to typing in 3.8 diff --git a/python/ngen_cal/src/ngen/cal/utils.py b/python/ngen_cal/src/ngen/cal/utils.py index 209c5fa1..6e5ba045 100644 --- a/python/ngen_cal/src/ngen/cal/utils.py +++ b/python/ngen_cal/src/ngen/cal/utils.py @@ -12,7 +12,7 @@ from pydantic.typing import CallableGenerator @contextmanager -def pushd(path: 'Path') -> None: +def pushd(path: Path) -> None: """Change working directory to `path` for duration of the context Args: diff --git a/python/ngen_cal/tests/conftest.py b/python/ngen_cal/tests/conftest.py index 26703035..64881ce9 100644 --- a/python/ngen_cal/tests/conftest.py +++ b/python/ngen_cal/tests/conftest.py @@ -8,10 +8,10 @@ from ngen.cal.configuration import General from ngen.cal.ngen import Ngen from ngen.cal.meta import JobMeta -from ngen.cal.calibration_cathment import CalibrationCatchment, EvaluatableCatchment, AdjustableCatchment +from ngen.cal.calibration_cathment import CalibrationCatchment from ngen.cal.model import EvaluationOptions from ngen.cal.agent import Agent -from hypy import Nexus, HydroLocation +from hypy import Nexus from .utils import * @@ -151,12 +151,12 @@ def catchment(nexus, fabric, workdir, mocker) -> Generator[CalibrationCatchment, output = nexus._hydro_location.get_data().rename(columns={'value':'sim_flow'}) output.set_index('value_time', inplace=True) #Override the output property so it doesn't try to reload output each time - mocker.patch(__name__+'.EvaluatableCatchment.output', + mocker.patch('ngen.cal.calibration_cathment.EvaluatableCatchment.output', new_callable=mocker.PropertyMock, return_value = output ) #Disable output saving for testing purpose - mocker.patch(__name__+'.AdjustableCatchment.save_output', + mocker.patch('ngen.cal.calibration_cathment.AdjustableCatchment.save_output', return_value=None) id = 'tst-1' @@ -217,4 +217,4 @@ def explicit_catchments(nexus, fabric, workdir) -> Generator[ List[ CalibrationC id = f"tst-{i}" cat = CalibrationCatchment(workdir, id, nexus, start, end, fabric, 'Q_Out', eval_options, data) catchments.append(cat) - yield catchments \ No newline at end of file + yield catchments diff --git a/python/ngen_cal/tests/test_configuration.py b/python/ngen_cal/tests/test_configuration.py deleted file mode 100644 index 009df7cf..00000000 --- a/python/ngen_cal/tests/test_configuration.py +++ /dev/null @@ -1,30 +0,0 @@ -import pytest -from typing import TYPE_CHECKING, Generator - -from .utils import config - -if TYPE_CHECKING: - from ngen.cal.configuration import Configuration -""" - Test suite for reading and manipulating ngen configration files -""" - -# TODO determmine if any unit tests of General config are appropriate -# pydantic is thoroughly tested upstream, but it may be a good idea to -# write some simple tests around the defaults -# may also want to implment some testing around the Model union/parsing - -# @pytest.mark.usefixtures("conf", "realization_config") -# def test_config_file(conf: 'Configuration', realization_config: str) -> None: -# """ -# Test configuration property `config_file` -# """ -# assert conf.config_file == realization_config - -# @pytest.mark.usefixtures("conf") -# def test_catchments(conf: 'Configuration') -> None: -# """ -# Ensure that only the catchment marked with "calibration" is used in the configuration -# """ -# assert len(conf.catchments) == 1 -# assert conf.catchments[0].id == 'test-catchment' diff --git a/python/ngen_cal/tests/test_model.py b/python/ngen_cal/tests/test_model.py index 54d9ba16..2f189fc0 100644 --- a/python/ngen_cal/tests/test_model.py +++ b/python/ngen_cal/tests/test_model.py @@ -1,4 +1,3 @@ -import pytest from typing import TYPE_CHECKING from ngen.cal.ngen import Ngen from ngen.cal.meta import JobMeta diff --git a/python/ngen_cal/tests/test_objectives.py b/python/ngen_cal/tests/test_objectives.py index 7c5a6018..c9221d91 100644 --- a/python/ngen_cal/tests/test_objectives.py +++ b/python/ngen_cal/tests/test_objectives.py @@ -1,9 +1,6 @@ import pytest from math import inf -from typing import TYPE_CHECKING import pandas as pd # type: ignore -if TYPE_CHECKING: - from pandas import DataFrame from ngen.cal.objectives import * #A data frame of "perfectly simulated" data diff --git a/python/ngen_cal/tests/test_search.py b/python/ngen_cal/tests/test_search.py index b8e26514..a00ea346 100644 --- a/python/ngen_cal/tests/test_search.py +++ b/python/ngen_cal/tests/test_search.py @@ -4,7 +4,6 @@ from ngen.cal.search import dds if TYPE_CHECKING: - from ngen.cal.meta import CalibrationMeta from ngen.cal.calibration_cathment import CalibrationCatchment from ngen.cal.agent import Agent diff --git a/python/ngen_conf/src/ngen/config/all_formulations.py b/python/ngen_conf/src/ngen/config/all_formulations.py index b26b9e46..9db1a578 100644 --- a/python/ngen_conf/src/ngen/config/all_formulations.py +++ b/python/ngen_conf/src/ngen/config/all_formulations.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Union from ngen.config.cfe import CFE diff --git a/python/ngen_conf/src/ngen/config/bmi_formulation.py b/python/ngen_conf/src/ngen/config/bmi_formulation.py index 7c5e1e8b..7ee163b8 100644 --- a/python/ngen_conf/src/ngen/config/bmi_formulation.py +++ b/python/ngen_conf/src/ngen/config/bmi_formulation.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pydantic import BaseModel, DirectoryPath, PyObject, Field, root_validator, validator from typing import Mapping, Optional, Union, Sequence, Any from typing_extensions import TypeAlias @@ -40,7 +42,7 @@ class BMIParams(BaseModel, smart_union=True, allow_population_by_field_name = Tr name: str model_name: str = Field(alias='model_type_name') main_output_variable: str - config: Union[Path] = Field(alias='init_config') #Bmi config, can be a file or a str pattern + config: Path = Field(alias='init_config') #Bmi config, can be a file or a str pattern #reasonable defaultable fields allow_exceed_end_time: bool = False @@ -62,7 +64,7 @@ class BMIParams(BaseModel, smart_union=True, allow_population_by_field_name = Tr _config_prefix: Optional[DirectoryPath] = Field(default=None, alias="config_prefix") _output_map: Optional[Mapping[str, str]] = Field(None, alias="output_map") - def resolve_paths(self, relative_to: Optional[Path]=None): + def resolve_paths(self, relative_to: Path | None=None): """Resolve relative paths into absolute paths Args: @@ -183,7 +185,7 @@ class BMILib(BMIParams): #optional _library_prefix: Optional[DirectoryPath] = Field(None, alias="library_prefix") - def resolve_paths(self, relative_to: Optional[Path]=None): + def resolve_paths(self, relative_to: Path | None=None): super().resolve_paths(relative_to) if relative_to is None: self.library = self.library.resolve() diff --git a/python/ngen_conf/src/ngen/config/cfe.py b/python/ngen_conf/src/ngen/config/cfe.py index 29eea7cc..9951464a 100644 --- a/python/ngen_conf/src/ngen/config/cfe.py +++ b/python/ngen_conf/src/ngen/config/cfe.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Optional, Literal from pydantic import BaseModel, Field diff --git a/python/ngen_conf/src/ngen/config/configurations.py b/python/ngen_conf/src/ngen/config/configurations.py index b1b3256c..dc11db1d 100644 --- a/python/ngen_conf/src/ngen/config/configurations.py +++ b/python/ngen_conf/src/ngen/config/configurations.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from enum import Enum from datetime import datetime from pydantic import BaseModel, conint, Field @@ -20,9 +22,9 @@ class Provider(str, Enum): file_pattern: Optional[Union[Path, str]] path: Path #reasonable? default - provider: Provider = Field(Provider.CSV) + provider: Forcing.Provider = Field(Provider.CSV) - def resolve_paths(self, relative_to: Optional[Path]=None): + def resolve_paths(self, relative_to: Path | None=None): if isinstance(self.file_pattern, Path): if relative_to is None: self.file_pattern = self.file_pattern.resolve() @@ -60,7 +62,7 @@ class Routing(BaseModel): #optional/not used TODO make default None? path: Optional[str] = Field('', alias='t_route_connection_path') #TODO deprecate this field? - def resolve_paths(self, relative_to: Optional[Path]=None): + def resolve_paths(self, relative_to: Path | None=None): if relative_to is None: self.config = self.config.resolve() else: diff --git a/python/ngen_conf/src/ngen/config/formulation.py b/python/ngen_conf/src/ngen/config/formulation.py index b4a83440..994e48cb 100644 --- a/python/ngen_conf/src/ngen/config/formulation.py +++ b/python/ngen_conf/src/ngen/config/formulation.py @@ -1,7 +1,8 @@ +from __future__ import annotations + from pydantic import BaseModel, validator -from typing import Any, Dict, TYPE_CHECKING, Union +from typing import Any, TYPE_CHECKING if TYPE_CHECKING: - from typing import Optional from pathlib import Path class Formulation(BaseModel, smart_union=True): @@ -15,12 +16,12 @@ class Formulation(BaseModel, smart_union=True): """ #TODO make this an enum? name: str - params: "KnownFormulations" + params: KnownFormulations @validator("params", pre=True) def _validate_params( - cls, value: Union[Dict[str, Any], "KnownFormulations"] - ) -> Union[Dict[str, Any], "KnownFormulations"]: + cls, value: dict[str, Any] | KnownFormulations + ) -> dict[str, Any] | KnownFormulations: if isinstance(value, BaseModel): return value @@ -35,7 +36,7 @@ def _validate_params( raise ValueError("'name' and 'model_type_name' fields are required when deserializing _into_ a 'Formulation'.") return value - def resolve_paths(self, relative_to: 'Optional[Path]'=None): + def resolve_paths(self, relative_to: Path | None=None): self.params.resolve_paths(relative_to) #NOTE To avoid circular import and support recrusive modules diff --git a/python/ngen_conf/src/ngen/config/hydrofabric.py b/python/ngen_conf/src/ngen/config/hydrofabric.py index 483b5ada..95fe2152 100644 --- a/python/ngen_conf/src/ngen/config/hydrofabric.py +++ b/python/ngen_conf/src/ngen/config/hydrofabric.py @@ -1,5 +1,6 @@ +from __future__ import annotations -from typing import List, Literal, Optional +from typing import Optional from pydantic import BaseModel, validator from geojson_pydantic import FeatureCollection, Point, MultiPolygon import re diff --git a/python/ngen_conf/src/ngen/config/init_config/cfe.py b/python/ngen_conf/src/ngen/config/init_config/cfe.py index fcbdc5bc..b8e17521 100644 --- a/python/ngen_conf/src/ngen/config/init_config/cfe.py +++ b/python/ngen_conf/src/ngen/config/init_config/cfe.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import TYPE_CHECKING, List, Literal, Optional, Union from ngen.init_config import serializer_deserializer as serde @@ -73,7 +75,7 @@ def _coerce_lists(cls, value): return [x.strip() for x in value.split(",")] class Config(serde.IniSerializerDeserializer.Config): - def _serialize_list(l: List[float]) -> str: + def _serialize_list(l: list[float]) -> str: return ",".join(map(lambda x: str(x), l)) field_serializers = { @@ -157,14 +159,14 @@ class Config(serde.IniSerializerDeserializer.Config): def dict( self, *, - include: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, - exclude: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, + include: AbstractSetIntStr | MappingIntStrAny | None = None, + exclude: AbstractSetIntStr | MappingIntStrAny | None = None, by_alias: bool = False, - skip_defaults: Optional[bool] = None, + skip_defaults: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, - ) -> "DictStrAny": + ) -> DictStrAny: serial = super().dict( include=include, exclude=exclude, diff --git a/python/ngen_conf/src/ngen/config/init_config/lstm.py b/python/ngen_conf/src/ngen/config/init_config/lstm.py index 30db9775..2d08b927 100644 --- a/python/ngen_conf/src/ngen/config/init_config/lstm.py +++ b/python/ngen_conf/src/ngen/config/init_config/lstm.py @@ -1,5 +1,6 @@ +from __future__ import annotations + from ngen.init_config import serializer_deserializer as serde -from pydantic import Extra from pathlib import Path from typing import Optional, Literal diff --git a/python/ngen_conf/src/ngen/config/init_config/noahowp.py b/python/ngen_conf/src/ngen/config/init_config/noahowp.py index bb1a8caa..e5f66490 100644 --- a/python/ngen_conf/src/ngen/config/init_config/noahowp.py +++ b/python/ngen_conf/src/ngen/config/init_config/noahowp.py @@ -1,8 +1,10 @@ +from __future__ import annotations + import warnings from datetime import datetime from enum import Enum from pathlib import Path, PosixPath, WindowsPath -from typing import ClassVar, Dict, List, Literal, Union +from typing import ClassVar, List, Literal from ngen.init_config import core from ngen.init_config import serializer_deserializer as serde @@ -34,7 +36,7 @@ USGS_NVEG = 27 -def _set_nveg_based_on_veg_class_name(parameters: "Parameters", structure: "Structure"): +def _set_nveg_based_on_veg_class_name(parameters: Parameters, structure: Structure): # don't set if `nveg` is provided if structure.nveg is not None: return @@ -49,13 +51,13 @@ def _set_nveg_based_on_veg_class_name(parameters: "Parameters", structure: "Stru class NoahOWP(serde.NamelistSerializerDeserializer): - timing: "Timing" - parameters: "Parameters" - location: "Location" - forcing: "Forcing" - model_options: "ModelOptions" - structure: "Structure" - initial_values: "InitialValues" + timing: Timing + parameters: Parameters + location: Location + forcing: Forcing + model_options: ModelOptions + structure: Structure + initial_values: InitialValues class Config(serde.NamelistSerializerDeserializer.Config): # NOTE: must explicitly specify Path subtype (i.e. `PosixPath`) here b.c. @@ -68,7 +70,7 @@ class Config(serde.NamelistSerializerDeserializer.Config): } @root_validator - def _validate(cls, values: Dict[str, BaseModel]) -> Dict[str, BaseModel]: + def _validate(cls, values: dict[str, BaseModel]) -> dict[str, BaseModel]: parameters: Parameters = values["parameters"] # type: ignore structure: Structure = values["structure"] # type: ignore _set_nveg_based_on_veg_class_name(parameters, structure) @@ -92,7 +94,7 @@ class Timing(core.Base): )(validate_str_len_lt(257)) @validator("startdate", "enddate", pre=True) - def _validate_dates(cls, value: Union[datetime, str]): + def _validate_dates(cls, value: datetime | str): if isinstance(value, datetime): return value return datetime.strptime(value, cls._datetime_format) @@ -272,7 +274,7 @@ class Config(core.Base.Config): def _warn_if_soil_or_veg_type_is_water_but_not_both( - parameters: "Parameters", structure: "Structure" + parameters: Parameters, structure: Structure ): if parameters.soil_class_name not in ("STAS", "STAS-RUC"): return diff --git a/python/ngen_conf/src/ngen/config/init_config/noahowp_options.py b/python/ngen_conf/src/ngen/config/init_config/noahowp_options.py index 13f5245d..b84b2c5d 100644 --- a/python/ngen_conf/src/ngen/config/init_config/noahowp_options.py +++ b/python/ngen_conf/src/ngen/config/init_config/noahowp_options.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from enum import Enum diff --git a/python/ngen_conf/src/ngen/config/init_config/pet.py b/python/ngen_conf/src/ngen/config/init_config/pet.py index eef1813c..2a8fc6ae 100644 --- a/python/ngen_conf/src/ngen/config/init_config/pet.py +++ b/python/ngen_conf/src/ngen/config/init_config/pet.py @@ -1,5 +1,7 @@ +from __future__ import annotations + from enum import Enum -from typing import Literal, Union +from typing import Literal from ngen.init_config import serializer_deserializer as serde from pydantic import validator @@ -66,8 +68,8 @@ class PET( @validator("pet_method", pre=True) def _coerce_pet_method( - cls, value: Union[str, int, PetMethod] - ) -> Union[int, PetMethod]: + cls, value: str | int | PetMethod + ) -> int | PetMethod: if isinstance(value, (PetMethod, int)): return value return int(value) diff --git a/python/ngen_conf/src/ngen/config/init_config/utils.py b/python/ngen_conf/src/ngen/config/init_config/utils.py index 45fcd834..9c517314 100644 --- a/python/ngen_conf/src/ngen/config/init_config/utils.py +++ b/python/ngen_conf/src/ngen/config/init_config/utils.py @@ -1,16 +1,13 @@ +from __future__ import annotations + import re from enum import Enum from typing import ( TYPE_CHECKING, Any, - Dict, Generic, - Optional, - Tuple, - Type, TypeVar, - Union, ) from pydantic.generics import GenericModel @@ -31,7 +28,7 @@ _FLOAT_UNIT_PAIR_RE = re.compile(_FLOAT_UNIT_PAIR_RE_PATTERN) -def _parse_float_unit_str(s: str) -> Tuple[float, str]: +def _parse_float_unit_str(s: str) -> tuple[float, str]: match = _FLOAT_UNIT_PAIR_RE.search(s) if match is None: @@ -53,7 +50,7 @@ class FloatUnitPair(GenericModel, Generic[L]): unit: L @classmethod - def __modify_schema__(cls, field_schema: Dict[str, Any]): + def __modify_schema__(cls, field_schema: dict[str, Any]): field_schema.update( type="string", pattern=_FLOAT_UNIT_PAIR_RE_PATTERN, @@ -61,7 +58,7 @@ def __modify_schema__(cls, field_schema: Dict[str, Any]): ) @classmethod - def validate(cls: Type[Self], value: Any) -> Self: + def validate(cls: type[Self], value: Any) -> Self: if isinstance(value, FloatUnitPair): return value @@ -85,14 +82,14 @@ def __str__(self): def dict( self, *, - include: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, - exclude: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, + include: AbstractSetIntStr | MappingIntStrAny | None = None, + exclude: AbstractSetIntStr | MappingIntStrAny | None = None, by_alias: bool = False, - skip_defaults: Optional[bool] = None, + skip_defaults: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, - ) -> "DictStrAny": + ) -> DictStrAny: return str(self) diff --git a/python/ngen_conf/src/ngen/config/init_config/validators.py b/python/ngen_conf/src/ngen/config/init_config/validators.py index e711d475..e6bdaf42 100644 --- a/python/ngen_conf/src/ngen/config/init_config/validators.py +++ b/python/ngen_conf/src/ngen/config/init_config/validators.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import sys if sys.version_info >= (3, 8): diff --git a/python/ngen_conf/src/ngen/config/init_config/value_unit_pair.py b/python/ngen_conf/src/ngen/config/init_config/value_unit_pair.py index e284c524..6475959e 100644 --- a/python/ngen_conf/src/ngen/config/init_config/value_unit_pair.py +++ b/python/ngen_conf/src/ngen/config/init_config/value_unit_pair.py @@ -1,5 +1,7 @@ +from __future__ import annotations + import re -from typing import TYPE_CHECKING, Any, Generic, List, Optional, Type, TypeVar, Union +from typing import TYPE_CHECKING, Any, Generic, List, TypeVar from pydantic import validator from pydantic.generics import GenericModel @@ -23,7 +25,7 @@ class ValueUnitPair(GenericModel, Generic[V, U]): @override @classmethod - def validate(cls: Type[Self], value: Any) -> Self: + def validate(cls: type[Self], value: Any) -> Self: if isinstance(value, ValueUnitPair): # return a shallow copy. this also validates / coerces mismatching generic types return cls(value=value.value, unit=value.unit) @@ -53,7 +55,7 @@ def validate(cls: Type[Self], value: Any) -> Self: @override @classmethod - def parse_obj(cls: Type[Self], obj: Any) -> Self: + def parse_obj(cls: type[Self], obj: Any) -> Self: return cls.validate(obj) def _serialize(self) -> str: @@ -63,14 +65,14 @@ def _serialize(self) -> str: def dict( self, *, - include: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, - exclude: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, + include: AbstractSetIntStr | MappingIntStrAny | None = None, + exclude: AbstractSetIntStr | MappingIntStrAny | None = None, by_alias: bool = False, - skip_defaults: Optional[bool] = None, + skip_defaults: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, - ) -> "DictStrAny": + ) -> DictStrAny: return self._serialize() @@ -81,7 +83,7 @@ def dict( class ListUnitPair(ValueUnitPair[List[T], U], Generic[T, U]): @validator("value", pre=True) - def _coerce_values(cls, value: Union[str, List[str]]) -> List[str]: + def _coerce_values(cls, value: str | list[str]) -> list[str]: if isinstance(value, list): return value if not isinstance(value, str): diff --git a/python/ngen_conf/src/ngen/config/lgar.py b/python/ngen_conf/src/ngen/config/lgar.py index 8dbb125e..cf183c4f 100644 --- a/python/ngen_conf/src/ngen/config/lgar.py +++ b/python/ngen_conf/src/ngen/config/lgar.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pydantic import Field from .bmi_formulation import BMICxx diff --git a/python/ngen_conf/src/ngen/config/lstm.py b/python/ngen_conf/src/ngen/config/lstm.py index c41ffe0a..14f53fe6 100644 --- a/python/ngen_conf/src/ngen/config/lstm.py +++ b/python/ngen_conf/src/ngen/config/lstm.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pydantic import PyObject, Field from typing import Literal, Union from .bmi_formulation import BMIPython diff --git a/python/ngen_conf/src/ngen/config/multi.py b/python/ngen_conf/src/ngen/config/multi.py index ee5f5fba..fb7c88c9 100644 --- a/python/ngen_conf/src/ngen/config/multi.py +++ b/python/ngen_conf/src/ngen/config/multi.py @@ -1,5 +1,6 @@ +from __future__ import annotations + from typing import Sequence, Mapping, Any, Optional, TYPE_CHECKING -from typing_extensions import Literal if TYPE_CHECKING: from pathlib import Path @@ -15,7 +16,7 @@ class MultiBMI(BMIParams, smart_union=True): #required #Due to a recursive formulation definition, have to postpone this #type definition and use `update_forward_refs` - modules: Sequence["Formulation"] + modules: Sequence[Formulation] #defaults name: str = Field("bmi_multi", const=True) @@ -32,7 +33,7 @@ class MultiBMI(BMIParams, smart_union=True): name_map: Mapping[str, str] = Field(None, const=True) #not relevant for multi-bmi model_params: Optional[Mapping[str, str]] = Field(None, const=True) #not relevant for multi-bmi - def resolve_paths(self, relative_to: Optional['Path']=None): + def resolve_paths(self, relative_to: Path | None=None): for m in self.modules: m.resolve_paths(relative_to) diff --git a/python/ngen_conf/src/ngen/config/noahowp.py b/python/ngen_conf/src/ngen/config/noahowp.py index d197cb58..c81e68b5 100644 --- a/python/ngen_conf/src/ngen/config/noahowp.py +++ b/python/ngen_conf/src/ngen/config/noahowp.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Literal from pydantic import BaseModel, Field diff --git a/python/ngen_conf/src/ngen/config/path_pair/_abc_mixins.py b/python/ngen_conf/src/ngen/config/path_pair/_abc_mixins.py index 665c395e..e5d7a36b 100644 --- a/python/ngen_conf/src/ngen/config/path_pair/_abc_mixins.py +++ b/python/ngen_conf/src/ngen/config/path_pair/_abc_mixins.py @@ -1,6 +1,8 @@ +from __future__ import annotations + from abc import ABC, abstractmethod -from typing import Generic, Iterable, Optional +from typing import Generic, Iterable from typing_extensions import Self from .typing import StrPath, T @@ -19,7 +21,7 @@ class AbstractPathPairMixin(ABC, Generic[T]): @property @abstractmethod - def inner(self) -> Optional[T]: + def inner(self) -> T | None: """Return the inner object if it exists.""" @abstractmethod @@ -34,7 +36,7 @@ def with_path(self, *args: StrPath) -> Self: """ @abstractmethod - def serialize(self) -> Optional[bytes]: + def serialize(self) -> bytes | None: """ If the inner `T` exists, return a serialized version. @@ -141,7 +143,7 @@ def serialize(self) -> Iterable[bytes]: @abstractmethod def deserialize( - self, data: Iterable[bytes], *, paths: Optional[Iterable[StrPath]] = None + self, data: Iterable[bytes], *, paths: Iterable[StrPath] | None = None ) -> bool: """ Deserialize iterable of bytes into `T`'s and wrap each `T` as a `PathPair[T]`. Replace diff --git a/python/ngen_conf/src/ngen/config/path_pair/_mixins.py b/python/ngen_conf/src/ngen/config/path_pair/_mixins.py index 950ab7df..da7ed9fe 100644 --- a/python/ngen_conf/src/ngen/config/path_pair/_mixins.py +++ b/python/ngen_conf/src/ngen/config/path_pair/_mixins.py @@ -1,10 +1,12 @@ +from __future__ import annotations + from itertools import zip_longest from pathlib import Path from ._abc_mixins import AbstractPathPairMixin, AbstractPathPairCollectionMixin from ._utils import path_unlink_37 -from typing import Optional, Iterable +from typing import Iterable from typing_extensions import Self from .typing import StrPath, T @@ -21,7 +23,7 @@ def parent(self) -> Path: return Path(self).parent @property - def inner(self) -> Optional[T]: + def inner(self) -> T | None: return self._inner def with_path(self, *args: StrPath) -> Self: @@ -34,7 +36,7 @@ def with_path(self, *args: StrPath) -> Self: deserializer=self._deserializer, ) - def serialize(self) -> Optional[bytes]: + def serialize(self) -> bytes | None: if self._serializer is None or self._inner is None: return None @@ -129,7 +131,7 @@ def serialize(self) -> Iterable[bytes]: yield self._serializer(item) def deserialize( - self, data: Iterable[bytes], *, paths: Optional[Iterable[StrPath]] = None + self, data: Iterable[bytes], *, paths: Iterable[StrPath] | None = None ) -> bool: """ Deserialize collection of bytes into T's and wrap each T as a `PathPair[T]`. Replace diff --git a/python/ngen_conf/src/ngen/config/path_pair/_utils.py b/python/ngen_conf/src/ngen/config/path_pair/_utils.py index de00f1ba..3edaa6e5 100644 --- a/python/ngen_conf/src/ngen/config/path_pair/_utils.py +++ b/python/ngen_conf/src/ngen/config/path_pair/_utils.py @@ -1,5 +1,10 @@ +from __future__ import annotations + import sys -from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path def path_unlink_37(p: Path, missing_ok: bool): diff --git a/python/ngen_conf/src/ngen/config/path_pair/common.py b/python/ngen_conf/src/ngen/config/path_pair/common.py index df5397ec..c505c39c 100644 --- a/python/ngen_conf/src/ngen/config/path_pair/common.py +++ b/python/ngen_conf/src/ngen/config/path_pair/common.py @@ -1,9 +1,13 @@ +from __future__ import annotations + from pydantic import BaseModel -from pathlib import Path from .protocol import Deserializer -from typing import Type, TypeVar +from typing import TypeVar, TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path M = TypeVar("M", bound=BaseModel) @@ -20,7 +24,7 @@ def pydantic_serializer(o: BaseModel) -> bytes: return o.json(by_alias=True).encode() -def pydantic_deserializer(m: Type[M]) -> Deserializer[M]: +def pydantic_deserializer(m: type[M]) -> Deserializer[M]: def deserialize(data: bytes) -> M: return m.parse_raw(data) diff --git a/python/ngen_conf/src/ngen/config/path_pair/path_pair.py b/python/ngen_conf/src/ngen/config/path_pair/path_pair.py index 365e1c31..5cc6528d 100644 --- a/python/ngen_conf/src/ngen/config/path_pair/path_pair.py +++ b/python/ngen_conf/src/ngen/config/path_pair/path_pair.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import os from pathlib import Path, PosixPath, WindowsPath @@ -6,7 +8,7 @@ from ._abc_mixins import AbstractPathPairMixin, AbstractPathPairCollectionMixin from ._mixins import PathPairMixin, PathPairCollectionMixin -from typing import Any, Dict, Generic, Optional, Union, List +from typing import Any, Generic from typing_extensions import Self from .typing import StrPath, T @@ -31,15 +33,15 @@ class PathPair(AbstractPathPairMixin[T], Path, Generic[T]): def __new__( cls, *args: StrPath, - inner: Optional[T] = None, + inner: T | None = None, reader: Reader = path_reader, writer: Writer = path_writer, - serializer: Optional[Serializer[T]] = None, - deserializer: Optional[Deserializer[T]] = None, + serializer: Serializer[T] | None = None, + deserializer: Deserializer[T] | None = None, **kwargs: Any, - ) -> Union["WindowsPathPair[T]", "PosixPathPair[T]"]: + ) -> WindowsPathPair[T] | PosixPathPair[T]: cls = WindowsPathPair[T] if os.name == "nt" else PosixPathPair[T] - self: Union[WindowsPathPair[T], PosixPathPair[T]] = Path.__new__( + self: WindowsPathPair[T] | PosixPathPair[T] = Path.__new__( cls, *args, **kwargs ) self._inner = inner @@ -57,9 +59,9 @@ def with_object( path: StrPath = "", reader: Reader = path_reader, writer: Writer = path_writer, - serializer: Optional[Serializer[T]] = None, - deserializer: Optional[Deserializer[T]] = None, - ) -> Union["PosixPathPair[T]", "WindowsPathPair[T]"]: + serializer: Serializer[T] | None = None, + deserializer: Deserializer[T] | None = None, + ) -> PosixPathPair[T] | WindowsPathPair[T]: return cls( Path(path), inner=obj, @@ -74,7 +76,7 @@ def __get_validators__(cls): yield cls.validate @classmethod - def __modify_schema__(cls, field_schema: Dict[str, Any]): + def __modify_schema__(cls, field_schema: dict[str, Any]): field_schema.clear() field_schema["format"] = "path" field_schema["type"] = "string" @@ -110,13 +112,13 @@ def __new__( cls, *args: StrPath, pattern: str, - inner: Optional[List["PosixPathPair[T]"]] = None, + inner: list[PosixPathPair[T]] | None = None, reader: Reader = path_reader, writer: Writer = path_writer, - serializer: Optional[Serializer[T]] = None, - deserializer: Optional[Deserializer[T]] = None, + serializer: Serializer[T] | None = None, + deserializer: Deserializer[T] | None = None, **kwargs: Any, - ) -> Union["WindowsPathPairCollection[T]", "PosixPathPairCollection[T]"]: + ) -> WindowsPathPairCollection[T] | PosixPathPairCollection[T]: cls = ( WindowsPathPairCollection[T] if os.name == "nt" @@ -135,9 +137,9 @@ def __new__( f"Filename not derived from template and pattern, {template_str.name!r} {pattern!r}: {item}" ) - self: Union[ - WindowsPathPairCollection[T], PosixPathPairCollection[T] - ] = Path.__new__( + self: ( + WindowsPathPairCollection[T] | PosixPathPairCollection[T] + ) = Path.__new__( cls, *args, inner=inner, @@ -157,7 +159,7 @@ def __new__( @staticmethod def _get_id( - p: "PosixPathPair[T]", + p: PosixPathPair[T], prefix: str, suffix: str, ) -> str: @@ -176,16 +178,16 @@ def home(cls) -> Path: @classmethod def with_objects( cls, - objs: List[T], + objs: list[T], *, path: Path, pattern: str, - ids: List[str], - reader: Optional[Reader] = path_reader, - writer: Optional[Writer] = path_writer, - serializer: Optional[Serializer[T]] = None, - deserializer: Optional[Deserializer[T]] = None, - ) -> Union["PosixPathPairCollection[T]", "WindowsPathPairCollection[T]"]: + ids: list[str], + reader: Reader | None = path_reader, + writer: Writer | None = path_writer, + serializer: Serializer[T] | None = None, + deserializer: Deserializer[T] | None = None, + ) -> PosixPathPairCollection[T] | WindowsPathPairCollection[T]: assert len(objs) == len(ids) prefix, _, suffix = path.name.partition(pattern) assert prefix != "" and suffix != "" @@ -218,7 +220,7 @@ def __get_validators__(cls): yield cls.validate @classmethod - def __modify_schema__(cls, field_schema: Dict[str, Any]): + def __modify_schema__(cls, field_schema: dict[str, Any]): field_schema.clear() field_schema["format"] = "path" field_schema["type"] = "string" diff --git a/python/ngen_conf/src/ngen/config/path_pair/protocol.py b/python/ngen_conf/src/ngen/config/path_pair/protocol.py index b1a46ce6..1e8d6ebe 100644 --- a/python/ngen_conf/src/ngen/config/path_pair/protocol.py +++ b/python/ngen_conf/src/ngen/config/path_pair/protocol.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import sys if sys.version_info >= (3, 8): @@ -5,9 +7,12 @@ else: from typing_extensions import Protocol -from pathlib import Path from .typing import T, S +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path class Writer(Protocol): diff --git a/python/ngen_conf/src/ngen/config/path_pair/typing.py b/python/ngen_conf/src/ngen/config/path_pair/typing.py index 53d3f6e2..79bfa68c 100644 --- a/python/ngen_conf/src/ngen/config/path_pair/typing.py +++ b/python/ngen_conf/src/ngen/config/path_pair/typing.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pathlib import Path from typing import Union, TypeVar from typing_extensions import TypeAlias diff --git a/python/ngen_conf/src/ngen/config/pet.py b/python/ngen_conf/src/ngen/config/pet.py index b9481764..44ae8e0a 100644 --- a/python/ngen_conf/src/ngen/config/pet.py +++ b/python/ngen_conf/src/ngen/config/pet.py @@ -1,5 +1,7 @@ -from pydantic import PyObject, Field -from typing import Literal, Union +from __future__ import annotations + +from pydantic import Field +from typing import Literal from .bmi_formulation import BMIC class PET(BMIC): diff --git a/python/ngen_conf/src/ngen/config/realization.py b/python/ngen_conf/src/ngen/config/realization.py index f6aa7460..590b2d4a 100644 --- a/python/ngen_conf/src/ngen/config/realization.py +++ b/python/ngen_conf/src/ngen/config/realization.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from pydantic import BaseModel, Field from typing import Optional, Mapping, Sequence, Any from pathlib import Path @@ -13,7 +15,7 @@ class Realization(BaseModel): forcing: Forcing calibration: Optional[ Mapping[ str, Sequence[ Any ]] ] - def resolve_paths(self, relative_to: Optional['Path']=None): + def resolve_paths(self, relative_to: Path | None=None): for f in self.formulations: f.resolve_paths(relative_to) if self.forcing : @@ -43,7 +45,7 @@ class Config: datetime: lambda v: v.strftime("%Y-%m-%d %H:%M:%S") } - def resolve_paths(self, relative_to: Optional['Path']=None): + def resolve_paths(self, relative_to: Path | None=None): """resolve possible relative paths in configuration """ self.global_config.resolve_paths(relative_to) diff --git a/python/ngen_conf/src/ngen/config/sloth.py b/python/ngen_conf/src/ngen/config/sloth.py index cc9cad1b..478d4bbf 100644 --- a/python/ngen_conf/src/ngen/config/sloth.py +++ b/python/ngen_conf/src/ngen/config/sloth.py @@ -1,5 +1,7 @@ +from __future__ import annotations + from typing import Literal, Optional, Mapping -from pydantic import BaseModel, Field +from pydantic import Field from .bmi_formulation import BMICxx diff --git a/python/ngen_conf/src/ngen/config/soil_freeze_thaw.py b/python/ngen_conf/src/ngen/config/soil_freeze_thaw.py index 3a51f459..94faeb22 100644 --- a/python/ngen_conf/src/ngen/config/soil_freeze_thaw.py +++ b/python/ngen_conf/src/ngen/config/soil_freeze_thaw.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Optional from pydantic import BaseModel, Field diff --git a/python/ngen_conf/src/ngen/config/soil_moisture_profile.py b/python/ngen_conf/src/ngen/config/soil_moisture_profile.py index c4ddf726..c08cfcff 100644 --- a/python/ngen_conf/src/ngen/config/soil_moisture_profile.py +++ b/python/ngen_conf/src/ngen/config/soil_moisture_profile.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Optional from pydantic import BaseModel, Field diff --git a/python/ngen_conf/src/ngen/config/topmod.py b/python/ngen_conf/src/ngen/config/topmod.py index 614dfe57..ab0a316f 100644 --- a/python/ngen_conf/src/ngen/config/topmod.py +++ b/python/ngen_conf/src/ngen/config/topmod.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Optional, Literal from pydantic import BaseModel, Field diff --git a/python/ngen_conf/src/ngen/config/utils.py b/python/ngen_conf/src/ngen/config/utils.py index 91847bb4..f2867c6c 100644 --- a/python/ngen_conf/src/ngen/config/utils.py +++ b/python/ngen_conf/src/ngen/config/utils.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from contextlib import contextmanager from os import getcwd, chdir from typing import TYPE_CHECKING @@ -5,7 +7,7 @@ from pathlib import Path @contextmanager -def pushd(path: 'Path') -> None: +def pushd(path: Path) -> None: """Change working directory to `path` for duration of the context Args: diff --git a/python/ngen_conf/src/ngen/config/validate.py b/python/ngen_conf/src/ngen/config/validate.py index 84cb3833..5c7d8d65 100644 --- a/python/ngen_conf/src/ngen/config/validate.py +++ b/python/ngen_conf/src/ngen/config/validate.py @@ -1,10 +1,11 @@ +from __future__ import annotations + from pydantic import BaseModel from pathlib import Path -from typing import List class MissingPath: - def __init__(self, models: List[BaseModel], name: str, value: Path): + def __init__(self, models: list[BaseModel], name: str, value: Path): """Chain of pydantic model instances the last of which `name` and `value` properties belong""" self.models = models """Pydantic model instance on which `name` and `value` properties belong""" @@ -26,16 +27,16 @@ def __str__(self) -> str: ) -def validate_paths(m: BaseModel) -> List[MissingPath]: +def validate_paths(m: BaseModel) -> list[MissingPath]: """ Recursively walk a pydantic model's fields and return a list of MissingPath instances for each encountered `pathlib.Path` instance that does not exist. The empty list means any `pathlib.Path` instances that were encountered exist. """ - paths_that_dont_exist: List[MissingPath] = [] + paths_that_dont_exist: list[MissingPath] = [] - def rec(mod: BaseModel, visited: List[BaseModel]): + def rec(mod: BaseModel, visited: list[BaseModel]): for f_name in mod.__fields__.keys(): f_value = getattr(mod, f_name) if isinstance(f_value, Path): diff --git a/python/ngen_conf/tests/test_multi.py b/python/ngen_conf/tests/test_multi.py index 8938443e..1e4857e5 100644 --- a/python/ngen_conf/tests/test_multi.py +++ b/python/ngen_conf/tests/test_multi.py @@ -1,4 +1,3 @@ -import pytest from ngen.config.formulation import Formulation from ngen.config.multi import MultiBMI diff --git a/python/ngen_conf/tests/test_noahowp.py b/python/ngen_conf/tests/test_noahowp.py index 0cf6db8f..8a6db140 100644 --- a/python/ngen_conf/tests/test_noahowp.py +++ b/python/ngen_conf/tests/test_noahowp.py @@ -1,4 +1,3 @@ -import pytest from ngen.config.formulation import Formulation from ngen.config.noahowp import NoahOWP diff --git a/python/ngen_conf/tests/test_realization.py b/python/ngen_conf/tests/test_realization.py index 872d4532..a2cc0cbc 100644 --- a/python/ngen_conf/tests/test_realization.py +++ b/python/ngen_conf/tests/test_realization.py @@ -3,7 +3,6 @@ from ngen.config.realization import Realization, NgenRealization from ngen.config.formulation import Formulation -import json @pytest.mark.parametrize("forcing",["csv", "netcdf"], indirect=True ) def test_realization(forcing, time, cfe): diff --git a/python/ngen_config_gen/src/ngen/config_gen/__init__.py b/python/ngen_config_gen/src/ngen/config_gen/__init__.py index 8dee4bf8..4aef7df1 100644 --- a/python/ngen_config_gen/src/ngen/config_gen/__init__.py +++ b/python/ngen_config_gen/src/ngen/config_gen/__init__.py @@ -1 +1,3 @@ +from __future__ import annotations + from ._version import __version__ diff --git a/python/ngen_config_gen/src/ngen/config_gen/file_writer.py b/python/ngen_config_gen/src/ngen/config_gen/file_writer.py index 0544acc2..975d76f9 100644 --- a/python/ngen_config_gen/src/ngen/config_gen/file_writer.py +++ b/python/ngen_config_gen/src/ngen/config_gen/file_writer.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import dataclasses import enum import hashlib @@ -20,7 +22,7 @@ class FileWriter(Protocol): - def __call__(self, id: Union[str, Literal["global"]], data: BaseModel): ... + def __call__(self, id: str | Literal["global"], data: BaseModel): ... class _Reader(Protocol): @@ -29,7 +31,7 @@ class _Reader(Protocol): EOF is empty bytes buffer of empty str. """ - def read(self, size: Union[int, None] = ...) -> Union[bytes, str]: ... + def read(self, size: int | None = ...) -> bytes | str: ... def _sha256_hexdigest(r: _Reader) -> str: @@ -103,7 +105,7 @@ def _get_file_extension(data: BaseModel) -> str: class DefaultFileWriter: - def __init__(self, root: Union[str, Path]): + def __init__(self, root: str | Path): root = Path(root) if not root.exists(): root.mkdir(parents=True) @@ -122,7 +124,7 @@ def _gen_alt_filename(p: Path) -> Path: i += 1 return f_name - def __call__(self, id: Union[str, Literal["global"]], data: BaseModel): + def __call__(self, id: str | Literal["global"], data: BaseModel): class_name = data.__class__.__name__ ext = _get_file_extension(data) output_file = self.__root / f"{class_name}_{id}.{ext}" @@ -206,7 +208,7 @@ def __exit__(self, exc_type, exc_value, traceback) -> None: self._filehandle.close() self._filehandle = None - def __call__(self, id: Union[str, Literal["global"]], data: BaseModel) -> None: + def __call__(self, id: str | Literal["global"], data: BaseModel) -> None: class_name = data.__class__.__name__ ext = _get_file_extension(data) output_file = f"{class_name}_{id}.{ext}" diff --git a/python/ngen_config_gen/src/ngen/config_gen/generate.py b/python/ngen_config_gen/src/ngen/config_gen/generate.py index d1268af3..48d47ad9 100644 --- a/python/ngen_config_gen/src/ngen/config_gen/generate.py +++ b/python/ngen_config_gen/src/ngen/config_gen/generate.py @@ -1,4 +1,6 @@ -from typing import Any, Collection, Dict, Iterable, Protocol, Union, TYPE_CHECKING +from __future__ import annotations + +from typing import Any, Collection, Iterable, Protocol, TYPE_CHECKING if TYPE_CHECKING: from .hook_providers import HookProvider @@ -9,17 +11,17 @@ class DivideIdHookObject: def __init__(self): - self.__divide_id: Union[str, None] = None + self.__divide_id: str | None = None def hydrofabric_hook( - self, version: str, divide_id: str, data: Dict[str, Any] + self, version: str, divide_id: str, data: dict[str, Any] ) -> None: self.__divide_id = divide_id - def visit(self, hook_provider: "HookProvider") -> None: + def visit(self, hook_provider: HookProvider) -> None: hook_provider.provide_hydrofabric_data(self) - def divide_id(self) -> Union[str, None]: + def divide_id(self) -> str | None: return self.__divide_id @@ -31,7 +33,7 @@ def __call__(self) -> BuilderVisitable: def generate_configs( - hook_providers: Iterable["HookProvider"], + hook_providers: Iterable[HookProvider], hook_objects: Collection[BuilderVisitableFn], file_writer: FileWriter, ): diff --git a/python/ngen_config_gen/src/ngen/config_gen/hook_providers.py b/python/ngen_config_gen/src/ngen/config_gen/hook_providers.py index 8f646069..54676bd3 100644 --- a/python/ngen_config_gen/src/ngen/config_gen/hook_providers.py +++ b/python/ngen_config_gen/src/ngen/config_gen/hook_providers.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, Protocol, Union, runtime_checkable +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable from typing_extensions import Self import geopandas as gpd @@ -52,8 +54,8 @@ def __init__(self, hf: gpd.GeoDataFrame, hf_lnk_data: pd.DataFrame): self.hf_iter = self.__hf.iterrows() self.hf_lnk_iter = self.__hf_lnk.iterrows() - self.hf_row: Union[Dict[str, Any], None] = None - self.hf_lnk_row: Union[Dict[str, Any], None] = None + self.hf_row: dict[str, Any] | None = None + self.hf_lnk_row: dict[str, Any] | None = None def provide_hydrofabric_data(self, hook: HydrofabricHook): if self.hf_row is None: diff --git a/python/ngen_config_gen/src/ngen/config_gen/hooks.py b/python/ngen_config_gen/src/ngen/config_gen/hooks.py index 036d337a..35432726 100644 --- a/python/ngen_config_gen/src/ngen/config_gen/hooks.py +++ b/python/ngen_config_gen/src/ngen/config_gen/hooks.py @@ -50,7 +50,9 @@ For a more hands on experience, see example implementations: https://github.com/NOAA-OWP/ngen-cal/tree/master/python/ngen_config_gen/examples """ -from typing import Any, Dict, Protocol, runtime_checkable, TYPE_CHECKING +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable, TYPE_CHECKING from pydantic import BaseModel if TYPE_CHECKING: @@ -69,7 +71,7 @@ def build(self) -> BaseModel: @runtime_checkable class Visitable(Protocol): - def visit(self, hook_provider: "HookProvider") -> None: + def visit(self, hook_provider: HookProvider) -> None: """ Classes that implement `visit` are assumed to also implement some or all hook methods (e.g. `hydrofabric_hook`). Classes that implement `visit` should call associated hook provider methods for each hook @@ -117,7 +119,7 @@ class HydrofabricHook(Protocol): """ def hydrofabric_hook( - self, version: str, divide_id: str, data: Dict[str, Any] + self, version: str, divide_id: str, data: dict[str, Any] ) -> None: """ Expect to receive a hydrofabric version, data (see class docs), and the associated divide_id. @@ -175,7 +177,7 @@ class HydrofabricLinkedDataHook(Protocol): """ def hydrofabric_linked_data_hook( - self, version: str, divide_id: str, data: Dict[str, Any] + self, version: str, divide_id: str, data: dict[str, Any] ) -> None: """ Expect to receive a hydrofabric linked data version, data (see class docs), and the associated divide_id. diff --git a/python/ngen_config_gen/src/ngen/config_gen/models/cfe.py b/python/ngen_config_gen/src/ngen/config_gen/models/cfe.py index b25b4f5d..039972aa 100644 --- a/python/ngen_config_gen/src/ngen/config_gen/models/cfe.py +++ b/python/ngen_config_gen/src/ngen/config_gen/models/cfe.py @@ -1,5 +1,7 @@ -from typing import Any, Dict, List, Union -from typing_extensions import TYPE_CHECKING +from __future__ import annotations + +from typing import Any +from typing import TYPE_CHECKING if TYPE_CHECKING: from ..hook_providers import HookProvider @@ -43,10 +45,10 @@ class Cfe: """ def __init__(self): - self.data: Dict[str, Union[FloatUnitPair[str], List[float]]] = {} + self.data: dict[str, FloatUnitPair[str] | list[float]] = {} def hydrofabric_linked_data_hook( - self, version: str, divide_id: str, data: Dict[str, Any] + self, version: str, divide_id: str, data: dict[str, Any] ) -> None: """ Implements `ngen.config_gen.hooks.hydrofabric_linked_data_hook`. @@ -141,7 +143,7 @@ def build(self) -> BaseModel: """ return CFEConfig(__root__=self.data) - def visit(self, hook_provider: "HookProvider") -> None: + def visit(self, hook_provider: HookProvider) -> None: """ Call associated `hook_provider` methods for all hooks implemented by Self. """ diff --git a/python/ngen_config_gen/src/ngen/config_gen/models/pet.py b/python/ngen_config_gen/src/ngen/config_gen/models/pet.py index 00f2c630..40bcad5c 100644 --- a/python/ngen_config_gen/src/ngen/config_gen/models/pet.py +++ b/python/ngen_config_gen/src/ngen/config_gen/models/pet.py @@ -1,4 +1,6 @@ -from typing import Any, Dict, TYPE_CHECKING, Union +from __future__ import annotations + +from typing import Any, TYPE_CHECKING from pydantic import BaseModel if TYPE_CHECKING: @@ -28,11 +30,11 @@ class Pet: """ def __init__(self, method: PetMethod = PetMethod.energy_balance): - self.data: Dict[str, Union[bool, float, int, str]] = {} + self.data: dict[str, bool | float | int | str] = {} self.__pet_method = method def hydrofabric_linked_data_hook( - self, version: str, divide_id: str, data: Dict[str, Any] + self, version: str, divide_id: str, data: dict[str, Any] ) -> None: """ Implements `ngen.config_gen.hooks.hydrofabric_linked_data_hook`. @@ -78,7 +80,7 @@ def build(self) -> BaseModel: """ return PetConfig(**self.data) - def visit(self, hook_provider: "HookProvider") -> None: + def visit(self, hook_provider: HookProvider) -> None: """ Call associated `hook_provider` methods for all hooks implemented by Self. diff --git a/python/ngen_init_config/src/ngen/init_config/_deserializers.py b/python/ngen_init_config/src/ngen/init_config/_deserializers.py index c6d31e05..f9837a2f 100644 --- a/python/ngen_init_config/src/ngen/init_config/_deserializers.py +++ b/python/ngen_init_config/src/ngen/init_config/_deserializers.py @@ -1,13 +1,15 @@ +from __future__ import annotations + import configparser -from typing import Any, Dict, Type +from typing import Any from .typing import M from .utils import try_import from ._constants import NO_SECTIONS -def from_ini_str(ini_str: str, m: Type[M]) -> M: +def from_ini_str(ini_str: str, m: type[M]) -> M: cp = configparser.ConfigParser(interpolation=None) # cp.optionxform = str cp.read_string(ini_str) @@ -17,7 +19,7 @@ def from_ini_str(ini_str: str, m: Type[M]) -> M: return m.parse_obj(values) -def from_ini_no_section_header_str(ini_str: str, m: Type[M]) -> M: +def from_ini_no_section_header_str(ini_str: str, m: type[M]) -> M: cp = configparser.ConfigParser(interpolation=None) cp.read_string(f"[{NO_SECTIONS}]\n" + ini_str) @@ -28,7 +30,7 @@ def from_ini_no_section_header_str(ini_str: str, m: Type[M]) -> M: return m.parse_obj(values) -def from_namelist_str(nl_str: str, m: Type[M]) -> M: +def from_namelist_str(nl_str: str, m: type[M]) -> M: f90nml = try_import("f90nml", extras_require_name="namelist") parser = f90nml.Parser() data: f90nml.Namelist = parser.reads(nl_str) @@ -36,19 +38,19 @@ def from_namelist_str(nl_str: str, m: Type[M]) -> M: return m.parse_obj(data.todict()) -def from_yaml_str(yaml_str: str, m: Type[M]) -> M: +def from_yaml_str(yaml_str: str, m: type[M]) -> M: yaml = try_import("yaml", extras_require_name="yaml") try: from yaml import CLoader as Loader except ImportError: from yaml import Loader - data: Dict[str, Any] = yaml.load(yaml_str, Loader=Loader) + data: dict[str, Any] = yaml.load(yaml_str, Loader=Loader) return m.parse_obj(data) -def from_toml_str(toml_str: str, m: Type[M]) -> M: +def from_toml_str(toml_str: str, m: type[M]) -> M: tomli = try_import("tomli", extras_require_name="toml") - data: Dict[str, Any] = tomli.loads(toml_str) + data: dict[str, Any] = tomli.loads(toml_str) return m.parse_obj(data) diff --git a/python/ngen_init_config/src/ngen/init_config/alias_generator.py b/python/ngen_init_config/src/ngen/init_config/alias_generator.py index 1b77a862..2318605b 100644 --- a/python/ngen_init_config/src/ngen/init_config/alias_generator.py +++ b/python/ngen_init_config/src/ngen/init_config/alias_generator.py @@ -1,3 +1,5 @@ +from __future__ import annotations + def lower_case(s: str) -> str: return s.lower() diff --git a/python/ngen_init_config/src/ngen/init_config/core.py b/python/ngen_init_config/src/ngen/init_config/core.py index a59a0c48..d1c39544 100644 --- a/python/ngen_init_config/src/ngen/init_config/core.py +++ b/python/ngen_init_config/src/ngen/init_config/core.py @@ -1,6 +1,8 @@ +from __future__ import annotations + from datetime import datetime from functools import lru_cache -from typing import TYPE_CHECKING, Optional, Type, Union +from typing import TYPE_CHECKING from pydantic import BaseModel from pydantic.main import BaseModel, _missing @@ -68,12 +70,12 @@ def _iter( self, to_dict: bool = False, by_alias: bool = False, - include: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, - exclude: Optional[Union["AbstractSetIntStr", "MappingIntStrAny"]] = None, + include: AbstractSetIntStr | MappingIntStrAny | None = None, + exclude: AbstractSetIntStr | MappingIntStrAny | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, - ) -> "TupleGenerator": + ) -> TupleGenerator: # NOTE: majority of this code was copy pasted from `pydantic==1.10.5`. Due to `pydantic`'s # stability and implementation of this method, it was easier and more straightforward to go # this route rather than an implementation overload. For readability and maintainability, @@ -201,21 +203,21 @@ def _iter( yield dict_key, v -def _get_field_type_serializers(t: Type[Base]) -> TypeSerializers: +def _get_field_type_serializers(t: type[Base]) -> TypeSerializers: return merge_class_attr(t, "Config.field_type_serializers", {}) # type: ignore -def _has_field_type_serializers(t: Type[Base]) -> bool: +def _has_field_type_serializers(t: type[Base]) -> bool: return bool(_get_field_serializers(t)) @lru_cache(maxsize=None) -def _get_field_serializers(t: Type[Base]) -> FieldSerializers: +def _get_field_serializers(t: type[Base]) -> FieldSerializers: return merge_class_attr(t, "Config.field_serializers", {}) # type: ignore @lru_cache(maxsize=None) -def _uses_composition(cls: Type[Base]) -> bool: +def _uses_composition(cls: type[Base]) -> bool: for v in cls.__fields__.values(): flat_type_hints = flatten_args(v.type_) for t in flat_type_hints: diff --git a/python/ngen_init_config/src/ngen/init_config/deserializer.py b/python/ngen_init_config/src/ngen/init_config/deserializer.py index 466bc6ca..850c055d 100644 --- a/python/ngen_init_config/src/ngen/init_config/deserializer.py +++ b/python/ngen_init_config/src/ngen/init_config/deserializer.py @@ -1,4 +1,4 @@ -from pathlib import Path +from __future__ import annotations from .core import Base from .utils import merge_class_attr @@ -10,7 +10,11 @@ from_toml_str, ) -from typing_extensions import Self +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from typing_extensions import Self + from pathlib import Path class IniDeserializer(Base): diff --git a/python/ngen_init_config/src/ngen/init_config/format_serializers.py b/python/ngen_init_config/src/ngen/init_config/format_serializers.py index 7ec436ee..54d113f8 100644 --- a/python/ngen_init_config/src/ngen/init_config/format_serializers.py +++ b/python/ngen_init_config/src/ngen/init_config/format_serializers.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import configparser from collections import OrderedDict from io import StringIO @@ -9,7 +11,7 @@ from .utils import try_import -def to_namelist_str(d: Dict[str, Any]) -> str: +def to_namelist_str(d: dict[str, Any]) -> str: """Serialize a dictionary as an namelist formatted string.""" f90nml = try_import("f90nml", extras_require_name="namelist") @@ -34,7 +36,7 @@ def to_namelist_str(d: Dict[str, Any]) -> str: def to_ini_str( - d: Dict[str, NON_NONE_JSON_DICT], + d: dict[str, NON_NONE_JSON_DICT], *, space_around_delimiters: bool = True, preserve_key_case: bool = False, @@ -98,7 +100,7 @@ def to_ini_no_section_header_str( return buff[buff.find("\n") + 1 :].rstrip() -def to_yaml_str(d: Dict[str, Any]) -> str: +def to_yaml_str(d: dict[str, Any]) -> str: """Serialize a dictionary as a yaml formatted string.""" yaml = try_import("yaml", extras_require_name="yaml") @@ -115,7 +117,7 @@ def increase_indent(self, flow=False, *args, **kwargs): return yaml.dump(d, Dumper=Dumper).rstrip() -def to_toml_str(d: Dict[str, Any]) -> str: +def to_toml_str(d: dict[str, Any]) -> str: """Serialize a dictionary as a toml formatted string.""" tomli_w = try_import("tomli_w", extras_require_name="toml") # drop eol chars diff --git a/python/ngen_init_config/src/ngen/init_config/root_validators.py b/python/ngen_init_config/src/ngen/init_config/root_validators.py index 750e5a31..494c0489 100644 --- a/python/ngen_init_config/src/ngen/init_config/root_validators.py +++ b/python/ngen_init_config/src/ngen/init_config/root_validators.py @@ -1,9 +1,11 @@ -from typing import Any, Dict, Type +from __future__ import annotations + +from typing import Any from .typing import M -def case_insensitive_keys(cls: Type[M], values: Dict[str, Any]) -> Dict[str, Any]: +def case_insensitive_keys(cls: type[M], values: dict[str, Any]) -> dict[str, Any]: """pydantic root validator that case insensitively remaps input `values` keys to model alias, if present, or field names. """ @@ -12,7 +14,7 @@ def case_insensitive_keys(cls: Type[M], values: Dict[str, Any]) -> Dict[str, Any # NOTE: only guarantees remapping defined fields to their case insensitive representation; # e.g. if `Config.extra = "allow"`, undefined fields will be included as is. - remapped: Dict[str, Any] = {} + remapped: dict[str, Any] = {} for k, v in values.items(): try: key = keys_map[k.casefold()] diff --git a/python/ngen_init_config/src/ngen/init_config/serializer.py b/python/ngen_init_config/src/ngen/init_config/serializer.py index 45755575..9b231668 100644 --- a/python/ngen_init_config/src/ngen/init_config/serializer.py +++ b/python/ngen_init_config/src/ngen/init_config/serializer.py @@ -1,7 +1,12 @@ +from __future__ import annotations + import os -import pathlib from . import core, format_serializers, utils +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + import pathlib class IniSerializer(core.Base): diff --git a/python/ngen_init_config/src/ngen/init_config/serializer_deserializer.py b/python/ngen_init_config/src/ngen/init_config/serializer_deserializer.py index 0be734a2..ca2237a2 100644 --- a/python/ngen_init_config/src/ngen/init_config/serializer_deserializer.py +++ b/python/ngen_init_config/src/ngen/init_config/serializer_deserializer.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from . import serializer as ser from . import deserializer as de diff --git a/python/ngen_init_config/src/ngen/init_config/typing.py b/python/ngen_init_config/src/ngen/init_config/typing.py index 3692a325..dddc7910 100644 --- a/python/ngen_init_config/src/ngen/init_config/typing.py +++ b/python/ngen_init_config/src/ngen/init_config/typing.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import sys # see: https://docs.python.org/3.8/library/typing.html#typing.get_args @@ -12,8 +14,6 @@ Any, Callable, Dict, - List, - Tuple, Type, TypeVar, Union, @@ -30,7 +30,7 @@ TypeSerializers: TypeAlias = Dict[Type[Any], FnJsonSerializable] -def flatten_args(t: Type[object]) -> Tuple[Type[object], ...]: +def flatten_args(t: type[object]) -> tuple[type[object], ...]: """Flatten and deduplicate nested type hints. Order is preserved. In line with `get_args`' behavior, `typing.Union` are excluded. @@ -40,8 +40,8 @@ def flatten_args(t: Type[object]) -> Tuple[Type[object], ...]: ``` """ # NOTE: python 3.7 >= dictionaries are ordered - flat: Dict[Type[object], None] = {} - horizon: List[Type[object]] = [t] + flat: dict[type[object], None] = {} + horizon: list[type[object]] = [t] while True: if not horizon: diff --git a/python/ngen_init_config/src/ngen/init_config/utils.py b/python/ngen_init_config/src/ngen/init_config/utils.py index d2efb1dc..16da3e86 100644 --- a/python/ngen_init_config/src/ngen/init_config/utils.py +++ b/python/ngen_init_config/src/ngen/init_config/utils.py @@ -1,8 +1,13 @@ +from __future__ import annotations + import importlib -from typing import Any, Type, Union -from types import ModuleType +from typing import TYPE_CHECKING from copy import deepcopy +if TYPE_CHECKING: + from typing import Any + from types import ModuleType + __SENTINEL = object() __MERGE_SENTINEL = object() @@ -11,7 +16,7 @@ def get_attr( __o: object, __name: str, __default: object = __SENTINEL -) -> Union[Any, object]: +) -> Any | object: value = __o partial_attrs = __name.split(".") @@ -30,8 +35,8 @@ def has_attr(__o: object, __name: str) -> bool: def merge_class_attr( - __t: Type[object], __name: str, __default: object = __SENTINEL -) -> Union[Any, object]: + __t: type[object], __name: str, __default: object = __SENTINEL +) -> Any | object: """ Walk a type's mro from back to front and merge or retrieve a _deep copy_ of a class attribute. Mergable attribute types (including subtypes) are, lists, dictionaries, and sets. Attributes are @@ -51,7 +56,7 @@ def merge_class_attr( """ mro = __t.mro() - final: Union[object, Any] = __MERGE_SENTINEL + final: object | Any = __MERGE_SENTINEL # unique id of source collection. # its healthy to think of this as a pointer to a source collection. diff --git a/python/ngen_init_config/src/ngen/init_config/validators.py b/python/ngen_init_config/src/ngen/init_config/validators.py index 957fdc16..c9dc3259 100644 --- a/python/ngen_init_config/src/ngen/init_config/validators.py +++ b/python/ngen_init_config/src/ngen/init_config/validators.py @@ -1,8 +1,10 @@ -from typing import Callable, List +from __future__ import annotations +from typing import Callable -def str_split(sep: str, *, strip: bool = False) -> Callable[[str], List[str]]: - def inner(items: str) -> List[str]: + +def str_split(sep: str, *, strip: bool = False) -> Callable[[str], list[str]]: + def inner(items: str) -> list[str]: split_items = items.split(sep) if strip: for idx, item in enumerate(split_items):