Causal activity bridge¶

CORNETO solves a small CARNIVAL signaling problem. AnnNet then stores the selected signal edges, node activities, and a solution slice next to the original prior network.

In [1]:

Copied!

import annnet as an

an.info()
import annnet as an

an.info()

Out[1]:

annnetv0.3.0graph: networkxplot: graphvizI/O: 11/12

Show environment details

Version	v0.3.0
License	BSD-3-Clause
Authors	Youssef Zerta ✉, Daniele Bottazzi ✉, Denes Turei ✉
Repository	https://github.com/saezlab/annnet
Documentation	https://saezlab.github.io/annnet/reference/
Default adapter backend	networkx
Default plot backend	graphviz
Graph backends	✓networkx✓igraph✗graph-tool✓pyg
Plot backends	✓graphviz✗pydot✓matplotlib
Tabular data backends	✓polars✓pandas✓pyarrow
I/O modules	✓annnet✓json/ndjson✓dataframes✓csv✓excel✓graphml/gexf✓sif✓cx2✓parquet✓zarr✗sbml✓scverse
Installable bundles	backendsplotiostorageall

Build the CORNETO prior and measurements¶

In [2]:

Copied!





import corneto as cn
from corneto.graph import Graph

pkn_tuples = [
    ('rec1', 1, 'a'),
    ('rec1', -1, 'b'),
    ('rec1', 1, 'f'),
    ('rec1', -1, 'c'),
    ('rec2', 1, 'b'),
    ('rec2', 1, 'tf2'),
    ('b', 1, 'g'),
    ('g', -1, 'd'),
    ('rec2', -1, 'd'),
    ('a', 1, 'c'),
    ('a', -1, 'd'),
    ('c', 1, 'd'),
    ('c', -1, 'e'),
    ('c', 1, 'tf3'),
    ('e', 1, 'a'),
    ('d', -1, 'c'),
    ('e', 1, 'tf1'),
    ('a', -1, 'tf1'),
    ('d', 1, 'tf2'),
    ('c', -1, 'tf2'),
    ('tf1', 1, 'tf2'),
    ('tf1', -1, 'rec2'),
    ('tf2', 1, 'rec1'),
    ('tf1', 1, 'f'),
]
corneto_graph = Graph.from_tuples(pkn_tuples)
samples = {
    'input_example': {
        'rec2': {'value': 1, 'mapping': 'node', 'role': 'input'},
        'tf1': {'value': -2, 'mapping': 'node', 'role': 'output'},
        'tf2': {'value': 1, 'mapping': 'node', 'role': 'output'},
    }
}
data = cn.Data.from_cdict(samples)

print(corneto_graph)
print(data)
import corneto as cn
from corneto.graph import Graph

pkn_tuples = [
    ('rec1', 1, 'a'),
    ('rec1', -1, 'b'),
    ('rec1', 1, 'f'),
    ('rec1', -1, 'c'),
    ('rec2', 1, 'b'),
    ('rec2', 1, 'tf2'),
    ('b', 1, 'g'),
    ('g', -1, 'd'),
    ('rec2', -1, 'd'),
    ('a', 1, 'c'),
    ('a', -1, 'd'),
    ('c', 1, 'd'),
    ('c', -1, 'e'),
    ('c', 1, 'tf3'),
    ('e', 1, 'a'),
    ('d', -1, 'c'),
    ('e', 1, 'tf1'),
    ('a', -1, 'tf1'),
    ('d', 1, 'tf2'),
    ('c', -1, 'tf2'),
    ('tf1', 1, 'tf2'),
    ('tf1', -1, 'rec2'),
    ('tf2', 1, 'rec1'),
    ('tf1', 1, 'f'),
]
corneto_graph = Graph.from_tuples(pkn_tuples)
samples = {
    'input_example': {
        'rec2': {'value': 1, 'mapping': 'node', 'role': 'input'},
        'tf1': {'value': -2, 'mapping': 'node', 'role': 'output'},
        'tf2': {'value': 1, 'mapping': 'node', 'role': 'output'},
    }
}
data = cn.Data.from_cdict(samples)

print(corneto_graph)
print(data)

<corneto.graph._graph.Graph object at 0x7fec840f17c0>
Data(n_samples=1, n_feats=[3])

Solve CARNIVAL and inspect the selected signal¶

In [3]:

Copied!





import numpy as np
import pandas as pd
import polars as pl
from corneto.methods.carnival import CarnivalFlow

model = CarnivalFlow(lambda_reg=1e-3)
problem = model.build_from_data(corneto_graph, data)
problem.solve(verbosity=0, solver='scipy')

edge_values = pd.DataFrame(
    problem.expr.edge_value.value,
    index=model.processed_graph.E,
    columns=['edge_activity'],
).astype(int)
node_values = pd.DataFrame(
    problem.expr.vertex_value.value,
    index=model.processed_graph.V,
    columns=['node_activity'],
).astype(int)
selected_idx = np.flatnonzero(problem.expr.edge_has_signal.value)

edge_rows = []
for edge_obj, row in edge_values.iterrows():
    source, target = edge_obj
    activity = int(row['edge_activity'])
    if activity == 0:
        continue
    edge_rows.append(
        {
            'edge': f"{', '.join(sorted(source)) or '(input)'} -> "
            f"{', '.join(sorted(target)) or '(output)'}",
            'activity': activity,
        }
    )

print('objective values:', [float(obj.value) for obj in problem.objectives])
display(pl.DataFrame(edge_rows))
display(
    pl.DataFrame(
        [
            {'node': str(node), 'activity': int(row['node_activity'])}
            for node, row in node_values.iterrows()
            if int(row['node_activity']) != 0
        ]
    )
)
import numpy as np
import pandas as pd
import polars as pl
from corneto.methods.carnival import CarnivalFlow

model = CarnivalFlow(lambda_reg=1e-3)
problem = model.build_from_data(corneto_graph, data)
problem.solve(verbosity=0, solver='scipy')

edge_values = pd.DataFrame(
    problem.expr.edge_value.value,
    index=model.processed_graph.E,
    columns=['edge_activity'],
).astype(int)
node_values = pd.DataFrame(
    problem.expr.vertex_value.value,
    index=model.processed_graph.V,
    columns=['node_activity'],
).astype(int)
selected_idx = np.flatnonzero(problem.expr.edge_has_signal.value)

edge_rows = []
for edge_obj, row in edge_values.iterrows():
    source, target = edge_obj
    activity = int(row['edge_activity'])
    if activity == 0:
        continue
    edge_rows.append(
        {
            'edge': f"{', '.join(sorted(source)) or '(input)'} -> "
            f"{', '.join(sorted(target)) or '(output)'}",
            'activity': activity,
        }
    )

print('objective values:', [float(obj.value) for obj in problem.objectives])
display(pl.DataFrame(edge_rows))
display(
    pl.DataFrame(
        [
            {'node': str(node), 'activity': int(row['node_activity'])}
            for node, row in node_values.iterrows()
            if int(row['node_activity']) != 0
        ]
    )
)

---------------------------------------------------------------------------
ImportError                               Traceback (most recent call last)
Cell In[3], line 4
      2 import pandas as pd
      3 import polars as pl
----> 4 from corneto.methods.carnival import CarnivalFlow
      6 model = CarnivalFlow(lambda_reg=1e-3)
      7 problem = model.build_from_data(corneto_graph, data)

ImportError: cannot import name 'CarnivalFlow' from 'corneto.methods.carnival' (/home/runner/work/annnet/annnet/.venv/lib/python3.12/site-packages/corneto/methods/carnival.py)

CORNETO selected subgraph¶

In [4]:

Copied!

model.processed_graph.edge_subgraph(selected_idx).plot()
model.processed_graph.edge_subgraph(selected_idx).plot()

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[4], line 1
----> 1 model.processed_graph.edge_subgraph(selected_idx).plot()

NameError: name 'model' is not defined

Store the CORNETO result in AnnNet¶

In [5]:

Copied!





G = an.AnnNet(directed=True)
edge_ids_by_pair = {}
for source, sign, target in pkn_tuples:
    effect = 'activates' if sign > 0 else 'inhibits'
    edge_id = f'{source}_{effect}_{target}'
    G.add_edges(
        source,
        target,
        edge_id=edge_id,
        weight=abs(sign),
        interaction=sign,
        effect=effect,
    )
    edge_ids_by_pair[(source, target)] = edge_id

selected_edges = []
for edge_obj, row in edge_values.iterrows():
    source, target = edge_obj
    activity = int(row['edge_activity'])
    if activity == 0 or len(source) != 1 or len(target) != 1:
        continue
    pair = (next(iter(source)), next(iter(target)))
    edge_id = edge_ids_by_pair.get(pair)
    if edge_id is None:
        continue
    selected_edges.append(edge_id)
    G.attrs.set_edge_attrs(edge_id, corneto_activity=activity)

for node, row in node_values.iterrows():
    if str(node) in G.nodes():
        G.attrs.set_node_attrs(str(node), corneto_activity=int(row['node_activity']))

G.slices.add('corneto_signal')
G.slices.add_edges('corneto_signal', selected_edges)

selected_labels = {
    row['edge_id']: f"{row['source']} -> {row['target']}"
    for row in G.views.edges().iter_rows(named=True)
}
print('AnnNet shape:', G.shape)
print('selected edges:', [selected_labels[eid] for eid in sorted(selected_edges)])
G.views.edges().select(
    ['edge_id', 'source', 'target', 'effect', 'corneto_activity']
)
G = an.AnnNet(directed=True)
edge_ids_by_pair = {}
for source, sign, target in pkn_tuples:
    effect = 'activates' if sign > 0 else 'inhibits'
    edge_id = f'{source}_{effect}_{target}'
    G.add_edges(
        source,
        target,
        edge_id=edge_id,
        weight=abs(sign),
        interaction=sign,
        effect=effect,
    )
    edge_ids_by_pair[(source, target)] = edge_id

selected_edges = []
for edge_obj, row in edge_values.iterrows():
    source, target = edge_obj
    activity = int(row['edge_activity'])
    if activity == 0 or len(source) != 1 or len(target) != 1:
        continue
    pair = (next(iter(source)), next(iter(target)))
    edge_id = edge_ids_by_pair.get(pair)
    if edge_id is None:
        continue
    selected_edges.append(edge_id)
    G.attrs.set_edge_attrs(edge_id, corneto_activity=activity)

for node, row in node_values.iterrows():
    if str(node) in G.nodes():
        G.attrs.set_node_attrs(str(node), corneto_activity=int(row['node_activity']))

G.slices.add('corneto_signal')
G.slices.add_edges('corneto_signal', selected_edges)

selected_labels = {
    row['edge_id']: f"{row['source']} -> {row['target']}"
    for row in G.views.edges().iter_rows(named=True)
}
print('AnnNet shape:', G.shape)
print('selected edges:', [selected_labels[eid] for eid in sorted(selected_edges)])
G.views.edges().select(
    ['edge_id', 'source', 'target', 'effect', 'corneto_activity']
)

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[5], line 17
     14     edge_ids_by_pair[(source, target)] = edge_id
     16 selected_edges = []
---> 17 for edge_obj, row in edge_values.iterrows():
     18     source, target = edge_obj
     19     activity = int(row['edge_activity'])

NameError: name 'edge_values' is not defined

AnnNet solution slice¶

In [6]:

Copied!

from annnet.utils import plotting

solution = G.subgraph_from_slice('corneto_signal')
plotting.plot(solution, backend='graphviz', show_edge_labels=True)
from annnet.utils import plotting

solution = G.subgraph_from_slice('corneto_signal')
plotting.plot(solution, backend='graphviz', show_edge_labels=True)

---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
Cell In[6], line 3
      1 from annnet.utils import plotting
----> 3 solution = G.subgraph_from_slice('corneto_signal')
      4 plotting.plot(solution, backend='graphviz', show_edge_labels=True)

File ~/work/annnet/annnet/annnet/core/_Ops.py:666, in Operations.subgraph_from_slice(self, slice_id, resolve_slice_weights)
    646 """Create a subgraph induced by a single slice.
    647 
    648 Parameters
   (...)    663     If the slice does not exist.
    664 """
    665 if slice_id not in self._slices:
--> 666     raise KeyError(f'slice {slice_id} not found')
    668 slice_meta = self._slices[slice_id]
    669 V = set(slice_meta['nodes'])

KeyError: 'slice corneto_signal not found'

The CORNETO solve remains reproducible in the notebook, while AnnNet keeps the result attached to a graph object that can be sliced, annotated, exported, or combined with other analyses.