From 8224cd92cec1e7679fb51c03ffb003c14db81850 Mon Sep 17 00:00:00 2001 From: viniciusdutra314 Date: Thu, 13 Aug 2026 16:20:46 -0300 Subject: [PATCH 1/2] feat: added hif_schema --- hypergraphx/readwrite/__init__.py | 5 +- hypergraphx/readwrite/hif.py | 263 +++++++++++++++++++++++-- pyproject.toml | 2 + tests/readwrite/hif_schema_v0.1.0.json | 84 ++++++++ tests/readwrite/test_hif.py | 210 ++++++++++++++++++++ 5 files changed, 543 insertions(+), 21 deletions(-) create mode 100644 tests/readwrite/hif_schema_v0.1.0.json create mode 100644 tests/readwrite/test_hif.py diff --git a/hypergraphx/readwrite/__init__.py b/hypergraphx/readwrite/__init__.py index 63dfcb47..b2ff7dec 100644 --- a/hypergraphx/readwrite/__init__.py +++ b/hypergraphx/readwrite/__init__.py @@ -11,8 +11,7 @@ ) from .load import load as load_any from .save import save_hypergraph -from .hif import read_hif -from .hif import write_hif +from .hif import from_hif_dict, read_hif, to_hif_dict, write_hif __all__ = [ "load_module", @@ -26,6 +25,8 @@ "load_hypergraph_from_server", "search_remote_datasets", "save_hypergraph", + "from_hif_dict", "read_hif", + "to_hif_dict", "write_hif", ] diff --git a/hypergraphx/readwrite/hif.py b/hypergraphx/readwrite/hif.py index 7c5b2d1d..28fa771d 100644 --- a/hypergraphx/readwrite/hif.py +++ b/hypergraphx/readwrite/hif.py @@ -1,10 +1,227 @@ +import copy import json -import logging +import math +import sys +from collections.abc import Hashable, Mapping +from os import PathLike +from typing import Any, Literal, TypeAlias, cast -from hypergraphx import Hypergraph +from hypergraphx import DirectedHypergraph, Hypergraph +if sys.version_info >= (3, 11): + from typing import NotRequired, TypedDict +else: # python 3.10 doesn't have NotRequired which is very useful for the HIF format + from typing_extensions import NotRequired, TypedDict -def read_hif(path: str) -> Hypergraph: +__all__ = [ + "HIFSchema", + "HIFEdgeRecord", + "HIFIncidenceRecord", + "HIFNodeRecord", + "from_hif_dict", + "read_hif", + "to_hif_dict", + "write_hif", +] + +HIF_ID: TypeAlias = str | int +Weight: TypeAlias = int | float +Metadata: TypeAlias = dict[str, Any] +NetworkType: TypeAlias = Literal["asc", "directed", "undirected"] +Direction: TypeAlias = Literal["head", "tail"] +HypergraphType: TypeAlias = Hypergraph | DirectedHypergraph +UndirectedEdge: TypeAlias = tuple[Hashable, ...] +DirectedEdge: TypeAlias = tuple[tuple[Hashable, ...], tuple[Hashable, ...]] +EdgeKey: TypeAlias = UndirectedEdge | DirectedEdge +UndirectedMembers: TypeAlias = list[Hashable] +DirectedMembers: TypeAlias = tuple[list[Hashable], list[Hashable]] +EdgeMembers: TypeAlias = UndirectedMembers | DirectedMembers + + +class HIFNodeRecord(TypedDict): + node: HIF_ID + weight: NotRequired[Weight] + attrs: NotRequired[Metadata] + + +class HIFEdgeRecord(TypedDict): + edge: HIF_ID + weight: NotRequired[Weight] + attrs: NotRequired[Metadata] + + +class HIFIncidenceRecord(TypedDict): + edge: HIF_ID + node: HIF_ID + weight: NotRequired[Weight] + direction: NotRequired[Direction] + attrs: NotRequired[Metadata] + + +# network-type is not a valid python identifier, so we need to create +# the typedict manually. +HIFSchema = TypedDict( + "HIFSchema", + { + "network-type": NotRequired[NetworkType], + "metadata": NotRequired[Metadata], + "incidences": list[HIFIncidenceRecord], + "nodes": NotRequired[list[HIFNodeRecord]], + "edges": NotRequired[list[HIFEdgeRecord]], + }, +) + + +def _record_metadata( + record: Mapping[str, Any], include_weight: bool = False +) -> Metadata: + """Copy the non-structural attributes from a HIF record.""" + attrs = copy.deepcopy(record.get("attrs", {})) + if not isinstance(attrs, dict): + raise TypeError("HIF record 'attrs' must be a dictionary.") + if include_weight and "weight" in record: + attrs["weight"] = record["weight"] + return attrs + + +def _add_metadata( + record: HIFNodeRecord | HIFEdgeRecord | HIFIncidenceRecord, + metadata: Metadata, + include_weight: bool = False, +) -> None: + attrs = copy.deepcopy(metadata) + if include_weight and "weight" in attrs: + record["weight"] = attrs.pop("weight") + if len(attrs) > 0: + record["attrs"] = attrs + + +def _replace_nan_with_none(value: Any) -> Any: + if isinstance(value, float) and math.isnan(value): + return None + if isinstance(value, dict): + return {key: _replace_nan_with_none(item) for key, item in value.items()} + if isinstance(value, list): + return [_replace_nan_with_none(item) for item in value] + if isinstance(value, tuple): + return tuple(_replace_nan_with_none(item) for item in value) + return value + + +def _empty_edges(H: HypergraphType) -> dict[HIF_ID, Metadata]: + """Copy empty-edge data exposed by HypergraphX's serialization API.""" + if not isinstance(H, Hypergraph): + return {} + return copy.deepcopy(H.expose_data_structures().get("empty_edges", {})) + + +def _edge_ids(H: HypergraphType) -> dict[EdgeKey, HIF_ID]: + """Return HIF edge IDs which do not collide with named empty edges.""" + used_ids: set[HIF_ID] = set(_empty_edges(H)) + edge_ids: dict[EdgeKey, HIF_ID] = {} + next_id = 0 + for edge, edge_id in H.get_edge_list().items(): + if edge_id in used_ids: + while next_id in used_ids: + next_id += 1 + edge_id = next_id + edge_ids[edge] = edge_id + used_ids.add(edge_id) + return edge_ids + + +def _edge_key(members: EdgeMembers, directed: bool) -> EdgeKey: + if directed: + source, target = cast(DirectedMembers, members) + return (tuple(source), tuple(target)) + return tuple(cast(UndirectedMembers, members)) + + +def from_hif_dict(data: HIFSchema) -> HypergraphType: + """Create a hypergraph from a dictionary following the HIF standard.""" + if not isinstance(data, dict): + raise TypeError("HIF data must be provided as a dictionary.") + network_type: NetworkType = data.get("network-type", "undirected") + is_directed = network_type == "directed" + metadata = copy.deepcopy(data.get("metadata", {})) + if not isinstance(metadata, dict): + raise TypeError("HIF 'metadata' must be a dictionary.") + edge_records: list[HIFEdgeRecord] = data.get("edges", []) + weighted = any("weight" in record for record in edge_records) + + if network_type == "undirected": + H: HypergraphType = Hypergraph(weighted=weighted, duplicate_policy="error") + elif is_directed: + H = DirectedHypergraph(weighted=weighted, duplicate_policy="error") + elif network_type == "asc": + raise NotImplementedError( + "HypergraphX does not support abstract simplicial complexes." + ) + else: + raise ValueError(f"Unknown hypergraph type: {network_type}") + + if "metadata" in data: + H.set_hypergraph_metadata(metadata) + + tmp_edges: dict[Hashable, EdgeMembers] = {} + for incidence in data["incidences"]: + edge = incidence["edge"] + node = incidence["node"] + + if edge not in tmp_edges: + tmp_edges[edge] = ([], []) if is_directed else [] + if is_directed: + directed_members = cast(DirectedMembers, tmp_edges[edge]) + direction = incidence.get("direction") + if direction == "tail": + directed_members[0].append(node) + elif direction == "head": + directed_members[1].append(node) + else: + raise ValueError( + "Directed HIF incidences require direction 'head' or 'tail'." + ) + else: + cast(UndirectedMembers, tmp_edges[edge]).append(node) + + for record in data.get("nodes", []): + node = record["node"] + H.add_node(node) + H.set_node_metadata(node, _record_metadata(record, include_weight=True)) + + added: dict[Hashable, EdgeKey] = {} + + for record in edge_records: + edge = record["edge"] + attrs = _record_metadata(record) + if edge in tmp_edges: + edge_key = _edge_key(tmp_edges[edge], is_directed) + H.add_edge(edge_key, weight=record.get("weight"), metadata=attrs) + added[edge] = edge_key + elif is_directed: + edge_key = ((), ()) + H.add_edge(edge_key, weight=record.get("weight"), metadata=attrs) + added[edge] = edge_key + else: + if "weight" in record: + attrs["weight"] = record["weight"] + H.add_empty_edge(edge, attrs) + + for incidence in data["incidences"]: + edge = incidence["edge"] + node = incidence["node"] + if edge not in added: + edge_key = _edge_key(tmp_edges[edge], is_directed) + H.add_edge(edge_key) + added[edge] = edge_key + attrs = _record_metadata(incidence, include_weight=True) + if attrs: + H.set_incidence_metadata(added[edge], node, attrs) + + return H + + +def read_hif(path: str | PathLike[str]) -> HypergraphType: """ Load a hypergraph from a HIF file. @@ -24,28 +241,36 @@ def read_hif(path: str) -> Hypergraph: nid = 0 with open(path) as file: - data = json.loads(file.read()) + data: HIFSchema = json.loads(file.read()) + return from_hif_dict(data, nodetype=nodetype, edgetype=edgetype) if "type" not in data: logging.getLogger(__name__).warning("No hypergraph type - assume undirected") data["type"] = "undirected" - if data["type"] == "undirected" or data["type"] == "asc": - H = Hypergraph() - elif data["type"] == "directed": - H = Hypergraph(directed=True) +def to_hif_dict(H: HypergraphType, convert_nans: bool = False) -> HIFData: + """Create a dictionary following the HIF standard from a hypergraph.""" + if isinstance(H, DirectedHypergraph): + network_type: NetworkType = "directed" + elif isinstance(H, Hypergraph): + network_type = "undirected" else: - raise ValueError(f"Unknown hypergraph type: {data['type']}") + raise TypeError( + "HIF conversion supports Hypergraph and DirectedHypergraph objects." + ) - if "metadata" in data: - H.set_hypergraph_metadata(data["metadata"]) + data: HIFData = { + "network-type": network_type, + "metadata": copy.deepcopy(H.get_hypergraph_metadata()), + "edges": [], + "nodes": [], + "incidences": [], + } - tmp_edges = {} - for incidence in data["incidences"]: - if incidence["edge"] not in edge_name_to_uid: - edge_name_to_uid[incidence["edge"]] = eid - eid += 1 - edge = edge_name_to_uid[incidence["edge"]] + for node, attrs in H.get_all_nodes_metadata().items(): + node_record: HIFNodeRecord = {"node": node} + _add_metadata(node_record, attrs, include_weight=True) + data["nodes"].append(node_record) if incidence["node"] not in node_name_to_uid: node_name_to_uid[incidence["node"]] = nid @@ -91,7 +316,7 @@ def read_hif(path: str) -> Hypergraph: return H -def write_hif(H: Hypergraph, path: str): +def write_hif(H: HypergraphType, path: str) -> None: """ Save a hypergraph to a HIF file. @@ -112,4 +337,4 @@ def write_hif(H: Hypergraph, path: str): } with open(path, "w") as file: - file.write(json.dumps(data)) + file.write(json.dumps(to_hif_dict(H, convert_nans=True))) diff --git a/pyproject.toml b/pyproject.toml index 48ea102a..cc4606e0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ dependencies = [ "networkx", "pandas", "tqdm", + "typing-extensions>=4.1; python_version < '3.11'", ] dynamic = ["version"] @@ -39,6 +40,7 @@ Changelog = "https://github.com/HGX-Team/hypergraphx/releases" [project.optional-dependencies] dev = [ "pytest", + "jsonschema>=4", "black>=24.3.0", "ruff>=0.6.0", "build>=1.2.1", diff --git a/tests/readwrite/hif_schema_v0.1.0.json b/tests/readwrite/hif_schema_v0.1.0.json new file mode 100644 index 00000000..1c5e10f1 --- /dev/null +++ b/tests/readwrite/hif_schema_v0.1.0.json @@ -0,0 +1,84 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "https://raw.githubusercontent.com/pszufe/HIF_validators/main/schemas/hif_schema_v0.1.0.json", + "title": "Hypergraph Interchange Format", + "version": "0.1.0", + "type": "object", + "properties": { + "network-type": { + "enum": ["undirected", "directed", "asc"] + }, + "metadata": { + "type": "object" + }, + "incidences": { + "type": "array", + "items": { + "type": "object", + "properties": { + "edge": { + "type": ["string", "integer"] + }, + "node": { + "type": ["string", "integer"] + }, + "weight": { + "type": "number" + }, + "direction": { + "enum": ["head", "tail"] + }, + "attrs": { + "type": "object" + } + }, + "unevaluatedProperties": false, + "additionalProperties": false, + "required": ["edge", "node"] + } + }, + "nodes": { + "type": "array", + "items": { + "type": "object", + "properties": { + "node": { + "type": ["string", "integer"] + }, + "weight": { + "type": "number" + }, + "attrs": { + "type": "object" + } + }, + "unevaluatedProperties": false, + "additionalProperties": false, + "required": ["node"] + } + }, + "edges": { + "type": "array", + "items": { + "type": "object", + "properties": { + "edge": { + "type": ["string", "integer"] + }, + "weight": { + "type": "number" + }, + "attrs": { + "type": "object" + } + }, + "unevaluatedProperties": false, + "additionalProperties": false, + "required": ["edge"] + } + } + }, + "unevaluatedProperties": false, + "additionalProperties": false, + "required": ["incidences"] +} diff --git a/tests/readwrite/test_hif.py b/tests/readwrite/test_hif.py new file mode 100644 index 00000000..d29cc146 --- /dev/null +++ b/tests/readwrite/test_hif.py @@ -0,0 +1,210 @@ +import json +from typing import Any + +import pytest + +from hypergraphx import DirectedHypergraph, Hypergraph +from hypergraphx.readwrite import ( + from_hif_dict, + read_hif, + to_hif_dict, + write_hif, +) + + +def test_from_hif_dict_matches_read_hif(tmp_path): + data = { + "network-type": "undirected", + "metadata": {"name": "in-memory"}, + "nodes": [{"node": "isolated", "weight": 3, "attrs": {"color": "red"}}], + "edges": [ + {"edge": "friendship", "weight": 2, "attrs": {"kind": "social"}}, + {"edge": "empty", "weight": 4}, + ], + "incidences": [ + { + "edge": "friendship", + "node": "alice", + "weight": 0.5, + "attrs": {"since": 2020}, + }, + {"edge": "friendship", "node": "bob"}, + ], + } + path = tmp_path / "network.json" + path.write_text(json.dumps(data), encoding="utf-8") + + from_dict = from_hif_dict(data) + from_file = read_hif(path) + + assert to_hif_dict(from_file) == to_hif_dict(from_dict) + assert from_dict.get_edges() == [("alice", "bob")] + assert from_dict.get_hypergraph_metadata() == {"name": "in-memory"} + assert from_dict.get_weight(("alice", "bob")) == 2 + assert from_dict.get_edge_metadata(("alice", "bob")) == {"kind": "social"} + assert from_dict.get_node_metadata("isolated") == { + "color": "red", + "weight": 3, + } + assert from_dict.get_incidence_metadata(("alice", "bob"), "alice") == { + "since": 2020, + "weight": 0.5, + } + assert (("alice", "bob"), "bob") not in from_dict.get_all_incidences_metadata() + assert {"edge": "empty", "weight": 4} in to_hif_dict(from_dict)["edges"] + + +def test_only_incidences_are_required_and_ids_are_preserved(): + H = from_hif_dict( + { + "incidences": [ + {"edge": "10", "node": "20"}, + {"edge": "10", "node": "30"}, + ], + } + ) + + assert H.get_edges() == [("20", "30")] + assert not H.is_weighted() + + +def test_weighted_metadata_does_not_control_hypergraph_type(): + H = from_hif_dict( + { + "metadata": {"weighted": True}, + "incidences": [{"edge": 0, "node": 0}], + } + ) + + assert not H.is_weighted() + assert H.get_hypergraph_metadata()["weighted"] is True + + +def test_to_hif_dict_matches_write_hif_and_uses_standard_fields(tmp_path): + H = Hypergraph(edge_list=[("alice", "bob")], weighted=False) + H.set_hypergraph_metadata({"name": "example"}) + H.set_node_metadata("alice", {"color": "blue", "weight": 2}) + H.set_edge_metadata(("alice", "bob"), {"kind": "social"}) + H.set_incidence_metadata(("alice", "bob"), "alice", {"weight": 0.5}) + path = tmp_path / "network.json" + expected = { + "network-type": "undirected", + "metadata": {"name": "example"}, + "nodes": [ + {"node": "alice", "weight": 2, "attrs": {"color": "blue"}}, + {"node": "bob"}, + ], + "edges": [{"edge": 0, "attrs": {"kind": "social"}}], + "incidences": [ + {"edge": 0, "node": "alice", "weight": 0.5}, + {"edge": 0, "node": "bob"}, + ], + } + + write_hif(H, path) + + assert to_hif_dict(H) == expected + assert json.loads(path.read_text(encoding="utf-8")) == expected + + +def test_directed_hif_roundtrip(): + data = { + "network-type": "directed", + "edges": [{"edge": "reaction", "weight": 2}], + "incidences": [ + {"edge": "reaction", "node": "a", "direction": "tail"}, + {"edge": "reaction", "node": "b", "direction": "head"}, + ], + } + + H = from_hif_dict(data) + converted = to_hif_dict(H) + + assert isinstance(H, DirectedHypergraph) + assert H.get_edges() == [(("a",), ("b",))] + assert H.get_weight((("a",), ("b",))) == 2 + assert converted["incidences"] == [ + {"edge": 0, "node": "a", "direction": "tail"}, + {"edge": 0, "node": "b", "direction": "head"}, + ] + assert from_hif_dict(converted).get_edges() == H.get_edges() + + +def test_directed_incidence_requires_a_direction(): + with pytest.raises(ValueError, match="require direction"): + from_hif_dict( + { + "network-type": "directed", + "incidences": [{"edge": 0, "node": 0}], + } + ) + + +@pytest.mark.parametrize( + "data", + [ + None, + {"metadata": [], "incidences": []}, + {"nodes": [{"node": 0, "attrs": []}], "incidences": []}, + ], +) +def test_from_hif_dict_rejects_invalid_dictionary_fields(data: Any): + with pytest.raises(TypeError, match="dictionary"): + from_hif_dict(data) + + +def test_abstract_simplicial_complex_is_not_silently_converted(): + with pytest.raises(NotImplementedError, match="simplicial complexes"): + from_hif_dict({"network-type": "asc", "incidences": []}) + + +def test_parallel_hif_edges_fail_instead_of_being_merged(): + with pytest.raises(ValueError, match="Duplicate edge"): + from_hif_dict( + { + "incidences": [ + {"edge": "first", "node": 0}, + {"edge": "second", "node": 0}, + ] + } + ) + + +def test_empty_edge_ids_do_not_collide_with_generated_ids(): + H = Hypergraph(edge_list=[(0, 1)], weighted=False) + H.add_empty_edge(0, {}) + + data = to_hif_dict(H) + + assert {record["edge"] for record in data["edges"]} == {0, 1} + + +def test_hif_dict_roundtrip_does_not_alias_metadata(): + H = Hypergraph( + edge_list=[(0, 1)], + weighted=False, + hypergraph_metadata={"nested": {"name": "original"}}, + ) + H.set_node_metadata(0, {"nested": {"color": "blue"}}) + data = to_hif_dict(H) + + restored = from_hif_dict(data) + data["metadata"]["nested"]["name"] = "changed" + data["nodes"][0]["attrs"]["nested"]["color"] = "red" + + assert restored.get_hypergraph_metadata()["nested"]["name"] == "original" + assert restored.get_node_metadata(0)["nested"]["color"] == "blue" + assert H.get_hypergraph_metadata()["nested"]["name"] == "original" + assert H.get_node_metadata(0)["nested"]["color"] == "blue" + + +def test_write_hif_rejects_nan(tmp_path): + H = Hypergraph(weighted=False) + H.set_hypergraph_metadata({"nested": {"missing": float("nan")}}) + path = tmp_path / "network.json" + path.write_text("existing data", encoding="utf-8") + + with pytest.raises(ValueError, match="Out of range float values"): + write_hif(H, path) + + assert path.read_text(encoding="utf-8") == "existing data" From 59f6a84f67e4c5668e50db641e939a8cacf54837 Mon Sep 17 00:00:00 2001 From: viniciusdutra314 Date: Thu, 13 Aug 2026 18:08:29 -0300 Subject: [PATCH 2/2] feat: added from_hif_dict and to_hif_dict and unit tested --- hypergraphx/readwrite/__init__.py | 17 +- hypergraphx/readwrite/hif.py | 479 +++++++++++++++++------------- tests/readwrite/test_hif.py | 206 +++++++------ 3 files changed, 408 insertions(+), 294 deletions(-) diff --git a/hypergraphx/readwrite/__init__.py b/hypergraphx/readwrite/__init__.py index b2ff7dec..72dd015a 100644 --- a/hypergraphx/readwrite/__init__.py +++ b/hypergraphx/readwrite/__init__.py @@ -1,4 +1,15 @@ from . import load as load_module +from .hif import ( + HIFEdgeRecord, + HIFIncidenceRecord, + HIFJson, + HIFNodeRecord, + JSONValue, + from_hif_dict, + read_hif, + to_hif_dict, + write_hif, +) from .load import ( download_remote_dataset, download_remote_datasets, @@ -11,7 +22,6 @@ ) from .load import load as load_any from .save import save_hypergraph -from .hif import from_hif_dict, read_hif, to_hif_dict, write_hif __all__ = [ "load_module", @@ -25,6 +35,11 @@ "load_hypergraph_from_server", "search_remote_datasets", "save_hypergraph", + "HIFJson", + "HIFEdgeRecord", + "HIFIncidenceRecord", + "HIFNodeRecord", + "JSONValue", "from_hif_dict", "read_hif", "to_hif_dict", diff --git a/hypergraphx/readwrite/hif.py b/hypergraphx/readwrite/hif.py index 28fa771d..b963d647 100644 --- a/hypergraphx/readwrite/hif.py +++ b/hypergraphx/readwrite/hif.py @@ -1,10 +1,8 @@ import copy import json -import math import sys -from collections.abc import Hashable, Mapping -from os import PathLike -from typing import Any, Literal, TypeAlias, cast +from itertools import count +from typing import Literal, TypeAlias, cast from hypergraphx import DirectedHypergraph, Hypergraph @@ -14,10 +12,11 @@ from typing_extensions import NotRequired, TypedDict __all__ = [ - "HIFSchema", + "HIFJson", "HIFEdgeRecord", "HIFIncidenceRecord", "HIFNodeRecord", + "JSONValue", "from_hif_dict", "read_hif", "to_hif_dict", @@ -26,16 +25,17 @@ HIF_ID: TypeAlias = str | int Weight: TypeAlias = int | float -Metadata: TypeAlias = dict[str, Any] +JSONValue: TypeAlias = ( + None | bool | int | float | str | list["JSONValue"] | dict[str, "JSONValue"] +) +Metadata: TypeAlias = dict[str, JSONValue] NetworkType: TypeAlias = Literal["asc", "directed", "undirected"] Direction: TypeAlias = Literal["head", "tail"] -HypergraphType: TypeAlias = Hypergraph | DirectedHypergraph -UndirectedEdge: TypeAlias = tuple[Hashable, ...] -DirectedEdge: TypeAlias = tuple[tuple[Hashable, ...], tuple[Hashable, ...]] +UndirectedEdge: TypeAlias = tuple[HIF_ID, ...] +DirectedEdge: TypeAlias = tuple[tuple[HIF_ID, ...], tuple[HIF_ID, ...]] EdgeKey: TypeAlias = UndirectedEdge | DirectedEdge -UndirectedMembers: TypeAlias = list[Hashable] -DirectedMembers: TypeAlias = tuple[list[Hashable], list[Hashable]] -EdgeMembers: TypeAlias = UndirectedMembers | DirectedMembers +UndirectedEdgeMembers: TypeAlias = list[HIF_ID] +DirectedEdgeMembers: TypeAlias = tuple[list[HIF_ID], list[HIF_ID]] class HIFNodeRecord(TypedDict): @@ -60,8 +60,8 @@ class HIFIncidenceRecord(TypedDict): # network-type is not a valid python identifier, so we need to create # the typedict manually. -HIFSchema = TypedDict( - "HIFSchema", +HIFJson = TypedDict( + "HIFJson", { "network-type": NotRequired[NetworkType], "metadata": NotRequired[Metadata], @@ -72,156 +72,215 @@ class HIFIncidenceRecord(TypedDict): ) -def _record_metadata( - record: Mapping[str, Any], include_weight: bool = False +def _get_record_metadata( + record: HIFNodeRecord | HIFEdgeRecord | HIFIncidenceRecord, + include_weight: bool = False, ) -> Metadata: - """Copy the non-structural attributes from a HIF record.""" + """Copy the attributes from a HIF record.""" attrs = copy.deepcopy(record.get("attrs", {})) if not isinstance(attrs, dict): raise TypeError("HIF record 'attrs' must be a dictionary.") + if "weight" in record and type(record["weight"]) not in (int, float): + raise TypeError("HIF weights must be integers or floats.") if include_weight and "weight" in record: attrs["weight"] = record["weight"] return attrs -def _add_metadata( +def _add_metadata_in_record( record: HIFNodeRecord | HIFEdgeRecord | HIFIncidenceRecord, metadata: Metadata, include_weight: bool = False, ) -> None: attrs = copy.deepcopy(metadata) if include_weight and "weight" in attrs: - record["weight"] = attrs.pop("weight") - if len(attrs) > 0: + weight = attrs.pop("weight") + if type(weight) not in (int, float): + raise TypeError("HIF weights must be integers or floats.") + record["weight"] = cast(Weight, weight) + if attrs: record["attrs"] = attrs -def _replace_nan_with_none(value: Any) -> Any: - if isinstance(value, float) and math.isnan(value): - return None - if isinstance(value, dict): - return {key: _replace_nan_with_none(item) for key, item in value.items()} - if isinstance(value, list): - return [_replace_nan_with_none(item) for item in value] - if isinstance(value, tuple): - return tuple(_replace_nan_with_none(item) for item in value) - return value - - -def _empty_edges(H: HypergraphType) -> dict[HIF_ID, Metadata]: - """Copy empty-edge data exposed by HypergraphX's serialization API.""" - if not isinstance(H, Hypergraph): - return {} - return copy.deepcopy(H.expose_data_structures().get("empty_edges", {})) - - -def _edge_ids(H: HypergraphType) -> dict[EdgeKey, HIF_ID]: - """Return HIF edge IDs which do not collide with named empty edges.""" - used_ids: set[HIF_ID] = set(_empty_edges(H)) - edge_ids: dict[EdgeKey, HIF_ID] = {} - next_id = 0 - for edge, edge_id in H.get_edge_list().items(): - if edge_id in used_ids: - while next_id in used_ids: - next_id += 1 - edge_id = next_id - edge_ids[edge] = edge_id - used_ids.add(edge_id) - return edge_ids - - -def _edge_key(members: EdgeMembers, directed: bool) -> EdgeKey: - if directed: - source, target = cast(DirectedMembers, members) - return (tuple(source), tuple(target)) - return tuple(cast(UndirectedMembers, members)) - - -def from_hif_dict(data: HIFSchema) -> HypergraphType: - """Create a hypergraph from a dictionary following the HIF standard.""" +def _get_hif_edge_ids( + internal_edge_ids: dict[EdgeKey, int], reserved_ids: set[HIF_ID] +) -> dict[EdgeKey, HIF_ID]: + """Keep internal IDs when possible and replace IDs reserved by empty edges.""" + used_ids = reserved_ids.copy() + available_ids = (candidate for candidate in count() if candidate not in used_ids) + hif_edge_ids: dict[EdgeKey, HIF_ID] = {} + + for edge_key, internal_id in internal_edge_ids.items(): + hif_edge_id = ( + internal_id if internal_id not in used_ids else next(available_ids) + ) + hif_edge_ids[edge_key] = hif_edge_id + used_ids.add(hif_edge_id) + + return hif_edge_ids + + +def _get_hif_incidences( + h: Hypergraph | DirectedHypergraph, edge_key: EdgeKey +) -> list[tuple[HIF_ID, Direction | None]]: + if isinstance(h, DirectedHypergraph): + source, target = cast(DirectedEdge, edge_key) + incidences: list[tuple[HIF_ID, Direction | None]] = [ + (node, "tail") for node in source + ] + incidences.extend((node, "head") for node in target) + return incidences + + members = cast(UndirectedEdge, edge_key) + return [(node, None) for node in members] + + +def _get_edge_weight(h: Hypergraph | DirectedHypergraph, edge_key: EdgeKey) -> Weight: + if isinstance(h, DirectedHypergraph): + return h.get_weight(cast(DirectedEdge, edge_key)) + return h.get_weight(cast(UndirectedEdge, edge_key)) + + +def from_hif_dict(data: HIFJson) -> Hypergraph | DirectedHypergraph: + """ + Create a hypergraph from a dictionary following the HIF standard. + + Parameters + ---------- + data : HIFJson + A HIF dictionary containing an ``incidences`` list and, optionally, + network type, metadata, nodes, and edges. + + Returns + ------- + Hypergraph or DirectedHypergraph + The hypergraph represented by ``data``. If ``network-type`` is omitted, + an undirected hypergraph is returned. + + Raises + ------ + TypeError + If any field has an invalid type according to the HIF schema. + NotImplementedError + If ``network-type`` is ``"asc"``. + + Notes + ----- + HypergraphX does not support parallel edges. Parallel HIF edges are + rejected rather than merged. Input metadata is deeply copied. + """ if not isinstance(data, dict): raise TypeError("HIF data must be provided as a dictionary.") network_type: NetworkType = data.get("network-type", "undirected") - is_directed = network_type == "directed" + edge_records: list[HIFEdgeRecord] = data.get("edges", []) + is_weighted = any("weight" in record for record in edge_records) + match network_type: + case "undirected": + h = Hypergraph(weighted=is_weighted, duplicate_policy="error") + case "directed": + h = DirectedHypergraph(weighted=is_weighted, duplicate_policy="error") + case "asc": + raise NotImplementedError( + "HypergraphX does not support abstract simplicial complexes." + ) + case _: + raise ValueError(f"Unknown hypergraph type: {network_type}") + metadata = copy.deepcopy(data.get("metadata", {})) if not isinstance(metadata, dict): raise TypeError("HIF 'metadata' must be a dictionary.") - edge_records: list[HIFEdgeRecord] = data.get("edges", []) - weighted = any("weight" in record for record in edge_records) - - if network_type == "undirected": - H: HypergraphType = Hypergraph(weighted=weighted, duplicate_policy="error") - elif is_directed: - H = DirectedHypergraph(weighted=weighted, duplicate_policy="error") - elif network_type == "asc": - raise NotImplementedError( - "HypergraphX does not support abstract simplicial complexes." - ) - else: - raise ValueError(f"Unknown hypergraph type: {network_type}") + h.set_hypergraph_metadata(metadata) - if "metadata" in data: - H.set_hypergraph_metadata(metadata) + for record in data.get("nodes", []): + node = record["node"] + h.add_node(node) + h.set_node_metadata(node, _get_record_metadata(record, include_weight=True)) - tmp_edges: dict[Hashable, EdgeMembers] = {} + undirected_members_by_hif_edge_id: dict[HIF_ID, UndirectedEdgeMembers] = {} + directed_members_by_hif_edge_id: dict[HIF_ID, DirectedEdgeMembers] = {} for incidence in data["incidences"]: - edge = incidence["edge"] + hif_edge_id = incidence["edge"] node = incidence["node"] - - if edge not in tmp_edges: - tmp_edges[edge] = ([], []) if is_directed else [] - if is_directed: - directed_members = cast(DirectedMembers, tmp_edges[edge]) - direction = incidence.get("direction") - if direction == "tail": - directed_members[0].append(node) - elif direction == "head": - directed_members[1].append(node) - else: + match h, incidence.get("direction"): + case DirectedHypergraph(), "tail": + tail, _ = directed_members_by_hif_edge_id.setdefault( + hif_edge_id, ([], []) + ) + tail.append(node) + case DirectedHypergraph(), "head": + _, head = directed_members_by_hif_edge_id.setdefault( + hif_edge_id, ([], []) + ) + head.append(node) + case DirectedHypergraph(), _: raise ValueError( "Directed HIF incidences require direction 'head' or 'tail'." ) - else: - cast(UndirectedMembers, tmp_edges[edge]).append(node) - - for record in data.get("nodes", []): - node = record["node"] - H.add_node(node) - H.set_node_metadata(node, _record_metadata(record, include_weight=True)) - - added: dict[Hashable, EdgeKey] = {} + case Hypergraph(), _: + undirected_members_by_hif_edge_id.setdefault(hif_edge_id, []).append( + node + ) + added_hif_edge_ids: set[HIF_ID] = set() for record in edge_records: - edge = record["edge"] - attrs = _record_metadata(record) - if edge in tmp_edges: - edge_key = _edge_key(tmp_edges[edge], is_directed) - H.add_edge(edge_key, weight=record.get("weight"), metadata=attrs) - added[edge] = edge_key - elif is_directed: - edge_key = ((), ()) - H.add_edge(edge_key, weight=record.get("weight"), metadata=attrs) - added[edge] = edge_key - else: - if "weight" in record: - attrs["weight"] = record["weight"] - H.add_empty_edge(edge, attrs) + hif_edge_id = record["edge"] + edge_metadata = _get_record_metadata(record) + edge_weight = record.get("weight") + match h: + case DirectedHypergraph(): + tail, head = directed_members_by_hif_edge_id.get(hif_edge_id, ([], [])) + directed_edge_key: DirectedEdge = (tuple(tail), tuple(head)) + h.add_edge( + directed_edge_key, + weight=edge_weight, + metadata=edge_metadata, + ) + case Hypergraph(): + members = undirected_members_by_hif_edge_id.get(hif_edge_id) + if members is None: + h.add_empty_edge( + hif_edge_id, + _get_record_metadata(record, include_weight=True), + ) + continue + undirected_edge_key: UndirectedEdge = tuple(members) + h.add_edge( + undirected_edge_key, + weight=edge_weight, + metadata=edge_metadata, + ) + added_hif_edge_ids.add(hif_edge_id) for incidence in data["incidences"]: - edge = incidence["edge"] + hif_edge_id = incidence["edge"] node = incidence["node"] - if edge not in added: - edge_key = _edge_key(tmp_edges[edge], is_directed) - H.add_edge(edge_key) - added[edge] = edge_key - attrs = _record_metadata(incidence, include_weight=True) - if attrs: - H.set_incidence_metadata(added[edge], node, attrs) - - return H + incidence_metadata = _get_record_metadata(incidence, include_weight=True) + match h: + case DirectedHypergraph(): + tail, head = directed_members_by_hif_edge_id[hif_edge_id] + directed_edge_key = (tuple(tail), tuple(head)) + if hif_edge_id not in added_hif_edge_ids: + h.add_edge(directed_edge_key) + added_hif_edge_ids.add(hif_edge_id) + if incidence_metadata: + h.set_incidence_metadata( + directed_edge_key, node, incidence_metadata + ) + case Hypergraph(): + undirected_edge_key = tuple( + undirected_members_by_hif_edge_id[hif_edge_id] + ) + if hif_edge_id not in added_hif_edge_ids: + h.add_edge(undirected_edge_key) + added_hif_edge_ids.add(hif_edge_id) + if incidence_metadata: + h.set_incidence_metadata( + undirected_edge_key, node, incidence_metadata + ) + return h -def read_hif(path: str | PathLike[str]) -> HypergraphType: +def read_hif(path: str) -> Hypergraph | DirectedHypergraph: """ Load a hypergraph from a HIF file. @@ -235,31 +294,51 @@ def read_hif(path: str | PathLike[str]) -> HypergraphType: Hypergraph The loaded hypergraph """ - edge_name_to_uid = {} - node_name_to_uid = {} - eid = 0 - nid = 0 - - with open(path) as file: - data: HIFSchema = json.loads(file.read()) - return from_hif_dict(data, nodetype=nodetype, edgetype=edgetype) - - if "type" not in data: - logging.getLogger(__name__).warning("No hypergraph type - assume undirected") - data["type"] = "undirected" - -def to_hif_dict(H: HypergraphType, convert_nans: bool = False) -> HIFData: - """Create a dictionary following the HIF standard from a hypergraph.""" - if isinstance(H, DirectedHypergraph): - network_type: NetworkType = "directed" - elif isinstance(H, Hypergraph): - network_type = "undirected" - else: - raise TypeError( - "HIF conversion supports Hypergraph and DirectedHypergraph objects." - ) + with open(path, encoding="utf-8") as file: + data: HIFJson = json.load(file) + return from_hif_dict(data) - data: HIFData = { + +def to_hif_dict(H: Hypergraph | DirectedHypergraph) -> HIFJson: + """ + Create a dictionary following the HIF standard from a hypergraph. + + Parameters + ---------- + H : Hypergraph or DirectedHypergraph + The hypergraph to convert. + + Returns + ------- + HIFJson + A HIF dictionary containing the network type, metadata, nodes, edges, + and incidences. + + Raises + ------ + TypeError + If ``H`` is not a supported hypergraph type or a metadata weight is not + an integer or float. + + Notes + ----- + Edge identifiers are generated for ordinary edges. Named empty edges in an + undirected hypergraph retain their identifiers. All metadata is deeply + copied into the returned dictionary. + """ + match H: + case DirectedHypergraph(): + network_type: NetworkType = "directed" + empty_edges: dict[HIF_ID, Metadata] = {} + case Hypergraph(): + network_type = "undirected" + empty_edges = H.expose_data_structures().get("empty_edges", {}) + case _: + raise TypeError( + "HIF conversion supports Hypergraph and DirectedHypergraph objects." + ) + + data: HIFJson = { "network-type": network_type, "metadata": copy.deepcopy(H.get_hypergraph_metadata()), "edges": [], @@ -267,74 +346,54 @@ def to_hif_dict(H: HypergraphType, convert_nans: bool = False) -> HIFData: "incidences": [], } - for node, attrs in H.get_all_nodes_metadata().items(): + for node, metadata in H.get_all_nodes_metadata().items(): node_record: HIFNodeRecord = {"node": node} - _add_metadata(node_record, attrs, include_weight=True) + _add_metadata_in_record(node_record, metadata, include_weight=True) data["nodes"].append(node_record) - if incidence["node"] not in node_name_to_uid: - node_name_to_uid[incidence["node"]] = nid - nid += 1 - node = node_name_to_uid[incidence["node"]] - - if edge not in tmp_edges: - tmp_edges[edge] = [] - tmp_edges[edge].append(node) - - for record in data["nodes"]: - node_name = record["node"] - if node_name not in node_name_to_uid: - node_name_to_uid[node_name] = nid - nid += 1 - node = node_name_to_uid[node_name] - H.add_node(node) - H.set_node_metadata(node, record) - - added = {} - - for record in data["edges"]: - edge_name = record["edge"] - if edge_name not in edge_name_to_uid: - edge_name_to_uid[edge_name] = eid - eid += 1 - edge = edge_name_to_uid[edge_name] - if edge in tmp_edges: - H.add_edge(tuple(sorted(tmp_edges[edge]))) - added[tuple(sorted(tmp_edges[edge]))] = True - H.set_edge_metadata(tuple(sorted(tmp_edges[edge])), record) - else: - H.add_empty_edge(edge_name, record) - - for incidence in data["incidences"]: - edge = edge_name_to_uid[incidence["edge"]] - node = node_name_to_uid[incidence["node"]] - if tuple(sorted(tmp_edges[edge])) not in added: - H.add_edge(tuple(sorted(tmp_edges[edge]))) - added[tuple(sorted(tmp_edges[edge]))] = True - H.set_incidence_metadata(tuple(sorted(tmp_edges[edge])), node, incidence) - - return H - - -def write_hif(H: HypergraphType, path: str) -> None: + hif_edge_ids = _get_hif_edge_ids(H.get_edge_list(), reserved_ids=set(empty_edges)) + incidence_metadata = H.get_all_incidences_metadata() + for edge_key, hif_edge_id in hif_edge_ids.items(): + edge_record: HIFEdgeRecord = {"edge": hif_edge_id} + _add_metadata_in_record( + edge_record, H.get_edge_metadata(edge_key), include_weight=True + ) + if H.is_weighted(): + edge_record["weight"] = _get_edge_weight(H, edge_key) + data["edges"].append(edge_record) + + for node, direction in _get_hif_incidences(H, edge_key): + incidence_record: HIFIncidenceRecord = { + "edge": hif_edge_id, + "node": node, + } + if direction is not None: + incidence_record["direction"] = direction + _add_metadata_in_record( + incidence_record, + incidence_metadata.get((edge_key, node), {}), + include_weight=True, + ) + data["incidences"].append(incidence_record) + + for hif_edge_id, metadata in empty_edges.items(): + edge_record: HIFEdgeRecord = {"edge": hif_edge_id} + _add_metadata_in_record(edge_record, metadata, include_weight=True) + data["edges"].append(edge_record) + return data + + +def write_hif(H: Hypergraph | DirectedHypergraph, path: str) -> None: """ Save a hypergraph to a HIF file. Parameters ---------- - H: Hypergraph + H : Hypergraph or DirectedHypergraph The hypergraph to save. - path: str + path : str The path to save the hypergraph to. """ - - data = { - "type": "undirected", - "metadata": H.get_hypergraph_metadata(), - "edges": H.get_all_edges_metadata(), - "nodes": H.get_all_nodes_metadata(), - "incidences": H.get_all_incidences_metadata(), - } - - with open(path, "w") as file: - file.write(json.dumps(to_hif_dict(H, convert_nans=True))) + serialized_data = json.dumps(to_hif_dict(H), allow_nan=False) + with open(path, "w", encoding="utf-8") as file: + file.write(serialized_data) diff --git a/tests/readwrite/test_hif.py b/tests/readwrite/test_hif.py index d29cc146..256b7dfe 100644 --- a/tests/readwrite/test_hif.py +++ b/tests/readwrite/test_hif.py @@ -1,43 +1,71 @@ +import copy import json +from pathlib import Path +from tempfile import TemporaryDirectory from typing import Any import pytest +from jsonschema import Draft7Validator from hypergraphx import DirectedHypergraph, Hypergraph from hypergraphx.readwrite import ( + HIFJson, from_hif_dict, read_hif, to_hif_dict, write_hif, ) +HIF_SCHEMA_PATH = Path(__file__).with_name("hif_schema_v0.1.0.json") +HIF_VALIDATOR = Draft7Validator(json.loads(HIF_SCHEMA_PATH.read_text(encoding="utf-8"))) -def test_from_hif_dict_matches_read_hif(tmp_path): - data = { + +def assert_valid_hif(data: HIFJson) -> None: + HIF_VALIDATOR.validate(data) + + +def assert_round_trip( + data: HIFJson, *, check_identity: bool = True +) -> Hypergraph | DirectedHypergraph: + assert_valid_hif(data) + H = from_hif_dict(data) + converted = to_hif_dict(H) + assert_valid_hif(converted) + if check_identity: + assert converted == data + with TemporaryDirectory() as directory: + path = Path(directory) / "network.json" + write_hif(H, path) + assert to_hif_dict(read_hif(path)) == converted + return H + + +def test_undirected_hif_roundtrip(): + data: HIFJson = { "network-type": "undirected", "metadata": {"name": "in-memory"}, - "nodes": [{"node": "isolated", "weight": 3, "attrs": {"color": "red"}}], + "nodes": [ + {"node": "isolated", "weight": 3, "attrs": {"color": "red"}}, + {"node": "alice"}, + {"node": "bob"}, + ], "edges": [ - {"edge": "friendship", "weight": 2, "attrs": {"kind": "social"}}, + {"edge": 0, "weight": 2, "attrs": {"kind": "social"}}, {"edge": "empty", "weight": 4}, ], "incidences": [ { - "edge": "friendship", + "edge": 0, "node": "alice", "weight": 0.5, "attrs": {"since": 2020}, }, - {"edge": "friendship", "node": "bob"}, + {"edge": 0, "node": "bob"}, ], } - path = tmp_path / "network.json" - path.write_text(json.dumps(data), encoding="utf-8") + from_dict = assert_round_trip(data) - from_dict = from_hif_dict(data) - from_file = read_hif(path) - - assert to_hif_dict(from_file) == to_hif_dict(from_dict) + assert isinstance(from_dict, Hypergraph) assert from_dict.get_edges() == [("alice", "bob")] assert from_dict.get_hypergraph_metadata() == {"name": "in-memory"} assert from_dict.get_weight(("alice", "bob")) == 2 @@ -51,43 +79,54 @@ def test_from_hif_dict_matches_read_hif(tmp_path): "weight": 0.5, } assert (("alice", "bob"), "bob") not in from_dict.get_all_incidences_metadata() - assert {"edge": "empty", "weight": 4} in to_hif_dict(from_dict)["edges"] + assert from_dict.expose_data_structures()["empty_edges"] == {"empty": {"weight": 4}} -def test_only_incidences_are_required_and_ids_are_preserved(): - H = from_hif_dict( - { - "incidences": [ - {"edge": "10", "node": "20"}, - {"edge": "10", "node": "30"}, - ], - } - ) +def test_only_incidences_are_required_and_node_ids_are_preserved(): + data: HIFJson = { + "incidences": [ + {"edge": "10", "node": "20"}, + {"edge": "10", "node": "30"}, + ], + } + H = assert_round_trip(data, check_identity=False) # network-type will be added assert H.get_edges() == [("20", "30")] assert not H.is_weighted() +@pytest.mark.parametrize( + ("data", "expected_type"), + [ + ({"incidences": []}, Hypergraph), + ({"network-type": "directed", "incidences": []}, DirectedHypergraph), + ], +) +def test_empty_incidence_list_creates_empty_hypergraph(data, expected_type): + H = assert_round_trip(data, check_identity=False) + assert isinstance(H, expected_type) + assert H.num_nodes() == 0 + assert H.num_edges() == 0 + + def test_weighted_metadata_does_not_control_hypergraph_type(): - H = from_hif_dict( - { - "metadata": {"weighted": True}, - "incidences": [{"edge": 0, "node": 0}], - } - ) + data: HIFJson = { + "metadata": {"weighted": True}, + "incidences": [{"edge": 0, "node": 0}], + } + H = assert_round_trip(data, check_identity=False) assert not H.is_weighted() assert H.get_hypergraph_metadata()["weighted"] is True -def test_to_hif_dict_matches_write_hif_and_uses_standard_fields(tmp_path): +def test_to_hif_dict_uses_standard_fields(): H = Hypergraph(edge_list=[("alice", "bob")], weighted=False) H.set_hypergraph_metadata({"name": "example"}) H.set_node_metadata("alice", {"color": "blue", "weight": 2}) H.set_edge_metadata(("alice", "bob"), {"kind": "social"}) H.set_incidence_metadata(("alice", "bob"), "alice", {"weight": 0.5}) - path = tmp_path / "network.json" - expected = { + expected: HIFJson = { "network-type": "undirected", "metadata": {"name": "example"}, "nodes": [ @@ -101,43 +140,35 @@ def test_to_hif_dict_matches_write_hif_and_uses_standard_fields(tmp_path): ], } - write_hif(H, path) - assert to_hif_dict(H) == expected - assert json.loads(path.read_text(encoding="utf-8")) == expected + assert_round_trip(expected) def test_directed_hif_roundtrip(): - data = { + data: HIFJson = { "network-type": "directed", - "edges": [{"edge": "reaction", "weight": 2}], + "metadata": {}, + "nodes": [{"node": "a"}, {"node": "b"}], + "edges": [{"edge": 0, "weight": 2}], "incidences": [ - {"edge": "reaction", "node": "a", "direction": "tail"}, - {"edge": "reaction", "node": "b", "direction": "head"}, + {"edge": 0, "node": "a", "direction": "tail"}, + {"edge": 0, "node": "b", "direction": "head"}, ], } - - H = from_hif_dict(data) - converted = to_hif_dict(H) + H = assert_round_trip(data) assert isinstance(H, DirectedHypergraph) assert H.get_edges() == [(("a",), ("b",))] assert H.get_weight((("a",), ("b",))) == 2 - assert converted["incidences"] == [ - {"edge": 0, "node": "a", "direction": "tail"}, - {"edge": 0, "node": "b", "direction": "head"}, - ] - assert from_hif_dict(converted).get_edges() == H.get_edges() def test_directed_incidence_requires_a_direction(): + data: HIFJson = { + "network-type": "directed", + "incidences": [{"edge": 0, "node": 0}], + } with pytest.raises(ValueError, match="require direction"): - from_hif_dict( - { - "network-type": "directed", - "incidences": [{"edge": 0, "node": 0}], - } - ) + from_hif_dict(data) @pytest.mark.parametrize( @@ -153,58 +184,67 @@ def test_from_hif_dict_rejects_invalid_dictionary_fields(data: Any): from_hif_dict(data) +@pytest.mark.parametrize( + "data", + [ + {"nodes": [{"node": 0, "weight": "heavy"}], "incidences": []}, + {"edges": [{"edge": 0, "weight": "heavy"}], "incidences": []}, + {"incidences": [{"edge": 0, "node": 0, "weight": "heavy"}]}, + {"nodes": [{"node": 0, "weight": True}], "incidences": []}, + ], +) +def test_from_hif_dict_rejects_non_numeric_weights(data: Any): + with pytest.raises(TypeError, match="integers or floats"): + from_hif_dict(data) + + def test_abstract_simplicial_complex_is_not_silently_converted(): + data: HIFJson = {"network-type": "asc", "incidences": []} + with pytest.raises(NotImplementedError, match="simplicial complexes"): - from_hif_dict({"network-type": "asc", "incidences": []}) + from_hif_dict(data) def test_parallel_hif_edges_fail_instead_of_being_merged(): + data: HIFJson = { + "incidences": [ + {"edge": "first", "node": 0}, + {"edge": "second", "node": 0}, + ] + } + with pytest.raises(ValueError, match="Duplicate edge"): - from_hif_dict( - { - "incidences": [ - {"edge": "first", "node": 0}, - {"edge": "second", "node": 0}, - ] - } - ) + from_hif_dict(data) def test_empty_edge_ids_do_not_collide_with_generated_ids(): H = Hypergraph(edge_list=[(0, 1)], weighted=False) H.add_empty_edge(0, {}) - data = to_hif_dict(H) - - assert {record["edge"] for record in data["edges"]} == {0, 1} + restored = assert_round_trip(data) + assert isinstance(restored, Hypergraph) + assert restored.get_edges() == [(0, 1)] + assert {record["edge"] for record in data.get("edges", [])} == {0, 1} -def test_hif_dict_roundtrip_does_not_alias_metadata(): +def test_hif_conversions_deep_copy_all_metadata(): H = Hypergraph( edge_list=[(0, 1)], weighted=False, hypergraph_metadata={"nested": {"name": "original"}}, ) H.set_node_metadata(0, {"nested": {"color": "blue"}}) + H.set_edge_metadata((0, 1), {"nested": {"kind": "ordinary"}}) + H.set_incidence_metadata((0, 1), 0, {"nested": {"role": "member"}}) + H.add_empty_edge("empty", {"nested": {"kind": "empty"}}) data = to_hif_dict(H) - + expected = copy.deepcopy(data) restored = from_hif_dict(data) data["metadata"]["nested"]["name"] = "changed" - data["nodes"][0]["attrs"]["nested"]["color"] = "red" - - assert restored.get_hypergraph_metadata()["nested"]["name"] == "original" - assert restored.get_node_metadata(0)["nested"]["color"] == "blue" - assert H.get_hypergraph_metadata()["nested"]["name"] == "original" - assert H.get_node_metadata(0)["nested"]["color"] == "blue" - - -def test_write_hif_rejects_nan(tmp_path): - H = Hypergraph(weighted=False) - H.set_hypergraph_metadata({"nested": {"missing": float("nan")}}) - path = tmp_path / "network.json" - path.write_text("existing data", encoding="utf-8") + for records in (data["nodes"], data["edges"], data["incidences"]): + for record in records: + if "attrs" in record: + record["attrs"]["nested"].clear() - with pytest.raises(ValueError, match="Out of range float values"): - write_hif(H, path) - - assert path.read_text(encoding="utf-8") == "existing data" + assert to_hif_dict(H) == expected + assert to_hif_dict(restored) == expected