diff --git a/changelog.md b/changelog.md index 8c9cdb68f..b0f0ab1f4 100644 --- a/changelog.md +++ b/changelog.md @@ -7,6 +7,7 @@ * `MeasurementValue.time_stamp` * `RelayInfo.curve_setting` * `RelayInfo.reclose_fast` +* Removed `TracedPhases`. `Terminal.normalPhases` and `Terminal.currentPhases` should be used instead of `Terminal.tracedPhases` going forward. (missed in 0.48.0) ### New Features * Added the following new CIM classes: diff --git a/src/zepben/ewb/__init__.py b/src/zepben/ewb/__init__.py index b5932b227..e2979ff9f 100644 --- a/src/zepben/ewb/__init__.py +++ b/src/zepben/ewb/__init__.py @@ -240,7 +240,6 @@ # END CIM MODEL # ################# -from zepben.ewb.model.phases import * from zepben.ewb.model.resistance_reactance import * from zepben.ewb.services.network.tracing.util import * diff --git a/src/zepben/ewb/model/cim/iec61970/base/core/terminal.py b/src/zepben/ewb/model/cim/iec61970/base/core/terminal.py index e5505b50f..899e3dd06 100644 --- a/src/zepben/ewb/model/cim/iec61970/base/core/terminal.py +++ b/src/zepben/ewb/model/cim/iec61970/base/core/terminal.py @@ -1,10 +1,12 @@ -# Copyright 2024 Zeppelin Bend Pty Ltd +# 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 __future__ import annotations +__all__ = ["Terminal"] + from typing import Optional, Generator from typing import TYPE_CHECKING from weakref import ref, ReferenceType @@ -13,16 +15,13 @@ from zepben.ewb.model.cim.iec61970.base.core.feeder import Feeder from zepben.ewb.model.cim.iec61970.base.core.phase_code import PhaseCode from zepben.ewb.model.cim.iec61970.base.wires.busbar_section import BusbarSection -from zepben.ewb.model.phases import TracedPhases from zepben.ewb.services.network.tracing.feeder.feeder_direction import FeederDirection -from zepben.ewb.services.network.tracing.phases.phase_status import PhaseStatus, NormalPhases, CurrentPhases +from zepben.ewb.services.network.tracing.phases.phase_status import PhaseStatus if TYPE_CHECKING: from zepben.ewb.model.cim.iec61970.base.core.conducting_equipment import ConductingEquipment from zepben.ewb.model.cim.iec61970.base.core.connectivity_node import ConnectivityNode -__all__ = ["Terminal"] - class Terminal(AcDcTerminal): """ @@ -36,10 +35,6 @@ class Terminal(AcDcTerminal): phases: PhaseCode = PhaseCode.ABC """Represents the normal network phasing condition. If the attribute is missing three phases (ABC) shall be assumed.""" - traced_phases: TracedPhases = TracedPhases() - """the phase object representing the traced phases in both the normal and current network. If properly configured you would expect the normal state phases - to match those in `phases`""" - sequence_number: int = 0 """The orientation of the terminal connections for a multiple terminal conducting equipment. The sequence numbering starts with 1 and additional terminals should follow in increasing order. The first terminal is the "starting point" for a two terminal branch.""" @@ -56,8 +51,16 @@ class Terminal(AcDcTerminal): """This is a weak reference to the connectivity node so if a Network object goes out of scope, holding a single conducting equipment reference does not cause everything connected to it in the network to stay in memory.""" + _normal_phases: PhaseStatus = None + _current_phases: PhaseStatus = None + def __init__(self, conducting_equipment: ConductingEquipment = None, connectivity_node: ConnectivityNode = None, **kwargs): super(Terminal, self).__init__(**kwargs) + + self._normal_phases = PhaseStatus(self) + + self._current_phases = PhaseStatus(self) + if conducting_equipment: self.conducting_equipment = conducting_equipment @@ -69,13 +72,13 @@ def __init__(self, conducting_equipment: ConductingEquipment = None, connectivit @property def normal_phases(self) -> PhaseStatus: - """ Convenience method for accessing the normal phases""" - return NormalPhases(self) + """The status of phases as traced for the normal state of the network.""" + return self._normal_phases @property def current_phases(self) -> PhaseStatus: - """ Convenience method for accessing the current phases""" - return CurrentPhases(self) + """The status of phases as traced for the current state of the network.gi""" + return self._current_phases @property def conducting_equipment(self): @@ -146,9 +149,10 @@ def other_terminals(self) -> Generator[Terminal]: * :return: A `Generator` of terminals that share the same `ConductingEquipment` as this `Terminal`. """ - for t in self.conducting_equipment.terminals: - if t is not self: - yield t + if self.conducting_equipment is not None: + for t in self.conducting_equipment.terminals: + if t is not self: + yield t def connect(self, connectivity_node: ConnectivityNode): self.connectivity_node = connectivity_node diff --git a/src/zepben/ewb/model/phases.py b/src/zepben/ewb/model/phases.py deleted file mode 100644 index f21a682e8..000000000 --- a/src/zepben/ewb/model/phases.py +++ /dev/null @@ -1,168 +0,0 @@ -# Copyright 2024 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 __future__ import annotations - -__all__ = ["get_phase", "set_phase", "TracedPhases"] - -from collections import defaultdict -from dataclasses import dataclass - -from zepben.ewb.exceptions import PhaseException -from zepben.ewb.model.cim.iec61970.base.core.phase_code import PhaseCode -from zepben.ewb.model.cim.iec61970.base.wires.single_phase_kind import SinglePhaseKind - -BITS_TO_PHASE = defaultdict(lambda: SinglePhaseKind.NONE) -BITS_TO_PHASE[0b0001] = SinglePhaseKind.A -BITS_TO_PHASE[0b0010] = SinglePhaseKind.B -BITS_TO_PHASE[0b0100] = SinglePhaseKind.C -BITS_TO_PHASE[0b1000] = SinglePhaseKind.N - -PHASE_TO_BITS = defaultdict(lambda: 0) -PHASE_TO_BITS[SinglePhaseKind.A] = 0b0001 -PHASE_TO_BITS[SinglePhaseKind.B] = 0b0010 -PHASE_TO_BITS[SinglePhaseKind.C] = 0b0100 -PHASE_TO_BITS[SinglePhaseKind.N] = 0b1000 - -NOMINAL_PHASE_MASKS = [0x000f, 0x00f0, 0x0f00, 0xf000] - - -def _valid_phase_check(nominal_phase): - if nominal_phase == SinglePhaseKind.NONE or nominal_phase == SinglePhaseKind.INVALID: - raise ValueError(f"INTERNAL ERROR: Phase {nominal_phase.name} is invalid. Must not be NONE or INVALID.") - - -def get_phase(status: int, nominal_phase: SinglePhaseKind): - return BITS_TO_PHASE[(status >> _byte_selector(nominal_phase)) & 0x0f] - - -def set_phase(status: int, nominal_phase: SinglePhaseKind, traced_phase: SinglePhaseKind) -> int: - if traced_phase == SinglePhaseKind.NONE: - return status & ~NOMINAL_PHASE_MASKS[nominal_phase.mask_index] - else: - return (status & ~NOMINAL_PHASE_MASKS[nominal_phase.mask_index]) | _shifted_value(nominal_phase, traced_phase) - - -def _byte_selector(nominal_phase: SinglePhaseKind) -> int: - return nominal_phase.mask_index * 4 - - -def _shifted_value(nominal_phase: SinglePhaseKind, traced_phase: SinglePhaseKind) -> int: - return PHASE_TO_BITS[traced_phase] << _byte_selector(nominal_phase) - - -# todo split file into correct packages - -@dataclass -class TracedPhases(object): - """ - Class that holds the traced phase statuses for the current and normal state of the network. - - Traced phase status: - | integer | - | 16 bits |16 bits | - | current | normal | - - See [TracedPhasesBitManipulation] for details on bit representation for normal and current status. - """ - - phase_status: int = 0 - """ - The underlying implementation value tracking the phase statuses for the current and normal state of the network. - It is primarily used for data serialisation and debugging within official evolve libraries and utilities. - - NOTE: This property should be considered evolve internal and not for public use as the underlying - data structure to store the status could change at any time (and thus be a breaking change). - Use at your own risk. - """ - - _NORMAL_MASK = 0x0000ffff - _CURRENT_MASK = 0xffff0000 - _CURRENT_SHIFT = 16 - - def __str__(self): - def to_string(select_phase): - return ', '.join([select_phase(phs).short_name for phs in PhaseCode.ABCN.single_phases]) - - return f"TracedPhases(normal={{{to_string(self.normal)}}}, current={{{to_string(self.current)}}})" - - def normal(self, nominal_phase: SinglePhaseKind) -> SinglePhaseKind: - """ - Get the phase that was traced in the normal state of the network on this nominal phase. - - `nominal_phase` The nominal phase to check. - - Returns The `zepben.protobuf.cim.iec61970.base.wires.SinglePhaseKind` traced in the normal state of the network for the nominal phase, or - SinglePhaseKind.NONE if the nominal phase is de-energised. - - Raises `NominalPhaseException` if the nominal phase is invalid. - """ - _valid_phase_check(nominal_phase) - return get_phase(self.phase_status, nominal_phase) - - def current(self, nominal_phase: SinglePhaseKind) -> SinglePhaseKind: - """ - Get the phase that was traced in the current state of the network on this nominal phase. - - `nominal_phase` The nominal phase to check. - - Returns The `zepben.protobuf.cim.iec61970.base.wires.SinglePhaseKind` traced in the current state of the network for the nominal phase, or - SinglePhaseKind.NONE if the nominal phase is de-energised. - - Raises `NominalPhaseException` if the nominal phase is invalid. - """ - _valid_phase_check(nominal_phase) - return get_phase(self.phase_status >> self._CURRENT_SHIFT, nominal_phase) - - def set_normal(self, nominal_phase: SinglePhaseKind, traced_phase: SinglePhaseKind) -> bool: - """ - Set the phase that was traced in the normal state of the network on this nominal phase. - - `nominal_phase` The nominal phase to use. - - `traced_phase` The traced phase to apply to the nominal phase. - - Returns True if there was a change, otherwise False. - - Raises `PhaseException` if phases cross. i.e. you try to apply more than one phase to a nominal phase. - - Raises `NominalPhaseException` if the nominal phase is invalid. - """ - it = self.normal(nominal_phase) - if it == traced_phase: - return False - elif (it == SinglePhaseKind.NONE) or (traced_phase == SinglePhaseKind.NONE): - self.phase_status = (self.phase_status & self._CURRENT_MASK) | set_phase(self.phase_status, nominal_phase, traced_phase) - return True - else: - raise PhaseException("Crossing Phases.") - - def set_current(self, nominal_phase: SinglePhaseKind, traced_phase: SinglePhaseKind) -> bool: - """ - Set the phase that was traced in the current state of the network on this nominal phase. - - `nominal_phase` The nominal phase to use. - - `traced_phase` The traced phase to apply to the nominal phase. - - Returns True if there was a change, otherwise False. - - Raises `PhaseException` if phases cross. i.e. you try to apply more than one phase to a nominal phase. - - Raises `NominalPhaseException` if the nominal phase is invalid. - """ - it = self.current(nominal_phase) - if it == traced_phase: - return False - elif (it == SinglePhaseKind.NONE) or (traced_phase == SinglePhaseKind.NONE): - self.phase_status = (self.phase_status & self._NORMAL_MASK) | ( - set_phase(self.phase_status >> self._CURRENT_SHIFT, nominal_phase, traced_phase) << self._CURRENT_SHIFT) - return True - else: - raise PhaseException("Crossing Phases.") - - @staticmethod - def copy(): - return TracedPhases() diff --git a/src/zepben/ewb/services/common/base_service_comparator.py b/src/zepben/ewb/services/common/base_service_comparator.py index 0741afa9f..90fdcec06 100644 --- a/src/zepben/ewb/services/common/base_service_comparator.py +++ b/src/zepben/ewb/services/common/base_service_comparator.py @@ -219,7 +219,7 @@ def _add_if_different(diff: ObjectDifference, name: str, difference: Optional[Di diff.differences[name] = difference @staticmethod - def _calculate_values_diff(prop: Union[MemberDescriptorType, property], diff: ObjectDifference) -> Optional[ValueDifference]: + def _calculate_values_diff(prop: Union[MemberDescriptorType, property], diff: ObjectDifference, to_comparable: Callable[[property], Any] = (lambda it: it)) -> Optional[ValueDifference]: if isinstance(prop, property): s_val = getattr(diff.source, prop.fget.__name__) if diff.source else None t_val = getattr(diff.target, prop.fget.__name__) if diff.target else None @@ -230,7 +230,7 @@ def _calculate_values_diff(prop: Union[MemberDescriptorType, property], diff: Ob if (type(s_val) == float) or (type(t_val) == float): raise TypeError(f"Using wrong comparator for {prop}, use _calculate_float_diff instead.") - if s_val == t_val: + if to_comparable(s_val) == to_comparable(t_val): return None else: return ValueDifference(s_val, t_val) diff --git a/src/zepben/ewb/services/network/network_service_comparator.py b/src/zepben/ewb/services/network/network_service_comparator.py index 895613f9d..c9fa7e591 100644 --- a/src/zepben/ewb/services/network/network_service_comparator.py +++ b/src/zepben/ewb/services/network/network_service_comparator.py @@ -764,8 +764,9 @@ def _compare_terminal(self, source: Terminal, target: Terminal) -> ObjectDiffere Terminal.sequence_number, Terminal.normal_feeder_direction, Terminal.current_feeder_direction, - Terminal.phases ) + self._add_if_different(diff, Terminal.normal_phases.fget.__name__, self._calculate_values_diff(Terminal.normal_phases, diff, to_comparable=lambda it: it._phase_status_internal)) + self._add_if_different(diff, Terminal.current_phases.fget.__name__, self._calculate_values_diff(Terminal.current_phases, diff, to_comparable=lambda it: it._phase_status_internal)) return self._compare_ac_dc_terminal(diff) diff --git a/src/zepben/ewb/services/network/tracing/phases/phase_status.py b/src/zepben/ewb/services/network/tracing/phases/phase_status.py index b0b1f8831..871a7cf17 100644 --- a/src/zepben/ewb/services/network/tracing/phases/phase_status.py +++ b/src/zepben/ewb/services/network/tracing/phases/phase_status.py @@ -5,61 +5,87 @@ from __future__ import annotations -__all__ = ["normal_phases", "current_phases", "PhaseStatus", "NormalPhases", "CurrentPhases"] +__all__ = ["PhaseStatus"] +from dataclasses import dataclass from typing import TYPE_CHECKING, Optional -if TYPE_CHECKING: - from zepben.ewb import Terminal from zepben.ewb.model.cim.iec61970.base.core.phase_code import phase_code_from_single_phases, PhaseCode - from zepben.ewb.model.cim.iec61970.base.wires.single_phase_kind import SinglePhaseKind -from abc import ABC, abstractmethod +from zepben.ewb.exceptions import PhaseException +from zepben.ewb.services.network.tracing.phases.traced_phases_bit_manipulation import TracedPhasesBitManipulation + +if TYPE_CHECKING: + from zepben.ewb.model.cim.iec61970.base.core.terminal import Terminal -def normal_phases(terminal: Terminal): - return NormalPhases(terminal) +def _validate_spk(spk: SinglePhaseKind): + if spk in (SinglePhaseKind.NONE, SinglePhaseKind.INVALID): + raise ValueError(f"INTERNAL ERROR: Phase {spk.name} is invalid") + return spk -def current_phases(terminal: Terminal): - return CurrentPhases(terminal) +SinglePhaseKind.validate = _validate_spk +@dataclass(slots=True) +class PhaseStatus: + """ + Class that holds the traced phase statuses for a nominal phase on a [Terminal]. -class PhaseStatus(ABC): + :var _phase_status_internal: + """ terminal: Terminal - def __init__(self, terminal: Terminal): - self.terminal = terminal + _phase_status_internal: int = 0 + """ + The underlying implementation value tracking the phase status for nominal phases of a terminal. + It is exposed internally for data serialisation and debugging within official EWB libraries and utilities. + + This property should be considered internal and not for public use as the underlying + data structure to store the status could change at any time (and thus be a breaking change). + Use directly at your own risk. + + See ``TracedPhasesBitManipulation`` for details on bit representation for phase_status_internal and how we track phases status. + """ - @abstractmethod def __getitem__(self, nominal_phase: SinglePhaseKind) -> SinglePhaseKind: """ - Get the traced phase for the specified `nominal_phase`. + Get the traced phase for the specified ``nominal_phase``. - `nominal_phase` The nominal phase you are interested in querying. + :param nominal_phase: The nominal phase you are interested in querying. - Returns the traced phase. + :returns: the traced phase. """ - raise NotImplementedError() + return TracedPhasesBitManipulation.get(self._phase_status_internal, nominal_phase.validate()) - @abstractmethod - def __setitem__(self, nominal_phase: SinglePhaseKind, traced_phase: SinglePhaseKind) -> bool: + def __setitem__(self, nominal_phase: SinglePhaseKind, single_phase_kind: SinglePhaseKind) -> bool: """ - Set the traced phase for the specified `nominal_phase`. - - `nominal_phase` The nominal phase you are interested in updating. + Set the traced phase for the specified ``nominal_phase``. - `traced_phase` The phase you wish to set for this `nominal_phase`. Specify `SinglePhaseKind.NONE` to clear the phase. + :param nominal_phase: The nominal phase you are interested in updating. + :param single_phase_kind: The phase you wish to set for this ``single_phase_kind``. Specify ``SinglePhaseKind.NONE`` to clear the phase. - Returns True if the phase is updated, otherwise False. + :returns: ``True`` if the phase is updated, otherwise False. """ - raise NotImplementedError() + it = self[nominal_phase] + if it == single_phase_kind: + return False + + elif it == SinglePhaseKind.NONE or single_phase_kind == SinglePhaseKind.NONE: + self._phase_status_internal = TracedPhasesBitManipulation.set( + self._phase_status_internal, + nominal_phase, + single_phase_kind + ) + return True + else: + raise PhaseException("Crossing Phases") def as_phase_code(self) -> Optional[PhaseCode]: """ - Get the traced phase for each nominal phase as a `PhaseCode`. + Get the traced phase for each nominal phase as a ``PhaseCode``. - Returns The `PhaseCode` if the combination of phases makes sense, otherwise `None`. + :returns: The ``PhaseCode`` if the combination of phases makes sense, otherwise ``None``. """ if self.terminal.phases == PhaseCode.NONE: return PhaseCode.NONE @@ -75,27 +101,3 @@ def as_phase_code(self) -> Optional[PhaseCode]: return phase_code_from_single_phases(phases) else: return None - - -class NormalPhases(PhaseStatus): - """ - The traced phases in the normal state of the network. - """ - - def __getitem__(self, nominal_phase: SinglePhaseKind) -> SinglePhaseKind: - return self.terminal.traced_phases.normal(nominal_phase) - - def __setitem__(self, nominal_phase: SinglePhaseKind, traced_phase: SinglePhaseKind) -> bool: - return self.terminal.traced_phases.set_normal(nominal_phase, traced_phase) - - -class CurrentPhases(PhaseStatus): - """ - The traced phases in the current state of the network. - """ - - def __getitem__(self, nominal_phase: SinglePhaseKind) -> SinglePhaseKind: - return self.terminal.traced_phases.current(nominal_phase) - - def __setitem__(self, nominal_phase: SinglePhaseKind, traced_phase: SinglePhaseKind) -> bool: - return self.terminal.traced_phases.set_current(nominal_phase, traced_phase) diff --git a/src/zepben/ewb/services/network/tracing/phases/remove_phases.py b/src/zepben/ewb/services/network/tracing/phases/remove_phases.py index 70ad22950..3be062e43 100644 --- a/src/zepben/ewb/services/network/tracing/phases/remove_phases.py +++ b/src/zepben/ewb/services/network/tracing/phases/remove_phases.py @@ -77,7 +77,8 @@ async def _run_with_network(network_service: NetworkService, network_state_opera """ for t in network_service.objects(Terminal): - t.traced_phases.phase_status = 0 + t.normal_phases._phase_status_internal = 0 + t.current_phases._phase_status_internal = 0 async def _run_with_terminal(self, terminal: Terminal, network_state_operators: Type[NetworkStateOperators] = NetworkStateOperators.NORMAL): """ diff --git a/src/zepben/ewb/services/network/tracing/phases/traced_phases_bit_manipulation.py b/src/zepben/ewb/services/network/tracing/phases/traced_phases_bit_manipulation.py new file mode 100644 index 000000000..8f3062bf1 --- /dev/null +++ b/src/zepben/ewb/services/network/tracing/phases/traced_phases_bit_manipulation.py @@ -0,0 +1,58 @@ +# 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/. + +__all__ = ["TracedPhasesBitManipulation"] + +from zepben.ewb.model.cim.iec61970.base.wires.single_phase_kind import SinglePhaseKind + + +_nominal_phase_masks = [0x000F, 0x00F0, 0x0F00, 0xF000] +"""Bitwise mask for selecting the actual phases from a nominal phase A/B/C/N, X/Y/N or s1/s2/N""" + +_phase_masks = [1, 2, 4, 8] +"""Bitwise mask for setting the presence of an actual phase""" + + +class TracedPhasesBitManipulation: + """ + Class that performs the bit manipulation for the traced phase statuses. + Each byte in an int is used to store all possible phases and directions for a nominal phase. + Each byte has 2 bits that represent the direction for a phase. If none of those bits are set the direction is equal to NONE. + Use the figures below as a reference. + + :: + + Network state phase status: + | 16 bits | + | 4 bits | 4 bits | 4 bits | 4 bits | + Nominal phase: | N | C | B/Y/s2 | A/X/s1 | + + :: + + Each nominal phase (actual phase): + | 4 bits | + | 1 bit | 1 bit | 1 bit | 1 bit | + Actual Phase: | N | C | B | A | + """ + + @staticmethod + def get(status: int, nominal_phase: SinglePhaseKind) -> SinglePhaseKind: + match ((status >> nominal_phase.byte_selector()) & 15): + case 1: return SinglePhaseKind.A + case 2: return SinglePhaseKind.B + case 4: return SinglePhaseKind.C + case 8: return SinglePhaseKind.N + case _: return SinglePhaseKind.NONE + + @staticmethod + def set(status: int, nominal_phase: SinglePhaseKind, single_phase_kind: SinglePhaseKind) -> int: + if single_phase_kind == SinglePhaseKind.NONE: + return status & (~_nominal_phase_masks[nominal_phase.mask_index]) + else: + return (status & (~_nominal_phase_masks[nominal_phase.mask_index])) | single_phase_kind.shifted_value(nominal_phase) + + +SinglePhaseKind.byte_selector = lambda self: self.mask_index * 4 +SinglePhaseKind.shifted_value = lambda self, nominal_phase: _phase_masks[self.mask_index] << nominal_phase.byte_selector() diff --git a/src/zepben/ewb/services/network/translator/network_cim2proto.py b/src/zepben/ewb/services/network/translator/network_cim2proto.py index 95c8eeb01..a13a5fec9 100644 --- a/src/zepben/ewb/services/network/translator/network_cim2proto.py +++ b/src/zepben/ewb/services/network/translator/network_cim2proto.py @@ -1166,7 +1166,7 @@ def terminal_to_pb(cim: Terminal) -> PBTerminal: sequenceNumber=cim.sequence_number, normalFeederDirection=_map_feeder_direction.to_pb(cim.normal_feeder_direction), currentFeederDirection=_map_feeder_direction.to_pb(cim.current_feeder_direction), - # phases=cim.pha + tracedPhases=(cim.current_phases._phase_status_internal << 16) + cim.normal_phases._phase_status_internal, ) diff --git a/src/zepben/ewb/services/network/translator/network_proto2cim.py b/src/zepben/ewb/services/network/translator/network_proto2cim.py index 35b0aaa95..b84e35555 100644 --- a/src/zepben/ewb/services/network/translator/network_proto2cim.py +++ b/src/zepben/ewb/services/network/translator/network_proto2cim.py @@ -1352,6 +1352,8 @@ def terminal_to_cim(pb: PBTerminal, network_service: NetworkService) -> Optional normal_feeder_direction=FeederDirection(pb.normalFeederDirection), current_feeder_direction=FeederDirection(pb.currentFeederDirection), ) + cim.normal_phases._phase_status_internal = (pb.tracedPhases & 0xFFFF) + cim.current_phases._phase_status_internal = (pb.tracedPhases >> 16) & 0xFFFF network_service.resolve_or_defer_reference(resolver.conducting_equipment(cim), pb.conductingEquipmentMRID) network_service.resolve_or_defer_reference(resolver.connectivity_node(cim), pb.connectivityNodeMRID) diff --git a/test/cim/cim_creators.py b/test/cim/cim_creators.py index 8858ea066..736fdec02 100644 --- a/test/cim/cim_creators.py +++ b/test/cim/cim_creators.py @@ -1012,7 +1012,6 @@ def create_substation(include_runtime: bool = True): def create_terminal(include_runtime: bool = True): runtime = { - "traced_phases": builds(TracedPhases) } if include_runtime else {} return builds( Terminal, diff --git a/test/cim/iec61970/base/core/test_terminal.py b/test/cim/iec61970/base/core/test_terminal.py index b49cd387e..3ea4cb346 100644 --- a/test/cim/iec61970/base/core/test_terminal.py +++ b/test/cim/iec61970/base/core/test_terminal.py @@ -9,7 +9,7 @@ from cim.iec61970.base.core.test_ac_dc_terminal import ac_dc_terminal_kwargs, verify_ac_dc_terminal_constructor_default, \ verify_ac_dc_terminal_constructor_kwargs, verify_ac_dc_terminal_constructor_args, ac_dc_terminal_args from util import mrid_strategy -from zepben.ewb import Terminal, ConnectivityNode, TracedPhases, ConductingEquipment, PhaseCode, generate_id +from zepben.ewb import Terminal, ConnectivityNode, ConductingEquipment, PhaseCode, generate_id, NetworkService, Junction from zepben.ewb.services.network.tracing.feeder.feeder_direction import FeederDirection terminal_kwargs = { @@ -19,7 +19,6 @@ "sequence_number": integers(min_value=MIN_32_BIT_INTEGER, max_value=MAX_32_BIT_INTEGER), "normal_feeder_direction": sampled_from(FeederDirection), "current_feeder_direction": sampled_from(FeederDirection), - "traced_phases": builds(TracedPhases, phase_status=integers(min_value=0, max_value=15)), "connectivity_node": builds(ConnectivityNode, mrid=mrid_strategy) } @@ -28,7 +27,6 @@ *ac_dc_terminal_args, ConductingEquipment(mrid=generate_id()), PhaseCode.XYN, - TracedPhases(1), 1, FeederDirection.UPSTREAM, FeederDirection.DOWNSTREAM, @@ -45,19 +43,19 @@ def test_terminal_constructor_default(): assert t.sequence_number == 0 assert t.normal_feeder_direction == FeederDirection.NONE assert t.current_feeder_direction == FeederDirection.NONE - assert t.traced_phases == TracedPhases() + assert t.normal_phases._phase_status_internal == 0 + assert t.current_phases._phase_status_internal == 0 assert not t.connectivity_node @given(**terminal_kwargs) -def test_terminal_constructor_kwargs(conducting_equipment, phases, sequence_number, normal_feeder_direction, current_feeder_direction, traced_phases, +def test_terminal_constructor_kwargs(conducting_equipment, phases, sequence_number, normal_feeder_direction, current_feeder_direction, connectivity_node, **kwargs): t = Terminal(conducting_equipment=conducting_equipment, phases=phases, sequence_number=sequence_number, normal_feeder_direction=normal_feeder_direction, current_feeder_direction=current_feeder_direction, - traced_phases=traced_phases, connectivity_node=connectivity_node, **kwargs) @@ -67,7 +65,6 @@ def test_terminal_constructor_kwargs(conducting_equipment, phases, sequence_numb assert t.sequence_number == sequence_number assert t.normal_feeder_direction == normal_feeder_direction assert t.current_feeder_direction == current_feeder_direction - assert t.traced_phases == traced_phases assert t.connectivity_node == connectivity_node @@ -78,10 +75,72 @@ def test_terminal_constructor_args(): expected_args = [ t.conducting_equipment, t.phases, - t.traced_phases, t.sequence_number, t.normal_feeder_direction, t.current_feeder_direction, t.connectivity_node ] assert (terminal_args[-len(expected_args):] == expected_args) + + +def test_connectivity(): + terminal = Terminal(mrid=generate_id()) + connectivity_node = ConnectivityNode(mrid=generate_id()) + + assert terminal.connectivity_node is None + assert terminal.connectivity_node_id is None + assert terminal.connected == False + + terminal.connect(connectivity_node) + + assert terminal.connectivity_node == connectivity_node + assert terminal.connectivity_node_id is connectivity_node.mrid + assert terminal.connected == True + + terminal.disconnect() + + assert terminal.connectivity_node is None + assert terminal.connectivity_node_id is None + assert terminal.connected == False + + +def test_connected_terminals(): + terminal1 = Terminal(mrid=generate_id()) + terminal2 = Terminal(mrid=generate_id()) + terminal3 = Terminal(mrid=generate_id()) + network_service = NetworkService() + + assert list(terminal1.connected_terminals()) == [] + + network_service.connect(terminal1, "cn1") + assert list(terminal1.connected_terminals()) == [] + + network_service.connect(terminal2, "cn1") + assert list(terminal1.connected_terminals()) == [terminal2] + + network_service.connect(terminal3, "cn1") + assert list(terminal1.connected_terminals()) == [terminal2, terminal3] + + +def test_other_terminals(): + terminal1 = Terminal(mrid=generate_id()) + terminal2 = Terminal(mrid=generate_id()) + terminal3 = Terminal(mrid=generate_id()) + ce = Junction(mrid=generate_id()) + + assert list(terminal1.other_terminals()) == [] + + ce.add_terminal(terminal1) + assert list(terminal1.other_terminals()) == [] + + ce.add_terminal(terminal2) + assert list(terminal1.other_terminals()) == [terminal2] + + ce.add_terminal(terminal3) + assert list(terminal1.other_terminals()) == [terminal2, terminal3] + + +def test_normal_and_current_phases_are_different_statuses(): + terminal = Terminal(mrid=generate_id()) + + assert terminal.normal_phases is not terminal.current_phases diff --git a/test/database/sqlite/network/test_network_database_schema.py b/test/database/sqlite/network/test_network_database_schema.py index 80b6fbc26..fbc1ecf47 100644 --- a/test/database/sqlite/network/test_network_database_schema.py +++ b/test/database/sqlite/network/test_network_database_schema.py @@ -648,7 +648,8 @@ async def test_schema_energy_source(self, energy_source: EnergySource): # Need to apply phases to match after the database load. network_service = SchemaNetworks().network_services_of(EnergySource, energy_source) - await Tracing.set_phases().run(network_service) + await Tracing.set_phases().run(network_service, network_state_operators=NetworkStateOperators.NORMAL) + await Tracing.set_phases().run(network_service, network_state_operators=NetworkStateOperators.CURRENT) await self._validate_schema(network_service) diff --git a/test/services/network/test_network_service_comparator.py b/test/services/network/test_network_service_comparator.py index 6947feacb..b1d9c0794 100644 --- a/test/services/network/test_network_service_comparator.py +++ b/test/services/network/test_network_service_comparator.py @@ -20,14 +20,15 @@ PowerElectronicsWindUnit, AcLineSegment, Breaker, BusbarSection, Connector, Disconnector, EnergyConnection, \ EnergyConsumer, PhaseShuntConnectionKind, SinglePhaseKind, EnergySource, EnergySourcePhase, Fuse, Jumper, Line, \ PowerTransformer, PowerTransformerEnd, VectorGroup, \ - ProtectedSwitch, Recloser, RegulatingCondEq, ShuntCompensator, Switch, ObjectDifference, ValueDifference, Circuit, Loop, NetworkService, TracedPhases, \ + ProtectedSwitch, Recloser, RegulatingCondEq, ShuntCompensator, Switch, ObjectDifference, ValueDifference, Circuit, Loop, NetworkService, \ FeederDirection, ShuntCompensatorInfo, TransformerConstructionKind, TransformerFunctionKind, LvFeeder, Sensor, \ CurrentTransformer, PotentialTransformer, CurrentTransformerInfo, PotentialTransformerInfo, PotentialTransformerKind, Ratio, SwitchInfo, RelayInfo, \ CurrentRelay, EvChargingUnit, PowerDirectionKind, RegulatingControl, TapChangerControl, RegulatingControlModeKind, \ TransformerCoolingType, ProtectionRelayFunction, ProtectionRelayScheme, RelaySetting, DistanceRelay, VoltageRelay, ProtectionKind, \ ProtectionRelaySystem, Ground, GroundDisconnector, SeriesCompensator, BatteryControl, BatteryControlMode, AssetFunction, PanDemandResponseFunction, \ StaticVarCompensator, SVCControlMode, PerLengthPhaseImpedance, ReactiveCapabilityCurve, Curve, CurveData, \ - PhaseImpedanceData, EarthFaultCompensator, GroundingImpedance, PetersenCoil, RotatingMachine, SynchronousMachine, SynchronousMachineKind, LvSubstation + PhaseImpedanceData, EarthFaultCompensator, GroundingImpedance, PetersenCoil, RotatingMachine, SynchronousMachine, SynchronousMachineKind, LvSubstation, \ + PhaseStatus from zepben.ewb.model.cim.extensions.iec61970.base.core.hv_customer import HvCustomer from zepben.ewb.model.cim.extensions.iec61970.base.protection.directional_current_relay import DirectionalCurrentRelay from zepben.ewb.model.cim.extensions.iec61970.base.protection.polarizing_quantity_type import PolarizingQuantityType @@ -926,15 +927,27 @@ def test_compare_terminal(self): self.validator.validate_property(Terminal.normal_feeder_direction, Terminal, lambda _: FeederDirection.UPSTREAM, lambda _: FeederDirection.DOWNSTREAM) self.validator.validate_property(Terminal.current_feeder_direction, Terminal, lambda _: FeederDirection.UPSTREAM, lambda _: FeederDirection.DOWNSTREAM) - for i in range(0, 32, 4): - # noinspection PyArgumentList - self.validator.validate_property(Terminal.phases, Terminal, lambda _: TracedPhases(0x00000001 << i), lambda _: TracedPhases(0x00000002 << i)) - # noinspection PyArgumentList - self.validator.validate_property(Terminal.phases, Terminal, lambda _: TracedPhases(0x00000004 << i), lambda _: TracedPhases(0x00000008 << i)) - # noinspection PyArgumentList - self.validator.validate_property(Terminal.phases, Terminal, lambda _: TracedPhases(0x00000010 << i), lambda _: TracedPhases(0x00000020 << i)) - # noinspection PyArgumentList - self.validator.validate_property(Terminal.phases, Terminal, lambda _: TracedPhases(0x00000040 << i), lambda _: TracedPhases(0x00000080 << i)) + def _change_state(prop, val): + setattr(prop, '_phase_status_internal', val) + + for first, second in ( + (0x00000001, 0x00000002), + (0x00000010, 0x00000020), + (0x00000100, 0x00000200), + (0x00001000, 0x00002000), + ): + #expected_diff = ObjectDifference() + #expected_diff.differences["normal_phases"] = ValueDifference(1, 2) + + #self.validator.validate_compare(closed_switch, open_switch, expect_modification=difference) + + for state in (Terminal.normal_phases, Terminal.current_phases): + self.validator.validate_val_property( + state, + Terminal, + lambda _, prop: _change_state(prop, first), + lambda _, prop: _change_state(prop, second), + ) self.validator.validate_val_property( Terminal.connectivity_node, diff --git a/test/services/network/tracing/connectivity/test_terminal_connectivity_connected.py b/test/services/network/tracing/connectivity/test_terminal_connectivity_connected.py index 67d402ef4..7b272f436 100644 --- a/test/services/network/tracing/connectivity/test_terminal_connectivity_connected.py +++ b/test/services/network/tracing/connectivity/test_terminal_connectivity_connected.py @@ -228,8 +228,8 @@ def _validate_connection_multi(self, t: Terminal, expected_phases: List[Tuple[Te @staticmethod async def _replace_normal_phases(terminal: Terminal, normal_phases: PhaseCode): for index, phase in enumerate(terminal.phases.single_phases): - terminal.traced_phases.set_normal(phase, Phase.NONE) - terminal.traced_phases.set_normal(phase, normal_phases.single_phases[index]) + terminal.normal_phases[phase] = Phase.NONE + terminal.normal_phases[phase] = normal_phases.single_phases[index] def _get_next_connectivity_node(self) -> ConnectivityNode: return self._network_service.add_connectivity_node(f"cn{self._network_service.len_of(ConnectivityNode)}") diff --git a/test/services/network/tracing/phases/test_phase_status.py b/test/services/network/tracing/phases/test_phase_status.py index 1f0978699..bf2dcb525 100644 --- a/test/services/network/tracing/phases/test_phase_status.py +++ b/test/services/network/tracing/phases/test_phase_status.py @@ -2,8 +2,9 @@ # 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.ewb import Terminal, SinglePhaseKind, PhaseCode, NetworkStateOperators, NormalPhases, CurrentPhases, generate_id +from zepben.ewb import Terminal, SinglePhaseKind, PhaseCode, NetworkStateOperators, generate_id, PhaseStatus, PhaseException def test_normal_and_current_phases(): @@ -34,8 +35,8 @@ def test_normal_and_current_phases(): def test_normal_and_current_phase_codes_three(): terminal = Terminal(mrid=generate_id(), phases=PhaseCode.ABCN) - normal_phases = NormalPhases(terminal) - current_phases = CurrentPhases(terminal) + normal_phases = PhaseStatus(terminal) + current_phases = PhaseStatus(terminal) assert normal_phases.as_phase_code() == PhaseCode.NONE assert current_phases.as_phase_code() == PhaseCode.NONE @@ -67,8 +68,8 @@ def test_normal_and_current_phase_codes_three(): def test_normal_and_current_phase_codes_single(): terminal = Terminal(mrid=generate_id(), phases=PhaseCode.BC) - normal_phases = NormalPhases(terminal) - current_phases = CurrentPhases(terminal) + normal_phases = PhaseStatus(terminal) + current_phases = PhaseStatus(terminal) assert normal_phases.as_phase_code() == PhaseCode.NONE assert current_phases.as_phase_code() == PhaseCode.NONE @@ -94,8 +95,8 @@ def test_normal_and_current_phase_codes_single(): def test_normal_and_current_phase_codes_none(): terminal = Terminal(mrid=generate_id(), phases=PhaseCode.NONE) - normal_phases = NormalPhases(terminal) - current_phases = CurrentPhases(terminal) + normal_phases = PhaseStatus(terminal) + current_phases = PhaseStatus(terminal) assert normal_phases.as_phase_code() == PhaseCode.NONE assert current_phases.as_phase_code() == PhaseCode.NONE @@ -106,3 +107,58 @@ def test_normal_and_current_phase_codes_none(): assert normal_phases.as_phase_code() == PhaseCode.NONE assert current_phases.as_phase_code() == PhaseCode.NONE + + +def as_phase_code_handles_changing_terminal_phases(): + terminal = Terminal(mrid=generate_id(), phases=PhaseCode.ABN) + phase_status = PhaseStatus(terminal) + + phase_status[SinglePhaseKind.A] = SinglePhaseKind.A + phase_status[SinglePhaseKind.B] = SinglePhaseKind.B + phase_status[SinglePhaseKind.C] = SinglePhaseKind.C + phase_status[SinglePhaseKind.N] = SinglePhaseKind.N + + assert phase_status[SinglePhaseKind.A] == SinglePhaseKind.A + assert phase_status[SinglePhaseKind.B] == SinglePhaseKind.B + assert phase_status[SinglePhaseKind.C] == SinglePhaseKind.C + assert phase_status[SinglePhaseKind.N] == SinglePhaseKind.N + assert phase_status.as_phase_code() == PhaseCode.ABN + + terminal.phases = PhaseCode.AC + + assert phase_status[SinglePhaseKind.A] == SinglePhaseKind.A + assert phase_status[SinglePhaseKind.B] == SinglePhaseKind.B + assert phase_status[SinglePhaseKind.C] == SinglePhaseKind.C + assert phase_status[SinglePhaseKind.N] == SinglePhaseKind.N + assert phase_status.as_phase_code() == PhaseCode.AC + + terminal.phases = PhaseCode.ABN + + assert phase_status.as_phase_code() == PhaseCode.ABN + + +def as_phase_code_does_not_drop_phases(): + terminal = Terminal(mrid=generate_id(), phases=PhaseCode.ABCN) + phase_status = PhaseStatus(terminal) + + phase_status[SinglePhaseKind.B] = SinglePhaseKind.A + phase_status[SinglePhaseKind.C] = SinglePhaseKind.A + + assert phase_status.as_phase_code() is None + + +def test_invalid_nominal_phase(): + terminal = Terminal(mrid=generate_id(), phases=PhaseCode.ABCN) + phase_status = PhaseStatus(terminal) + + with pytest.raises(ValueError, match="INTERNAL ERROR: Phase INVALID is invalid"): + phase_status[SinglePhaseKind.INVALID] + + +def test_crossing_phases_exception(): + terminal = Terminal(mrid=generate_id(), phases=PhaseCode.ABCN) + phase_status = PhaseStatus(terminal) + + with pytest.raises(PhaseException, match="Crossing Phases"): + phase_status[SinglePhaseKind.A] = SinglePhaseKind.A + phase_status[SinglePhaseKind.A] = SinglePhaseKind.B diff --git a/test/services/network/tracing/phases/test_traced_phases.py b/test/services/network/tracing/phases/test_traced_phases.py deleted file mode 100644 index f4e56511d..000000000 --- a/test/services/network/tracing/phases/test_traced_phases.py +++ /dev/null @@ -1,71 +0,0 @@ -# Copyright 2024 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.ewb import TracedPhases, SinglePhaseKind as SPK -from zepben.ewb.exceptions import PhaseException - -traced_phases = TracedPhases() - - -def test_set_and_get(): - # -- Setting -- - assert traced_phases.set_normal(SPK.A, SPK.N) - assert traced_phases.set_normal(SPK.B, SPK.C) - assert traced_phases.set_normal(SPK.C, SPK.B) - assert traced_phases.set_normal(SPK.N, SPK.A) - - assert traced_phases.set_current(SPK.A, SPK.A) - assert traced_phases.set_current(SPK.B, SPK.B) - assert traced_phases.set_current(SPK.C, SPK.C) - assert traced_phases.set_current(SPK.N, SPK.N) - - # -- Getting Phase-- - assert traced_phases.normal(SPK.A) == SPK.N - assert traced_phases.normal(SPK.B) == SPK.C - assert traced_phases.normal(SPK.C) == SPK.B - assert traced_phases.normal(SPK.N) == SPK.A - - assert traced_phases.current(SPK.A) == SPK.A - assert traced_phases.current(SPK.B) == SPK.B - assert traced_phases.current(SPK.C) == SPK.C - assert traced_phases.current(SPK.N) == SPK.N - - # -- Setting Unchanged -- - assert not traced_phases.set_normal(SPK.A, SPK.N) - assert not traced_phases.set_normal(SPK.B, SPK.C) - assert not traced_phases.set_normal(SPK.C, SPK.B) - assert not traced_phases.set_normal(SPK.N, SPK.A) - - assert not traced_phases.set_current(SPK.A, SPK.A) - assert not traced_phases.set_current(SPK.B, SPK.B) - assert not traced_phases.set_current(SPK.C, SPK.C) - assert not traced_phases.set_current(SPK.N, SPK.N) - - -def test_invalid_nominal_phase_normal(): - with pytest.raises(ValueError) as e_info: - traced_phases.normal(SPK.INVALID) - - assert str(e_info.value) == "INTERNAL ERROR: Phase INVALID is invalid. Must not be NONE or INVALID." - - -def test_crossing_phases_exception_normal(): - with pytest.raises(PhaseException) as e_info: - traced_phases.set_normal(SPK.A, SPK.A) - traced_phases.set_normal(SPK.A, SPK.B) - - assert str(e_info.value) == "Crossing Phases." - - -def test_invalid_nominal_phase_current(): - with pytest.raises(ValueError): - traced_phases.current(SPK.INVALID) - - -def test_crossing_phases_exception_current(): - with pytest.raises(PhaseException): - traced_phases.set_current(SPK.A, SPK.A) - traced_phases.set_current(SPK.A, SPK.B) diff --git a/test/services/network/tracing/phases/test_traced_phases_bit_manipulation.py b/test/services/network/tracing/phases/test_traced_phases_bit_manipulation.py new file mode 100644 index 000000000..b1ce57f05 --- /dev/null +++ b/test/services/network/tracing/phases/test_traced_phases_bit_manipulation.py @@ -0,0 +1,54 @@ +# 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 zepben.ewb import SinglePhaseKind +from zepben.ewb.services.network.tracing.phases.traced_phases_bit_manipulation import TracedPhasesBitManipulation + + +def test_get(): + assert TracedPhasesBitManipulation().get(0x0001, SinglePhaseKind.A) == SinglePhaseKind.A + assert TracedPhasesBitManipulation().get(0x0002, SinglePhaseKind.A) == SinglePhaseKind.B + assert TracedPhasesBitManipulation().get(0x0004, SinglePhaseKind.A) == SinglePhaseKind.C + assert TracedPhasesBitManipulation().get(0x0008, SinglePhaseKind.A) == SinglePhaseKind.N + + assert TracedPhasesBitManipulation().get(0x0010, SinglePhaseKind.B) == SinglePhaseKind.A + assert TracedPhasesBitManipulation().get(0x0020, SinglePhaseKind.B) == SinglePhaseKind.B + assert TracedPhasesBitManipulation().get(0x0040, SinglePhaseKind.B) == SinglePhaseKind.C + assert TracedPhasesBitManipulation().get(0x0080, SinglePhaseKind.B) == SinglePhaseKind.N + + assert TracedPhasesBitManipulation().get(0x0100, SinglePhaseKind.C) == SinglePhaseKind.A + assert TracedPhasesBitManipulation().get(0x0200, SinglePhaseKind.C) == SinglePhaseKind.B + assert TracedPhasesBitManipulation().get(0x0400, SinglePhaseKind.C) == SinglePhaseKind.C + assert TracedPhasesBitManipulation().get(0x0800, SinglePhaseKind.C) == SinglePhaseKind.N + + assert TracedPhasesBitManipulation().get(0x1000, SinglePhaseKind.N) == SinglePhaseKind.A + assert TracedPhasesBitManipulation().get(0x2000, SinglePhaseKind.N) == SinglePhaseKind.B + assert TracedPhasesBitManipulation().get(0x4000, SinglePhaseKind.N) == SinglePhaseKind.C + assert TracedPhasesBitManipulation().get(0x8000, SinglePhaseKind.N) == SinglePhaseKind.N + + +def test_set(): + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.A, SinglePhaseKind.A) == 0x0001 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.A, SinglePhaseKind.B) == 0x0002 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.A, SinglePhaseKind.C) == 0x0004 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.A, SinglePhaseKind.N) == 0x0008 + assert TracedPhasesBitManipulation().set(0xFFFF, SinglePhaseKind.A, SinglePhaseKind.NONE) == 0xFFF0 + + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.B, SinglePhaseKind.A) == 0x0010 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.B, SinglePhaseKind.B) == 0x0020 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.B, SinglePhaseKind.C) == 0x0040 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.B, SinglePhaseKind.N) == 0x0080 + assert TracedPhasesBitManipulation().set(0xFFFF, SinglePhaseKind.B, SinglePhaseKind.NONE) == 0xFF0F + + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.C, SinglePhaseKind.A) == 0x0100 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.C, SinglePhaseKind.B) == 0x0200 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.C, SinglePhaseKind.C) == 0x0400 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.C, SinglePhaseKind.N) == 0x0800 + assert TracedPhasesBitManipulation().set(0xFFFF, SinglePhaseKind.C, SinglePhaseKind.NONE) == 0xF0FF + + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.N, SinglePhaseKind.A) == 0x1000 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.N, SinglePhaseKind.B) == 0x2000 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.N, SinglePhaseKind.C) == 0x4000 + assert TracedPhasesBitManipulation().set(0x0000, SinglePhaseKind.N, SinglePhaseKind.N) == 0x8000 + assert TracedPhasesBitManipulation().set(0xFFFF, SinglePhaseKind.N, SinglePhaseKind.NONE) == 0x0FFF diff --git a/test/services/network/tracing/phases/util.py b/test/services/network/tracing/phases/util.py index 6acd185f2..9587526f9 100644 --- a/test/services/network/tracing/phases/util.py +++ b/test/services/network/tracing/phases/util.py @@ -5,7 +5,7 @@ import logging from typing import Iterable, Optional, Union -from zepben.ewb import ConductingEquipment, NetworkService, SinglePhaseKind as Phase, Terminal, PhaseStatus, PhaseCode, Tracing, Traversal +from zepben.ewb import ConductingEquipment, NetworkService, SinglePhaseKind as Phase, Terminal, PhaseStatus, PhaseCode, Tracing, Traversal, SinglePhaseKind from zepben.ewb.services.network.tracing.networktrace.network_trace_step import NetworkTraceStep logger = logging.getLogger("phase_logger.py") @@ -95,7 +95,7 @@ def _log_equipment(step: NetworkTraceStep, _: bool): "\n", ce) - def phase_info(term, phase): + def phase_info(term: Terminal, phase: SinglePhaseKind) -> str: nps = term.normal_phases[phase] cps = term.current_phases[phase] @@ -110,7 +110,8 @@ def phase_info(term, phase): ) -def _do_phase_validation(terminal: Terminal, phase_status: PhaseStatus, expected_phases: Union[Iterable[Phase], PhaseCode]): +def _do_phase_validation(terminal: Terminal, phase_status: PhaseStatus, expected_phases: Union[Iterable[Phase], PhaseCode]) -> None: + """:raises AssertionError: if `expected_phases` is `None`.""" if list(expected_phases) == [Phase.NONE]: for nominal_phase in terminal.phases.single_phases: assert phase_status[nominal_phase] == Phase.NONE, \