File size: 5,049 Bytes
55e3496
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7477316
 
55e3496
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7477316
 
 
 
 
 
 
 
 
 
 
 
 
 
55e3496
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7477316
 
55e3496
 
 
 
7477316
 
55e3496
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7477316
55e3496
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
"""
Grid Loader — IEEE Test Systems via pandapower
Loads IEEE 33-bus (case33bw) and IEEE 118-bus (case118) systems directly
from pandapower's built-in network library.  No external file I/O required.
"""
from __future__ import annotations

import copy
from typing import Literal

import pandas as pd
import pandapower as pp
import pandapower.networks as pn


_LOADERS = {
    "case33bw": pn.case33bw,
    "case118": pn.case118,
}

# IEEE 33-bus tie line indices (out of service in default config)
IEEE33_TIE_LINES = [32, 33, 34, 35, 36]
# Default open lines for IEEE 33-bus in OptiQ (override to 3 OOS)
IEEE33_DEFAULT_OPEN_LINES = [32, 33, 34]


def load_network(system: Literal["case33bw", "case118"] = "case33bw") -> pp.pandapowerNet:
    """Return a fresh pandapower network for the given IEEE test system.

    Parameters
    ----------
    system : str
        One of ``"case33bw"`` (IEEE 33-bus distribution) or
        ``"case118"`` (IEEE 118-bus transmission).

    Returns
    -------
    pp.pandapowerNet
        A ready-to-simulate network object.
    """
    loader = _LOADERS.get(system)
    if loader is None:
        raise ValueError(f"Unknown system '{system}'. Choose from {list(_LOADERS)}")
    net = loader()
    net["optiq_system"] = system
    net["optiq_is_distribution"] = system == "case33bw"

    if system == "case33bw":
        # Ensure only 3 lines are open by default for 33-bus
        net["optiq_tie_lines"] = list(IEEE33_TIE_LINES)
        net["optiq_default_open_lines"] = list(IEEE33_DEFAULT_OPEN_LINES)
        net.line["in_service"] = True
        for idx in IEEE33_DEFAULT_OPEN_LINES:
            if idx in net.line.index:
                net.line.at[idx, "in_service"] = False

    return net


def clone_network(net: pp.pandapowerNet) -> pp.pandapowerNet:
    """Deep-copy a pandapower network (useful for parallel scenarios)."""
    return copy.deepcopy(net)


def get_line_info(net: pp.pandapowerNet) -> dict:
    """Identify switchable lines (all lines), in-service and out-of-service.

    In IEEE 33-bus, reconfiguration is modelled by toggling ``line.in_service``.
    Lines that are out of service in the default config are the tie lines.

    Returns
    -------
    dict with keys:
        ``"all"``          – all line indices
        ``"in_service"``   – indices of lines currently in service (closed)
        ``"out_of_service"`` – indices of lines currently out of service (open/tie)
        ``"tie_lines"``    – the default tie line indices (for IEEE 33-bus)
    """
    all_idx = net.line.index.tolist()
    in_svc = net.line.index[net.line.in_service].tolist()
    out_svc = net.line.index[~net.line.in_service].tolist()
    tie_lines = net.get("optiq_tie_lines", out_svc)
    n_required_open = len(net.line) - (len(net.bus) - 1)
    return {
        "all": all_idx,
        "in_service": in_svc,
        "out_of_service": out_svc,
        "tie_lines": list(tie_lines),
        "n_required_open": n_required_open,
    }


def get_network_summary(net: pp.pandapowerNet) -> dict:
    """Return a JSON-serialisable summary of the network structure."""
    line_info = get_line_info(net)
    return {
        "n_buses": len(net.bus),
        "n_lines": len(net.line),
        "n_lines_in_service": len(line_info["in_service"]),
        "n_tie_lines": len(line_info["out_of_service"]),
        "n_generators": len(net.gen) + len(net.ext_grid),
        "n_loads": len(net.load),
        "total_load_mw": float(net.load.p_mw.sum()),
        "total_load_mvar": float(net.load.q_mvar.sum()),
        "tie_line_indices": line_info["out_of_service"],
    }


def get_topology_data(net: pp.pandapowerNet) -> list[dict]:
    """Return line data for topology visualization.

    Each entry has: index, from_bus, to_bus, r_ohm, x_ohm, in_service, is_tie.
    """
    line_info = get_line_info(net)
    tie_set = set(line_info["tie_lines"])
    rows = []
    for idx, row in net.line.iterrows():
        rows.append({
            "index": int(idx),
            "from_bus": int(row["from_bus"]),
            "to_bus": int(row["to_bus"]),
            "r_ohm_per_km": float(row["r_ohm_per_km"]),
            "x_ohm_per_km": float(row["x_ohm_per_km"]),
            "length_km": float(row["length_km"]),
            "in_service": bool(row["in_service"]),
            "is_tie": idx in tie_set,
        })
    return rows


def get_bus_data(net: pp.pandapowerNet) -> list[dict]:
    """Return bus data for visualization."""
    rows = []
    for idx, row in net.bus.iterrows():
        load_p = 0.0
        load_q = 0.0
        if len(net.load) > 0:
            bus_loads = net.load[net.load.bus == idx]
            load_p = float(bus_loads.p_mw.sum())
            load_q = float(bus_loads.q_mvar.sum())
        rows.append({
            "index": int(idx),
            "name": str(row.get("name", f"Bus {idx}")),
            "vn_kv": float(row["vn_kv"]),
            "load_mw": load_p,
            "load_mvar": load_q,
            "is_slack": int(idx) in net.ext_grid.bus.values,
        })
    return rows