Tracing

Once you have a connected network model you can start following the connectivity to do some interesting analyses, from simple tasks such as adding up cable lengths along a line to advanced functions like a full load flow analysis.

The EWB SDK provides an API to make following connectivity (what we will call tracing) less cumbersome and more productive. The Tracing class is the primary entry point: its static factory methods return ready-made, state-aware traces for common jobs. Under the covers those traces are NetworkTrace instances driven by NetworkStateOperators and stop/queue conditions, which you can also assemble yourself for advanced use cases. The tracing package is a great place to get an overview of the tools we provide as part of the SDK.

Basic Connectivity

So, let's start at the absolute basic use case. "What is directly connected to this piece of equipment?". Obviously we provide functions to let you find out:

from zepben.ewb import Breaker, ConnectivityResult, PhaseCode, connected_equipment

a_switch = network.get("aSwitch", Breaker)
connectivity: list[ConnectivityResult] = connected_equipment(a_switch, PhaseCode.ABCN)  # network.get: base_service.py#L164
for cr in connectivity:
    print(f"{cr.from_equip} is connected to {cr.to_equip} via terminals {cr.from_terminal} -> {cr.to_terminal}")
    for phase_path in cr.nominal_phase_paths:
        print(f"  phase {phase_path.from_phase} is wired to phase {phase_path.to_phase}")

connected_equipment (defined in network_service.py) is a module-level function and is exported directly from zepben.ewb, as are the equipment, terminal, node and phase types used above. network.get("aSwitch", Breaker) is BaseService.get. Each ConnectivityResult links two terminals: from_equip and to_equip are properties that look up the owning ConductingEquipment, and nominal_phase_paths is a tuple attribute of NominalPhasePath items, not a method.

Sometimes you want to be a bit more specific about your connectivity, such as "what is connected only to one side of the equipment?". You can do this by asking about a specific terminal:

from zepben.ewb import Breaker, PhaseCode, connected_terminals

a_switch = network.get("aSwitch", Breaker)
connectivity = connected_terminals(a_switch.get_terminal_by_sn(1), PhaseCode.ABCN)

connected_terminals also lives in zepben.ewb.services.network.network_service (here) and takes a Terminal (found with ConductingEquipment.get_terminal_by_sn), not the equipment itself. This produces a subset of the first example's results: the same kind of ConnectivityResult instances, but only for connections made through the given terminal.

Sometimes you do not care about the phase connectivity at all, and are just interested in terminals that share a connectivity node:

for terminal in a_switch.get_terminal_by_sn(1).connected_terminals():
    print(terminal)

This uses Terminal.connected_terminals and returns a Generator of all the terminals that share the same ConnectivityNode, regardless of any phase connectivity.

Tracing

Feeder Direction

UPSTREAM and DOWNSTREAM are concepts that only apply to equipment on feeders (HV or LV). These directions are purely for finding the feeder head, and have not been calculated with any power flow analysis. The FeederDirection enum also defines BOTH when the feeder head can be reached in either direction (e.g. in a loop or dual-feed situation), CONNECTOR for single-terminal Connector equipment, and NONE for terminals that are not on a feeder.

Terminals start out with no direction: it must be computed first, by Tracing.set_direction() or Tracing.assign_equipment_to_feeders() (see Setting Feeder Direction). Until you do, a downstream() or upstream() trace condition (see Stop and Queue Conditions) will queue nothing.

Network Traces

To trace a network, we need to repeat the above process for each piece of equipment in the connectivity result list. In computer science, stepping between nodes in a data structure like this is known as a traversal. Conveniently, the SDK provides a Traversal interface with a number of implementations for different use cases. The one you will use most is NetworkTrace, created through the Tracing factory methods:

• Tracing.network_trace() — a trace that steps between connected terminals, backed by a depth-first queue and defaulting to the normal state of the network (see State-Aware Tracing).
• Tracing.network_trace_branching() — the same, except a new 'branch' is created each time a step can move to two or more terminals, so loops can be traced both ways around.

Each step the trace takes is a NetworkTraceStep, which carries the path just walked (its from_terminal, to_terminal and any traversed_ac_line_segment), access to the equipment stepped onto via step.path.to_equipment, and data for the step. During a run the trace calls your add_step_action actions and checks your add_stop_condition conditions at each step, and running a trace is async, so you await it. Now, say we wanted to identify the nearest circuit breakers to a piece of equipment, capturing all equipment between those breakers. We can do that with the following:

from zepben.ewb import AcLineSegment, Breaker, NetworkService, Tracing

network = NetworkService()

# Populate network service with your network

acls = network.get("my conductor", AcLineSegment)
breakers = []
equipment_between_breakers = []

def capture_step(step, context):
    equipment = step.path.to_equipment
    if isinstance(equipment, Breaker):
        breakers.append(equipment)
    elif not context.is_stopping:
        equipment_between_breakers.append(equipment)

trace = Tracing.network_trace()
trace.add_stop_condition(lambda step, context: isinstance(step.path.to_equipment, Breaker))
trace.add_step_action(capture_step)
await trace.run(acls)

print(f"The closest circuit breakers are: {breakers}")
print(f"The equipment between the above breakers: {equipment_between_breakers}")

Let's break down what is happening with the trace:

• add_stop_condition is returning true if the equipment at the current step (step.path.to_equipment) is a Breaker. This will make the trace stop at the current step if the equipment is a breaker. Note that this will not terminate the trace, it simply stops any more traversing from the current step. That is, no more equipment will be queued to be stepped to from the equipment at this step. If there are other paths / branches in the queue, they will still be stepped to (traversal.py).
• add_step_action is called with the step and its StepContext. context.is_stopping (here) says whether the trace is stopping at this step, so the action captures breakers as breakers, and otherwise captures "between breakers" equipment if it is not stopping (as stopping happens on the breaker).

As you can see this provides an extremely simple but powerful way to perform useful traces over your network.

Step Data

Every NetworkTraceStep can carry arbitrary data for that step. Pass a callback as the compute_data argument of a Tracing factory, and the trace calls it with (current_step, context, next_path) when queueing each next step (ComputeData.compute_next); whatever it returns becomes the next step's step.data. This is how you can add up cable lengths along a line:

from zepben.ewb import Tracing

def add_segment_length(current_step, context, next_path):
    segment = next_path.traversed_ac_line_segment
    length = segment.length if segment else None
    return current_step.data + (length or 0)

trace = Tracing.network_trace(compute_data=add_segment_length)
trace.add_step_action(lambda step, context: print(f"{step.data}m of cable traced so far at {step.path.to_equipment}"))
await trace.run(acls, 0)

AcLineSegment.length is optional (conductor.py), hence the length or 0 guard, and the second positional argument to run is the data attached to the start step (0 here, as the starting cable length).

State-Aware Tracing: Normal vs Current

Network objects in an EWB model have two states: the normal state (how the network is designed to run) and the current state (how it is running right now). Switches have both a normally-open and a currently-open state, equipment has a normally-in-service and a currently-in-service flag, and terminals have normal and current feeder directions. Rather than duplicating trace code for each state, every trace is made state-aware with a NetworkStateOperators instance: NetworkStateOperators.NORMAL for the normal state and NetworkStateOperators.CURRENT for the current state. The same trace definition, conditions and actions work in either state; you just pick the operators when you create the trace:

from zepben.ewb import Tracing, NetworkStateOperators, stop_at_open, downstream

# Directions must have been computed for the same state you are tracing (see below).
await Tracing.assign_equipment_to_feeders().run(network, NetworkStateOperators.CURRENT)
await Tracing.set_direction().run(network, NetworkStateOperators.CURRENT)

# Trace downstream of `acls` in the current state, stopping at currently open switches.
current_downstream = (
    Tracing.network_trace(NetworkStateOperators.CURRENT)
    .add_condition(stop_at_open())
    .add_condition(downstream())
)
await current_downstream.run(acls)

Because this trace uses CURRENT, stop_at_open() checks each switch's current open state and downstream() follows each terminal's current feeder direction. The exact same object created with NetworkStateOperators.NORMAL instead would check the normally-open and normal direction instead. One behaviour applies to every trace, in both states: equipment that is not in service is never stepped onto, because the trace's next-path provider asks the state operators is_in_service for every candidate step (here).

Stop and Queue Conditions

There are two kinds of conditions a trace checks. A stop condition answers (item, context) -> bool for the step the trace is currently on (ShouldStop), and a true answer stops further stepping from that item. A queue condition answers (next_item, next_context, current_item, current_context) -> bool for each candidate next step (ShouldQueue), and all queue conditions must answer true for the next step to be queued.

The SDK ships several ready-made conditions in conditions.py, all exported from zepben.ewb:

• stop_at_open() does not queue a step that walks through an open switch. It is a queue condition checked on the internal step through the switch (OpenCondition), because stepping through a switch is internal to the switch itself. It also checks the switch in whichever state the trace is running.
• upstream(), downstream() and with_direction(direction) only queue steps whose terminal feeder direction matches, using the trace's state operators (DirectionCondition).
• limit_equipment_steps(limit, equipment_type=None) stops a path after the given number of equipment steps, optionally counting only one equipment type.
• stop_on_shunt_compensator_ground() stops a path trying to traverse a ShuntCompensator through its ground terminal.

These factories are DSL: call them with no arguments and NetworkTrace.add_condition binds the returned condition to the trace's own NetworkStateOperators (network_trace.py). You can also call the operator forms directly, e.g. NetworkStateOperators.NORMAL.downstream() (here). Queue-shaped conditions (including stop_at_open(), which is a queue condition) go through add_condition or add_queue_condition; stop-shaped conditions such as limit_equipment_steps(...) or a 2-argument lambda go through add_stop_condition:

from zepben.ewb import AcLineSegment, Breaker, Tracing, limit_equipment_steps, stop_at_open, upstream

trace = Tracing.network_trace()
trace.add_condition(stop_at_open())
trace.add_condition(upstream())  # requires feeder directions to have been computed first
trace.add_stop_condition(limit_equipment_steps(10, AcLineSegment))
trace.add_stop_condition(lambda step, context: isinstance(step.path.to_equipment, Breaker))
await trace.run(acls)

By default a NetworkTrace actions each piece of equipment only once per trace (FIRST_STEP_ON_EQUIPMENT, here); pass action_step_type=NetworkTraceActionType.ALL_STEPS (here) to a Tracing factory if you want every terminal step, internal and external, actioned.

Phasing

Tracing.set_phases() returns a SetPhases instance. This provides a way to dynamically set phases at runtime based on a network's energy sources and the nominal phase connectivity within the network:

from zepben.ewb import PhaseCode, Tracing, NetworkStateOperators

# Flow phases through the whole network from all of its EnergySources, in the normal state.
await Tracing.set_phases().run(network, network_state_operators=NetworkStateOperators.NORMAL)

# Or flow phases from a single terminal. The number of phases applied must match the number
# of nominal phases on that terminal, so `PhaseCode.ABC` on an ABC terminal:
await Tracing.set_phases().run(acls.get_terminal_by_sn(1), phases=PhaseCode.ABC)

To apply phases against the current state of the network, pass network_state_operators=NetworkStateOperators.CURRENT; note that applying a phase set whose size differs from the start terminal's nominal phases raises a TracingException (set_phases.py). There are also Tracing.remove_phases() and Tracing.phase_inferrer() for, respectively, removing phases and inferring what phases equipment should have.

Setting Feeder Direction

The direction conditions need computed feeder directions, and the same Tracing class provides the tools to compute them. Assign equipment to its (LV) feeder first, then apply directions from the feeder heads:

from zepben.ewb import AcLineSegment, NetworkService, Tracing

network = NetworkService()

# Populate network service with your network, including Feeder objects with head terminals

acls = network.get("my conductor", AcLineSegment)
await Tracing.assign_equipment_to_feeders().run(network)      # tracing.py line 105
await Tracing.assign_equipment_to_lv_feeders().run(network)   # tracing.py line 110
await Tracing.set_direction().run(network)                    # tracing.py line 95

# To re-apply directions after a state change, clear them from a terminal and re-run from the
# feeder heads that clearing encountered:
head_terminals = await Tracing.clear_direction().run(acls.get_terminal_by_sn(1))  # tracing.py line 100
for head_terminal in head_terminals:
    await Tracing.set_direction().run(head_terminal)

Tracing.set_direction() returns a SetDirection whose run applies DOWNSTREAM direction from its start item, or from every feeder head in the network when passed a NetworkService; Tracing.clear_direction() (ClearDirection) clears directions starting from a Terminal and returns the feeder head terminals it encountered. All four feeder factories accept network_state_operators so you can set directions for either network state.

Building an Equipment Tree

A branching trace plus the EquipmentTreeBuilder step action gives you the tree/equipment-branch view of a network: the builder records the path taken through the network as a tree of TreeNode objects, one node per piece of equipment (identified_object, parent, children):

from zepben.ewb import AcLineSegment, EquipmentTreeBuilder, NetworkService, Tracing, downstream, stop_at_open

network = NetworkService()

# Populate network service with your network and compute directions (see above)

acls = network.get("my conductor", AcLineSegment)
builder = EquipmentTreeBuilder(calculate_leaves=True)
trace = (
    Tracing.network_trace_branching()
    .add_condition(downstream())
    .add_condition(stop_at_open())
    .add_step_action(builder)
)
await trace.run(acls)

for root in builder.roots.values():
    print(f"Tree from {root.identified_object} has {len(root.children)} children")
for leaf in builder.leaves:
    print(f"A branch ended at {leaf.identified_object}")

builder.roots maps each start item's equipment to its root TreeNode; reading builder.leaves requires calculate_leaves=True on the builder (here).