Flexible edge orientation¶

Some edges shouldn't be statically directed. Their orientation depends on the current value of an attribute — a temperature gradient flips a flow direction; a regulatory signal switches an activation to an inhibition; a measurement crosses a threshold.

AnnNet supports this through edge direction policies: declare the policy at edge creation, then update the watched attribute later and the incidence column flips automatically.

Anatomy of a policy¶

A policy dict has these keys:

key	meaning
var	name of the attribute to watch
threshold	the cut-off value
scope	'edge' (watch an edge attribute) or 'vertex' (watch a vertex attribute on src/tgt)
above	which direction wins when x > threshold: 's->t' or 't->s'
tie	what to do at equality: 'keep', 'undirected', 's->t', 't->s'

1. Edge-scope policy¶

Watch an edge attribute. Above the threshold ⇒ orient src→tgt; below ⇒ orient tgt→src.

In [1]:

from annnet import AnnNet

G = AnnNet(directed=True)
G.add_vertices(['A', 'B'])

# Edge AB whose orientation is governed by its 'temperature' attribute.
G.add_edges(
    'A',
    'B',
    edge_id='heat_flow',
    weight=1.0,
    flexible={
        'var': 'temperature',
        'threshold': 10.0,
        'scope': 'edge',
        'above': 's->t',
        'tie': 'keep',
    },
)
print('policy registered on:', list(G.edge_direction_policy))

from annnet import AnnNet G = AnnNet(directed=True) G.add_vertices(['A', 'B']) # Edge AB whose orientation is governed by its 'temperature' attribute. G.add_edges( 'A', 'B', edge_id='heat_flow', weight=1.0, flexible={ 'var': 'temperature', 'threshold': 10.0, 'scope': 'edge', 'above': 's->t', 'tie': 'keep', }, ) print('policy registered on:', list(G.edge_direction_policy))

policy registered on: ['heat_flow']

In [2]:

# Helper to inspect the incidence column, via the public API.
def show_column(graph, eid):
    col = graph.edge_to_idx[eid]  # edge id -> matrix column
    incidence = graph.ops.incidence(values=True)  # dense: vertices x edges
    labels = list(graph.ops.incidence_as_lists())  # vertex ids, in row order
    print(f'incidence column for {eid}:')
    for row, vid in enumerate(labels):
        print(f'  {vid}: {incidence[row, col]:+.2f}')


show_column(G, 'heat_flow')

# Helper to inspect the incidence column, via the public API. def show_column(graph, eid): col = graph.edge_to_idx[eid] # edge id -> matrix column incidence = graph.ops.incidence(values=True) # dense: vertices x edges labels = list(graph.ops.incidence_as_lists()) # vertex ids, in row order print(f'incidence column for {eid}:') for row, vid in enumerate(labels): print(f' {vid}: {incidence[row, col]:+.2f}') show_column(G, 'heat_flow')

incidence column for heat_flow:
  A: +1.00
  B: -1.00

In [3]:

# Push temperature above the threshold → orient src→tgt (+w on A, -w on B).
G.attrs.set_edge_attrs('heat_flow', temperature=20.0)
show_column(G, 'heat_flow')

# Push temperature above the threshold → orient src→tgt (+w on A, -w on B). G.attrs.set_edge_attrs('heat_flow', temperature=20.0) show_column(G, 'heat_flow')

incidence column for heat_flow:
  A: +1.00
  B: -1.00

In [4]:

# Push temperature below the threshold → orient tgt→src (-w on A, +w on B).
G.attrs.set_edge_attrs('heat_flow', temperature=5.0)
show_column(G, 'heat_flow')

# Push temperature below the threshold → orient tgt→src (-w on A, +w on B). G.attrs.set_edge_attrs('heat_flow', temperature=5.0) show_column(G, 'heat_flow')

incidence column for heat_flow:
  A: -1.00
  B: +1.00

2. Vertex-scope policy¶

Watch a vertex attribute on src and tgt. The condition fires when xs - xt > 0 (or < 0, depending on above).

In [5]:

H = AnnNet(directed=True)
H.add_vertices(['A', 'B'])
H.add_edges(
    'A',
    'B',
    edge_id='diffusion',
    weight=1.0,
    flexible={
        'var': 'concentration',
        'threshold': 0.0,  # threshold on the difference (xs - xt)
        'scope': 'vertex',
        'above': 's->t',
        'tie': 'keep',
    },
)

# A high, B low → expect s->t orientation
H.attrs.set_vertex_attrs('A', concentration=10.0)
H.attrs.set_vertex_attrs('B', concentration=2.0)
show_column(H, 'diffusion')

H = AnnNet(directed=True) H.add_vertices(['A', 'B']) H.add_edges( 'A', 'B', edge_id='diffusion', weight=1.0, flexible={ 'var': 'concentration', 'threshold': 0.0, # threshold on the difference (xs - xt) 'scope': 'vertex', 'above': 's->t', 'tie': 'keep', }, ) # A high, B low → expect s->t orientation H.attrs.set_vertex_attrs('A', concentration=10.0) H.attrs.set_vertex_attrs('B', concentration=2.0) show_column(H, 'diffusion')

incidence column for diffusion:
  A: +1.00
  B: -1.00

In [6]:

# Flip the gradient → expect t->s orientation
H.attrs.set_vertex_attrs('A', concentration=1.0)
H.attrs.set_vertex_attrs('B', concentration=9.0)
show_column(H, 'diffusion')

# Flip the gradient → expect t->s orientation H.attrs.set_vertex_attrs('A', concentration=1.0) H.attrs.set_vertex_attrs('B', concentration=9.0) show_column(H, 'diffusion')

incidence column for diffusion:
  A: -1.00
  B: +1.00

3. Tie handling¶

When the watched value equals the threshold, the tie setting decides the outcome.

• 'keep' — do nothing (incidence stays as it was)
• 'undirected' — force +w on both endpoints
• 's->t' — force src→tgt
• 't->s' — force tgt→src

In [7]:

K = AnnNet(directed=True)
K.add_vertices(['A', 'B'])

for tie_mode in ('keep', 'undirected', 's->t', 't->s'):
    K.remove_edges(K.edges(), errors='ignore')
    K.add_edges(
        'A',
        'B',
        edge_id=f'tie_{tie_mode}',
        weight=1.0,
        flexible={'var': 'x', 'threshold': 5.0, 'scope': 'edge', 'tie': tie_mode},
    )
    # Set x exactly at the threshold.
    K.attrs.set_edge_attrs(f'tie_{tie_mode}', x=5.0)
    print(f'tie={tie_mode}:')
    show_column(K, f'tie_{tie_mode}')
    print()

K = AnnNet(directed=True) K.add_vertices(['A', 'B']) for tie_mode in ('keep', 'undirected', 's->t', 't->s'): K.remove_edges(K.edges(), errors='ignore') K.add_edges( 'A', 'B', edge_id=f'tie_{tie_mode}', weight=1.0, flexible={'var': 'x', 'threshold': 5.0, 'scope': 'edge', 'tie': tie_mode}, ) # Set x exactly at the threshold. K.attrs.set_edge_attrs(f'tie_{tie_mode}', x=5.0) print(f'tie={tie_mode}:') show_column(K, f'tie_{tie_mode}') print()

tie=keep:
incidence column for tie_keep:
  A: +1.00
  B: -1.00

tie=undirected:
incidence column for tie_undirected:
  A: +1.00
  B: +1.00

tie=s->t:
incidence column for tie_s->t:
  A: +1.00
  B: -1.00

tie=t->s:
incidence column for tie_t->s:
  A: -1.00
  B: +1.00

4. Bulk attribute changes¶

A bulk call (set_vertex_attrs_bulk, set_edge_attrs_bulk) collects all the policies that need re-evaluation and applies them once at the end. No need to call any orientation method by hand.

In [8]:

B = AnnNet(directed=True)
B.add_vertices(['A', 'B', 'C'])
B.add_edges(
    'A',
    'B',
    edge_id='ab',
    weight=1.0,
    flexible={'var': 'level', 'threshold': 5.0, 'scope': 'vertex'},
)
B.add_edges(
    'B',
    'C',
    edge_id='bc',
    weight=1.0,
    flexible={'var': 'level', 'threshold': 5.0, 'scope': 'vertex'},
)

B.attrs.set_vertex_attrs_bulk(
    {
        'A': {'level': 10.0},
        'B': {'level': 8.0},
        'C': {'level': 1.0},
    }
)

for eid in ('ab', 'bc'):
    show_column(B, eid)
    print()

B = AnnNet(directed=True) B.add_vertices(['A', 'B', 'C']) B.add_edges( 'A', 'B', edge_id='ab', weight=1.0, flexible={'var': 'level', 'threshold': 5.0, 'scope': 'vertex'}, ) B.add_edges( 'B', 'C', edge_id='bc', weight=1.0, flexible={'var': 'level', 'threshold': 5.0, 'scope': 'vertex'}, ) B.attrs.set_vertex_attrs_bulk( { 'A': {'level': 10.0}, 'B': {'level': 8.0}, 'C': {'level': 1.0}, } ) for eid in ('ab', 'bc'): show_column(B, eid) print()

incidence column for ab:
  A: +1.00
  B: -1.00
  C: +0.00

incidence column for bc:
  A: +0.00
  B: +1.00
  C: -1.00

When to use this¶

• Signal-direction switching driven by phosphorylation state
• Flow direction driven by a measured gradient
• Conditional activation / inhibition based on a regulator's concentration

For static directed edges, just pass directed=True to add_edges — no policy needed.