flpq_data.graphs.readwrite.mtx#

Read (and write) a graph from (and to) a directory of MatrixMarket files.

Functions

`filename_to_label`(filename)	Returns the edge label of a MatrixMarket file name.
`graph_from_mtx_dir`(path)	Loads a graph from a directory of MatrixMarket files.
`graph_to_mtx_dir`(graph, path)	Saves the `graph` to a directory of MatrixMarket files by `path`.
`iter_edges_from_mtx_dir`(path)	Yields the edges of a directory of MatrixMarket files as a stream.
`label_to_filename`(label)	Returns the canonical MatrixMarket file name of an edge label.
`mtx_dir_from_edges`(edges, path, *[, dimension])	Writes an edge stream to a directory of MatrixMarket files.

filename_to_label(filename: Path | str) → str[source]#

Returns the edge label of a MatrixMarket file name.

The file name is the label itself (load_5.mtx holds the edges labeled load_5).

Parameters:

filenameUnion[Path, str]: The name of a MatrixMarket file.

Returns:

labelstr: The edge label of the file.

Examples

>>> filename_to_label("type.mtx")
'type'
>>> filename_to_label("alloc_r.mtx")
'alloc_r'
>>> filename_to_label("load_5.mtx")
'load_5'

graph_from_mtx_dir(path: Path | str) → MultiDiGraph[source]#

Loads a graph from a directory of MatrixMarket files.

Each *.mtx file in the directory is a Boolean pattern matrix with one entry per edge of a single label (0-based indices, no values); the label is derived from the file name by filename_to_label().

Parameters:

pathUnion[Path, str]: The path to the directory with the MatrixMarket files.

Returns:

gMultiDiGraph: Loaded graph.

Examples

>>> import pathlib, tempfile
>>> d = pathlib.Path(tempfile.mkdtemp()) / "graph"
>>> _ = d.mkdir(parents=True)
>>> _ = (d / "a.mtx").write_text(
...     "%%MatrixMarket matrix coordinate pattern general\n"
...     "%%GraphBLAS type bool\n3 3 2\n0 1\n1 2\n"
... )
>>> _ = (d / "b_5.mtx").write_text(
...     "%%MatrixMarket matrix coordinate pattern general\n"
...     "%%GraphBLAS type bool\n3 3 1\n2 0\n"
... )
>>> g = graph_from_mtx_dir(d)
>>> sorted((u, v, e["label"]) for u, v, e in g.edges(data=True))
[(0, 1, 'a'), (1, 2, 'a'), (2, 0, 'b_5')]

graph_to_mtx_dir(graph: MultiDiGraph, path: Path | str) → Path[source]#

Saves the graph to a directory of MatrixMarket files by path.

One file per edge label (the canonical name from label_to_filename()); each file is a Boolean pattern matrix with 0-based indices and no values. The node ids must be non-negative integers (the matrix dimensions are the largest node id plus one).

Parameters:

graphMultiDiGraph: Graph to save.
pathUnion[Path, str]: The path to the directory where the MatrixMarket files will be saved.

Returns:

pathPath: Path to the directory where the graph will be saved.

Examples

>>> import pathlib, tempfile
>>> g = nx.MultiDiGraph()
>>> _ = g.add_edges_from(
...     [(0, 1, {"label": "a"}), (1, 2, {"label": "a"}),
...      (2, 0, {"label": "b_5"})]
... )
>>> d = pathlib.Path(tempfile.mkdtemp())
>>> path = graph_to_mtx_dir(g, d / "graph")
>>> g2 = graph_from_mtx_dir(path)
>>> sorted((u, v, e["label"]) for u, v, e in g2.edges(data=True)) == \
...     sorted((u, v, e["label"]) for u, v, e in g.edges(data=True))
True

iter_edges_from_mtx_dir(path: Path | str) → Iterator[Tuple[int, str, int]][source]#

Yields the edges of a directory of MatrixMarket files as a stream.

Each *.mtx file is a Boolean pattern matrix with one entry per edge of a single label (0-based indices, no values); the label is derived from the file name by filename_to_label(). The files are read line by line, so the memory stays O(1) in the number of edges.

Parameters:

pathUnion[Path, str]: The path to the directory with the MatrixMarket files.

Returns:

edgesIterator[Tuple[int, str, int]]: The (u, label, v) edge tuples in file order.

Raises:

ValueError: If a file has an unexpected header or its entry count differs from the declared nnz.

Examples

>>> import pathlib, tempfile
>>> d = pathlib.Path(tempfile.mkdtemp()) / "graph"
>>> _ = d.mkdir(parents=True)
>>> _ = (d / "a.mtx").write_text(
...     "%%MatrixMarket matrix coordinate pattern general\n"
...     "%%GraphBLAS type bool\n3 3 2\n0 1\n1 2\n"
... )
>>> _ = (d / "b_5.mtx").write_text(
...     "%%MatrixMarket matrix coordinate pattern general\n"
...     "%%GraphBLAS type bool\n3 3 1\n2 0\n"
... )
>>> list(iter_edges_from_mtx_dir(d))
[(0, 'a', 1), (1, 'a', 2), (2, 'b_5', 0)]

label_to_filename(label: str) → str[source]#

Returns the canonical MatrixMarket file name of an edge label.

The canonical style is <label>.mtx; reading it back with filename_to_label() returns the same label.

Parameters:

labelstr: The edge label.

Returns:

filenamestr: The file name of the label.

Examples

>>> label_to_filename("load_5")
'load_5.mtx'
>>> label_to_filename("alloc_r")
'alloc_r.mtx'

mtx_dir_from_edges(edges: Iterable[Tuple[Any, str, Any]], path: Path | str, *, dimension: int | None = None) → Path[source]#

Writes an edge stream to a directory of MatrixMarket files.

One Boolean pattern matrix per label (the canonical name from label_to_filename()), 0-based indices and no values — the layout of iter_edges_from_mtx_dir(). The stream is consumed in a single pass: each label’s edges are appended to its own temporary file (at most one of which is open at a time, so the number of labels is unbounded), and after the stream ends every file is finalized with its three-line header. The memory stays O(1) in the number of edges; the disk holds one extra copy of the edge data until the files are finalized.

Parameters:

edgesIterable[Tuple[Any, str, Any]]: The (u, label, v) edge tuples to write; node ids must be non-negative integers (digit strings are accepted).
pathUnion[Path, str]: The path to the directory where the MatrixMarket files will be saved.
dimensionint, optional: The matrix dimensions to declare in every file header. Defaults to the largest node index plus one; give it explicitly when the graph has isolated nodes that carry no edges (e.g. an in-memory graph).

Returns:

pathPath: Path to the directory where the MatrixMarket files will be saved.

Raises:

TypeError: If a node id is not a non-negative integer (or digit string).
ValueError: If dimension is given but smaller than the largest node index plus one.

Examples

>>> import pathlib, tempfile
>>> d = pathlib.Path(tempfile.mkdtemp()) / "graph"
>>> _ = mtx_dir_from_edges([(0, "a", 1)], d)
>>> (d / "a.mtx").read_text().splitlines()[2:]
['2 2 1', '0 1']