diff --git a/CHANGELOG.rst b/CHANGELOG.rst index bb5af3dc7..4f6fc35e1 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -3,6 +3,7 @@ Change Log [upcoming release] - 2026-..-.. ------------------------------- +- [ADDED] toolbox: :code:`compute_switch_flows` computes power flow through zero-impedance bus-bus switches via nodal balance - [FIXED] runopp(init="results") now preserves the warm-start vector in the PIPS-backed AC OPF solver - [ADDED] added more functions to diagnostic - [ADDED] check to check if vkr_percent values are reasonable (see issue #786). diff --git a/doc/elements/switch.rst b/doc/elements/switch.rst index 8df2f0fc8..ef795b387 100644 --- a/doc/elements/switch.rst +++ b/doc/elements/switch.rst @@ -1,4 +1,4 @@ -.. _switch_model: +.. _switch_model: ============= Switch @@ -41,6 +41,13 @@ This has the following advantages compared to modelling the switch as a small im - there is no voltage drop over the switch (ideal switch) - no convergence problems due to small impedances / large admittances - less buses in the admittance matrix + +.. note:: + + Because fused buses share one internal node, ``res_switch`` contains NaN for + bus-bus switches with ``z_ohm=0`` after ``runpp()``. To compute the power + flow through these switches via nodal balance, call + :func:`pandapower.toolbox.compute_switch_flows` after the load flow. *Bus-Element-Switches:* diff --git a/doc/toolbox.rst b/doc/toolbox.rst index c69421e7e..626a7602b 100644 --- a/doc/toolbox.rst +++ b/doc/toolbox.rst @@ -47,6 +47,8 @@ Result Information .. autofunction:: pandapower.toolbox.clear_result_tables +.. autofunction:: pandapower.toolbox.compute_switch_flows + .. autofunction:: pandapower.toolbox.res_power_columns ==================================== diff --git a/pandapower/test/toolbox/test_compute_switch_flows.py b/pandapower/test/toolbox/test_compute_switch_flows.py new file mode 100644 index 000000000..1f92a7391 --- /dev/null +++ b/pandapower/test/toolbox/test_compute_switch_flows.py @@ -0,0 +1,515 @@ +# -*- coding: utf-8 -*- + +# Copyright (c) 2016-2026 by University of Kassel and Fraunhofer Institute for Energy Economics +# and Energy System Technology (IEE), Kassel. All rights reserved. + +import numpy as np +import pytest + +import pandapower as pp +from pandapower.create import ( + create_empty_network, create_bus, create_ext_grid, create_line_from_parameters, + create_load, create_switch, create_gen, create_transformer_from_parameters, + create_sgen, create_shunt, +) +from pandapower.run import runpp +from pandapower.toolbox import compute_switch_flows + + +def _make_two_bus_coupler_net(): + """Two buses connected by a zero-impedance switch, load on one side. + + ext_grid (bus 0) --line-- (bus 1) --switch-- (bus 2) --load + """ + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0, name="slack") + b1 = create_bus(net, vn_kv=20.0, name="bus1") + b2 = create_bus(net, vn_kv=20.0, name="bus2") + + create_ext_grid(net, b0, vm_pu=1.0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + create_switch(net, bus=b1, element=b2, et="b", closed=True) + create_load(net, b2, p_mw=1.0, q_mvar=0.5) + return net, b0, b1, b2 + + +class TestComputeSwitchFlowsBasic: + """Basic scenarios: single coupler, open switches, not-converged.""" + + def test_single_coupler_power_balance(self): + """The switch must carry the entire load since it's the only path.""" + net, b0, b1, b2 = _make_two_bus_coupler_net() + runpp(net) + assert np.isnan(net.res_switch.at[0, "p_from_mw"]) + + compute_switch_flows(net) + + p_from = net.res_switch.at[0, "p_from_mw"] + q_from = net.res_switch.at[0, "q_from_mvar"] + p_to = net.res_switch.at[0, "p_to_mw"] + q_to = net.res_switch.at[0, "q_to_mvar"] + i_ka = net.res_switch.at[0, "i_ka"] + + assert not np.isnan(p_from) + assert not np.isnan(i_ka) + + # Power balance: from + to ≈ 0 (zero-impedance, no losses) + assert np.isclose(p_from + p_to, 0, atol=1e-10) + assert np.isclose(q_from + q_to, 0, atol=1e-10) + + # Switch must carry approximately the load power + load_p = net.res_load.at[0, "p_mw"] + load_q = net.res_load.at[0, "q_mvar"] + assert np.isclose(abs(p_from), load_p, atol=1e-6) + assert np.isclose(abs(q_from), load_q, atol=1e-6) + + # Current must be positive + assert i_ka > 0 + + def test_open_switch_zero_flow(self): + """An open bus-bus switch must have zero flow (set by runpp already).""" + net, b0, b1, b2 = _make_two_bus_coupler_net() + net.switch.at[0, "closed"] = False + # Add line so bus 2 is still connected (otherwise non-convergence) + create_line_from_parameters(net, b1, b2, length_km=1, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + runpp(net) + compute_switch_flows(net) + + # Open switches should remain zero / not be filled with flow + assert net.res_switch.at[0, "i_ka"] == 0 + + def test_not_converged_raises(self): + """Must raise when load flow has not converged.""" + net, _, _, _ = _make_two_bus_coupler_net() + # Don't run load flow + net.converged = False + with pytest.raises(UserWarning, match="did not converge"): + compute_switch_flows(net) + + def test_no_switches(self): + """Net without switches should return silently.""" + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + create_load(net, b1, p_mw=1.0) + runpp(net) + compute_switch_flows(net) # should not raise + + def test_impedance_switch_not_overwritten(self): + """Switches with z_ohm > 0 already have results; they must not change.""" + net, b0, b1, b2 = _make_two_bus_coupler_net() + net.switch.at[0, "z_ohm"] = 0.001 + runpp(net) + + p_before = net.res_switch.at[0, "p_from_mw"] + compute_switch_flows(net) + p_after = net.res_switch.at[0, "p_from_mw"] + + assert np.isclose(p_before, p_after) + + +class TestComputeSwitchFlowsChain: + """Multiple couplers in series (chain topology).""" + + def test_three_bus_chain(self): + """ext_grid -- (b0) --line-- (b1) --sw0-- (b2) --sw1-- (b3) --load + + sw0 must carry the full load; sw1 must also carry the full load. + """ + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + b3 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + sw0 = create_switch(net, bus=b1, element=b2, et="b") + sw1 = create_switch(net, bus=b2, element=b3, et="b") + create_load(net, b3, p_mw=2.0, q_mvar=1.0) + + runpp(net) + compute_switch_flows(net) + + load_p = net.res_load.at[0, "p_mw"] + load_q = net.res_load.at[0, "q_mvar"] + + # Both switches carry the full load + for sw in [sw0, sw1]: + assert np.isclose(abs(net.res_switch.at[sw, "p_from_mw"]), load_p, atol=1e-6) + assert np.isclose(abs(net.res_switch.at[sw, "q_from_mvar"]), load_q, atol=1e-6) + assert net.res_switch.at[sw, "i_ka"] > 0 + + def test_chain_with_intermediate_load(self): + """(b0)--line--(b1)--sw0--(b2)--sw1--(b3) + ext_grid load1=1MW load2=2MW + + sw0 must carry 3 MW, sw1 must carry 2 MW. + """ + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + b3 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + sw0 = create_switch(net, bus=b1, element=b2, et="b") + sw1 = create_switch(net, bus=b2, element=b3, et="b") + create_load(net, b2, p_mw=1.0) + create_load(net, b3, p_mw=2.0) + + runpp(net) + compute_switch_flows(net) + + # sw1 carries only load at b3 + assert np.isclose(abs(net.res_switch.at[sw1, "p_from_mw"]), + net.res_load.at[1, "p_mw"], atol=1e-6) + # sw0 carries both loads + total_load = net.res_load.at[0, "p_mw"] + net.res_load.at[1, "p_mw"] + assert np.isclose(abs(net.res_switch.at[sw0, "p_from_mw"]), total_load, atol=1e-6) + + +class TestComputeSwitchFlowsBranching: + """Tree topologies with branches.""" + + def test_t_junction(self): + """ load1=1MW + | + ext_grid--(b0)--line--(b1)--sw0--(b2)--sw1--(b3)--load2=2MW + | + sw2 + | + (b4)--load3=3MW + + sw0 carries 1+2+3=6 MW, sw1 carries 2 MW, sw2 carries 3 MW. + """ + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + b3 = create_bus(net, vn_kv=20.0) + b4 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + sw0 = create_switch(net, bus=b1, element=b2, et="b") + sw1 = create_switch(net, bus=b2, element=b3, et="b") + sw2 = create_switch(net, bus=b2, element=b4, et="b") + + create_load(net, b2, p_mw=1.0) + create_load(net, b3, p_mw=2.0) + create_load(net, b4, p_mw=3.0) + + runpp(net) + compute_switch_flows(net) + + l0 = net.res_load.at[0, "p_mw"] + l1 = net.res_load.at[1, "p_mw"] + l2 = net.res_load.at[2, "p_mw"] + + assert np.isclose(abs(net.res_switch.at[sw0, "p_from_mw"]), l0 + l1 + l2, atol=1e-6) + assert np.isclose(abs(net.res_switch.at[sw1, "p_from_mw"]), l1, atol=1e-6) + assert np.isclose(abs(net.res_switch.at[sw2, "p_from_mw"]), l2, atol=1e-6) + + +class TestComputeSwitchFlowsGenAndBranch: + """Scenarios with generators and outgoing branches within a fused group.""" + + def test_generator_on_fused_bus(self): + """Generator on one fused bus, load on the other. + + (b0)--line--(b1)--sw--(b2) + ext_grid gen=5MW load=3MW + + The switch should carry |gen - load| = 2 MW toward the line. + But sign depends on direction: the gen produces 5MW at b1, + load consumes 3MW at b2. Net demand at b2 is 3MW, so the + switch carries 3MW from b1 to b2. + """ + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + sw = create_switch(net, bus=b1, element=b2, et="b") + create_gen(net, b1, p_mw=5.0, vm_pu=1.0) + create_load(net, b2, p_mw=3.0) + + runpp(net) + compute_switch_flows(net) + + load_p = net.res_load.at[0, "p_mw"] + assert np.isclose(abs(net.res_switch.at[sw, "p_from_mw"]), load_p, atol=1e-6) + + def test_branch_leaving_fused_group(self): + """Branch exits from a fused bus. + + ext_grid--(b0)--line0--(b1)--sw--(b2)--line1--(b3)--load + """ + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + b3 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + sw = create_switch(net, bus=b1, element=b2, et="b") + create_line_from_parameters(net, b2, b3, length_km=5, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + create_load(net, b3, p_mw=2.0, q_mvar=1.0) + + runpp(net) + compute_switch_flows(net) + + # The switch must carry what the outgoing line takes from b2 + line1_p = abs(net.res_line.at[1, "p_from_mw"]) + line1_q = abs(net.res_line.at[1, "q_from_mvar"]) + sw_p = abs(net.res_switch.at[sw, "p_from_mw"]) + sw_q = abs(net.res_switch.at[sw, "q_from_mvar"]) + + assert np.isclose(sw_p, line1_p, atol=1e-6) + assert np.isclose(sw_q, line1_q, atol=1e-6) + + def test_sgen_and_shunt(self): + """Ensure static generators and shunts are accounted for.""" + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + sw = create_switch(net, bus=b1, element=b2, et="b") + create_sgen(net, b2, p_mw=2.0, q_mvar=0.0) + create_shunt(net, b2, q_mvar=-0.5, p_mw=0.0) + create_load(net, b2, p_mw=5.0, q_mvar=1.0) + + runpp(net) + compute_switch_flows(net) + + # Net demand at b2: load - sgen + shunt + net_p_b2 = net.res_load.at[0, "p_mw"] - net.res_sgen.at[0, "p_mw"] + net.res_shunt.at[0, "p_mw"] + assert np.isclose(abs(net.res_switch.at[sw, "p_from_mw"]), abs(net_p_b2), atol=1e-4) + + +class TestComputeSwitchFlowsCycleDetection: + """Cycle detection for parallel zero-impedance paths.""" + + def test_cycle_raises(self): + """Two parallel zero-impedance switches between the same buses must raise.""" + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + # Two parallel switches b1 <-> b2 + create_switch(net, bus=b1, element=b2, et="b") + create_switch(net, bus=b1, element=b2, et="b") + create_load(net, b2, p_mw=1.0) + + runpp(net) + with pytest.raises(ValueError, match="cycle"): + compute_switch_flows(net) + + def test_loop_of_three_raises(self): + """Three buses in a loop: b1--sw--b2--sw--b3--sw--b1.""" + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + b3 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + create_switch(net, bus=b1, element=b2, et="b") + create_switch(net, bus=b2, element=b3, et="b") + create_switch(net, bus=b3, element=b1, et="b") + create_load(net, b2, p_mw=1.0) + + runpp(net) + with pytest.raises(ValueError, match="cycle"): + compute_switch_flows(net) + + +class TestComputeSwitchFlowsSignConvention: + """Verify sign convention matches Pandapower's from/to convention.""" + + def test_sign_convention(self): + """Power flows from bus column toward element column when load is on element side.""" + net, b0, b1, b2 = _make_two_bus_coupler_net() + runpp(net) + compute_switch_flows(net) + + # Switch: bus=b1, element=b2; load on b2 + # Power should flow from b1 to b2, so p_from > 0, p_to < 0 + p_from = net.res_switch.at[0, "p_from_mw"] + p_to = net.res_switch.at[0, "p_to_mw"] + assert p_from > 0, "Power should flow from bus toward element (load side)" + assert p_to < 0 + + +class TestComputeSwitchFlowsValidation: + """Cross-validate against z_ohm > 0 results for small impedance.""" + + def test_cross_validate_with_impedance(self): + """Results from nodal balance should be close to results with small z_ohm. + + Uses z_ohm=0.01 (not smaller, since very low impedance branches can + cause convergence issues in Newton-Raphson for small test networks). + """ + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + create_switch(net, bus=b1, element=b2, et="b") + create_load(net, b2, p_mw=1.5, q_mvar=0.5) + + # Run with z_ohm=0 + nodal balance + runpp(net) + compute_switch_flows(net) + p_nodal = net.res_switch.at[0, "p_from_mw"] + q_nodal = net.res_switch.at[0, "q_from_mvar"] + i_nodal = net.res_switch.at[0, "i_ka"] + + # Run with small z_ohm for comparison + net.switch.at[0, "z_ohm"] = 0.01 + runpp(net) + p_z = net.res_switch.at[0, "p_from_mw"] + q_z = net.res_switch.at[0, "q_from_mvar"] + i_z = net.res_switch.at[0, "i_ka"] + + # Tolerances account for the non-zero impedance introducing small losses + assert np.isclose(p_nodal, p_z, rtol=1e-2), \ + f"P mismatch: nodal={p_nodal:.6f}, z_ohm={p_z:.6f}" + assert np.isclose(q_nodal, q_z, rtol=1e-2), \ + f"Q mismatch: nodal={q_nodal:.6f}, z_ohm={q_z:.6f}" + assert np.isclose(i_nodal, i_z, rtol=5e-2), \ + f"I mismatch: nodal={i_nodal:.6f}, z_ohm={i_z:.6f}" + + +class TestComputeSwitchFlowsMultipleGroups: + """Independent fused groups should be computed independently.""" + + def test_two_independent_groups(self): + """Two separate fused groups with different loads.""" + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + b3 = create_bus(net, vn_kv=20.0) + b4 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + create_line_from_parameters(net, b0, b3, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + + sw0 = create_switch(net, bus=b1, element=b2, et="b") + sw1 = create_switch(net, bus=b3, element=b4, et="b") + + create_load(net, b2, p_mw=1.0) + create_load(net, b4, p_mw=3.0) + + runpp(net) + compute_switch_flows(net) + + assert np.isclose(abs(net.res_switch.at[sw0, "p_from_mw"]), + net.res_load.at[0, "p_mw"], atol=1e-6) + assert np.isclose(abs(net.res_switch.at[sw1, "p_from_mw"]), + net.res_load.at[1, "p_mw"], atol=1e-6) + + +class TestComputeSwitchFlowsLoadingPercent: + """Loading percent computation when in_ka is available.""" + + def test_loading_percent(self): + net, b0, b1, b2 = _make_two_bus_coupler_net() + in_ka = 0.1 + net.switch.at[0, "in_ka"] = in_ka + runpp(net) + compute_switch_flows(net) + + i_ka = net.res_switch.at[0, "i_ka"] + expected_loading = i_ka / in_ka * 100 + assert np.isclose(net.res_switch.at[0, "loading_percent"], expected_loading, atol=1e-6) + + +class TestComputeSwitchFlowsDcline: + """DC line as a branch leaving a fused group.""" + + def test_dcline_outflow(self): + """Power leaving through a dcline must be accounted for. + + ext_grid--(b0)--line--(b1)--sw--(b2)--dcline--(b3)--load + """ + from pandapower.create import create_dcline + + net = create_empty_network() + b0 = create_bus(net, vn_kv=20.0) + b1 = create_bus(net, vn_kv=20.0) + b2 = create_bus(net, vn_kv=20.0) + b3 = create_bus(net, vn_kv=20.0) + + create_ext_grid(net, b0) + create_ext_grid(net, b3, vm_pu=1.0) + create_line_from_parameters(net, b0, b1, length_km=10, r_ohm_per_km=0.1, + x_ohm_per_km=0.1, c_nf_per_km=0, max_i_ka=1) + sw = create_switch(net, bus=b1, element=b2, et="b") + create_dcline(net, from_bus=b2, to_bus=b3, p_mw=5.0, loss_percent=1.0, + loss_mw=0.1, vm_from_pu=1.0, vm_to_pu=1.0) + create_load(net, b3, p_mw=10.0) + + runpp(net) + compute_switch_flows(net) + + # Switch must carry the dcline from-side flow + p_dcline = abs(net.res_dcline.at[0, "p_from_mw"]) + p_sw = abs(net.res_switch.at[sw, "p_from_mw"]) + assert np.isclose(p_sw, p_dcline, atol=1e-4), \ + f"Switch P={p_sw:.4f} should match dcline from P={p_dcline:.4f}" + assert net.res_switch.at[sw, "i_ka"] > 0 + + +class TestComputeSwitchFlowsDeenergized: + """De-energized fused groups should be skipped gracefully.""" + + def test_deenergized_group_skipped(self): + """Fused group with vm=0 should not produce results (no crash).""" + net, b0, b1, b2 = _make_two_bus_coupler_net() + runpp(net) + + # Manually set bus voltage to 0 to simulate de-energized state + net.res_bus.at[b1, "vm_pu"] = 0.0 + net.res_bus.at[b2, "vm_pu"] = 0.0 + + compute_switch_flows(net) + # Should remain NaN (skipped), not crash + assert np.isnan(net.res_switch.at[0, "p_from_mw"]) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/pandapower/toolbox/result_info.py b/pandapower/toolbox/result_info.py index 11484f958..2c826912b 100644 --- a/pandapower/toolbox/result_info.py +++ b/pandapower/toolbox/result_info.py @@ -413,6 +413,245 @@ def clear_result_tables(net): net[key] = net[key].drop(net[key].index) +def compute_switch_flows(net): + """Compute power flow through zero-impedance bus-bus switches via nodal balance. + + After a converged load flow, switches with ``z_ohm=0`` (the default) have + ``NaN`` values in ``res_switch`` because Pandapower fuses the adjacent buses + into a single internal node. This function reconstructs the individual + switch flows by calculating the net local injection at each bus within a + fused group and propagating the residual through the switch tree. + + The function writes ``p_from_mw``, ``q_from_mvar``, ``p_to_mw``, + ``q_to_mvar``, and ``i_ka`` into ``net.res_switch`` for every closed + bus-bus switch that has ``z_ohm <= 0``. Results for open switches are set + to zero. Switches that already have results (``z_ohm > 0``) are not + modified. + + **Assumption:** Within each fused-bus group the zero-impedance switches + must form a *tree* (no parallel zero-impedance paths between the same pair + of buses). If a cycle is detected a ``ValueError`` is raised, since the + flow split is physically indeterminate without impedance information. + + Parameters + ---------- + net : pandapowerNet + A pandapower network with valid load flow results + (``net.converged is True``). + + Raises + ------ + UserWarning + If ``net.converged`` is ``False``. + ValueError + If zero-impedance bus-bus switches form a cycle within a fused group. + + Examples + -------- + >>> import pandapower as pp + >>> import pandapower.networks as pn + >>> net = pn.example_simple() + >>> pp.runpp(net) + >>> from pandapower.toolbox import compute_switch_flows + >>> compute_switch_flows(net) + >>> net.res_switch # p_from_mw, q_from_mvar etc. now populated + """ + from collections import defaultdict + + if not net.converged: + raise UserWarning("Power flow did not converge, results are invalid.") + + if len(net.switch) == 0: + return + + bus_lookup = getattr(net, "_pd2ppc_lookups", {}).get("bus") + if bus_lookup is None: + return + + # Identify fused-bus groups: ppc_bus_id -> set of original bus indices + fused_groups = defaultdict(set) + for orig_bus, ppc_bus in enumerate(bus_lookup): + fused_groups[int(ppc_bus)].add(orig_bus) + + multi_groups = {k: v for k, v in fused_groups.items() if len(v) > 1} + if not multi_groups: + return + + # Per-bus net local consumption: positive = power leaving the bus + bus_p = defaultdict(float) + bus_q = defaultdict(float) + + _bus_element_names = [ + ("load", 1.0), ("motor", 1.0), + ("shunt", 1.0), ("ward", 1.0), ("xward", 1.0), + ("ext_grid", -1.0), ("gen", -1.0), ("sgen", -1.0), ("storage", -1.0), + ] + for tbl_name, sign in _bus_element_names: + tbl = net.get(tbl_name) + res = net.get("res_%s" % tbl_name) + if tbl is None or res is None or len(tbl) == 0 or len(res) == 0: + continue + in_service = tbl["in_service"].values if "in_service" in tbl.columns else np.ones(len(tbl), dtype=bool) + buses = tbl["bus"].values + for i, idx in enumerate(tbl.index): + if not in_service[i]: + continue + b = int(buses[i]) + try: + bus_p[b] += sign * float(res.at[idx, "p_mw"]) + bus_q[b] += sign * float(res.at[idx, "q_mvar"]) + except (KeyError, ValueError): + pass + + # Map every bus to its fused group + bus_to_group = {} + for grp_id, buses in fused_groups.items(): + for b in buses: + bus_to_group[b] = grp_id + + # Add power carried by branches that leave the fused group + _branch_specs = [ + ("line", "res_line", [("from_bus", "p_from_mw", "q_from_mvar"), + ("to_bus", "p_to_mw", "q_to_mvar")]), + ("trafo", "res_trafo", [("hv_bus", "p_hv_mw", "q_hv_mvar"), + ("lv_bus", "p_lv_mw", "q_lv_mvar")]), + ("trafo3w", "res_trafo3w", [("hv_bus", "p_hv_mw", "q_hv_mvar"), + ("mv_bus", "p_mv_mw", "q_mv_mvar"), + ("lv_bus", "p_lv_mw", "q_lv_mvar")]), + ("impedance", "res_impedance", [("from_bus", "p_from_mw", "q_from_mvar"), + ("to_bus", "p_to_mw", "q_to_mvar")]), + ("dcline", "res_dcline", [("from_bus", "p_from_mw", "q_from_mvar"), + ("to_bus", "p_to_mw", "q_to_mvar")]), + ] + for tbl_name, res_name, bus_pairs in _branch_specs: + tbl = net.get(tbl_name) + res = net.get(res_name) + if tbl is None or res is None or len(tbl) == 0 or len(res) == 0: + continue + in_service = tbl["in_service"].values if "in_service" in tbl.columns else np.ones(len(tbl), dtype=bool) + for i, idx in enumerate(tbl.index): + if not in_service[i]: + continue + ends = [(int(tbl.at[idx, bc]), pc, qc) for bc, pc, qc in bus_pairs] + for j, (this_bus, p_col, q_col) in enumerate(ends): + this_grp = bus_to_group.get(this_bus) + if this_grp is None or this_grp not in multi_groups: + continue + other_buses = [ends[k][0] for k in range(len(ends)) if k != j] + if all(bus_to_group.get(ob) == this_grp for ob in other_buses): + continue + try: + bus_p[this_bus] += float(res.at[idx, p_col]) + bus_q[this_bus] += float(res.at[idx, q_col]) + except (KeyError, ValueError): + pass + + # Index closed zero-impedance bus-bus switches by fused group + sw_by_group = defaultdict(list) + z_ohm = net.switch["z_ohm"].values if "z_ohm" in net.switch.columns else np.zeros(len(net.switch)) + for sw_idx in net.switch.index: + row = net.switch.loc[sw_idx] + if row["et"] != "b" or not row["closed"] or z_ohm[sw_idx] > 0: + continue + a = int(row["bus"]) + b = int(row["element"]) + grp = bus_to_group.get(a) + if grp is not None and grp in multi_groups: + sw_by_group[grp].append((sw_idx, a, b)) + + computed_switches = set() + + # For each fused group, build the coupler subgraph and compute flows + for grp_id, grp_buses in multi_groups.items(): + couplers = sw_by_group.get(grp_id, []) + if not couplers: + continue + + # Skip de-energized fused groups (vm ≈ 0) + sample_bus = next(iter(grp_buses)) + if sample_bus in net.res_bus.index: + vm_pu = net.res_bus.at[sample_bus, "vm_pu"] + if vm_pu == 0 or np.isnan(vm_pu): + continue + + adj = defaultdict(list) + coupler_buses = set() + for sw_idx, a, b in couplers: + adj[a].append((b, sw_idx)) + adj[b].append((a, sw_idx)) + coupler_buses.add(a) + coupler_buses.add(b) + + # A tree with N nodes has exactly N-1 edges; more means a cycle + if len(couplers) >= len(coupler_buses): + raise ValueError( + "Zero-impedance bus-bus switches form a cycle in fused " + "group containing buses %s. The flow split is " + "indeterminate without impedance values." % sorted(grp_buses)) + + # DFS to build tree order + root = next(iter(grp_buses)) + visited = set() + stack = [(root, None, None)] + order = [] + while stack: + bus, parent, parent_sw = stack.pop() + if bus in visited: + continue + visited.add(bus) + order.append((bus, parent, parent_sw)) + for nb, sw_idx in adj.get(bus, []): + if nb not in visited: + stack.append((nb, bus, sw_idx)) + + # Accumulate subtree demand from leaves to root + subtree_p = {} + subtree_q = {} + for bus, parent, parent_sw in reversed(order): + sp = bus_p.get(bus, 0.0) + sq = bus_q.get(bus, 0.0) + for nb, sw_idx in adj.get(bus, []): + if nb != parent and nb in subtree_p: + sp += subtree_p[nb] + sq += subtree_q[nb] + subtree_p[bus] = sp + subtree_q[bus] = sq + if parent_sw is not None: + sw_bus_col = int(net.switch.at[parent_sw, "bus"]) + sw_elem_col = int(net.switch.at[parent_sw, "element"]) + if bus == sw_elem_col: + p_from, q_from = sp, sq + else: + p_from, q_from = -sp, -sq + p_to, q_to = -p_from, -q_from + + net.res_switch.at[parent_sw, "p_from_mw"] = p_from + net.res_switch.at[parent_sw, "q_from_mvar"] = q_from + net.res_switch.at[parent_sw, "p_to_mw"] = p_to + net.res_switch.at[parent_sw, "q_to_mvar"] = q_to + + # Derive current from apparent power and bus voltage + s_mva = np.sqrt(p_from ** 2 + q_from ** 2) + b_from = int(net.switch.at[parent_sw, "bus"]) + vm_pu = net.res_bus.at[b_from, "vm_pu"] if b_from in net.res_bus.index else np.nan + vn_kv = net.bus.at[b_from, "vn_kv"] if b_from in net.bus.index else np.nan + vm_kv = vm_pu * vn_kv + if vm_kv > 0: + i_ka = s_mva / (vm_kv * np.sqrt(3)) + else: + i_ka = np.nan + net.res_switch.at[parent_sw, "i_ka"] = i_ka + computed_switches.add(parent_sw) + + if computed_switches and "in_ka" in net.switch.columns and \ + "loading_percent" in net.res_switch.columns: + for sw_idx in computed_switches: + in_val = net.switch.at[sw_idx, "in_ka"] + if not np.isnan(in_val) and in_val > 0: + i_val = net.res_switch.at[sw_idx, "i_ka"] + net.res_switch.at[sw_idx, "loading_percent"] = i_val / in_val * 100 + + def res_power_columns(element_type, side=0): """Returns columns names of result tables for active and reactive power