From 4a01211765179e5bcc5964c8143b7344d364e696 Mon Sep 17 00:00:00 2001 From: Anthony Charlton Date: Thu, 26 Feb 2026 15:23:49 +1100 Subject: [PATCH] Neutral only network Signed-off-by: Anthony Charlton --- changelog.md | 12 +- .../cim/iec61970/base/core/phase_code.py | 6 +- .../iec61970/base/wires/shunt_compensator.py | 33 +- .../terminal_connectivity_internal.py | 46 +- .../networktrace/conditions/conditions.py | 12 +- .../conditions/shunt_compensator_condition.py | 40 ++ test/cim/fill_fields.py | 39 +- .../cim/iec61970/base/core/test_phase_code.py | 7 + .../common/service_comparator_validator.py | 241 ++++--- .../common/translator/base_test_translator.py | 67 +- .../test_network_service_comparator.py | 8 +- .../test_terminal_connectivity_internal.py | 605 +++++++++++------- .../actions/test_equipment_tree_builder.py | 39 +- .../conditions/test_conditions.py | 7 +- ...est_equipment_step_limit_condition_test.py | 4 + .../conditions/test_open_condition.py | 26 +- .../test_shunt_compensator_condition.py | 79 +++ .../network/tracing/networktrace/util.py | 8 + 18 files changed, 915 insertions(+), 364 deletions(-) create mode 100644 src/zepben/ewb/services/network/tracing/networktrace/conditions/shunt_compensator_condition.py create mode 100644 test/services/network/tracing/networktrace/conditions/test_shunt_compensator_condition.py diff --git a/changelog.md b/changelog.md index 51bfe6e6b..fefc3e751 100644 --- a/changelog.md +++ b/changelog.md @@ -14,9 +14,18 @@ * `ProtectionRelayFunction.relay_info` * `ShuntCompensator.shunt_compensator_info` * `Switch.switch_info` +* The `ShuntCompensator.groundingTerminal` must now: + * Belong to the `ShuntCompensator`. Assigning a `Terminal` to `ShuntCompensator.groundingTerminal` will now set the terminals `conductingEquipment` to the + `ShuntCompensator` if it isn't set, and throw an `IllegalArgumentException` if it is assigned to a different `ConductingEquipment`. + * Be in the `ShuntCompensator.terminals` collection, and will be added automatically if it is missing on assignment, which in turn will update the + `sequenceNumber` of the `Terminal` if it is `0`. + * Have phases `N`. +* Phase paths through a `ShuntCompensator` now add paths for mismatched phases between the grounding and normal terminals. This works in the same way as the + `PowerTransformer`. This will only impact traces that are tracking the included phase paths, and will allow traces that previously stopped at the + `ShuntCompensator` to continue. You should use the new `stopOnShuntCompensatorGround` condition to maintain current behaviour. ### New Features -* None. +* Added `Conditions.stopOnShuntCompensatorGround`, a new condition to prevent tracing through a `ShuntCompensator` using its grounding terminal. ### Enhancements * Added sequence unpacking support for `UnresolvedReference` and `ObjectDifference`. @@ -29,6 +38,7 @@ ### Fixes * Fixed the packing and unpacking of timestamps for `Agreement.validity_interval` in gRPC messages. Fix also ensures all other timestamps correctly support `None` when optional. +* Fixed an error in `PhaseCode` when adding `NONE` which previously resulted in `NONE` instead of the existing `PhaseCode`. ### Notes * None. diff --git a/src/zepben/ewb/model/cim/iec61970/base/core/phase_code.py b/src/zepben/ewb/model/cim/iec61970/base/core/phase_code.py index 0471ac9db..4c9c8f88f 100644 --- a/src/zepben/ewb/model/cim/iec61970/base/core/phase_code.py +++ b/src/zepben/ewb/model/cim/iec61970/base/core/phase_code.py @@ -199,4 +199,8 @@ def phase_code_from_single_phases(single_phases: Set[SinglePhaseKind]) -> PhaseC # The IDE is detecting `it` as a `SinglePhaseKind` rather than a `PhaseCode` # noinspection PyUnresolvedReferences -_PHASE_CODE_BY_PHASES = {frozenset(it.single_phases): it for it in PhaseCode} +_PHASE_CODE_BY_PHASES = { + k: v for k, v in + [(frozenset(it.single_phases), it) for it in PhaseCode] + + [(frozenset(it.single_phases + [SinglePhaseKind.NONE]), it) for it in PhaseCode] +} diff --git a/src/zepben/ewb/model/cim/iec61970/base/wires/shunt_compensator.py b/src/zepben/ewb/model/cim/iec61970/base/wires/shunt_compensator.py index 2a1916225..185b46341 100644 --- a/src/zepben/ewb/model/cim/iec61970/base/wires/shunt_compensator.py +++ b/src/zepben/ewb/model/cim/iec61970/base/wires/shunt_compensator.py @@ -3,6 +3,8 @@ # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at https://mozilla.org/MPL/2.0/. +from __future__ import annotations + __all__ = ["ShuntCompensator"] import sys @@ -12,8 +14,10 @@ else: from typing_extensions import deprecated +from zepben.ewb.model.cim.iec61970.base.core.phase_code import PhaseCode from zepben.ewb.model.cim.iec61970.base.wires.phase_shunt_connection_kind import PhaseShuntConnectionKind from zepben.ewb.model.cim.iec61970.base.wires.regulating_cond_eq import RegulatingCondEq +from zepben.ewb.util import require if TYPE_CHECKING: from zepben.ewb.model.cim.iec61968.assetinfo.shunt_compensator_info import ShuntCompensatorInfo @@ -47,9 +51,14 @@ class ShuntCompensator(RegulatingCondEq): grounded: Optional[bool] = None nom_u: Optional[int] = None phase_connection: PhaseShuntConnectionKind = PhaseShuntConnectionKind.UNKNOWN - grounding_terminal: 'Terminal | None' = None + _grounding_terminal: 'Terminal | None' = None sections: Optional[float] = None + def __init__(self, grounding_terminal=None, **kwargs): + super(ShuntCompensator, self).__init__(**kwargs) + if grounding_terminal is not None: + self.grounding_terminal = grounding_terminal + @property @deprecated("use asset_info instead.") def shunt_compensator_info(self) -> Optional['ShuntCompensatorInfo']: @@ -64,3 +73,25 @@ def shunt_compensator_info(self, sci: Optional['ShuntCompensatorInfo']): `sci` The `ShuntCompensatorInfo` for this `ShuntCompensator` """ self.asset_info = sci + + @property + def grounding_terminal(self) -> Optional[Terminal]: + """[ZBEX] The terminal connecting to grounded network.""" + return self._grounding_terminal + + @grounding_terminal.setter + def grounding_terminal(self, terminal: Optional[Terminal]): + if terminal is None: + self._grounding_terminal = None + return + + if terminal.conducting_equipment is None: + terminal.conducting_equipment = self + + require(terminal.conducting_equipment == self, lambda: "The grounding terminal must belong to this ShuntCompensator.") + require(terminal.phases == PhaseCode.N, lambda: "The grounding terminal must used nominal phases `N`.") + + if terminal not in self.terminals: + self.add_terminal(terminal) + + self._grounding_terminal = terminal diff --git a/src/zepben/ewb/services/network/tracing/connectivity/terminal_connectivity_internal.py b/src/zepben/ewb/services/network/tracing/connectivity/terminal_connectivity_internal.py index 60d9c984c..8e7c5ea42 100644 --- a/src/zepben/ewb/services/network/tracing/connectivity/terminal_connectivity_internal.py +++ b/src/zepben/ewb/services/network/tracing/connectivity/terminal_connectivity_internal.py @@ -5,9 +5,9 @@ __all__ = ["TerminalConnectivityInternal"] -from typing import Set, Optional +from typing import Set, Optional, Iterable, cast -from zepben.ewb import Terminal, PowerTransformer, SinglePhaseKind, ConnectivityResult, NominalPhasePath +from zepben.ewb import Terminal, PowerTransformer, SinglePhaseKind, ConnectivityResult, NominalPhasePath, ShuntCompensator from zepben.ewb.services.network.tracing.connectivity.transformer_phase_paths import transformer_phase_paths @@ -36,29 +36,43 @@ def between( include_phases = set(terminal.phases.single_phases) if isinstance(terminal.conducting_equipment, PowerTransformer): - return self._transformer_terminal_connectivity(terminal, other_terminal, include_phases) + paths = self._find_transformer_phase_paths(terminal, other_terminal, include_phases) + elif isinstance(terminal.conducting_equipment, ShuntCompensator): + paths = self._find_shunt_compensator_phase_paths(terminal, other_terminal, include_phases) + else: + paths = self._find_straight_phase_paths(terminal, other_terminal, include_phases) - return self._straight_terminal_connectivity(terminal, other_terminal, include_phases) + return ConnectivityResult(terminal, other_terminal, paths) @staticmethod - def _transformer_terminal_connectivity( + def _find_transformer_phase_paths( terminal: Terminal, other_terminal: Terminal, include_phases: Set[SinglePhaseKind] - ) -> ConnectivityResult: - paths = [it for it in transformer_phase_paths.get(terminal.phases, {}).get(other_terminal.phases, []) - if (it.from_phase in include_phases) or (it.from_phase == SinglePhaseKind.NONE)] + ) -> Iterable[NominalPhasePath]: + return [it for it in transformer_phase_paths.get(terminal.phases, {}).get(other_terminal.phases, []) + if (it.from_phase in include_phases) or (it.from_phase == SinglePhaseKind.NONE)] - return ConnectivityResult(terminal, other_terminal, paths) + def _find_shunt_compensator_phase_paths( + self, + terminal: Terminal, + other_terminal: Terminal, + include_phases: Set[SinglePhaseKind] + ) -> Iterable[NominalPhasePath]: + grounding_terminal = cast(ShuntCompensator, terminal.conducting_equipment).grounding_terminal + + if grounding_terminal == terminal: + return [NominalPhasePath(SinglePhaseKind.NONE, it) for it in other_terminal.phases.single_phases] + elif grounding_terminal == other_terminal: + return [NominalPhasePath(SinglePhaseKind.NONE, SinglePhaseKind.N)] + else: + return self._find_straight_phase_paths(terminal, other_terminal, include_phases) @staticmethod - def _straight_terminal_connectivity( + def _find_straight_phase_paths( terminal: Terminal, other_terminal: Terminal, include_phases: Set[SinglePhaseKind] - ) -> ConnectivityResult: - # noinspection PyArgumentList - paths = [NominalPhasePath(it, it) for it in set(terminal.phases.single_phases).intersection(set(other_terminal.phases.single_phases)) - if it in include_phases] - - return ConnectivityResult(terminal, other_terminal, paths) + ) -> Iterable[NominalPhasePath]: + return [NominalPhasePath(it, it) for it in set(terminal.phases.single_phases).intersection(set(other_terminal.phases.single_phases)) + if it in include_phases] diff --git a/src/zepben/ewb/services/network/tracing/networktrace/conditions/conditions.py b/src/zepben/ewb/services/network/tracing/networktrace/conditions/conditions.py index a99254fc7..7bd47a44f 100644 --- a/src/zepben/ewb/services/network/tracing/networktrace/conditions/conditions.py +++ b/src/zepben/ewb/services/network/tracing/networktrace/conditions/conditions.py @@ -5,7 +5,7 @@ from __future__ import annotations -__all__ = ['upstream', 'downstream', 'with_direction', 'limit_equipment_steps', 'stop_at_open'] +__all__ = ['upstream', 'downstream', 'with_direction', 'limit_equipment_steps', 'stop_at_open', 'stop_on_shunt_compensator_ground'] from typing import TYPE_CHECKING, TypeVar, Type, Callable @@ -13,6 +13,7 @@ from zepben.ewb.services.network.tracing.networktrace.conditions.direction_condition import DirectionCondition from zepben.ewb.services.network.tracing.networktrace.conditions.equipment_step_limit_condition import EquipmentStepLimitCondition from zepben.ewb.services.network.tracing.networktrace.conditions.equipment_type_step_limit_condition import EquipmentTypeStepLimitCondition +from zepben.ewb.services.network.tracing.networktrace.conditions.shunt_compensator_condition import _ShuntCompensatorCondition T = TypeVar('T') @@ -71,3 +72,12 @@ def limit_equipment_steps(limit: int, equipment_type: Type[ConductingEquipment] def stop_at_open(): return lambda state_operator: state_operator.stop_at_open() + +def stop_on_shunt_compensator_ground() -> QueueCondition[NetworkTraceStep[T]]: + """ + Creates a `NetworkTrace` condition that stops tracing a path if it attempts to traverse a `ShuntCompensator` using its grounding terminal. + + :return: A `NetworkTraceQueueCondition` that results in not queueing when a path attempts to traverse a `ShuntCompensator` using its grounding terminal. + """ + # noinspection PyProtectedMember + return _ShuntCompensatorCondition._StopOnGround() diff --git a/src/zepben/ewb/services/network/tracing/networktrace/conditions/shunt_compensator_condition.py b/src/zepben/ewb/services/network/tracing/networktrace/conditions/shunt_compensator_condition.py new file mode 100644 index 000000000..2f59eb2b6 --- /dev/null +++ b/src/zepben/ewb/services/network/tracing/networktrace/conditions/shunt_compensator_condition.py @@ -0,0 +1,40 @@ +# Copyright 2026 Zeppelin Bend Pty Ltd +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at https://mozilla.org/MPL/2.0/. +from abc import ABC +from typing import Generic, cast, TypeVar + +from zepben.ewb.model.cim.iec61970.base.wires.shunt_compensator import ShuntCompensator +from zepben.ewb.services.network.tracing.networktrace.conditions.network_trace_queue_condition import NetworkTraceQueueCondition +from zepben.ewb.services.network.tracing.networktrace.network_trace_step import NetworkTraceStep +from zepben.ewb.services.network.tracing.traversal.step_context import StepContext + +T = TypeVar('T') + + +class _ShuntCompensatorCondition(ABC): + class _StopOnGround(NetworkTraceQueueCondition[T], Generic[T]): + """ + A `NetworkTraceQueueCondition` that prevents the network trace from queueing between the normal and grounding terminals + of a `ShuntCompensator`. + """ + + def __init__(self): + super().__init__(NetworkTraceStep.Type.INTERNAL) + + # noinspection PyUnusedLocal + def should_queue_matched_step( + self, + next_item: NetworkTraceStep[T], + next_context: StepContext, + current_item: NetworkTraceStep[T], + current_context: StepContext + ) -> bool: + # Queue everything that isn't an internal traversal across a `ShuntCompensator` involving its `grounding_terminal`. + if not isinstance(next_item.path.to_equipment, ShuntCompensator): + return True + sc = cast(ShuntCompensator, next_item.path.to_equipment) + + return next_item.path.traced_externally or ( + (next_item.path.to_terminal != sc.grounding_terminal) and (next_item.path.from_terminal != sc.grounding_terminal)) diff --git a/test/cim/fill_fields.py b/test/cim/fill_fields.py index 43ae4b012..2f3b511ff 100644 --- a/test/cim/fill_fields.py +++ b/test/cim/fill_fields.py @@ -75,7 +75,7 @@ from zepben.ewb import * from hypothesis.strategies import builds, text, integers, sampled_from, booleans, uuids, datetimes, one_of, none, just, \ - lists as hypo_lists, floats as hypo_floats + lists as hypo_lists, floats as hypo_floats, composite from zepben.ewb.model.cim.extensions.iec61970.base.feeder.lv_substation import LvSubstation from zepben.ewb.model.cim.iec61968.assetinfo.wire_insulation_kind import WireInsulationKind @@ -2193,14 +2193,47 @@ def series_compensator_kwargs(include_runtime: bool = True): def shunt_compensator_kwargs(include_runtime: bool): + # + # NOTE: Grounding terminal must have phase N, so we need to fiddle with the terminals. Since we can't reference other strategies via + # Hypothesis, we need to build our own to deal with the terminals. To do this, we replace the original `terminals` strategy + # with our wrapper, which will call the original. + # + regulating_cond_eq = regulating_cond_eq_kwargs(include_runtime) + original_terminals_strategy = regulating_cond_eq["terminals"] + + last_terminal = None + + @composite + def terminals_wrapper(draw): + """ + A custom strategy that calls the original `terminals` strategy, but stores the generated terminals so we can pick the last later. + """ + nonlocal last_terminal + terminals = draw(original_terminals_strategy) + last_terminal = terminals[-1] + return terminals + + # noinspection PyUnusedLocal + @composite + def last_terminal_with_n(draw): + """ + A custom strategy that returns the last terminal generated, replacing its phases with N. + """ + last_terminal.phases = PhaseCode.N + + return last_terminal + + # Replace the standard `terminals` strategy with our wrapper, so we can reuse a terminal as the `grounding_terminal`. + regulating_cond_eq["terminals"] = terminals_wrapper() + return { - **regulating_cond_eq_kwargs(include_runtime), + **regulating_cond_eq, "asset_info": builds(ShuntCompensatorInfo, **identified_object_kwargs(include_runtime)), "grounded": booleans(), "nom_u": integers(min_value=MIN_32_BIT_INTEGER, max_value=MAX_32_BIT_INTEGER), "phase_connection": sampled_phase_shunt_connection_kind(), "sections": floats(min_value=FLOAT_MIN, max_value=FLOAT_MAX), - "grounding_terminal": builds(Terminal, **identified_object_kwargs(include_runtime)), + "grounding_terminal": last_terminal_with_n(), } diff --git a/test/cim/iec61970/base/core/test_phase_code.py b/test/cim/iec61970/base/core/test_phase_code.py index ad2cb8cc8..f15cabbdd 100644 --- a/test/cim/iec61970/base/core/test_phase_code.py +++ b/test/cim/iec61970/base/core/test_phase_code.py @@ -85,6 +85,10 @@ def test_plus(self): assert PhaseCode.ABCN + SinglePhaseKind.X == PhaseCode.NONE assert PhaseCode.ABCN + PhaseCode.X == PhaseCode.NONE + # Can add with NONE. + assert PhaseCode.NONE + SinglePhaseKind.A == PhaseCode.A + assert PhaseCode.B + SinglePhaseKind.NONE == PhaseCode.B + def test_minus(self): assert PhaseCode.ABCN - SinglePhaseKind.B == PhaseCode.ACN assert PhaseCode.ABCN - PhaseCode.AN == PhaseCode.BC @@ -95,3 +99,6 @@ def test_minus(self): assert PhaseCode.AB - PhaseCode.C == PhaseCode.AB assert PhaseCode.ABCN - PhaseCode.ABCN == PhaseCode.NONE + + assert PhaseCode.NONE - SinglePhaseKind.A == PhaseCode.NONE + assert PhaseCode.B - SinglePhaseKind.NONE == PhaseCode.B diff --git a/test/services/common/service_comparator_validator.py b/test/services/common/service_comparator_validator.py index 1424dbe6c..bacde7358 100644 --- a/test/services/common/service_comparator_validator.py +++ b/test/services/common/service_comparator_validator.py @@ -17,6 +17,7 @@ TAddr: TypeAlias = Callable[[TIdentifiedObject, R], TIdentifiedObject] | Callable[[R], TIdentifiedObject] + # # NOTE: Should be using the following with TCreator[TIdentifiedObject] in the functions, but Python # 3.10 and PyCharm don't like passing TCreator[TIdentifiedObject] to bind correctly, so until @@ -41,7 +42,7 @@ def validate_name_types( expect_modification: Optional[ObjectDifference] = None, expect_missing_from_target: Optional[NameType] = None, expect_missing_from_source: Optional[NameType] = None, - options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions() + options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), ): source_service = self.create_service() target_service = self.create_service() @@ -73,13 +74,16 @@ def validate_compare( expect_modification: Optional[ObjectDifference] = None, options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), options_stop_compare: bool = False, - expected_differences: Set[str] = None + expected_differences: Set[str] = None, ): if not expect_modification: expect_modification = ObjectDifference(source, target) diff = self.create_comparator(NetworkServiceComparatorOptions()).compare_objects(source, target) if expected_differences: + found_expected = {k for (k, v) in diff.differences.items() if k in expected_differences} + assert len(found_expected) == len(expected_differences), \ + f"Expected: {expected_differences}, only found {found_expected}. What happened to {expected_differences - found_expected}?" diff.differences = {k: v for (k, v) in diff.differences.items() if k not in expected_differences} assert diff == expect_modification, f"Actual:\n{diff}\n vs\nExpected:\n{expect_modification}" @@ -91,12 +95,12 @@ def validate_compare( def validate_property( self, prop: Property | R, # Isn't actually R, but that is what the type checker thinks when passing class member references. - creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. + creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. create_value: Callable[[TIdentifiedObject], R], create_other_value: Callable[[TIdentifiedObject], R], options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), options_stop_compare: bool = False, - expected_differences: Set[str] = None + expected_differences: Set[str] = None, ): subject = creator("mRID") matching = creator("mRID") @@ -108,21 +112,24 @@ def validate_property( self.validate_compare(subject, matching, options=options, options_stop_compare=options_stop_compare) - diff = ObjectDifference(subject, modified, { - _prop_name(prop): self._get_value_or_reference_difference(_get_prop(subject, prop), _get_prop(modified, prop)) - }) + diff = ObjectDifference( + subject, modified, + { + _prop_name(prop): self._get_value_or_reference_difference(_get_prop(subject, prop), _get_prop(modified, prop)) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences) def validate_val_property( self, prop: Property, - creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. + creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. change_state: Callable[[TIdentifiedObject, R], Any], other_change_state: Callable[[TIdentifiedObject, R], Any], options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), options_stop_compare: bool = False, - expected_differences: Set[str] = None + expected_differences: Set[str] = None, ): subject = creator("mRID") matching = creator("mRID") @@ -134,21 +141,24 @@ def validate_val_property( self.validate_compare(subject, matching, options=options, options_stop_compare=options_stop_compare) - diff = ObjectDifference(subject, modified, { - _prop_name(prop): self._get_value_or_reference_difference(_get_prop(subject, prop), _get_prop(modified, prop)) - }) + diff = ObjectDifference( + subject, modified, + { + _prop_name(prop): self._get_value_or_reference_difference(_get_prop(subject, prop), _get_prop(modified, prop)) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences) def validate_collection( self, prop: Property, add_to_collection: TAddr[TIdentifiedObject, R], - creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. + creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. create_item: Callable[[TIdentifiedObject], R], create_other_item: Callable[[TIdentifiedObject], R], options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), options_stop_compare: bool = False, - expected_differences: Set[str] = None + expected_differences: Set[str] = None, ): source_empty = creator("mRID") target_empty = creator("mRID") @@ -163,30 +173,42 @@ def validate_collection( self.validate_compare(source_empty, target_empty, options=options, options_stop_compare=options_stop_compare) self.validate_compare(in_source, in_target, options=options, options_stop_compare=options_stop_compare) - diff = ObjectDifference(in_source, target_empty, { - _prop_name(prop): CollectionDifference(missing_from_target=[next(_get_prop(in_source, prop))]) - }) + diff = ObjectDifference( + in_source, + target_empty, + { + _prop_name(prop): CollectionDifference(missing_from_target=[next(_get_prop(in_source, prop))]) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) - diff = ObjectDifference(source_empty, in_target, { - _prop_name(prop): CollectionDifference(missing_from_source=[next(_get_prop(in_target, prop))]) - }) + diff = ObjectDifference( + source_empty, + in_target, + { + _prop_name(prop): CollectionDifference(missing_from_source=[next(_get_prop(in_target, prop))]) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) - diff = ObjectDifference(in_source, in_target_difference, { - _prop_name(prop): CollectionDifference( - missing_from_source=[next(_get_prop(in_target_difference, prop))], - missing_from_target=[next(_get_prop(in_source, prop))] - ) - }) + diff = ObjectDifference( + in_source, + in_target_difference, + { + _prop_name(prop): CollectionDifference( + missing_from_source=[next(_get_prop(in_target_difference, prop))], + missing_from_target=[next(_get_prop(in_source, prop))], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences) def validate_name_collection( self, - creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. + creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), options_stop_compare: bool = False, - expected_differences: Set[str] = None + expected_differences: Set[str] = None, ): source_empty = creator("mRID") target_empty = creator("mRID") @@ -204,34 +226,46 @@ def validate_name_collection( self.validate_compare(source_empty, target_empty, options=options, options_stop_compare=options_stop_compare) self.validate_compare(in_source, in_target, options=options, options_stop_compare=options_stop_compare) - diff = ObjectDifference(in_source, target_empty, { - _prop_name(IdentifiedObject.names): CollectionDifference(missing_from_target=[next(_get_prop(in_source, IdentifiedObject.names))]) - }) + diff = ObjectDifference( + in_source, + target_empty, + { + _prop_name(IdentifiedObject.names): CollectionDifference(missing_from_target=[next(_get_prop(in_source, IdentifiedObject.names))]) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) - diff = ObjectDifference(source_empty, in_target, { - _prop_name(IdentifiedObject.names): CollectionDifference(missing_from_source=[next(_get_prop(in_target, IdentifiedObject.names))]) - }) + diff = ObjectDifference( + source_empty, + in_target, + { + _prop_name(IdentifiedObject.names): CollectionDifference(missing_from_source=[next(_get_prop(in_target, IdentifiedObject.names))]) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) - diff = ObjectDifference(in_source, in_target_difference, { - _prop_name(IdentifiedObject.names): CollectionDifference( - missing_from_source=[next(_get_prop(in_target_difference, IdentifiedObject.names))], - missing_from_target=[next(_get_prop(in_source, IdentifiedObject.names))] - ) - }) + diff = ObjectDifference( + in_source, + in_target_difference, + { + _prop_name(IdentifiedObject.names): CollectionDifference( + missing_from_source=[next(_get_prop(in_target_difference, IdentifiedObject.names))], + missing_from_target=[next(_get_prop(in_source, IdentifiedObject.names))], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences) def validate_indexed_collection( self, prop: Property, add_to_collection: TAddr, - creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. + creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. create_item: Callable[[TIdentifiedObject], R], create_other_item: Callable[[TIdentifiedObject], R], options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), options_stop_compare: bool = False, - expected_differences: Set[str] = None + expected_differences: Set[str] = None, ): source_empty = creator("mRID") target_empty = creator("mRID") @@ -249,25 +283,43 @@ def validate_indexed_collection( def get_item(obj) -> Optional[R]: return next(_get_prop(obj, prop), None) - diff = ObjectDifference(in_source, target_empty, { - _prop_name(prop): CollectionDifference(missing_from_target=[ - IndexedDifference(0, self._get_value_or_reference_difference(get_item(in_source), None)) - ]) - }) + diff = ObjectDifference( + in_source, + target_empty, + { + _prop_name(prop): CollectionDifference( + missing_from_target=[ + IndexedDifference(0, self._get_value_or_reference_difference(get_item(in_source), None)) + ], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) - diff = ObjectDifference(source_empty, in_target, { - _prop_name(prop): CollectionDifference(missing_from_source=[ - IndexedDifference(0, self._get_value_or_reference_difference(None, get_item(in_target))) - ]) - }) + diff = ObjectDifference( + source_empty, + in_target, + { + _prop_name(prop): CollectionDifference( + missing_from_source=[ + IndexedDifference(0, self._get_value_or_reference_difference(None, get_item(in_target))) + ], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) - diff = ObjectDifference(in_source, target_different, { - _prop_name(prop): CollectionDifference(modifications=[ - IndexedDifference(0, self._get_value_or_reference_difference(get_item(in_source), get_item(target_different))) - ]) - }) + diff = ObjectDifference( + in_source, + target_different, + { + _prop_name(prop): CollectionDifference( + modifications=[ + IndexedDifference(0, self._get_value_or_reference_difference(get_item(in_source), get_item(target_different))) + ], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) K = TypeVar('K') @@ -277,13 +329,13 @@ def validate_unordered_collection( self, prop: Property, add_to_collection: TAddr, - creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. + creator: Type[TIdentifiedObject] | Callable[[str], TIdentifiedObject], # Update to TCreator[TIdentifiedObject] when available. create_item_1: Callable[[K], R], create_item_2: Callable[[K], R], create_diff_item_1: Callable[[K], R], options: NetworkServiceComparatorOptions = NetworkServiceComparatorOptions(), options_stop_compare: bool = False, - expected_differences: Set[str] = None + expected_differences: Set[str] = None, ) -> TIdentifiedObject: source_empty = creator("mRID") target_empty = creator("mRID") @@ -320,20 +372,32 @@ def get_item_1(it): def get_item_2(it): return list(_get_prop(it, prop))[-1] - diff = ObjectDifference(in_source, target_empty, { - _prop_name(prop): CollectionDifference(missing_from_target=[ - self._get_value_or_reference_difference(get_item_1(in_source), None), - self._get_value_or_reference_difference(get_item_2(in_source), None) - ]) - }) + diff = ObjectDifference( + in_source, + target_empty, + { + _prop_name(prop): CollectionDifference( + missing_from_target=[ + self._get_value_or_reference_difference(get_item_1(in_source), None), + self._get_value_or_reference_difference(get_item_2(in_source), None) + ], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) - diff = ObjectDifference(source_empty, in_target_same_order, { - _prop_name(prop): CollectionDifference(missing_from_source=[ - self._get_value_or_reference_difference(None, get_item_1(in_target_same_order)), - self._get_value_or_reference_difference(None, get_item_2(in_target_same_order)) - ]) - }) + diff = ObjectDifference( + source_empty, + in_target_same_order, + { + _prop_name(prop): CollectionDifference( + missing_from_source=[ + self._get_value_or_reference_difference(None, get_item_1(in_target_same_order)), + self._get_value_or_reference_difference(None, get_item_2(in_target_same_order)) + ], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) target_different = creator("mRID") @@ -343,11 +407,17 @@ def get_item_2(it): item2 = create_item_2(target_different) add_to_collection(target_different, item2) - diff = ObjectDifference(in_source, target_different, { - _prop_name(prop): CollectionDifference(modifications=[ - self._get_value_or_reference_difference(get_item_1(in_source), get_item_1(target_different)), - ]) - }) + diff = ObjectDifference( + in_source, + target_different, + { + _prop_name(prop): CollectionDifference( + modifications=[ + self._get_value_or_reference_difference(get_item_1(in_source), get_item_1(target_different)), + ], + ) + }, + ) self._validate_expected(diff, options, options_stop_compare, expected_differences=expected_differences) # This is being returned to bind the TIdentifiedObject correctly. @@ -360,10 +430,21 @@ def _get_value_or_reference_difference(source: Optional[R], target: Optional[R]) else: return ValueDifference(source, target) - def _validate_expected(self, diff: ObjectDifference, options: NetworkServiceComparatorOptions, options_stop_compare: bool = False, - expected_differences: Set[str] = None): - self.validate_compare(diff.source, diff.target, expect_modification=diff, options=options, options_stop_compare=options_stop_compare, - expected_differences=expected_differences) + def _validate_expected( + self, + diff: ObjectDifference, + options: NetworkServiceComparatorOptions, + options_stop_compare: bool = False, + expected_differences: Set[str] = None, + ): + self.validate_compare( + diff.source, + diff.target, + expect_modification=diff, + options=options, + options_stop_compare=options_stop_compare, + expected_differences=expected_differences, + ) def _prop_name(prop: Property) -> str: diff --git a/test/services/common/translator/base_test_translator.py b/test/services/common/translator/base_test_translator.py index e0543a8e1..d92525779 100644 --- a/test/services/common/translator/base_test_translator.py +++ b/test/services/common/translator/base_test_translator.py @@ -3,17 +3,30 @@ # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at https://mozilla.org/MPL/2.0/. from traceback import print_tb -from typing import TypeVar, Type, Set, Any +from typing import TypeVar, Type, Set, Any, cast, Callable from hypothesis import given from hypothesis.strategies import SearchStrategy - -from zepben.ewb import IdentifiedObject, BaseService, BaseServiceComparator, EquipmentContainer, OperationalRestriction, ConnectivityNode, TableVersion, \ - TableMetadataDataSources, TableNameTypes, TableNames, SqliteTable, NetworkService, CustomerService, DiagramService, Feeder, LvFeeder -from zepben.ewb.database.sqlite.common.base_database_tables import BaseDatabaseTables from zepben.protobuf.cim.iec61970.base.core.IdentifiedObject_pb2 import IdentifiedObject as PBIdentifiedObject -from zepben.ewb.model.cim.extensions.iec61970.base.feeder.lv_substation import LvSubstation +from zepben.ewb.database.sqlite.common.base_database_tables import BaseDatabaseTables +from zepben.ewb.database.sqlite.tables.iec61970.base.core.table_name_types import TableNameTypes +from zepben.ewb.database.sqlite.tables.iec61970.base.core.table_names import TableNames +from zepben.ewb.database.sqlite.tables.sqlite_table import SqliteTable +from zepben.ewb.database.sqlite.tables.table_metadata_data_sources import TableMetadataDataSources +from zepben.ewb.database.sqlite.tables.table_version import TableVersion +from zepben.ewb.model.cim.iec61968.operations.operational_restriction import OperationalRestriction +from zepben.ewb.model.cim.iec61970.base.core.connectivity_node import ConnectivityNode +from zepben.ewb.model.cim.iec61970.base.core.equipment_container import EquipmentContainer +from zepben.ewb.model.cim.iec61970.base.core.identified_object import IdentifiedObject +from zepben.ewb.model.cim.iec61970.base.core.phase_code import PhaseCode +from zepben.ewb.model.cim.iec61970.base.core.terminal import Terminal +from zepben.ewb.services.common.base_service import BaseService +from zepben.ewb.services.common.base_service_comparator import BaseServiceComparator +from zepben.ewb.services.common.reference_resolvers import shunt_compensator_to_terminal_resolver, UnresolvedReference +from zepben.ewb.services.customer.customers import CustomerService +from zepben.ewb.services.diagram.diagrams import DiagramService +from zepben.ewb.services.network.network_service import NetworkService T = TypeVar("T", bound=IdentifiedObject) @@ -102,6 +115,7 @@ def run_test(cim): print(diffs) assert not diffs + def _format_validation_error(description: str, classes: Set[str]) -> str: return f"\n{description}: {classes}\n" if classes else "" @@ -114,28 +128,45 @@ def _remove_unsent_references(cim: T): if isinstance(cim, OperationalRestriction): cim.clear_equipment() - if isinstance(cim, Feeder): - cim.clear_current_equipment() - if isinstance(cim, ConnectivityNode): cim.clear_terminals() - if isinstance(cim, LvFeeder): - cim.clear_current_equipment() - - if isinstance(cim, LvSubstation): - cim.clear_current_equipment() - def _add_with_unresolved_references(service: BaseService, cim: T) -> T: # We need to convert the populated item before we check the differences so we can complete the unresolved references. # noinspection PyUnresolvedReferences converted_cim = service.add_from_pb(cim.to_pb()) - for ref in service.unresolved_references(): + + def resolve(unresolved_reference: UnresolvedReference): + _, to_mrid, resolver, _ = unresolved_reference try: - service.add(ref.resolver.to_class(ref.to_mrid)) + io = unresolved_reference.resolver.to_class(unresolved_reference.to_mrid) + + # Special case for the shunt compensator grounding terminal which must have phase N. + if resolver is shunt_compensator_to_terminal_resolver: + cast(Terminal, io).phases = PhaseCode.N + + service.add(io) except Exception as e: # If this fails you need to add a concrete type mapping to the abstractCreators map at the top of this class. - assert False, f"Failed to create unresolved reference for {ref.resolver.to_class.__name__}. {e}" + raise TypeError(f"Failed to create unresolved reference for {resolver.to_class}.", e) + + def resolve_all(predicate: Callable[[UnresolvedReference], bool]): + # Collect to a list to prevent hte underlying collection changing as we iterate. + for it in [ref for ref in service.unresolved_references() if predicate(ref)]: + resolve(it) + + # + # NOTE: We need a special case to exclude any `ShuntCompensator.groundingTerminal` resolvers to ensure it is added after any other + # terminals, to prevent assigning incorrect sequence numbers when we create unresolved terminals. This is complicated by + # having a matching resolver being added in the `ConductingEquipment.terminals` that also needs to be delayed. + # + delay_ids = {it.to_mrid for it in service.unresolved_references() if it.resolver is shunt_compensator_to_terminal_resolver} + if delay_ids: + resolve_all(lambda it: it.to_mrid not in delay_ids) + # Make sure we resolve the `grounding_terminal` reference before its matching the `terminals` one, otherwise we will end up with the wrong phases. + resolve_all(lambda it: it.resolver is shunt_compensator_to_terminal_resolver) + + resolve_all(lambda _: True) return converted_cim diff --git a/test/services/network/test_network_service_comparator.py b/test/services/network/test_network_service_comparator.py index 2d9825d13..d13157c57 100644 --- a/test/services/network/test_network_service_comparator.py +++ b/test/services/network/test_network_service_comparator.py @@ -1541,7 +1541,13 @@ def _compare_shunt_compensator(self, creator: Type[ShuntCompensator]): lambda _: ShuntCompensatorInfo(mrid="asci1"), lambda _: ShuntCompensatorInfo(mrid="asci2"), ) - self.validator.validate_property(ShuntCompensator.grounding_terminal, creator, lambda _: Terminal(mrid="t1"), lambda _: Terminal(mrid="t2")) + self.validator.validate_property( + ShuntCompensator.grounding_terminal, + creator, + lambda _: Terminal(mrid="t1", phases=PhaseCode.N), + lambda _: Terminal(mrid="t2", phases=PhaseCode.N), + expected_differences = {"terminals"}, # The terminals should be different as setting the groundingTerminal also adds the terminal. + ) def test_compare_static_var_compensator(self): self._compare_conducting_equipment(StaticVarCompensator) diff --git a/test/services/network/tracing/connectivity/test_terminal_connectivity_internal.py b/test/services/network/tracing/connectivity/test_terminal_connectivity_internal.py index c893c5445..cc62edcbd 100644 --- a/test/services/network/tracing/connectivity/test_terminal_connectivity_internal.py +++ b/test/services/network/tracing/connectivity/test_terminal_connectivity_internal.py @@ -2,265 +2,408 @@ # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at https://mozilla.org/MPL/2.0/. -from collections import Counter +from typing import Optional, Type, Callable, TypeVar -from zepben.ewb import TerminalConnectivityInternal, PhaseCode, PowerTransformer, Terminal, generate_id +from zepben.ewb import TerminalConnectivityInternal, PhaseCode, PowerTransformer, Terminal, generate_id, require, ConnectivityResult, ConductingEquipment, \ + LinearShuntCompensator, AcLineSegment + +TConductingEquipment = TypeVar("TConductingEquipment", bound=ConductingEquipment) class TestTerminalConnectivityInternal: _connectivity = TerminalConnectivityInternal() def test_paths_through_hv3_tx(self): - self._validate_tx_paths(PhaseCode.ABC, PhaseCode.ABC) - self._validate_tx_paths(PhaseCode.ABC, PhaseCode.ABCN) - self._validate_tx_paths(PhaseCode.ABCN, PhaseCode.ABC) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABC, returns=PhaseCode.ABC), + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABCN, returns=PhaseCode.ABCN), + ExpectedPaths(from_phases=PhaseCode.ABCN, to_phases=PhaseCode.ABC, returns=PhaseCode.ABC), + ) def test_paths_through_hv1_hv1_tx(self): - self._validate_tx_paths(PhaseCode.AB, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.AC) - - self._validate_tx_paths(PhaseCode.AB, PhaseCode.XY) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.XY) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.XY, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.AC) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.XY) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + ) def test_paths_through_hv1_lv2_tx(self): - self._validate_tx_paths(PhaseCode.AB, PhaseCode.ABN) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.BCN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.ACN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.XYN) - - self._validate_tx_paths(PhaseCode.BC, PhaseCode.ABN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.BCN) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.ACN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.XYN) - - self._validate_tx_paths(PhaseCode.AC, PhaseCode.ABN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.BCN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.ACN) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.XYN) - - self._validate_tx_paths(PhaseCode.XY, PhaseCode.ABN) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.ACN) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.BCN) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.XYN) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.ABN, returns=PhaseCode.ABN), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.BCN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.ACN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.ABN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.BCN, returns=PhaseCode.BCN), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.ACN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.ABN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.BCN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.ACN, returns=PhaseCode.ACN), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.ABN, returns=PhaseCode.ABN), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.ACN, returns=PhaseCode.ACN), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.BCN, returns=PhaseCode.BCN), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + ) def test_paths_through_hv1_lv1_tx(self): - self._validate_tx_paths(PhaseCode.AB, PhaseCode.AN) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.BN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.CN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.XN) - - self._validate_tx_paths(PhaseCode.BC, PhaseCode.AN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.BN) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.CN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.XN) - - self._validate_tx_paths(PhaseCode.AC, PhaseCode.AN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.BN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.CN) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.XN) - - self._validate_tx_paths(PhaseCode.XY, PhaseCode.AN) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.BN) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.CN) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.XN) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.AN, returns=PhaseCode.AN), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.BN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.CN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.AN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.BN, returns=PhaseCode.BN), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.CN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.AN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.BN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.CN, returns=PhaseCode.CN), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.AN, returns=PhaseCode.AN), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.BN, returns=PhaseCode.BN), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.CN, returns=PhaseCode.CN), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + ) def test_paths_through_lv2_lv2_tx(self): - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.ABN) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.BCN) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.ACN) - - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.XYN) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.XYN) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.XYN) - - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.ABN) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.BCN) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.ACN) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.XYN) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.ABN, returns=PhaseCode.ABN), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.BCN, returns=PhaseCode.BCN), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.ACN, returns=PhaseCode.ACN), + + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.ABN, returns=PhaseCode.ABN), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.BCN, returns=PhaseCode.BCN), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.ACN, returns=PhaseCode.ACN), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + ) def test_paths_through_lv2_hv1_tx(self): - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.BC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.AC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.AB, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.AC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.AB, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.BC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.AC) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.AC) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.XY) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.BC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.AC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.AB, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.AC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.AB, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.BC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + ) def test_paths_through_lv1_hv1_tx(self): - self._validate_tx_paths(PhaseCode.AN, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.AN, PhaseCode.BC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AN, PhaseCode.AC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AN, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.BN, PhaseCode.AB, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BN, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.BN, PhaseCode.AC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BN, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.CN, PhaseCode.AB, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.CN, PhaseCode.BC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.CN, PhaseCode.AC) - self._validate_tx_paths(PhaseCode.CN, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.XN, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.XN, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.XN, PhaseCode.AC) - self._validate_tx_paths(PhaseCode.XN, PhaseCode.XY) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.BC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.AC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.AB, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.AC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.AB, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.BC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + ) def test_paths_through_hv1_swer_tx(self): - self._validate_tx_paths(PhaseCode.AB, PhaseCode.A) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.B, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.C, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AB, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.BC, PhaseCode.A, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.B) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.C, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BC, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.AC, PhaseCode.A, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.B, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.C) - self._validate_tx_paths(PhaseCode.AC, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.XY, PhaseCode.A) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.B) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.C) - self._validate_tx_paths(PhaseCode.XY, PhaseCode.X) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AB, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BC, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.AC, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.XY, to_phases=PhaseCode.X, returns=PhaseCode.X), + ) def test_paths_through_swer_hv1_tx(self): - self._validate_tx_paths(PhaseCode.A, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.A, PhaseCode.BC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.A, PhaseCode.AC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.A, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.B, PhaseCode.AB, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.B, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.B, PhaseCode.AC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.B, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.C, PhaseCode.AB, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.C, PhaseCode.BC, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.C, PhaseCode.AC) - self._validate_tx_paths(PhaseCode.C, PhaseCode.XY) - - self._validate_tx_paths(PhaseCode.X, PhaseCode.AB) - self._validate_tx_paths(PhaseCode.X, PhaseCode.BC) - self._validate_tx_paths(PhaseCode.X, PhaseCode.AC) - self._validate_tx_paths(PhaseCode.X, PhaseCode.XY) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.BC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.AC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.AB, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.AC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.AB, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.BC, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.BC, returns=PhaseCode.BC), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.AC, returns=PhaseCode.AC), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.XY, returns=PhaseCode.XY), + ) def test_paths_through_swer_lv1_tx(self): - self._validate_tx_paths(PhaseCode.A, PhaseCode.AN) - self._validate_tx_paths(PhaseCode.A, PhaseCode.BN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.A, PhaseCode.CN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.A, PhaseCode.XN) - - self._validate_tx_paths(PhaseCode.B, PhaseCode.AN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.B, PhaseCode.BN) - self._validate_tx_paths(PhaseCode.B, PhaseCode.CN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.B, PhaseCode.XN) - - self._validate_tx_paths(PhaseCode.C, PhaseCode.AN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.C, PhaseCode.BN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.C, PhaseCode.CN) - self._validate_tx_paths(PhaseCode.C, PhaseCode.XN) - - self._validate_tx_paths(PhaseCode.X, PhaseCode.AN) - self._validate_tx_paths(PhaseCode.X, PhaseCode.BN) - self._validate_tx_paths(PhaseCode.X, PhaseCode.CN) - self._validate_tx_paths(PhaseCode.X, PhaseCode.XN) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.AN, returns=PhaseCode.AN), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.BN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.CN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.AN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.BN, returns=PhaseCode.BN), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.CN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.AN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.BN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.CN, returns=PhaseCode.CN), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.AN, returns=PhaseCode.AN), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.BN, returns=PhaseCode.BN), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.CN, returns=PhaseCode.CN), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.XN, returns=PhaseCode.XN), + ) def test_paths_through_lv1_swer_tx(self): - self._validate_tx_paths(PhaseCode.AN, PhaseCode.A) - self._validate_tx_paths(PhaseCode.AN, PhaseCode.B, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AN, PhaseCode.C, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.AN, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.BN, PhaseCode.A, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BN, PhaseCode.B) - self._validate_tx_paths(PhaseCode.BN, PhaseCode.C, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BN, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.CN, PhaseCode.A, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.CN, PhaseCode.B, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.CN, PhaseCode.C) - self._validate_tx_paths(PhaseCode.CN, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.XN, PhaseCode.A) - self._validate_tx_paths(PhaseCode.XN, PhaseCode.B) - self._validate_tx_paths(PhaseCode.XN, PhaseCode.C) - self._validate_tx_paths(PhaseCode.XN, PhaseCode.X) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.AN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.CN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.XN, to_phases=PhaseCode.X, returns=PhaseCode.X), + ) def test_paths_through_swer_lv2_tx(self): - self._validate_tx_paths(PhaseCode.A, PhaseCode.ABN) - self._validate_tx_paths(PhaseCode.A, PhaseCode.BCN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.A, PhaseCode.ACN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.A, PhaseCode.XYN) - - self._validate_tx_paths(PhaseCode.B, PhaseCode.ABN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.B, PhaseCode.BCN) - self._validate_tx_paths(PhaseCode.B, PhaseCode.ACN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.B, PhaseCode.XYN) - - self._validate_tx_paths(PhaseCode.C, PhaseCode.ABN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.C, PhaseCode.BCN, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.C, PhaseCode.ACN) - self._validate_tx_paths(PhaseCode.C, PhaseCode.XYN) - - self._validate_tx_paths(PhaseCode.X, PhaseCode.ABN) - self._validate_tx_paths(PhaseCode.X, PhaseCode.BCN) - self._validate_tx_paths(PhaseCode.X, PhaseCode.ACN) - self._validate_tx_paths(PhaseCode.X, PhaseCode.XYN) + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.ABN, returns=PhaseCode.ABN), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.BCN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.ACN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.A, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.ABN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.BCN, returns=PhaseCode.BCN), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.ACN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.B, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.ABN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.BCN, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.ACN, returns=PhaseCode.ACN), + ExpectedPaths(from_phases=PhaseCode.C, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.ABN, returns=PhaseCode.ABN), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.BCN, returns=PhaseCode.BCN), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.ACN, returns=PhaseCode.ACN), + ExpectedPaths(from_phases=PhaseCode.X, to_phases=PhaseCode.XYN, returns=PhaseCode.XYN), + ) def test_paths_through_lv2_swer_tx(self): - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.A) - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.B, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.C, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ABN, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.A, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.B) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.C, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.BCN, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.A, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.B, PhaseCode.NONE) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.C) - self._validate_tx_paths(PhaseCode.ACN, PhaseCode.X) - - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.A) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.B) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.C) - self._validate_tx_paths(PhaseCode.XYN, PhaseCode.X) - - def _validate_tx_paths(self, primary: PhaseCode, secondary: PhaseCode, traced: PhaseCode = None): - traced = traced or secondary - primary_terminal = Terminal(mrid=generate_id(), phases=primary) - secondary_terminal = Terminal(mrid=generate_id(), phases=secondary) - PowerTransformer(mrid=generate_id(), terminals=[primary_terminal, secondary_terminal]) - - if traced != PhaseCode.NONE: - assert Counter([it.to_phase for it in self._connectivity.between(primary_terminal, secondary_terminal).nominal_phase_paths]) == \ - Counter(traced.single_phases) - else: - assert not self._connectivity.between(primary_terminal, secondary_terminal).nominal_phase_paths + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.X, returns=PhaseCode.X), + ) + + def test_paths_through_lv2_swer_tx2(self): + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ABN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.C, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.BCN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.A, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.B, returns=PhaseCode.NONE), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.ACN, to_phases=PhaseCode.X, returns=PhaseCode.X), + + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.A, returns=PhaseCode.A), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.B, returns=PhaseCode.B), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.C, returns=PhaseCode.C), + ExpectedPaths(from_phases=PhaseCode.XYN, to_phases=PhaseCode.X, returns=PhaseCode.X), + ) + + def test_can_filter_transformer_paths(self): + self._validate_paths_through( + PowerTransformer, + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABC, using=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.ABCN, to_phases=PhaseCode.ABC, using=PhaseCode.AC, returns=PhaseCode.AC), + + # Neutral is picked up as it is added by the transformer, so not filtered out by the included phases. + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABCN, using=PhaseCode.BC, returns=PhaseCode.BCN), + ) + + def test_check_shunt_compensator_paths(self): + def set_grounding_terminal(lsc: LinearShuntCompensator): + lsc.grounding_terminal = next((it for it in lsc.terminals if it.phases == PhaseCode.N), None) + + self._validate_paths_through( + LinearShuntCompensator, + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABC, returns=PhaseCode.ABC), + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.N, returns=PhaseCode.N), + ExpectedPaths(from_phases=PhaseCode.N, to_phases=PhaseCode.ABC, returns=PhaseCode.ABC), + + # + # NOTE: When moving to/from the grounding terminal, all phases are added by the shunt compensator, so + # will not be impacted by the included phases. + # + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABC, using=PhaseCode.AB, returns=PhaseCode.AB), + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.N, using=PhaseCode.AB, returns=PhaseCode.N), + ExpectedPaths(from_phases=PhaseCode.N, to_phases=PhaseCode.ABC, using=PhaseCode.N, returns=PhaseCode.ABC), + + additional_setup=set_grounding_terminal, + ) + + def test_check_straight_paths(self): + self._validate_paths_through( + AcLineSegment, + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABC, returns=PhaseCode.ABC), + ExpectedPaths(from_phases=PhaseCode.ABC, to_phases=PhaseCode.ABC, using=PhaseCode.AB, returns=PhaseCode.AB), + ) + + def _validate_paths_through( + self, + builder: Type[TConductingEquipment], + *paths: 'ExpectedPaths', + additional_setup: Optional[Callable[[TConductingEquipment], None]] = None, + ): + ce = builder(mrid=generate_id()) + terminal = Terminal(mrid=generate_id()) + other_terminal = Terminal(mrid=generate_id()) + + ce.add_terminal(terminal) + ce.add_terminal(other_terminal) + + for from_phases, to_phases, expected, included in paths: + terminal.phases = from_phases + other_terminal.phases = to_phases + + if additional_setup: + additional_setup(ce) + + actual_phases = {it.to_phase for it in self._find_paths_between(terminal, other_terminal, included).nominal_phase_paths} + expected_phases = set(expected.single_phases) if expected != PhaseCode.NONE else set() + + assert actual_phases == expected_phases, f"{builder.__name__} from_phases {from_phases} to {to_phases} using {included} should return {expected}" + + def _find_paths_between(self, terminal: Terminal, other_terminal: Terminal, included: PhaseCode) -> ConnectivityResult: + return self._connectivity.between(terminal, other_terminal, set(included.single_phases)) + + +class ExpectedPaths: + + def __init__( + self, + from_phases: PhaseCode, + to_phases: PhaseCode, + using: Optional[PhaseCode] = None, + returns: Optional[PhaseCode] = None, + ): + self.from_phases = from_phases + self.to_phases = to_phases + self.expected = returns if returns is not None else to_phases + self.included = using if using is not None else from_phases + + require( + set(self.included.single_phases).issubset(set(self.from_phases.single_phases)), + lambda: "`included` must only contain phases in `from`", + ) + + def __getitem__(self, item): + return (self.from_phases, self.to_phases, self.expected, self.included)[item] diff --git a/test/services/network/tracing/networktrace/actions/test_equipment_tree_builder.py b/test/services/network/tracing/networktrace/actions/test_equipment_tree_builder.py index f23756e96..4c00601d5 100644 --- a/test/services/network/tracing/networktrace/actions/test_equipment_tree_builder.py +++ b/test/services/network/tracing/networktrace/actions/test_equipment_tree_builder.py @@ -10,7 +10,7 @@ from services.network.test_data.looping_network import create_looping_network from services.network.tracing.feeder.direction_logger import log_directions -from zepben.ewb import ConductingEquipment, Tracing, NetworkStateOperators +from zepben.ewb import ConductingEquipment, Tracing, NetworkStateOperators, TestNetworkBuilder from zepben.ewb import downstream, NetworkTraceActionType from zepben.ewb.services.network.tracing.networktrace.actions.equipment_tree_builder import EquipmentTreeBuilder from zepben.ewb.services.network.tracing.networktrace.actions.tree_node import TreeNode @@ -19,8 +19,10 @@ def test_accessing_leaves_when_not_calculated_raises_exception(): builder = EquipmentTreeBuilder() with pytest.raises(AttributeError): + # noinspection PyStatementEffect builder.leaves + @pytest.mark.asyncio async def test_equipment_tree_builder_leaves(): n = create_looping_network() @@ -39,7 +41,7 @@ async def test_equipment_tree_builder_leaves(): trace = ( Tracing.network_trace_branching( network_state_operators=normal, - action_step_type=NetworkTraceActionType.FIRST_STEP_ON_EQUIPMENT + action_step_type=NetworkTraceActionType.FIRST_STEP_ON_EQUIPMENT, ) .add_condition(downstream()) .add_step_action(tree_builder) @@ -71,7 +73,7 @@ async def test_downstream_tree(): trace = ( Tracing.network_trace_branching( network_state_operators=normal, - action_step_type=NetworkTraceActionType.FIRST_STEP_ON_EQUIPMENT + action_step_type=NetworkTraceActionType.FIRST_STEP_ON_EQUIPMENT, ) .add_condition(downstream()) .add_step_action(tree_builder) @@ -190,11 +192,40 @@ async def test_downstream_tree(): assert _find_node_depths(root, "ac16") == [8, 9, 11, 14] +@pytest.mark.asyncio +async def test_builds_tree_for_empty_feeder(): + n = await ( + TestNetworkBuilder() + .from_breaker() # b0 + .add_feeder("b0") + .build() + ) + + b0 = n.get("b0", ConductingEquipment) + await log_directions(b0) + + start: ConductingEquipment = n["b0"] + assert start is not None + + tree_builder = EquipmentTreeBuilder() + await ( + Tracing.network_trace_branching() + .add_condition(downstream()) + .add_step_action(tree_builder) + .run(start) + ) + + root = tree_builder.roots[start] + + assert root is not None + _verify_tree_asset(root, n["b0"], None, []) + + def _verify_tree_asset( tree_node: TreeNode, expected_asset: Optional[ConductingEquipment], expected_parent: Optional[ConductingEquipment], - expected_children: List[ConductingEquipment] + expected_children: List[ConductingEquipment], ): assert tree_node.identified_object is expected_asset diff --git a/test/services/network/tracing/networktrace/conditions/test_conditions.py b/test/services/network/tracing/networktrace/conditions/test_conditions.py index 4f6f309c4..bbc86d0d2 100644 --- a/test/services/network/tracing/networktrace/conditions/test_conditions.py +++ b/test/services/network/tracing/networktrace/conditions/test_conditions.py @@ -4,12 +4,13 @@ # file, You can obtain one at https://mozilla.org/MPL/2.0/. from typing import Optional, Callable -from zepben.ewb import NetworkStateOperators, FeederDirection, SinglePhaseKind, Switch, PowerTransformer +from zepben.ewb import NetworkStateOperators, FeederDirection, SinglePhaseKind, Switch, PowerTransformer, stop_on_shunt_compensator_ground from zepben.ewb.services.network.tracing.networktrace.conditions.conditions import limit_equipment_steps from zepben.ewb.services.network.tracing.networktrace.conditions.direction_condition import DirectionCondition from zepben.ewb.services.network.tracing.networktrace.conditions.equipment_step_limit_condition import EquipmentStepLimitCondition from zepben.ewb.services.network.tracing.networktrace.conditions.equipment_type_step_limit_condition import EquipmentTypeStepLimitCondition from zepben.ewb.services.network.tracing.networktrace.conditions.open_condition import OpenCondition +from zepben.ewb.services.network.tracing.networktrace.conditions.shunt_compensator_condition import _ShuntCompensatorCondition class TestCondition: @@ -58,3 +59,7 @@ def test_limit_equipment_type_steps(self): assert isinstance(condition, EquipmentTypeStepLimitCondition) assert condition.limit == 1 assert condition.equipment_type is PowerTransformer + + def test_stop_on_shunt_compensator_ground_condition(self): + condition = stop_on_shunt_compensator_ground() + assert isinstance(condition, _ShuntCompensatorCondition._StopOnGround) diff --git a/test/services/network/tracing/networktrace/conditions/test_equipment_step_limit_condition_test.py b/test/services/network/tracing/networktrace/conditions/test_equipment_step_limit_condition_test.py index b0dab2726..ca1a8fe30 100644 --- a/test/services/network/tracing/networktrace/conditions/test_equipment_step_limit_condition_test.py +++ b/test/services/network/tracing/networktrace/conditions/test_equipment_step_limit_condition_test.py @@ -8,6 +8,10 @@ from zepben.ewb.services.network.tracing.networktrace.conditions.equipment_step_limit_condition import EquipmentStepLimitCondition +# +# TODO: This should be moved into utils, but there is already a copy there that seems busted, so some time and +# understanding is required to resolve this. +# def mock_nts(num_terminal_steps=0, num_equipment_steps=0): return NetworkTraceStep(MagicMock(spec=NetworkTraceStep.Path), num_terminal_steps, num_equipment_steps, None) diff --git a/test/services/network/tracing/networktrace/conditions/test_open_condition.py b/test/services/network/tracing/networktrace/conditions/test_open_condition.py index adab61d58..5e27e344b 100644 --- a/test/services/network/tracing/networktrace/conditions/test_open_condition.py +++ b/test/services/network/tracing/networktrace/conditions/test_open_condition.py @@ -2,15 +2,17 @@ # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at https://mozilla.org/MPL/2.0/. -from typing import Callable from unittest.mock import MagicMock from zepben.ewb import Switch, SinglePhaseKind, NetworkTraceStep, ConductingEquipment, StepContext from zepben.ewb.services.network.tracing.networktrace.conditions.open_condition import OpenCondition - -def mock_nts(step_type: NetworkTraceStep.Type=None, path:NetworkTraceStep.Path=None) -> NetworkTraceStep: +# +# TODO: This should be moved into utils, but there is already a copy there that seems busted, so some time and +# understanding is required to resolve this. +# +def mock_nts(step_type: NetworkTraceStep.Type = None, path: NetworkTraceStep.Path = None) -> NetworkTraceStep: next_step = MagicMock(spec=NetworkTraceStep) if step_type: next_step.type = lambda: step_type @@ -20,20 +22,32 @@ def mock_nts(step_type: NetworkTraceStep.Type=None, path:NetworkTraceStep.Path=N return next_step -def mock_nts_path(to_equipment: ConductingEquipment=None) -> NetworkTraceStep.Path: + +# +# TODO: This should be moved into utils, but there is already a copy there that seems busted, so some time and +# understanding is required to resolve this. +# +def mock_nts_path(to_equipment: ConductingEquipment = None) -> NetworkTraceStep.Path: next_path = MagicMock(spec=NetworkTraceStep.Path) if to_equipment: next_path.to_equipment = to_equipment return next_path -def should_queue_params(next_step, next_context=None, current_step=None, current_context=None - ) -> (NetworkTraceStep, StepContext, NetworkTraceStep, StepContext): + +def should_queue_params( + next_step, + next_context=None, + current_step=None, + current_context=None, +) -> tuple[NetworkTraceStep, StepContext, NetworkTraceStep, StepContext]: return next_step, next_context or MagicMock(), current_step or MagicMock(), current_context or MagicMock() + def _is_open(switch: Switch, phase: SinglePhaseKind) -> bool: pass + class TestOpenCondition: def test_always_queues_external_steps(self): is_open = _is_open diff --git a/test/services/network/tracing/networktrace/conditions/test_shunt_compensator_condition.py b/test/services/network/tracing/networktrace/conditions/test_shunt_compensator_condition.py new file mode 100644 index 000000000..8b94a7a39 --- /dev/null +++ b/test/services/network/tracing/networktrace/conditions/test_shunt_compensator_condition.py @@ -0,0 +1,79 @@ +# Copyright 2026 Zeppelin Bend Pty Ltd +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at https://mozilla.org/MPL/2.0/. +from unittest.mock import MagicMock + +from zepben.ewb import LinearShuntCompensator, Terminal, generate_id, PhaseCode, NetworkTraceStep, ConductingEquipment +# noinspection PyProtectedMember +from zepben.ewb.services.network.tracing.networktrace.conditions.shunt_compensator_condition import _ShuntCompensatorCondition + + +# +# TODO: This should be moved into utils, but there is already a copy there that seems busted, so some time and +# understanding is required to resolve this. +# +def mock_nts(step_type: NetworkTraceStep.Type) -> NetworkTraceStep: + next_step = MagicMock(spec=NetworkTraceStep) + + next_step.type = lambda: step_type + + return next_step + + +# +# TODO: This should be moved into utils, but there is already a copy there that seems busted, so some time and +# understanding is required to resolve this. +# +def mock_nts_path( + to_equipment: ConductingEquipment, + traced_internally: bool, +) -> NetworkTraceStep.Path: + next_path = MagicMock(spec=NetworkTraceStep.Path) + + next_path.to_equipment = to_equipment + next_path.traced_internally = traced_internally + + return next_path + + +class TestShuntCompensatorCondition: + + def setup_method(self): + self.shunt_compensator = LinearShuntCompensator(mrid="sc") + self.from_term = Terminal(mrid=generate_id()) + self.to_term = Terminal(mrid=generate_id()) + self.ground_term = Terminal(mrid=generate_id()) + + self.shunt_compensator.add_terminal(self.from_term) + self.shunt_compensator.add_terminal(self.to_term) + self.ground_term.phases = PhaseCode.N + self.shunt_compensator.grounding_terminal = self.ground_term + + def test_always_queues_external_steps(self): + self._validate_queues(mock_nts(step_type=NetworkTraceStep.Type.EXTERNAL)) + + def test_always_queues_non_shunt_compensator_equipment(self): + self._validate_queues( + self._step_of( + mock_nts_path(to_equipment=ConductingEquipment(mrid="non-shunt-compensator"), traced_internally=True) + ) + ) + + def test_queues_shunt_compensator_paths_that_dont_use_the_grounding_terminal(self): + self._validate_queues(self._step_of(NetworkTraceStep.Path(self.from_term, self.to_term))) + + def test_does_not_queue_from_grounding_terminal(self): + self._validate_queues(self._step_of(NetworkTraceStep.Path(self.ground_term, self.to_term)), should_queue=False) + + def test_does_not_queue_onto_grounding_terminal(self): + self._validate_queues(self._step_of(NetworkTraceStep.Path(self.from_term, self.ground_term)), should_queue=False) + + @staticmethod + def _step_of(path: NetworkTraceStep.Path) -> NetworkTraceStep: + return NetworkTraceStep(path, 0, 0, None) + + @staticmethod + def _validate_queues(next_step: NetworkTraceStep, should_queue: bool = True): + result = _ShuntCompensatorCondition._StopOnGround().should_queue(next_step, MagicMock(), MagicMock(), MagicMock()) + assert result == should_queue diff --git a/test/services/network/tracing/networktrace/util.py b/test/services/network/tracing/networktrace/util.py index 802b3967b..2e1a1726e 100644 --- a/test/services/network/tracing/networktrace/util.py +++ b/test/services/network/tracing/networktrace/util.py @@ -7,6 +7,10 @@ from zepben.ewb import NetworkTraceStep, ConductingEquipment, StepContext +# +# TODO: This seems busted, and there are other copies of this lying around. Some time and understanding is required to +# merge these all into the same call that works as expected. +# def mock_nts(path: NetworkTraceStep.Path=None, num_terminal_steps=0, num_equipment_steps=0, @@ -21,6 +25,10 @@ def mock_nts(path: NetworkTraceStep.Path=None, return nts +# +# TODO: This seems busted, and there are other copies of this lying around. Some time and understanding is required to +# merge these all into the same call that works as expected. +# def mock_nts_path(to_equipment: ConductingEquipment=None, traced_internally: bool=None): if traced_internally: