I/O and Interoperability¶

This notebook introduces the package-level import/export helpers.

The goal is not to exhaust every format, but to show the main API patterns:

• dataframe round-trips
• NetworkX conversion
• native .annnet persistence

The examples below stay small so they can be adapted directly into docs pages.

In [1]:

from pathlib import Path
import sys

NOTEBOOK_DIR = Path.cwd()
PACKAGE_ROOT = (NOTEBOOK_DIR / "..").resolve()
if str(PACKAGE_ROOT) not in sys.path:
    sys.path.insert(0, str(PACKAGE_ROOT))

import annnet as an

from pathlib import Path import sys NOTEBOOK_DIR = Path.cwd() PACKAGE_ROOT = (NOTEBOOK_DIR / "..").resolve() if str(PACKAGE_ROOT) not in sys.path: sys.path.insert(0, str(PACKAGE_ROOT)) import annnet as an

In [2]:

G = an.AnnNet(directed=True)
G.add_vertices(
    [
        {"vertex_id": "A", "family": "input"},
        {"vertex_id": "B", "family": "relay"},
        {"vertex_id": "C", "family": "output"},
    ]
)
G.add_edge("A", "B", edge_id="e_ab", score=0.9)
G.add_edge("B", "C", edge_id="e_bc", score=0.8)
G.add_hyperedge(members=["A", "B", "C"], edge_id="complex_abc", kind="complex")

G = an.AnnNet(directed=True) G.add_vertices( [ {"vertex_id": "A", "family": "input"}, {"vertex_id": "B", "family": "relay"}, {"vertex_id": "C", "family": "output"}, ] ) G.add_edge("A", "B", edge_id="e_ab", score=0.9) G.add_edge("B", "C", edge_id="e_bc", score=0.8) G.add_hyperedge(members=["A", "B", "C"], edge_id="complex_abc", kind="complex")

Out[2]:

'complex_abc'

Convert to tabular data¶

annnet.to_dataframes() is the simplest way to expose an AnnNet object as explicit tables.

In [3]:

tables = an.to_dataframes(G)
sorted(tables)

tables = an.to_dataframes(G) sorted(tables)

Out[3]:

['edges', 'hyperedges', 'nodes', 'slice_weights', 'slices']

In [4]:

tables["nodes"]

tables["nodes"]

Out[4]:

shape: (3, 2)

vertex_id	family
str	str
"A"	"input"
"B"	"relay"
"C"	"output"

In [5]:

tables["edges"]

tables["edges"]

Out[5]:

shape: (2, 7)

edge_id	source	target	weight	directed	edge_type	score
str	str	str	f64	bool	str	f64
"e_ab"	"A"	"B"	1.0	true	"regular"	0.9
"e_bc"	"B"	"C"	1.0	true	"regular"	0.8

In [6]:

tables["hyperedges"]

tables["hyperedges"]

Out[6]:

shape: (1, 6)

edge_id	directed	weight	head	tail	members
str	bool	f64	null	null	list[str]
"complex_abc"	false	1.0	null	null	["B", "C", "A"]

Build a new graph from dataframe tables¶

This is convenient when your source data already lives in pandas or Polars dataframes.

In [7]:

H = an.from_dataframes(nodes=tables["nodes"], edges=tables["edges"], hyperedges=tables["hyperedges"])
print(H.shape)
H.edges_view()

H = an.from_dataframes(nodes=tables["nodes"], edges=tables["edges"], hyperedges=tables["hyperedges"]) print(H.shape) H.edges_view()

(3, 3)

Out[7]:

shape: (3, 12)

edge_id	kind	directed	global_weight	source	target	edge_type	head	tail	members	score	effective_weight
str	str	bool	f64	str	str	str	list[str]	list[str]	list[str]	f64	f64
"e_ab"	"binary"	true	1.0	"A"	"B"	"regular"	null	null	null	0.9	1.0
"e_bc"	"binary"	true	1.0	"B"	"C"	"regular"	null	null	null	0.8	1.0
"complex_abc"	"hyper"	false	1.0	null	null	null	null	null	["A", "B", "C"]	null	1.0

Convert to NetworkX¶

The NetworkX adapter is optional. If the extra dependency is not installed, the import will fail with a clear error.

In [8]:

try:
    nx_result = an.to_nx(G, directed=True, hyperedge_mode="skip")
    if isinstance(nx_result, tuple):
        nx_graph, manifest = nx_result
    else:
        nx_graph, manifest = nx_result, None
    print(type(nx_graph).__name__)
    print(nx_graph.number_of_nodes(), nx_graph.number_of_edges())
    if manifest is not None:
        print(sorted(manifest))
except ModuleNotFoundError as exc:
    print(exc)

try: nx_result = an.to_nx(G, directed=True, hyperedge_mode="skip") if isinstance(nx_result, tuple): nx_graph, manifest = nx_result else: nx_graph, manifest = nx_result, None print(type(nx_graph).__name__) print(nx_graph.number_of_nodes(), nx_graph.number_of_edges()) if manifest is not None: print(sorted(manifest)) except ModuleNotFoundError as exc: print(exc)

MultiDiGraph
3 2
['edge_attrs', 'edge_directed', 'edges', 'manifest_version', 'multilayer', 'slice_weights', 'slices', 'vertex_attrs', 'weights']

Native .annnet format¶

The package-level entry points are annnet.write() and annnet.read().

The code below is left as a minimal template because the native writer was not fully verified in this notebook authoring environment.

In [9]:

from pathlib import Path

archive_path = Path("example.annnet")

an.write(G, archive_path, overwrite=True)
restored = an.read(archive_path)
restored.shape

from pathlib import Path archive_path = Path("example.annnet") an.write(G, archive_path, overwrite=True) restored = an.read(archive_path) restored.shape

Out[9]:

(3, 3)

In [ ]: