Skip to content

Commit

Permalink
Update .model.transport type hints for Python >= 3.9
Browse files Browse the repository at this point in the history
- Use standard collections, e.g. list[str] instead of typing.List[str].
- Import certain types from collections.abc, instead of deprecated aliases in
  typing.
  • Loading branch information
khaeru committed Nov 20, 2024
1 parent 538255b commit 24b599e
Show file tree
Hide file tree
Showing 21 changed files with 140 additions and 154 deletions.
4 changes: 2 additions & 2 deletions message_ix_models/model/transport/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from functools import partial
from itertools import pairwise, product
from pathlib import Path
from typing import TYPE_CHECKING, Any, List, Optional
from typing import TYPE_CHECKING, Any, Optional

import genno
import numpy as np
Expand Down Expand Up @@ -423,7 +423,7 @@ def to_csv(


def format_share_constraints(
qty: "AnyQuantity", config: dict, *, kind: str, groupby: List[str] = []
qty: "AnyQuantity", config: dict, *, kind: str, groupby: list[str] = []
) -> pd.DataFrame:
"""Produce values for :file:`ue_share_constraints.xlsx`.
Expand Down
6 changes: 3 additions & 3 deletions message_ix_models/model/transport/build.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import logging
from importlib import import_module
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple
from typing import TYPE_CHECKING, Any, Optional

import pandas as pd
from genno import Computer, KeyExistsError, Quantity, quote
Expand Down Expand Up @@ -179,7 +179,7 @@ def add_exogenous_data(c: Computer, info: ScenarioInfo) -> None:
# Identify appropriate source keyword arguments for loading GDP and population data
source = str(config.ssp)
if config.ssp in SSP_2017:
source_kw: Tuple[Dict[str, Any], ...] = (
source_kw: tuple[dict[str, Any], ...] = (
dict(measure="GDP", model="IIASA GDP"),
dict(measure="POP", model="IIASA GDP"),
)
Expand Down Expand Up @@ -471,7 +471,7 @@ def get_computer(
def main(
context: Context,
scenario: Scenario,
options: Optional[Dict] = None,
options: Optional[dict] = None,
**option_kwargs,
):
"""Build MESSAGEix-Transport on `scenario`.
Expand Down
30 changes: 15 additions & 15 deletions message_ix_models/model/transport/config.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import logging
from dataclasses import InitVar, dataclass, field, replace
from enum import Enum
from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Tuple, Union
from typing import TYPE_CHECKING, Literal, Optional, Union

import message_ix
from genno import Quantity
Expand Down Expand Up @@ -63,7 +63,7 @@ class Config(ConfigHelper):
#: for transport modes *other than* LDV. See :func:`non_ldv.growth_new_capacity`.
#: "* initial_*_up"
#: Base value for growth constraints. These values are arbitrary.
constraint: Dict = field(
constraint: dict = field(
default_factory=lambda: {
"LDV growth_activity_lo": -0.0192,
"LDV growth_activity_up": 0.0192 * 3.0,
Expand Down Expand Up @@ -93,7 +93,7 @@ class Config(ConfigHelper):
#: 'updateTRPdata', with the comment "Original data from Sei (PAO)."
#: - This probably refers to some source that gave relative costs of different
#: buses, in PAO, for this year; it is applied across all years.
cost: Dict = field(
cost: dict = field(
default_factory=lambda: {
#
"ldv nga": 0.85,
Expand All @@ -114,7 +114,7 @@ class Config(ConfigHelper):
#: specified in the corresponding technology.yaml file.
#:
#: .. todo:: Read directly from technology.yaml
demand_modes: List[str] = field(
demand_modes: list[str] = field(
default_factory=lambda: ["LDV", "2W", "AIR", "BUS", "RAIL"]
)

Expand All @@ -129,7 +129,7 @@ class Config(ConfigHelper):
dummy_supply: bool = False

#: Various efficiency factors.
efficiency: Dict = field(
efficiency: dict = field(
default_factory=lambda: {
"*": 0.2,
"hev": 0.2,
Expand All @@ -150,7 +150,7 @@ class Config(ConfigHelper):
emission_relations: bool = True

#: Various other factors.
factor: Dict = field(default_factory=dict)
factor: dict = field(default_factory=dict)

#: If :obj:`True` (the default), do not record/preserve parameter data when removing
#: set elements from the base model.
Expand All @@ -171,7 +171,7 @@ class Config(ConfigHelper):
#:
#: ``F ROAD``: similar to IEA “Future of Trucks” (2017) values; see
#: .transport.freight. Alternately use 5.0, similar to Roadmap 2017 values.
load_factor: Dict = field(
load_factor: dict = field(
default_factory=lambda: {
"F ROAD": 10.0,
"F RAIL": 10.0,
Expand All @@ -183,7 +183,7 @@ class Config(ConfigHelper):

#: Period in which LDV costs match those of a reference region.
#: Dimensions: (node,).
ldv_cost_catch_up_year: Dict = field(default_factory=dict)
ldv_cost_catch_up_year: dict = field(default_factory=dict)

#: Method for calibrating LDV stock and sales:
#:
Expand All @@ -193,7 +193,7 @@ class Config(ConfigHelper):

#: Tuples of (node, technology (transport mode), commodity) for which minimum
#: activity should be enforced. See :func:`.non_ldv.bound_activity_lo`.
minimum_activity: Dict[Tuple[str, Tuple[str, ...], str], float] = field(
minimum_activity: dict[tuple[str, tuple[str, ...], str], float] = field(
default_factory=dict
)

Expand All @@ -202,15 +202,15 @@ class Config(ConfigHelper):
mode_share: str = "default"

#: List of modules containing model-building calculations.
modules: List[str] = field(
modules: list[str] = field(
default_factory=lambda: (
"groups demand freight ikarus ldv disutility non_ldv plot data"
).split()
)

#: Used by :func:`.get_USTIMES_MA3T` to map MESSAGE regions to U.S. census divisions
#: appearing in MA³T.
node_to_census_division: Dict = field(default_factory=dict)
node_to_census_division: dict = field(default_factory=dict)

#: **Temporary** setting for the SSP 2024 project: indicates whether the base
#: scenario used is a policy (carbon pricing) scenario, or not. This currently does
Expand All @@ -226,7 +226,7 @@ class Config(ConfigHelper):
#:
#: :mod:`.transport.build` and :mod:`.transport.report` code will respond to these
#: settings in documented ways.
project: Dict[str, Enum] = field(
project: dict[str, Enum] = field(
default_factory=lambda: dict(
futures=FUTURES_SCENARIO.BASE, navigate=NAVIGATE_SCENARIO.REF
)
Expand All @@ -236,7 +236,7 @@ class Config(ConfigHelper):
scaling: float = 1.0

#: Mapping from nodes to other nodes towards which share weights should converge.
share_weight_convergence: Dict = field(default_factory=dict)
share_weight_convergence: dict = field(default_factory=dict)

#: Specification for the structure of MESSAGEix-Transport, processed from contents
#: of :file:`set.yaml` and :file:`technology.yaml`.
Expand Down Expand Up @@ -282,7 +282,7 @@ class Config(ConfigHelper):
#: space-delimited string (:py:`"module_a -module_b"`) or sequence of strings.
#: Values prefixed with a hyphen (:py:`"-module_b"`) are *removed* from
#: :attr:`.modules`.
extra_modules: InitVar[Union[str, List[str]]] = None
extra_modules: InitVar[Union[str, list[str]]] = None

#: Identifier of a Transport Futures scenario, used to update :attr:`project` via
#: :meth:`.ScenarioFlags.parse_futures`.
Expand Down Expand Up @@ -315,7 +315,7 @@ def from_context(
cls,
context: Context,
scenario: Optional[message_ix.Scenario] = None,
options: Optional[Dict] = None,
options: Optional[dict] = None,
) -> "Config":
"""Configure `context` for building MESSAGEix-Transport.
Expand Down
23 changes: 12 additions & 11 deletions message_ix_models/model/transport/data.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,11 @@

import logging
from collections import defaultdict
from collections.abc import Callable, Mapping
from copy import deepcopy
from functools import partial
from operator import le
from typing import TYPE_CHECKING, Callable, Dict, List, Mapping, Optional, Set, Tuple
from typing import TYPE_CHECKING, Optional

import pandas as pd
from genno import Computer, Key, Quantity
Expand Down Expand Up @@ -51,8 +52,8 @@ def prepare_computer(c: Computer):


def conversion(
nodes: List[str], years: List[int], config: dict
) -> Dict[str, pd.DataFrame]:
nodes: list[str], years: list[int], config: dict
) -> dict[str, pd.DataFrame]:
"""Input and output data for conversion technologies:
The technologies are named 'transport {service} load factor'.
Expand All @@ -72,7 +73,7 @@ def conversion(
("pax", 1.0, "Gp km / a"),
]

data0: Mapping[str, List] = defaultdict(list)
data0: Mapping[str, list] = defaultdict(list)
for service, factor, output_unit in service_info:
i_o = make_io(
(f"transport {service} vehicle", "useful", "Gv km"),
Expand All @@ -98,7 +99,7 @@ def conversion(
return data1


def dummy_supply(technologies: List["Code"], info, config) -> Dict[str, pd.DataFrame]:
def dummy_supply(technologies: list["Code"], info, config) -> dict[str, pd.DataFrame]:
"""Dummy fuel supply for the bare RES."""
if not config["transport"].dummy_supply:
return dict()
Expand All @@ -116,7 +117,7 @@ def dummy_supply(technologies: List["Code"], info, config) -> Dict[str, pd.DataF
else:
raise TypeError(type(commodity))

result: Dict[str, pd.DataFrame] = dict()
result: dict[str, pd.DataFrame] = dict()
common = dict(mode="all", time="year", time_dest="year", unit="GWa")
values = dict(output=1.0, var_cost=1.0)

Expand All @@ -133,7 +134,7 @@ def dummy_supply(technologies: List["Code"], info, config) -> Dict[str, pd.DataF
return result


def misc(info: ScenarioInfo, nodes: List[str], y: List[int]):
def misc(info: ScenarioInfo, nodes: list[str], y: list[int]):
"""Miscellaneous bounds for calibration/vetting."""

# Limit activity of methanol LDVs in the model base year
Expand All @@ -157,8 +158,8 @@ def misc(info: ScenarioInfo, nodes: List[str], y: List[int]):


def navigate_ele(
nodes: List[str], techs: List["Code"], t_groups, years: List[int], config
) -> Dict[str, pd.DataFrame]:
nodes: list[str], techs: list["Code"], t_groups, years: list[int], config
) -> dict[str, pd.DataFrame]:
"""Return constraint data for :attr:`ScenarioFlags.ELE`.
The text reads as follows as of 2023-02-15:
Expand Down Expand Up @@ -307,7 +308,7 @@ class MaybeAdaptR11Source(ExoDataSource):
"""

#: Set of measures recognized by a subclass.
measures: Set[str] = set()
measures: set[str] = set()

#: Mapping from :attr:`.measures` entries to file names.
filename: Mapping[str, str] = dict()
Expand Down Expand Up @@ -364,7 +365,7 @@ def __call__(self):
def __repr__(self) -> str:
return self._repr

def get_keys(self) -> Tuple[Key, Key]:
def get_keys(self) -> tuple[Key, Key]:
"""Return the target keys for the (1) raw and (2) transformed data."""
k = self.key or Key(
self.name or self.measure.lower(), ("n", "y") + self.extra_dims
Expand Down
8 changes: 4 additions & 4 deletions message_ix_models/model/transport/demand.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Demand calculation for MESSAGEix-Transport."""

import logging
from typing import TYPE_CHECKING, Dict, List
from typing import TYPE_CHECKING

import genno
import numpy as np
Expand Down Expand Up @@ -48,8 +48,8 @@


def dummy(
commodities: List, nodes: List[str], y: List[int], config: dict
) -> Dict[str, pd.DataFrame]:
commodities: list, nodes: list[str], y: list[int], config: dict
) -> dict[str, pd.DataFrame]:
"""Dummy demands.
Expand Down Expand Up @@ -241,7 +241,7 @@ def pdt_per_capita(c: Computer) -> None:
# Add `y` dimension. Here for the future fixed point we use y=2 * max(y), e.g.
# 4220 for y=2110. The value doesn't matter, we just need to avoid overlap with y
# in the model.
def _future(qty: "AnyQuantity", years: List[int]) -> "AnyQuantity":
def _future(qty: "AnyQuantity", years: list[int]) -> "AnyQuantity":
return qty.expand_dims(y=[years[-1] * 2])

# Same, but adding y0
Expand Down
26 changes: 8 additions & 18 deletions message_ix_models/model/transport/factor.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,20 +11,10 @@
import operator
import re
from abc import ABC, abstractmethod
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass, field
from functools import partial
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
List,
Mapping,
Optional,
Sequence,
Tuple,
Union,
)
from typing import TYPE_CHECKING, Any, Optional, Union

import pandas as pd
from genno import Computer, Key, Quantity
Expand Down Expand Up @@ -95,7 +85,7 @@ class Constant(Layer):
value: Quantity

#: Dimensions of the result.
dims: Tuple[str, ...]
dims: tuple[str, ...]

operation = operator.mul

Expand Down Expand Up @@ -183,12 +173,12 @@ class Map(Layer):
"""

dim: str
values: Dict[str, Layer]
values: dict[str, Layer]

operation = operator.mul

def __init__(
self, dim: str, values: Optional[Dict[str, Layer]] = None, **value_kwargs: Layer
self, dim: str, values: Optional[dict[str, Layer]] = None, **value_kwargs: Layer
):
self.dim = dim
self.values = values or value_kwargs
Expand All @@ -215,7 +205,7 @@ class ScenarioSetting(Layer):
"""

#: Mapping from scenario identifier to setting label.
setting: Dict[Any, str]
setting: dict[Any, str]

#: Default setting.
default: str
Expand Down Expand Up @@ -310,7 +300,7 @@ class Factor:
"""

#: Ordered list of :class:`.Layer`.
layers: List[Layer] = field(default_factory=list)
layers: list[Layer] = field(default_factory=list)

def __hash__(self):
return hash(tuple(self.layers))
Expand Down Expand Up @@ -360,7 +350,7 @@ def add_tasks(
)

def __call__(
self, config, *coords, dims: Tuple[str, ...], scenario_expr: str
self, config, *coords, dims: tuple[str, ...], scenario_expr: str
) -> Quantity:
"""Invoke :meth:`quantify`, for use with :mod:`genno`."""
kw = dict(zip(dims, coords))
Expand Down
Loading

0 comments on commit 24b599e

Please sign in to comment.