[DEV-6155] Neutral only network

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.

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]
+}

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

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]

src/zepben/ewb/services/network/tracing/networktrace/conditions/conditions.py

@@ -5,14 +5,15 @@
 
 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
 
 from zepben.ewb.services.network.tracing.feeder.feeder_direction import FeederDirection
 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()

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))

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(),
     }
 
 

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