More tests, again

Author: chestm007

src/zepben/evolve/services/network/tracing/feeder/clear_direction.py

@@ -8,7 +8,9 @@
 
 from zepben.evolve.model.cim.iec61970.base.core.terminal import Terminal
 
-from zepben.evolve import FeederDirection, Tracing, WeightedPriorityQueue, Traversal
+from zepben.evolve import FeederDirection, Traversal
+from zepben.evolve.services.network.tracing.networktrace.tracing import Tracing
+from zepben.evolve.services.network.tracing.traversal.weighted_priority_queue import WeightedPriorityQueue
 from zepben.evolve.services.network.tracing.networktrace.network_trace import NetworkTrace
 from zepben.evolve.services.network.tracing.networktrace.network_trace_action_type import NetworkTraceActionType
 from zepben.evolve.services.network.tracing.networktrace.operators.network_state_operators import NetworkStateOperators

src/zepben/evolve/services/network/tracing/networktrace/conditions/open_condition.py

@@ -27,7 +27,12 @@ def __init__(self, is_open: Callable[[Switch, SinglePhaseKind], bool], phase: Si
         self._phase = phase
 
     def should_queue_matched_step(self, next_item: NetworkTraceStep[T], next_context: StepContext, current_item: NetworkTraceStep[T], current_context: StepContext) -> bool:
-        return not self._is_open(next_item.path.to_equipment, self._phase)
+        from zepben.evolve.model.cim.iec61970.base.wires.switch import Switch
+        equip = next_item.path.to_equipment
+        if isinstance(equip, Switch):
+            return not self._is_open(equip, self._phase)
+        else:
+            return True
 
     def should_queue_start_item(self, item: T) -> bool:
         return True

src/zepben/evolve/services/network/tracing/networktrace/operators/open_state_operators.py

@@ -16,6 +16,7 @@
 
 if TYPE_CHECKING:
     from zepben.evolve.model.cim.iec61970.base.wires.switch import Switch
+    from zepben.evolve.model.cim.iec61970.base.core.conducting_equipment import ConductingEquipment
 
     T = TypeVar('T')
 
@@ -27,27 +28,42 @@ class OpenStateOperators(StateOperator):
 
     @staticmethod
     @abstractmethod
-    def is_open(switch: Switch, phase: SinglePhaseKind=None) -> bool:
+    def is_open_switch(switch: Switch, phase: SinglePhaseKind=None) -> bool:
         """
         Checks if the specified switch is open. Optionally checking the state of a specific phase.
 
         `switch` The switch to check open state.
         `phase` The specific phase to check, or `null` to check if any phase is open.
-        Returns `true` if open; `false` otherwise.
+        Returns `True` if open; `False` otherwise.
         """
         raise NotImplementedError()
 
+    @classmethod
+    def is_open(cls, conducting_equipment: ConductingEquipment, phase: SinglePhaseKind=None) -> bool:
+        """
+        Convenience method that checks if the `conducting_equipment` is a `Switch` before checking if its open
+
+        :param conducting_equipment: The conducting equipment to check open state
+        :param phase: The specified phase to check, or 'None' to check if any phase is open
+        Returns `True` if conducting equipment is a switch and its open; `False`  otherwise
+        """
+        from zepben.evolve.model.cim.iec61970.base.wires.switch import Switch  # FIXME: circular import
+
+        if isinstance(conducting_equipment, Switch):
+            return cls.is_open_switch(conducting_equipment, phase)
+        return False
+
     @staticmethod
     @abstractmethod
     def set_open(switch: Switch, is_open: bool, phase: SinglePhaseKind=None) -> None:
         """
         Sets the open state of the specified switch. Optionally applies the state to a specific phase.
 
         `switch` The switch for which to set the open state.
-        `isOpen` The desired open state (`true` for open, `false` for closed).
-        `phase` The specific phase to set, or `null` to apply to all phases.
+        `isOpen` The desired open state (`True` for open, `False` for closed).
+        `phase` The specific phase to set, or `None` to apply to all phases.
         """
-        pass
+        raise NotImplementedError()
 
     @classmethod
     def stop_at_open(cls) -> NetworkTraceQueueCondition[T]:
@@ -59,11 +75,8 @@ class NormalOpenStateOperators(OpenStateOperators):
     Operates on the normal state of the `Switch`
     """
     @staticmethod
-    def is_open(switch: Switch, phase:SinglePhaseKind=None) -> Optional[bool]:
-        try:
-            return switch.is_normally_open(phase)
-        except AttributeError:
-            return False
+    def is_open_switch(switch: Switch, phase:SinglePhaseKind=None) -> Optional[bool]:
+        return switch.is_normally_open(phase)
 
     @staticmethod
     def set_open(switch: Switch, is_open: bool, phase: SinglePhaseKind = None) -> None:
@@ -75,16 +88,13 @@ class CurrentOpenStateOperators(OpenStateOperators):
     Operates on the current state of the `Switch`
     """
     @staticmethod
-    def is_open(switch: Switch, phase: SinglePhaseKind = None) -> Optional[bool]:
-        try:
-            return switch.is_open(phase)
-        except AttributeError:
-            return False
+    def is_open_switch(switch: Switch, phase: SinglePhaseKind = None) -> Optional[bool]:
+        return switch.is_open(phase)
 
     @staticmethod
     def set_open(switch: Switch, is_open: bool, phase: SinglePhaseKind = None) -> None:
         switch.set_open(is_open, phase)
 
 
 OpenStateOperators.NORMAL = NormalOpenStateOperators()
-OpenStateOperators.CURRENT = CurrentOpenStateOperators()
+OpenStateOperators.CURRENT = CurrentOpenStateOperators()

src/zepben/evolve/services/network/tracing/networktrace/tracing.py

@@ -37,7 +37,7 @@ def network_trace(network_state_operators: NetworkStateOperators=NetworkStateOpe
         return NetworkTrace.non_branching(network_state_operators, queue, action_step_type, compute_data)
 
     @staticmethod
-    def network_trace_branching(network_state_operators: NetworkStateOperators,
+    def network_trace_branching(network_state_operators: NetworkStateOperators=NetworkStateOperators.NORMAL,
                                 action_step_type: NetworkTraceActionType=NetworkTraceActionType.FIRST_STEP_ON_EQUIPMENT,
                                 queue_factory: Callable[[], TraversalQueue[NetworkTraceStep[T]]]=lambda: TraversalQueue.depth_first(),
                                 branch_queue_factory: Callable[[], TraversalQueue[NetworkTraceStep[T]]]=lambda: TraversalQueue.breadth_first(),
@@ -57,6 +57,7 @@ def set_direction():
 
     @staticmethod
     def clear_direction():
+        from zepben.evolve.services.network.tracing.feeder.clear_direction import ClearDirection
         return ClearDirection()
 
     @staticmethod

test/database/sqlite/network/test_network_database_schema.py

@@ -22,7 +22,8 @@
     NetworkDatabaseReader, NetworkServiceComparator, LvFeeder, CurrentTransformerInfo, PotentialTransformerInfo, CurrentTransformer, \
     PotentialTransformer, SwitchInfo, RelayInfo, CurrentRelay, EvChargingUnit, TapChangerControl, DistanceRelay, VoltageRelay, ProtectionRelayScheme, \
     ProtectionRelaySystem, Ground, GroundDisconnector, SeriesCompensator, NetworkService, StreetAddress, TownDetail, StreetDetail, GroundingImpedance, \
-    PetersenCoil, ReactiveCapabilityCurve, SynchronousMachine, PanDemandResponseFunction, BatteryControl, StaticVarCompensator, Tracing, NetworkStateOperators
+    PetersenCoil, ReactiveCapabilityCurve, SynchronousMachine, PanDemandResponseFunction, BatteryControl, StaticVarCompensator, Tracing, NetworkStateOperators, \
+    NetworkTraceStep
 from zepben.evolve.model.cim.iec61970.base.wires.clamp import Clamp
 from zepben.evolve.model.cim.iec61970.base.wires.cut import Cut
 from zepben.evolve.model.cim.iec61970.base.wires.per_length_phase_impedance import PerLengthPhaseImpedance
@@ -48,6 +49,25 @@
 from database.sqlite.schema_utils import SchemaNetworks
 
 
+# FIXME: see Line [305]
+
+class PatchedNetworkTraceStepPath(NetworkTraceStep.Path):
+    @property
+    def from_equipment(self):
+        try:
+            return super().from_equipment
+        except AttributeError:
+            return
+
+    @property
+    def to_equipment(self):
+        try:
+            return super().to_equipment
+        except AttributeError:
+            return
+
+NetworkTraceStep.Path = PatchedNetworkTraceStepPath
+
 # pylint: disable=too-many-public-methods
 class TestNetworkDatabaseSchema(CimDatabaseSchemaCommonTests[NetworkService, NetworkDatabaseWriter, NetworkDatabaseReader, NetworkServiceComparator]):
 

test/services/network/tracing/networktrace/operators/test_feeder_direction_state_operators.py

@@ -0,0 +1,57 @@
+#  Copyright 2025 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 zepben.evolve import Terminal, FeederDirection
+from zepben.evolve.services.network.tracing.networktrace.operators.feeder_direction_state_operations import FeederDirectionStateOperations
+
+
+class TestFeederDirectionStateOperators:
+
+    normal = FeederDirectionStateOperations.NORMAL
+    current = FeederDirectionStateOperations.CURRENT
+
+    def test_get_direction(self):
+        for operations, attr in ((self.normal, 'normal_feeder_direction'), (self.current, 'current_feeder_direction')):
+            terminal = Terminal()
+            setattr(terminal, attr, FeederDirection.UPSTREAM)
+            assert operations.get_direction(terminal) == FeederDirection.UPSTREAM
+
+
+    def test_set_direction(self):
+        for operations, attr in ((self.normal, 'normal_feeder_direction'), (self.current, 'current_feeder_direction')):
+            terminal = Terminal()
+            setattr(terminal, attr, FeederDirection.NONE)
+            assert operations.set_direction(terminal, FeederDirection.UPSTREAM)
+            assert getattr(terminal, attr) == FeederDirection.UPSTREAM
+
+            # Attempting to add a direction to the terminal already has should return False
+            assert not operations.set_direction(terminal, FeederDirection.UPSTREAM)
+
+            # Setting direction should replace the existing direction
+            assert operations.set_direction(terminal, FeederDirection.DOWNSTREAM)
+            assert getattr(terminal, attr) == FeederDirection.DOWNSTREAM
+
+    def test_add_direction(self):
+        for operations, attr in ((self.normal, 'normal_feeder_direction'), (self.current, 'current_feeder_direction')):
+            terminal = Terminal()
+            setattr(terminal, attr, FeederDirection.NONE)
+            assert operations.add_direction(terminal, FeederDirection.UPSTREAM)
+            assert getattr(terminal, attr) == FeederDirection.UPSTREAM
+
+            # Attempting to add a direction the terminal already has should return False
+            assert not operations.add_direction(terminal, FeederDirection.UPSTREAM)
+
+            # Adding a direction should end up with a combination of the directions
+            assert operations.add_direction(terminal, FeederDirection.DOWNSTREAM)
+            assert getattr(terminal, attr) == FeederDirection.BOTH
+
+    def test_remove_direction(self):
+        for operations, attr in ((self.normal, 'normal_feeder_direction'), (self.current, 'current_feeder_direction')):
+            terminal = Terminal()
+            setattr(terminal, attr, FeederDirection.BOTH)
+            assert operations.remove_direction(terminal, FeederDirection.UPSTREAM)
+            assert getattr(terminal, attr) == FeederDirection.DOWNSTREAM
+
+            # Attempting to remove a direction the terminal does not have should return False
+            assert not operations.remove_direction(terminal, FeederDirection.UPSTREAM)

test/services/network/tracing/networktrace/operators/test_in_service_state_operators.py

@@ -0,0 +1,31 @@
+#  Copyright 2025 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.evolve import Equipment
+from zepben.evolve.services.network.tracing.networktrace.operators.in_service_state_operators import InServiceStateOperators
+
+
+class TestInServiceStateOperators:
+    normal = InServiceStateOperators.NORMAL
+    current = InServiceStateOperators.CURRENT
+
+    def test_is_in_service(self):
+        for operator, attr in ((self.normal, 'normally_in_service'), (self.current, 'in_service')):
+            for _bool in (True, False):
+                equipment = MagicMock(Equipment)
+                setattr(equipment, attr, _bool)
+
+                assert operator.is_in_service(equipment) == _bool
+
+    def test_set_in_service(self):
+        for operator, attr in ((self.normal, 'normally_in_service'), (self.current, 'in_service')):
+            for _bool in (True, False):
+                equipment = MagicMock(Equipment)
+                assert getattr(equipment, attr)
+
+                operator.set_in_service(equipment, False)
+
+                assert not getattr(equipment, attr)

test/services/network/tracing/networktrace/operators/test_open_state_operators.py

@@ -0,0 +1,38 @@
+#  Copyright 2025 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.evolve import Switch, SinglePhaseKind
+from zepben.evolve.services.network.tracing.networktrace.operators.open_state_operators import OpenStateOperators
+
+
+class FlipFlopper:
+    def __init__(self, state):
+        self.state = state
+
+    def __call__(self, *args, **kwargs):
+        _state = self.state
+        self.state = not _state
+        return _state
+
+class TestOpenStateOperators:
+
+    normal = OpenStateOperators.NORMAL
+    current = OpenStateOperators.CURRENT
+
+    def test_is_open_check_swith_open_state(self):
+        for operators, attr in ((self.normal, 'is_normally_open'), (self.current, 'is_open')):
+            switch = MagicMock(Switch)
+            flopper = FlipFlopper(False)
+            setattr(switch, attr, lambda spk: flopper())
+
+            assert not operators.is_open(switch, SinglePhaseKind.A)
+            assert operators.is_open(switch, SinglePhaseKind.A)
+
+    def test_set_open(self):
+        for operators, attr in ((self.normal, 'is_normally_open'), (self.current, 'is_open')):
+            switch = MagicMock(Switch)
+            operators.set_open(switch, True, SinglePhaseKind.A)
+            assert getattr(switch, attr)

test/services/network/tracing/networktrace/operators/test_phase_state_operators.py

@@ -0,0 +1,18 @@
+#  Copyright 2025 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 zepben.evolve import Terminal
+from zepben.evolve.services.network.tracing.networktrace.operators.phase_state_operators import PhaseStateOperators
+
+
+class TestPhaseStateOperators:
+
+    normal = PhaseStateOperators.NORMAL
+    current = PhaseStateOperators.CURRENT
+
+    def test_phase_status(self):
+        for operators, attr in ((self.normal, 'normal_phases'), (self.current, 'current_phases')):
+            terminal = Terminal()
+            # FIXME: should be comparing the actual PhaseStatus object, but Terminal makes a new one on every call
+            assert operators.phase_status(terminal).terminal is getattr(terminal, attr).terminal

test/services/network/tracing/networktrace/test_network_trace.py

@@ -0,0 +1,27 @@
+#  Copyright 2025 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/.
+import pytest
+
+from zepben.evolve.services.network.tracing.networktrace.tracing import Tracing
+from zepben.evolve.testing.test_network_builder import TestNetworkBuilder
+
+
+class TestNetworkTrace:
+
+    @pytest.mark.asyncio
+    async def test_can_run_large_branching_traces(self):
+        builder = TestNetworkBuilder()
+        network = builder.network
+
+        builder.from_junction(num_terminals=1) \
+               .to_acls()
+
+        for i in range(250):
+            builder.to_junction(mrid=f'junc-{i}', num_terminals=3) \
+                   .to_acls(mrid=f'acls-{i}-top') \
+                   .from_acls(mrid=f'acls-{i}-bottom') \
+                   .connect(f'junc-{i}', f'acls-{i}-bottom', 2, 1)
+
+        await Tracing.network_trace_branching().run(network['j0'].get_terminal_by_sn(1))