diff --git a/hypergraphx/readwrite/__init__.py b/hypergraphx/readwrite/__init__.py index 63dfcb4..72dd015 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,8 +22,6 @@ ) from .load import load as load_any from .save import save_hypergraph -from .hif import read_hif -from .hif import write_hif __all__ = [ "load_module", @@ -26,6 +35,13 @@ "load_hypergraph_from_server", "search_remote_datasets", "save_hypergraph", + "HIFJson", + "HIFEdgeRecord", + "HIFIncidenceRecord", + "HIFNodeRecord", + "JSONValue", + "from_hif_dict", "read_hif", + "to_hif_dict", "write_hif", ] diff --git a/hypergraphx/readwrite/hif.py b/hypergraphx/readwrite/hif.py index 7c5b2d1..b963d64 100644 --- a/hypergraphx/readwrite/hif.py +++ b/hypergraphx/readwrite/hif.py @@ -1,10 +1,286 @@ +import copy import json -import logging +import sys +from itertools import count +from typing import 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__ = [ + "HIFJson", + "HIFEdgeRecord", + "HIFIncidenceRecord", + "HIFNodeRecord", + "JSONValue", + "from_hif_dict", + "read_hif", + "to_hif_dict", + "write_hif", +] + +HIF_ID: TypeAlias = str | int +Weight: TypeAlias = int | float +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"] +UndirectedEdge: TypeAlias = tuple[HIF_ID, ...] +DirectedEdge: TypeAlias = tuple[tuple[HIF_ID, ...], tuple[HIF_ID, ...]] +EdgeKey: TypeAlias = UndirectedEdge | DirectedEdge +UndirectedEdgeMembers: TypeAlias = list[HIF_ID] +DirectedEdgeMembers: TypeAlias = tuple[list[HIF_ID], list[HIF_ID]] + + +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. +HIFJson = TypedDict( + "HIFJson", + { + "network-type": NotRequired[NetworkType], + "metadata": NotRequired[Metadata], + "incidences": list[HIFIncidenceRecord], + "nodes": NotRequired[list[HIFNodeRecord]], + "edges": NotRequired[list[HIFEdgeRecord]], + }, +) + + +def _get_record_metadata( + record: HIFNodeRecord | HIFEdgeRecord | HIFIncidenceRecord, + include_weight: bool = False, +) -> Metadata: + """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_in_record( + record: HIFNodeRecord | HIFEdgeRecord | HIFIncidenceRecord, + metadata: Metadata, + include_weight: bool = False, +) -> None: + attrs = copy.deepcopy(metadata) + if include_weight and "weight" in attrs: + 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 _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") + 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.") + 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)) + + 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"]: + hif_edge_id = incidence["edge"] + node = incidence["node"] + 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'." + ) + 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: + 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"]: + hif_edge_id = incidence["edge"] + node = incidence["node"] + 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) -> Hypergraph | DirectedHypergraph: """ Load a hypergraph from a HIF file. @@ -18,98 +294,106 @@ def read_hif(path: str) -> Hypergraph: Hypergraph The loaded hypergraph """ - edge_name_to_uid = {} - node_name_to_uid = {} - eid = 0 - nid = 0 + with open(path, encoding="utf-8") as file: + data: HIFJson = json.load(file) + return from_hif_dict(data) + + +def to_hif_dict(H: Hypergraph | DirectedHypergraph) -> HIFJson: + """ + Create a dictionary following the HIF standard from a hypergraph. - with open(path) as file: - data = json.loads(file.read()) + Parameters + ---------- + H : Hypergraph or DirectedHypergraph + The hypergraph to convert. - if "type" not in data: - logging.getLogger(__name__).warning("No hypergraph type - assume undirected") - data["type"] = "undirected" + Returns + ------- + HIFJson + A HIF dictionary containing the network type, metadata, nodes, edges, + and incidences. - if data["type"] == "undirected" or data["type"] == "asc": - H = Hypergraph() - elif data["type"] == "directed": - H = Hypergraph(directed=True) - else: - raise ValueError(f"Unknown hypergraph type: {data['type']}") + Raises + ------ + TypeError + If ``H`` is not a supported hypergraph type or a metadata weight is not + an integer or float. - if "metadata" in data: - H.set_hypergraph_metadata(data["metadata"]) + 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." + ) - 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"]] - - 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) + data: HIFJson = { + "network-type": network_type, + "metadata": copy.deepcopy(H.get_hypergraph_metadata()), + "edges": [], + "nodes": [], + "incidences": [], + } - 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) + for node, metadata in H.get_all_nodes_metadata().items(): + node_record: HIFNodeRecord = {"node": node} + _add_metadata_in_record(node_record, metadata, include_weight=True) + data["nodes"].append(node_record) + + 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) - return H + 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, path: str): + +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(data)) + 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/pyproject.toml b/pyproject.toml index 48ea102..cc4606e 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 0000000..1c5e10f --- /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 0000000..256b7df --- /dev/null +++ b/tests/readwrite/test_hif.py @@ -0,0 +1,250 @@ +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 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"}}, + {"node": "alice"}, + {"node": "bob"}, + ], + "edges": [ + {"edge": 0, "weight": 2, "attrs": {"kind": "social"}}, + {"edge": "empty", "weight": 4}, + ], + "incidences": [ + { + "edge": 0, + "node": "alice", + "weight": 0.5, + "attrs": {"since": 2020}, + }, + {"edge": 0, "node": "bob"}, + ], + } + from_dict = assert_round_trip(data) + + 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 + 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 from_dict.expose_data_structures()["empty_edges"] == {"empty": {"weight": 4}} + + +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(): + 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_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}) + expected: HIFJson = { + "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"}, + ], + } + + assert to_hif_dict(H) == expected + assert_round_trip(expected) + + +def test_directed_hif_roundtrip(): + data: HIFJson = { + "network-type": "directed", + "metadata": {}, + "nodes": [{"node": "a"}, {"node": "b"}], + "edges": [{"edge": 0, "weight": 2}], + "incidences": [ + {"edge": 0, "node": "a", "direction": "tail"}, + {"edge": 0, "node": "b", "direction": "head"}, + ], + } + H = assert_round_trip(data) + + assert isinstance(H, DirectedHypergraph) + assert H.get_edges() == [(("a",), ("b",))] + assert H.get_weight((("a",), ("b",))) == 2 + + +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(data) + + +@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) + + +@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(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(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) + 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_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" + for records in (data["nodes"], data["edges"], data["incidences"]): + for record in records: + if "attrs" in record: + record["attrs"]["nested"].clear() + + assert to_hif_dict(H) == expected + assert to_hif_dict(restored) == expected