diff --git a/changelog.md b/changelog.md index f3ef9eed..b49d1903 100644 --- a/changelog.md +++ b/changelog.md @@ -10,7 +10,7 @@ * None. ### Fixes -* None. +* Fixed errors in handling phase energisation of `LinearShuntCompensator` instances with a `grounding_terminal`. ### Notes * None. diff --git a/src/zepben/ewb/services/network/tracing/phases/set_phases.py b/src/zepben/ewb/services/network/tracing/phases/set_phases.py index 7bb0f2f0..10ac3f96 100644 --- a/src/zepben/ewb/services/network/tracing/phases/set_phases.py +++ b/src/zepben/ewb/services/network/tracing/phases/set_phases.py @@ -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 @@ -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, ): """ @@ -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. @@ -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` @@ -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) @@ -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`. @@ -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 @@ -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): @@ -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) ) @@ -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) @@ -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): @@ -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: @@ -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: @@ -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 @@ -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) @@ -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) @@ -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) @@ -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 @@ -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): @@ -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) @@ -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}' @@ -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.", ) diff --git a/test/services/network/tracing/phases/test_set_phases.py b/test/services/network/tracing/phases/test_set_phases.py index 55ccd70a..95f875e8 100644 --- a/test/services/network/tracing/phases/test_set_phases.py +++ b/test/services/network/tracing/phases/test_set_phases.py @@ -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 @@ -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(): """ @@ -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,