fixed most mypy issues

Author: KS-HTK

pandapower/analysis/LODF.py

@@ -9,6 +9,7 @@
 
 import pandas as pd
 import numpy as np
+import numpy.typing as npt
 
 from pandapower import pandapowerNet
 from pandapower.analysis.PTDF import _makePTDF_ppci, _get_PTDF_perturb
@@ -29,7 +30,7 @@ def _get_LODF_direct(
     outage_branch_ix=None,
     using_sparse_solver=True,
     random_verify=True,
-    branch_dict: dict[str, Union[list[int], None]] | None = None,
+    branch_dict: dict[str, list[int] | None] | None = None,
     reduced=True,
 ) -> dict:
     """
@@ -41,8 +42,8 @@ def _get_LODF_direct(
         using_sparse_solver = True
 
     # If branch_dict not None compute list of ppci branch indices and its branch type intervals as lookup
-    branch_ppci_lookup = None
-    branch_id = None
+    branch_ppci_lookup: dict | None = None
+    branch_id: list | None = None
     if branch_dict is not None:
         branch_id, branch_ppci_lookup = branch_dict_to_ppci_branch_list(net=net, branch_dict=branch_dict)
     else:
@@ -68,6 +69,7 @@ def _get_LODF_direct(
     # Set results to default value of bridge branch
     lodf_ppci[:, bridge_branch_mask] = np.nan
     if branch_id is not None and not reduced:
+        assert branch_ppci_lookup is not None # force mypy type narrowing
         branch_id_complement = [x for x in range(list(branch_ppci_lookup.values())[-1][1]) if x not in branch_id]
         lodf_ppci[:, branch_id_complement] = np.nan
         lodf_ppci[branch_id_complement, :] = np.nan
@@ -123,7 +125,7 @@ def _init_LODF_pp_np(
             )
 
     for data in lodf_pp.values():
-        data[:] = np.NaN
+        data[:] = np.nan
     return lodf_pp
 
 
@@ -134,12 +136,12 @@ def _LODF_ppci_to_pp(
 ):
     # convert the branch sensitivity of the ppci layer to pandapower net layer
     if branch_ppci_lookup is not None:
-        pp_ppci_branch_lookups = {
+        pp_ppci_branch_lookups: dict[str, npt.NDArray | range | None] = {
             br_type: range(branch_ppci_lookup[br_type][0], branch_ppci_lookup[br_type][1])
             for br_type in ("line", "trafo", "impedance")
             if br_type in branch_ppci_lookup.keys()
         }
-        pp_ppci_trafo3w_lookups = {
+        pp_ppci_trafo3w_lookups: dict[str, npt.NDArray | range] | None = {
             type: range(branch_ppci_lookup[type][0], branch_ppci_lookup[type][1])
             for type in ("trafo3w_hv", "trafo3w_mv", "trafo3w_lv")
             if type in branch_ppci_lookup.keys()
@@ -237,7 +239,7 @@ def _get_LODF_perturb(
     )
 
     # number of out of service buses
-    num_out_of_service_bus = np.sum(np.isnan(net_mod.res_bus.va_degree.to_numpy()))
+    num_out_of_service_bus: int = np.sum(np.isnan(net_mod.res_bus.va_degree.to_numpy()))
     # list with the indices of out of service buses
     # list_out_of_service_bus = list(net_mod.res_bus.va_degree[np.isnan(net_mod.res_bus.va_degree.to_numpy())].index)
 
@@ -556,7 +558,7 @@ def _get_dc_n1_perturb(
     outage_branch_ix = _get_outage_branch_ix(net_mod, outage_branch_type, outage_branch_ix)
 
     rundcpp(net_mod, distributed_slack=distributed_slack)
-    num_out_of_service_bus = np.sum(np.isnan(net_mod.res_bus.va_degree.to_numpy()))
+    num_out_of_service_bus: int = np.sum(np.isnan(net_mod.res_bus.va_degree.to_numpy()))
     outage_br_p0_series = net_mod["res_" + outage_branch_type][
         "p_" + THIS_RES_BR_SIDE_MAPPING[outage_branch_type] + "_mw"
     ].copy()

pandapower/analysis/PSDF.py

@@ -4,56 +4,54 @@
 # and Energy System Technology (IEE), Kassel. All rights reserved.
 
 # Builds the DC PSDF matrix based on the DC PTDF
-import scipy as sp
+import logging
 from math import pi
+
+import scipy as sp
 from scipy.sparse import csr_matrix, csc_matrix
+import pandas as pd
+import numpy as np
+import numpy.typing as npt
 
 from pandapower.analysis.LODF import _LODF_ppci_to_pp, _LODF_pp_np_to_df
 from pandapower.analysis.PTDF import _makePTDF_ppci
 from pandapower.pypower.idx_brch import F_BUS, T_BUS
 from pandapower.pypower.idx_bus import BUS_TYPE, REF
 from pandapower.pypower.makeBdc import calc_b_from_branch
 from numpy import ones, r_, real, int64, arange, flatnonzero as find, isscalar
-
-from typing import Union, Tuple
-
-import pandas as pd
-import numpy as np
-
 from pandapower import pandapowerNet
 from pandapower.analysis.utils import branch_dict_to_ppci_branch_list, _get_outage_branch_ix, ELE_IX_TYPE
 
-import logging
 logger = logging.getLogger(__name__)
 
-
 def makePSDF(
         baseMVA: float,
-        PTDF: np.ndarray,
-        bus: np.ndarray,
-        branch: np.ndarray,
+        PTDF: npt.NDArray,
+        bus: npt.NDArray,
+        branch: npt.NDArray,
         using_sparse_solver: bool = False,
         branch_id: int | None = None,
         reduced: bool = False,
-        slack: Union[int, np.ndarray] | None = None
+        slack: int | npt.NDArray | None = None
 ):
-    """Builds the DC PSDF matrix based on the DC PTDF
+    """
+    Builds the DC PSDF matrix based on the DC PTDF
+
     Returns the DC PSDF matrix . The matrix is
     C{nbr x nbr}, where C{nbr} is the number of branches. The DC PSDF is independent from the selected slack.
     To restrict the PSDF computation to a subset of branches, supply a list of ppci branch indices in C{branch_id}.
     If C{reduced==True}, the output is reduced to the branches given in C{branch_id}, otherwise the complement rows are set to NaN.
     @see: L{makeLODF}
     """
-    if reduced and not branch_id:
+    if reduced and branch_id is None:
         raise ValueError("'reduced=True' is only valid if branch_id is not None")
 
     ## Select csc/csr B matrix
     sparse = csr_matrix if using_sparse_solver else csc_matrix
 
     ## use reference bus for slack by default
     if slack is None:
-        slack = find(bus[:, BUS_TYPE] == REF)
-        slack = slack[0]
+        slack = find(bus[:, BUS_TYPE] == REF)[0]
 
     ## set the slack bus to be used to compute initial PTDF
     if isscalar(slack):
@@ -90,11 +88,11 @@ def makePSDF(
 
 def _get_PSDF_direct(
     net: pandapowerNet,
-    phase_shift_branch_type: str,
+    phase_shift_branch_type: str | None,
     phase_shift_branch_ix: ELE_IX_TYPE | None = None,
     using_sparse_solver: bool = True,
     random_verify=False,
-    branch_dict: dict[str, Union[list[int], None]] | None = None,
+    branch_dict: dict[str, list[int] | None] | None = None,
     reduced=True,
 ):
     """
@@ -105,8 +103,8 @@ def _get_PSDF_direct(
         logger.warning("Calculating lodf for large network, switched to sparse solver!")
 
     # If branch_dict not None compute list of ppci branch indices and its branch type intervals as lookup
-    branch_ppci_lookup = None
-    branch_id = None
+    branch_ppci_lookup: dict | None = None
+    branch_id: int | None = None
     if branch_dict is not None:
         branch_id, branch_ppci_lookup = branch_dict_to_ppci_branch_list(net=net, branch_dict=branch_dict)
     else:
@@ -159,11 +157,11 @@ def _get_PSDF_direct(
 
 def _get_PSDF_perturb(
     net: pandapowerNet,
-    phase_shift_branch_type: str,
+    phase_shift_branch_type: str | None,
     phase_shift_branch_ix: ELE_IX_TYPE | None = None,
     distributed_slack=True,
-    recycle="lodf",
-) -> dict[Tuple[str, str], pd.DataFrame]:
+    recycle: str | None = "lodf",
+) -> dict[tuple[str, str], pd.DataFrame]:
     """
     this function calculates PSDF (ratio without unit) of branch to
     a pp branch with perturb method (brute-force)
@@ -173,15 +171,15 @@ def _get_PSDF_perturb(
 
 def run_PSDF(
     net: pandapowerNet,
-    phase_shift_branch_type: Union[None, str],
+    phase_shift_branch_type: str | None,
     phase_shift_branch_ix: ELE_IX_TYPE | None = None,
     distributed_slack: bool = True,
     perturb: bool = False,
-    recycle: Union[str, None] = None,
+    recycle: str | None = None,
     using_sparse_solver: bool = True,
-    branch_dict: dict[str, Union[list[int], None]] | None = None,
+    branch_dict: dict[str, list[int] | None] | None = None,
     reduced: bool = True,
-) -> dict[Tuple[str, str], pd.DataFrame]:
+) -> dict[tuple[str, str], pd.DataFrame]:
     """
     this function is a wrapper of calculating PSDF of a pp branch from the phase shift through a pp branch
     with pypower matrix function or perturb function.
@@ -207,10 +205,10 @@ def run_PSDF(
         DataFrame(data=psdf, index=goal_branch_pp_index, columns=phase_shift_branch_ix)}
     """
     # ToDo: check if distributed slack makes any difference
-    if perturb and phase_shift_branch_type is None:
-        logger.info("If a lot of branch required in psdf, please set perturb to False!")
 
     if perturb:
+        if phase_shift_branch_type is None:
+            logger.info("If a lot of branch required in psdf, please set perturb to False!")
         if recycle == "lodf" and distributed_slack == True:
             logger.warning("distributed_slack deactivated! recycling does not allow distributed slack")
         psdf = _get_PSDF_perturb(

pandapower/analysis/PTDF.py

@@ -2,14 +2,13 @@
 
 # Copyright (c) 2016-2025 by University of Kassel and Fraunhofer Institute for Energy Economics
 # and Energy System Technology (IEE), Kassel. All rights reserved.
-
-from typing import Union, List, Dict
 from copy import deepcopy
 
 import pandas as pd
 import numpy as np
+import numpy.typing as npt
 
-from pandapower import pandapowerNet
+from pandapower.auxiliary import pandapowerNet
 from pandapower.analysis.utils import _get_bus_lookup, _get_branch_lookup, _get_trafo3w_lookup, \
     branch_dict_to_ppci_branch_list, _get_source_bus_ix, DISCONNECTED_PADDING_VALUE, BR_SIDE_MAPPING, BR_SIDE_MAPPING_1, \
     ELE_IX_TYPE
@@ -22,19 +21,18 @@
 # replace pandapower makePTDF with custom function
 from pandapower.pypower.makePTDF import makePTDF
 from pandapower.analysis.utils import get_dist_slack, get_ppci_dist_slack, LOAD_REFRENCE
-from pandapower.auxiliary import pandapowerNet
 
 import logging
 logger = logging.getLogger(__name__)
 
 
 def _get_PTDF_direct(
         net: pandapowerNet,
-        source_bus: Union[int, np.ndarray] | None = None,
+        source_bus: int | npt.NDArray | None = None,
         result_side=0,
         using_sparse_solver: bool = True,
         random_verify: bool = True,
-        branch_dict: dict[str, Union[list[int], None]] | None = None,
+        branch_dict: dict[str, list[int] | None] | None = None,
         reduced: bool = True
 ):
     """
@@ -87,7 +85,7 @@ def _get_PTDF_direct(
 
 def _get_PTDF_perturb(
         net: pandapowerNet,
-        source_bus: Union[int, np.ndarray] | None = None,
+        source_bus: int | npt.NDArray | None = None,
         result_side: int = 0,
         distributed_slack: bool=True
 ):
@@ -332,13 +330,13 @@ def _PTDF_ppci_to_pp(net, ptdf_ppci, result_side, branch_ppci_lookup=None):
 
 def run_PTDF(
     net: pandapowerNet,
-    source_bus: Union[int, np.ndarray] | None = None,
+    source_bus: int | npt.NDArray | None = None,
     distributed_slack: bool = True,
     result_side: int = 0,
     perturb: bool = False,
     using_sparse_solver: bool = True,
     random_verify: bool = False,
-    branch_dict: dict[str, Union[list[int], None]] | None = None,
+    branch_dict: dict[str, list[int] | None] | None = None,
     reduced: bool = True,
 ):
     """