Skip to content

[DEV-8827] Fixed errors in handling phase energisation of LinearShuntCompensator instances with a grounding_terminal #224

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
charlta merged 1 commit into from
May 8, 2026
Merged

[DEV-8827] Fixed errors in handling phase energisation of LinearShuntCompensator instances with a grounding_terminal #224

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion changelog.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
* None.

### Fixes
* None.
* Fixed errors in handling phase energisation of `LinearShuntCompensator` instances with a `grounding_terminal`.

### Notes
* None.
Expand Down
71 changes: 38 additions & 33 deletions src/zepben/ewb/services/network/tracing/phases/set_phases.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from functools import singledispatchmethod
from typing import Union, Set, Iterable, List, Type, TYPE_CHECKING, Optional, Callable, Any

from zepben.ewb import PhaseStatus, add_neutral
from zepben.ewb import PhaseStatus, add_neutral, stop_on_shunt_compensator_ground
from zepben.ewb.exceptions import TracingException, PhaseException
from zepben.ewb.model.cim.iec61970.base.core.phase_code import PhaseCode
from zepben.ewb.model.cim.iec61970.base.core.terminal import Terminal
Expand Down Expand Up @@ -55,7 +55,7 @@ async def run(
self,
target: Union[NetworkService, Terminal],
phases: Union[PhaseCode, Iterable[SinglePhaseKind]] = None,
network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL
network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL,
):
"""

Expand All @@ -70,7 +70,7 @@ async def run(
async def _(
self,
network: NetworkService,
network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL
network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL,
):
"""
Apply phases and flow from all energy sources in the network.
Expand All @@ -94,7 +94,7 @@ async def _(
start_terminal: Terminal,
phases: Union[PhaseCode, List[SinglePhaseKind], Set[SinglePhaseKind]] = None,
network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL,
seed_terminal: Terminal = None
seed_terminal: Terminal = None,
):
"""
Apply phases to the `start_terminal` and flow, optionally specifying a `seed_terminal`. If specified, the `seed_terminal`
Expand Down Expand Up @@ -124,7 +124,7 @@ async def _(
if len(phases) != len(start_terminal.phases.single_phases):
raise TracingException(
f"Attempted to apply phases [{', '.join(phase.name for phase in phases)}] to {start_terminal} with nominal phases {start_terminal.phases.name}. "
f"Number of phases to apply must match the number of nominal phases. Found {len(phases)}, expected {len(start_terminal.phases.single_phases)}"
f"Number of phases to apply must match the number of nominal phases. Found {len(phases)}, expected {len(start_terminal.phases.single_phases)}",
)
self._apply_phases(phases, start_terminal, network_state_operators)
await self._run_terminals([start_terminal], network_state_operators=network_state_operators)
Expand All @@ -137,7 +137,7 @@ def spread_phases(
from_terminal: Terminal,
to_terminal: Terminal,
phases: List[SinglePhaseKind] = None,
network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL
network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL,
):
"""
Apply nominal phases from the `from_terminal` to the `to_terminal`.
Expand Down Expand Up @@ -190,8 +190,8 @@ async def _run_terminal_trace(self, terminal: Terminal, network_trace: NetworkTr
await network_trace.run(
terminal,
self.PhasesToFlow(
[NominalPhasePath(SinglePhaseKind.NONE, it) for it in terminal.phases]
), can_stop_on_start_item=False
[NominalPhasePath(SinglePhaseKind.NONE, it) for it in terminal.phases],
), can_stop_on_start_item=False,
)

# This is called in a loop so we need to reset it for each call. We choose to do this after to release the memory
Expand All @@ -207,7 +207,7 @@ def _nominal_phase_path_to_phases(nominal_phase_paths: list[NominalPhasePath]) -
def _create_network_trace(
self,
state_operators: Type[NetworkStateOperators],
partially_energised_transformers: Set[PowerTransformer]
partially_energised_transformers: Set[PowerTransformer],
) -> NetworkTrace[PhasesToFlow]:

def step_action(nts, ctx):
Expand All @@ -232,11 +232,10 @@ def step_action(nts, ctx):
name=f'SetPhases({state_operators.description})',
queue_factory=lambda: WeightedPriorityQueue.process_queue(lambda it: it.path.to_terminal.phases.num_phases),
branch_queue_factory=lambda: WeightedPriorityQueue.branch_queue(lambda it: it.path.to_terminal.phases.num_phases),
compute_data=self._compute_next_phases_to_flow(state_operators)
)
.add_queue_condition(
lambda next_step, x, y, z: len(next_step.data.nominal_phase_paths) > 0
compute_data=self._compute_next_phases_to_flow(state_operators),
)
.add_queue_condition(lambda next_step, x, y, z: len(next_step.data.nominal_phase_paths) > 0)
.add_queue_condition(stop_on_shunt_compensator_ground())
.add_step_action(step_action)
)

Expand All @@ -250,8 +249,8 @@ def inner(step, _, next_path):
state_operators,
next_path.from_terminal,
next_path.to_terminal,
self._nominal_phase_path_to_phases(step.data.nominal_phase_paths)
)
self._nominal_phase_path_to_phases(step.data.nominal_phase_paths),
),
)

return ComputeData(inner)
Expand All @@ -260,7 +259,7 @@ def inner(step, _, next_path):
def _apply_phases(
phases: List[SinglePhaseKind],
terminal: Terminal,
state_operators: Type[NetworkStateOperators]
state_operators: Type[NetworkStateOperators],
):
traced_phases = state_operators.phase_status(terminal)
for i, nominal_phase in enumerate(terminal.phases.single_phases):
Expand All @@ -271,7 +270,7 @@ def _get_nominal_phase_paths(
state_operators: Type[NetworkStateOperators],
from_terminal: Terminal,
to_terminal: Terminal,
phases: Sequence[SinglePhaseKind] = None
phases: Sequence[SinglePhaseKind] = None,
) -> List[NominalPhasePath]:

if phases is None:
Expand All @@ -290,7 +289,7 @@ def _get_phases_to_flow(
state_operators: Type[NetworkStateOperators],
terminal: Terminal,
phases: Sequence[SinglePhaseKind],
internal_flow: bool
internal_flow: bool,
) -> Set[SinglePhaseKind]:

if internal_flow:
Expand All @@ -304,7 +303,7 @@ def _flow_phases(
state_operators: Type[NetworkStateOperators],
from_terminal: Terminal,
to_terminal: Terminal,
nominal_phase_paths: List[NominalPhasePath]
nominal_phase_paths: List[NominalPhasePath],
) -> bool:

if (from_terminal.conducting_equipment == to_terminal.conducting_equipment
Expand All @@ -319,7 +318,7 @@ def _flow_straight_phases(
state_operators: Type[NetworkStateOperators],
from_terminal: Terminal,
to_terminal: Terminal,
nominal_phase_paths: List[NominalPhasePath]
nominal_phase_paths: List[NominalPhasePath],
) -> bool:

from_phases = state_operators.phase_status(from_terminal)
Expand All @@ -338,7 +337,7 @@ def _flow_transformer_phases(
from_terminal: Terminal,
to_terminal: Terminal,
nominal_phase_paths: List[NominalPhasePath] = None,
allow_suspect_flow: bool = False
allow_suspect_flow: bool = False,
) -> bool:

paths = nominal_phase_paths or self._get_nominal_phase_paths(state_operators, from_terminal, to_terminal)
Expand All @@ -361,12 +360,16 @@ def _flow_transformer_phases(
flow_phases = (p for p in paths if p.from_phase == SinglePhaseKind.NONE)
add_phases = (p for p in paths if p.from_phase != SinglePhaseKind.NONE)
for p in flow_phases:
self._try_add_phase(from_terminal, from_phases, to_terminal, to_phases, p.to_phase, allow_suspect_flow,
lambda: updated_phases.append(True))
self._try_add_phase(
from_terminal, from_phases, to_terminal, to_phases, p.to_phase, allow_suspect_flow,
lambda: updated_phases.append(True),
)

for p in add_phases:
self._try_set_phase(from_phases[p.from_phase], from_terminal, from_phases, p.from_phase,
to_terminal, to_phases, p.to_phase, lambda: updated_phases.append(True))
self._try_set_phase(
from_phases[p.from_phase], from_terminal, from_phases, p.from_phase,
to_terminal, to_phases, p.to_phase, lambda: updated_phases.append(True),
)

return any(updated_phases)

Expand All @@ -375,11 +378,13 @@ def _flow_transformer_phases_adding_neutral(
state_operators: Type[NetworkStateOperators],
from_terminal: Terminal,
to_terminal: Terminal,
paths: List[NominalPhasePath]
paths: List[NominalPhasePath],
) -> bool:

updated_phases = self._flow_straight_phases(state_operators, from_terminal, to_terminal,
[it for it in paths if it != add_neutral])
updated_phases = self._flow_straight_phases(
state_operators, from_terminal, to_terminal,
[it for it in paths if it != add_neutral],
)

# Only add the neutral if we added a phases to the transformer, otherwise you will flag an energised neutral
# with no active phases. We check to see if we need to add the neutral to prevent adding it when we traverse
Expand All @@ -399,7 +404,7 @@ def _try_set_phase(
to_terminal: Terminal,
to_phases: PhaseStatus,
to_: SinglePhaseKind,
on_success: Callable[[], Any]
on_success: Callable[[], Any],
):
try:
if phase != SinglePhaseKind.NONE and to_phases.__setitem__(to_, phase):
Expand All @@ -417,13 +422,13 @@ def _try_add_phase(
to_phases: PhaseStatus,
to_: SinglePhaseKind,
allow_suspect_flow: bool,
on_success: Callable[[], Any]
on_success: Callable[[], Any],
):
# The phases that can be added are ABCN and Y, so for all cases other than Y we can just use the added phase. For
# Y we need to look at what the phases on the other side of the transformer are to determine what has been added.

phase = _unless_none(
to_phases[to_], _to_y_phase(from_phases[from_terminal.phases.single_phases[0]], allow_suspect_flow)
to_phases[to_], _to_y_phase(from_phases[from_terminal.phases.single_phases[0]], allow_suspect_flow),
) if to_ == SinglePhaseKind.Y else to_

self._try_set_phase(phase, from_terminal, from_phases, SinglePhaseKind.NONE, to_terminal, to_phases, to_, on_success)
Expand All @@ -435,7 +440,7 @@ def _throw_cross_phase_exception(
from_: SinglePhaseKind,
to_terminal: Terminal,
to_phases: PhaseStatus,
to_: SinglePhaseKind
to_: SinglePhaseKind,
):
phase_desc = f'{from_.name}' if from_ == to_ else f'path {from_.name} to {to_.name}'

Expand All @@ -452,7 +457,7 @@ def get_ce_details(terminal: Terminal):
raise PhaseException(
f"Attempted to flow conflicting phase {from_phases[from_].name} onto {to_phases[to_].name} on nominal phase {phase_desc}. This occurred while " +
f"flowing {terminal_desc}. This is often caused by missing open points, or incorrect phases in upstream equipment that should be " +
"corrected in the source data."
"corrected in the source data.",
)


Expand Down
51 changes: 43 additions & 8 deletions test/services/network/tracing/phases/test_set_phases.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@

from network_fixtures import phase_swap_loop_network # noqa (Fixtures)
from services.network.tracing.phases.util import connected_equipment_trace_with_logging, validate_phases, validate_phases_from_term_or_equip, get_t
from zepben.ewb import SetPhases, EnergySource, ConductingEquipment, SinglePhaseKind as SPK, TestNetworkBuilder, PhaseCode, Breaker, NetworkStateOperators
from zepben.ewb import SetPhases, EnergySource, ConductingEquipment, SinglePhaseKind as SPK, TestNetworkBuilder, PhaseCode, Breaker, NetworkStateOperators, \
LinearShuntCompensator, Terminal
from zepben.ewb.exceptions import TracingException, PhaseException


Expand Down Expand Up @@ -551,13 +552,6 @@ async def test_can_back_trace_through_xn_xy_transformer_spur():
validate_phases_from_term_or_equip(network_service, "tx3", PhaseCode.AN, PhaseCode.AB)


def _set_normal_phase(terminal_index, from_phase: SPK, to_phase: SPK):
def action(ce: ConductingEquipment):
list(ce.terminals)[terminal_index].normal_phases[from_phase] = to_phase

return action


@pytest.mark.asyncio
async def test_can_set_phases_from_an_unknown_nominal_phase():
"""
Expand Down Expand Up @@ -626,6 +620,47 @@ async def test_energises_around_dropped_phase_dual_transformer_loop():
validate_phases_from_term_or_equip(ns, 'c11', PhaseCode.ABN, PhaseCode.ABN)


@pytest.mark.asyncio
async def test_doesnt_set_phases_either_way_through_a_grounding_terminal_of_a_shunt_compensator():
#
# s0 11--c1--21 lsc2 21--c3--2
#
# s4 11--c5--21 lsc6 21--c7--2
#

def set_grounding_terminal(lsc: LinearShuntCompensator, terminal_index: int):
terminal = list(lsc.terminals)[terminal_index]
terminal.phases = PhaseCode.N
lsc.grounding_terminal = terminal

ns = await (TestNetworkBuilder()
.from_source() # s0
.to_acls() # c1
.to_other(LinearShuntCompensator, default_mrid_prefix = "lsc", action= lambda it: set_grounding_terminal(it, -1)) # lsc2
.to_acls(PhaseCode.N) # c3
.from_source(PhaseCode.N) # s4
.to_acls(PhaseCode.N) # c5
.to_other(LinearShuntCompensator, default_mrid_prefix = "lsc", action= lambda it: set_grounding_terminal(it, 0)) # lcs6
.to_acls() # c7
.build()
)

validate_phases_from_term_or_equip(ns, "c1", PhaseCode.ABC, PhaseCode.ABC)
validate_phases_from_term_or_equip(ns, "lsc2", PhaseCode.ABC, PhaseCode.NONE)
validate_phases_from_term_or_equip(ns, "c3", PhaseCode.NONE, PhaseCode.NONE)
validate_phases_from_term_or_equip(ns, "c5", PhaseCode.N, PhaseCode.N)
validate_phases_from_term_or_equip(ns, "lsc6", PhaseCode.N, PhaseCode.NONE)
validate_phases_from_term_or_equip(ns, "c7", PhaseCode.NONE, PhaseCode.NONE)


def _set_normal_phase(terminal_index: int, from_phase: SPK, to_phase: SPK):
def action(ce: ConductingEquipment):
terminal = list(ce.terminals)[terminal_index]
terminal.normal_phases[from_phase] = to_phase

return action


async def _validate_tx_phases(
source_phases: PhaseCode,
tx_phase_1: PhaseCode,
Expand Down
Loading