chore: update to python 3.11 syntax

Refs: ECALC-1797

src/ecalc_cli/infrastructure/file_resource_service.py

@@ -1,6 +1,5 @@
 from collections.abc import Callable
 from pathlib import Path
-from typing import Dict
 
 from libecalc.common.errors.exceptions import EcalcError, InvalidHeaderException
 from libecalc.common.logger import logger
@@ -24,8 +23,8 @@ def _read_resource(resource_name: Path, *args, read_func: Callable[..., MemoryRe
             raise EcalcError("Failed to read resource", f"Failed to read {resource_name.name}: {str(exc)}") from exc
 
     @classmethod
-    def _read_resources(cls, configuration: YamlValidator, working_directory: Path) -> Dict[str, MemoryResource]:
-        resources: Dict[str, MemoryResource] = {}
+    def _read_resources(cls, configuration: YamlValidator, working_directory: Path) -> dict[str, MemoryResource]:
+        resources: dict[str, MemoryResource] = {}
         for timeseries_resource in configuration.timeseries_resources:
             resources[timeseries_resource.name] = cls._read_resource(
                 working_directory / timeseries_resource.name,
@@ -40,5 +39,5 @@ def _read_resources(cls, configuration: YamlValidator, working_directory: Path)
             )
         return resources
 
-    def get_resources(self, configuration: YamlValidator) -> Dict[str, Resource]:
+    def get_resources(self, configuration: YamlValidator) -> dict[str, Resource]:
         return self._read_resources(configuration=configuration, working_directory=self._working_directory)

src/ecalc_cli/io/output.py

@@ -1,7 +1,6 @@
 import json
 import sys
 from pathlib import Path
-from typing import Dict, List
 
 import libecalc.common.time_utils
 from ecalc_cli.errors import EcalcCLIError
@@ -152,7 +151,7 @@ def export_tsv(
     prognosis_filter = config.filter(frequency=frequency)
     result = prognosis_filter.filter(ExportableGraphResult(results), resampled_periods)
 
-    row_based_data: Dict[str, List[str]] = CSVFormatter(
+    row_based_data: dict[str, list[str]] = CSVFormatter(
         separation_character="\t", index_formatters=PeriodFormatterConfig.get_row_index_formatters()
     ).format_groups(result)
 

src/ecalc_cli/logger.py

@@ -4,7 +4,7 @@
 from datetime import date
 from logging.config import dictConfig
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 
 class LogLevel(str, enum.Enum):
@@ -57,7 +57,7 @@ def set_log_path(self, log_path: Path):
         self.__configure_logger()
 
     def __configure_logger(self):
-        log_config: Dict[str, Any] = {
+        log_config: dict[str, Any] = {
             "version": 1,
             "disable_existing_loggers": True,  # ie keep root logger ++
             "formatters": {

src/ecalc_neqsim_wrapper/mappings.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import Dict
 
 from ecalc_neqsim_wrapper import neqsim
 from libecalc.common.errors.exceptions import EcalcError
@@ -51,7 +50,7 @@ class NeqsimEoSModelType(Enum):
 }
 
 
-def map_fluid_composition_to_neqsim(fluid_composition: FluidComposition) -> Dict[str, float]:
+def map_fluid_composition_to_neqsim(fluid_composition: FluidComposition) -> dict[str, float]:
     component_dict = {}
     for component_name, value in fluid_composition.model_dump().items():
         if value is not None and value > 0:

src/ecalc_neqsim_wrapper/thermo.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from functools import lru_cache
 from pathlib import Path
-from typing import Dict, List, Tuple, Union
+from typing import Union
 
 from py4j.protocol import Py4JJavaError
 from pydantic import BaseModel
@@ -60,8 +60,8 @@ class NeqsimFluidState(BaseModel):
     A representation of a NeqSim Fluid State for a given composition and properties
     """
 
-    fluid_composition: List[NeqsimFluidComponent]
-    fluid_properties: List[NeqsimFluidProperty]
+    fluid_composition: list[NeqsimFluidComponent]
+    fluid_properties: list[NeqsimFluidProperty]
 
 
 class NeqsimFluid:
@@ -84,8 +84,8 @@ def json(self) -> str:
 
         :return: the JSON representation of the Neqsim state, as an unformatted string
         """
-        fluid_composition: List[NeqsimFluidComponent] = []
-        fluid_properties: List[NeqsimFluidProperty] = []
+        fluid_composition: list[NeqsimFluidComponent] = []
+        fluid_properties: list[NeqsimFluidProperty] = []
 
         try:
             for string_vector in self._thermodynamic_system.createTable("ECALC_DUMP"):
@@ -167,8 +167,8 @@ def create_thermo_system(
     @staticmethod
     @lru_cache(maxsize=512)
     def _init_thermo_system(
-        components: List[str],
-        molar_fraction: List[float],
+        components: list[str],
+        molar_fraction: list[float],
         eos_model: NeqsimEoSModelType,
         temperature_kelvin: float,
         pressure_bara: float,
@@ -385,10 +385,10 @@ def mix_neqsim_streams(
     pressure: float,
     temperature: float,
     eos_model: EoSModel = EoSModel.SRK,
-) -> Tuple[FluidComposition, NeqsimFluid]:
+) -> tuple[FluidComposition, NeqsimFluid]:
     """Mixing two streams (NeqsimFluids) with same pressure and temperature."""
 
-    composition_dict: Dict[str, float] = {}
+    composition_dict: dict[str, float] = {}
 
     stream_1 = NeqsimFluid.create_thermo_system(
         composition=stream_composition_1,

src/libecalc/application/energy_calculator.py

@@ -1,5 +1,4 @@
 from collections import defaultdict
-from typing import Dict
 
 import numpy as np
 
@@ -49,14 +48,14 @@ def __init__(
     ):
         self._graph = graph
 
-    def evaluate_energy_usage(self, variables_map: VariablesMap) -> Dict[str, EcalcModelResult]:
+    def evaluate_energy_usage(self, variables_map: VariablesMap) -> dict[str, EcalcModelResult]:
         component_ids = list(reversed(self._graph.sorted_node_ids))
         component_dtos = [self._graph.get_node(component_id) for component_id in component_ids]
 
-        consumer_results: Dict[str, EcalcModelResult] = {}
+        consumer_results: dict[str, EcalcModelResult] = {}
 
         for component_dto in component_dtos:
-            if isinstance(component_dto, (ElectricityConsumerDTO, FuelConsumerDTO)):
+            if isinstance(component_dto, ElectricityConsumerDTO | FuelConsumerDTO):
                 consumer = Consumer(
                     id=component_dto.id,
                     name=component_dto.name,
@@ -110,7 +109,7 @@ def evaluate_energy_usage(self, variables_map: VariablesMap) -> Dict[str, EcalcM
                 )
                 optimizer = PriorityOptimizer()
 
-                results_per_period: Dict[str, Dict[Period, ComponentResult]] = defaultdict(dict)
+                results_per_period: dict[str, dict[Period, ComponentResult]] = defaultdict(dict)
                 priorities_used = []
                 for period in variables_map.periods:
                     consumers_for_period = [
@@ -183,8 +182,8 @@ def evaluator(priority: PriorityID):
         return Numbers.format_results_to_precision(consumer_results, precision=6)
 
     def evaluate_emissions(
-        self, variables_map: VariablesMap, consumer_results: Dict[str, EcalcModelResult]
-    ) -> Dict[str, Dict[str, EmissionResult]]:
+        self, variables_map: VariablesMap, consumer_results: dict[str, EcalcModelResult]
+    ) -> dict[str, dict[str, EmissionResult]]:
         """
         Calculate emissions for fuel consumers and emitters
 
@@ -194,9 +193,9 @@ def evaluate_emissions(
 
         Returns: a mapping from consumer_id to emissions
         """
-        emission_results: Dict[str, Dict[str, EmissionResult]] = {}
+        emission_results: dict[str, dict[str, EmissionResult]] = {}
         for consumer_dto in self._graph.nodes.values():
-            if isinstance(consumer_dto, (FuelConsumerDTO, GeneratorSetDTO)):
+            if isinstance(consumer_dto, FuelConsumerDTO | GeneratorSetDTO):
                 fuel_model = FuelModel(consumer_dto.fuel)
                 energy_usage = consumer_results[consumer_dto.id].component_result.energy_usage
                 emission_results[consumer_dto.id] = fuel_model.evaluate_emissions(
@@ -210,7 +209,7 @@ def evaluate_emissions(
                     emission_results[consumer_dto.id] = fuel_model.evaluate_emissions(
                         expression_evaluator=variables_map, fuel_rate=np.asarray(energy_usage.values)
                     )
-            elif isinstance(consumer_dto, (YamlDirectTypeEmitter, YamlOilTypeEmitter)):
+            elif isinstance(consumer_dto, YamlDirectTypeEmitter | YamlOilTypeEmitter):
                 installation_id = self._graph.get_parent_installation_id(consumer_dto.id)
                 installation = self._graph.get_node(installation_id)
 

src/libecalc/application/graph_result.py

@@ -1,5 +1,3 @@
-from typing import Dict
-
 from pydantic import BaseModel
 
 from libecalc.common.variables import VariablesMap
@@ -9,17 +7,17 @@
 
 
 class EnergyCalculatorResult(BaseModel):
-    consumer_results: Dict[str, EcalcModelResult]
-    emission_results: Dict[str, Dict[str, EmissionResult]]
+    consumer_results: dict[str, EcalcModelResult]
+    emission_results: dict[str, dict[str, EmissionResult]]
     variables_map: VariablesMap
 
 
 class GraphResult:
     def __init__(
         self,
         graph: ComponentGraph,
-        consumer_results: Dict[str, EcalcModelResult],
-        emission_results: Dict[str, Dict[str, EmissionResult]],
+        consumer_results: dict[str, EcalcModelResult],
+        emission_results: dict[str, dict[str, EmissionResult]],
         variables_map: VariablesMap,
     ):
         self.graph = graph
@@ -52,7 +50,7 @@ def get_energy_result(self, component_id: str) -> ComponentResult:
     def periods(self):
         return self.variables_map.periods
 
-    def get_emissions(self, component_id: str) -> Dict[str, EmissionResult]:
+    def get_emissions(self, component_id: str) -> dict[str, EmissionResult]:
         return self.emission_results[component_id]
 
     def get_results(self) -> EnergyCalculatorResult:

src/libecalc/common/graph.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Dict, Generic, List, Protocol, Self, TypeVar
+from typing import Generic, Protocol, Self, TypeVar
 
 import networkx as nx
 
@@ -18,7 +18,7 @@ def id(self) -> NodeID: ...
 class Graph(Generic[TNode]):
     def __init__(self):
         self.graph = nx.DiGraph()
-        self.nodes: Dict[NodeID, TNode] = {}
+        self.nodes: dict[NodeID, TNode] = {}
 
     def add_node(self, node: TNode) -> Self:
         self.graph.add_node(node.id)
@@ -37,7 +37,7 @@ def add_subgraph(self, subgraph: Graph) -> Self:
         self.graph = nx.compose(self.graph, subgraph.graph)
         return self
 
-    def get_successors(self, node_id: NodeID, recursively=False) -> List[NodeID]:
+    def get_successors(self, node_id: NodeID, recursively=False) -> list[NodeID]:
         if recursively:
             return [
                 successor_id
@@ -64,10 +64,10 @@ def get_node(self, node_id: NodeID) -> TNode:
         return self.nodes[node_id]
 
     @property
-    def sorted_node_ids(self) -> List[NodeID]:
+    def sorted_node_ids(self) -> list[NodeID]:
         return list(nx.topological_sort(self.graph))
 
-    def breadth_first_search_tree(self, source_id: NodeID) -> List[NodeID]:
+    def breadth_first_search_tree(self, source_id: NodeID) -> list[NodeID]:
         """
         Create a tree with source as root by searching edges close to the source first. Breadth first orders the nodes
         that are closer to the source before nodes that are further out.

src/libecalc/common/list/list_utils.py

@@ -1,6 +1,6 @@
 from collections import defaultdict
 from collections.abc import Sequence
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Optional, Union, cast
 
 import numpy as np
 from numpy import float64
@@ -15,13 +15,13 @@
 """
 
 
-def transpose(a: List[List[Union[str, int, float]]]) -> List[List[Union[str, int, float]]]:
+def transpose(a: list[list[Union[str, int, float]]]) -> list[list[Union[str, int, float]]]:
     """Easily transpose from row based to column based data, and other
     way around, in order to use the format that best fits a certain
     purpose to work with such a list/dataframe.
 
     Args:
-        a: List to be transposed
+        a:list to be transposed
 
     Returns:
         Transposed list
@@ -30,7 +30,7 @@ def transpose(a: List[List[Union[str, int, float]]]) -> List[List[Union[str, int
     return list(map(list, zip(*a)))
 
 
-def group_data_by_value_at_index(index: int, row_based_data: List[List[Any]]) -> Dict[Union[float, int], List[Any]]:
+def group_data_by_value_at_index(index: int, row_based_data: list[list[Any]]) -> dict[Union[float, int], list[Any]]:
     """Given an index of the list, group the list by the value corresponding to that index and
     return a dict with lists, where the keys correspond to the different values at the index provided.
 
@@ -107,7 +107,7 @@ def elementwise_multiplication(
     return result
 
 
-def array_to_list(result_array: Union[NDArray[np.float64], List[NDArray[np.float64]], None]) -> Optional[List]:
+def array_to_list(result_array: Union[NDArray[np.float64], list[NDArray[np.float64]], None]) -> Optional[list]:
     """Method to convert numpy arrays and list of numpy arrays into lists (or list of lists). Method is used recursively on lists so needs to handle None as well.
 
     Args:

src/libecalc/common/math/math_utils.py

@@ -1,4 +1,4 @@
-from typing import Dict, TypeVar
+from typing import TypeVar
 
 from libecalc.common.errors.exceptions import (
     DifferentLengthsError,
@@ -11,7 +11,7 @@
 
 class MathUtil:
     @staticmethod
-    def elementwise_subtraction_by_key(this: Dict[T, float], that: Dict[T, float]) -> Dict[T, float]:
+    def elementwise_subtraction_by_key(this: dict[T, float], that: dict[T, float]) -> dict[T, float]:
         """For compatible dicts, with the same keys, subtract corresponding numbers in lists for each key
         :param this:
         :param that:
@@ -20,14 +20,14 @@ def elementwise_subtraction_by_key(this: Dict[T, float], that: Dict[T, float]) -
         :throws: IncompatibleDataError if dicts cannot be subtracted
         """
         if this is None or that is None:
-            raise IncompatibleDataError(f"A or B is None. Dict A has value {this}, and B has {that}")
+            raise IncompatibleDataError(f"A or B is None.dict A has value {this}, and B has {that}")
 
         if len(this.items()) != len(that.items()):
             raise DifferentLengthsError(
-                f"Subtracting values A-B failed due to different vector lengths. Dict A has value {len(this.items())}, but B has {len(that.items())}"
+                f"Subtracting values A-B failed due to different vector lengths.dict A has value {len(this.items())}, but B has {len(that.items())}"
             )
 
-        delta_dict: Dict[T, float] = {}
+        delta_dict: dict[T, float] = {}
 
         for item_key, this_value in this.items():
             that_value = that.get(item_key)

src/libecalc/common/priorities.py

@@ -1,7 +1,7 @@
-from typing import Dict, TypeVar
+from typing import TypeVar
 
 TPriorityValue = TypeVar("TPriorityValue")
 
 PriorityID = str
 
-Priorities = Dict[PriorityID, TPriorityValue]
+Priorities = dict[PriorityID, TPriorityValue]