Make dcat properties IRIs

pyflowbatt/battinfo.py

@@ -17,6 +17,15 @@
 
 logger = logging.getLogger(__name__)
 
+BDF_SCHEMA_IRI = "https://w3id.org/battery-data-alliance/ontology/battery-data-format/schema"
+# DCAT expects dcat:mediaType values from the IANA register, as IRIs in this form
+IANA_MEDIA_TYPES = "http://www.iana.org/assignments/media-types/"
+
+
+def _media_type_node(rel_file_path: str) -> dict[str, str]:
+    """IANA media type IRI of a file, as a JSON-LD node reference."""
+    return {"@id": IANA_MEDIA_TYPES + media_type(rel_file_path)}
+
 
 blank_flow_cell = {
     "@context": [
@@ -555,13 +564,13 @@ def add_input_data(
     dist: dict = {
         "@id": quote(rel_file_path),
         "@type": "dcat:Distribution",
-        "dcat:mediaType": media_type(rel_file_path),
+        "dcat:mediaType": _media_type_node(rel_file_path),
     }
     download_url = zenodo_download_url(
         rel_file_path, zenodo_doi_url, package=package, zip_filename=zip_filename
     )
     if download_url:
-        dist["dcat:downloadURL"] = download_url
+        dist["dcat:downloadURL"] = {"@id": download_url}
     if comment:
         dist["rdfs:comment"] = comment
 
@@ -599,7 +608,7 @@ def add_data(
     extras = extras or {}
     if rel_file_path.endswith(".parquet"):
         additions: dict[str, str | list | dict] = {
-            "csvw:tableSchema": "https://w3id.org/battery-data-alliance/ontology/battery-data-format/schema",
+            "csvw:tableSchema": {"@id": BDF_SCHEMA_IRI},
             "rdfs:comment": "Time series electrochemical data using Battery Data Format (bdf) columns",
         }
         if Path(rel_file_path).name.startswith("eis_"):
@@ -608,7 +617,7 @@ def add_data(
             )
     elif rel_file_path.endswith(".csv"):
         additions = {
-            "csvw:tableSchema": "https://w3id.org/battery-data-alliance/ontology/battery-data-format/schema",
+            "csvw:tableSchema": {"@id": BDF_SCHEMA_IRI},
             "csvw:dialect": {"@type": "csvw:Dialect", "csvw:delimiter": ",", "csvw:skipRows": 0},
             "rdfs:comment": "Time series electrochemical data using Battery Data Format (bdf) columns",
         }
@@ -636,15 +645,15 @@ def add_data(
     dist: dict = {
         "@id": quote(rel_file_path),
         "@type": "dcat:Distribution",
-        "dcat:mediaType": media_type(rel_file_path),
+        "dcat:mediaType": _media_type_node(rel_file_path),
         **additions,
         **extras,
     }
     download_url = zenodo_download_url(
         rel_file_path, zenodo_doi_url, package=package, zip_filename=zip_filename
     )
     if download_url:
-        dist["dcat:downloadURL"] = download_url
+        dist["dcat:downloadURL"] = {"@id": download_url}
 
     return {
         "@type": "BatteryTest",
@@ -677,8 +686,8 @@ def add_zenodo_url(
         "@context": [{"@base": base}],
         "@type": "BatteryTest",
         "hasOutput": {
-            "dcat:accessURL": zenodo_doi_url,
-            "dcat:endpointURL": f"https://zenodo.org/api/records/{record_id}",
+            "dcat:accessURL": {"@id": zenodo_doi_url},
+            "dcat:endpointURL": {"@id": f"https://zenodo.org/api/records/{record_id}"},
         },
     }
 

pyproject.toml

@@ -26,6 +26,7 @@ dev = [
     "pytest>=8.4.2",
     "ruff>=0.13.2",
     "pre-commit>=4.2.0",
+    "rdflib>=7.6.0",
 ]
 
 [project.scripts]

tests/test_config.py

@@ -556,12 +556,15 @@ def test_media_types_agree_between_crate_and_battinfo(
     }
     metadata_path = sample / "metadata.empa__fcid000001.json"
     metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
-    battinfo_types = {
-        d["@id"]: d["dcat:mediaType"]
+    iana = "http://www.iana.org/assignments/media-types/"
+    battinfo_iris = [
+        (d["@id"], d["dcat:mediaType"]["@id"])
         for section in ("hasOutput", "hasInput")
         for d in metadata[section]["dcat:distribution"]
         if d.get("dcat:mediaType")
-    }
+    ]
+    assert all(iri.startswith(iana) for _, iri in battinfo_iris)
+    battinfo_types = {file_id: iri.removeprefix(iana) for file_id, iri in battinfo_iris}
 
     assert crate_types["sample/sample_02_PEIS_A.mpr"] == "application/octet-stream"
     assert crate_types["sample/sample_protocol.mps"] == "text/plain"
@@ -733,12 +736,15 @@ def _fake_convert(_file: Path) -> dict:
 
 
 def _all_download_urls(obj: object) -> set[str]:
-    """Recursively collect every "dcat:downloadURL" string value."""
+    """Recursively collect every "dcat:downloadURL" IRI."""
     found: set[str] = set()
     if isinstance(obj, dict):
         url_value = obj.get("dcat:downloadURL")
-        if isinstance(url_value, str):
-            found.add(url_value)
+        if isinstance(url_value, dict):
+            found.add(url_value["@id"])
+        elif isinstance(url_value, str):
+            msg = "dcat:downloadURL is plain string - it should be an object with @id"
+            pytest.fail(msg)
         for v in obj.values():
             found |= _all_download_urls(v)
     elif isinstance(obj, list):
@@ -1774,6 +1780,65 @@ def test_battinfo_eis_data_comment(rel_path: str, frequency_domain: bool) -> Non
     assert comment.startswith("Frequency-domain" if frequency_domain else "Time series")
 
 
+DCAT_NS = "http://www.w3.org/ns/dcat#"
+BDF_SCHEMA = "https://w3id.org/battery-data-alliance/ontology/battery-data-format/schema"
+
+
+@pytest.mark.parametrize(
+    ("rel_path", "media_type", "table_schema"),
+    [
+        ("s/results/gcpl.bdf.csv", "text/csv", BDF_SCHEMA),
+        ("s/results/gcpl.bdf.parquet", "application/vnd.apache.parquet", BDF_SCHEMA),
+        ("s/s_02_GCPL_A.mpr", "application/octet-stream", None),
+    ],
+)
+def test_battinfo_distribution_values_are_iris(
+    rel_path: str, media_type: str, table_schema: str | None
+) -> None:
+    """mediaType, tableSchema, and downloadURL are IRIs."""
+    from rdflib import Graph, Namespace, URIRef
+
+    from pyflowbatt.battinfo import add_data
+
+    dcat = Namespace(DCAT_NS)
+    csvw = Namespace("http://www.w3.org/ns/csvw#")
+    dist = add_data(rel_path, "https://doi.org/10.5281/zenodo.12345678")
+    doc = {
+        "@context": {"@base": "https://example.org/", "dcat": str(dcat), "csvw": str(csvw)},
+        **dist["hasOutput"]["dcat:distribution"],
+    }
+    graph = Graph().parse(data=json.dumps(doc), format="json-ld")
+    file = URIRef(f"https://example.org/{rel_path}")
+
+    assert graph.value(file, dcat.mediaType) == URIRef(
+        f"http://www.iana.org/assignments/media-types/{media_type}"
+    )
+    expected_schema = URIRef(table_schema) if table_schema else None
+    assert graph.value(file, csvw.tableSchema) == expected_schema
+    download_url = graph.value(file, dcat.downloadURL)
+    assert isinstance(download_url, URIRef)
+    assert str(download_url).startswith("https://zenodo.org/")
+
+
+def test_battinfo_zenodo_access_urls_are_iris() -> None:
+    """Zenodo accessURL and endpointURL are IRIs."""
+    from rdflib import Graph, Namespace, URIRef
+
+    from pyflowbatt.battinfo import add_zenodo_url
+
+    dcat = Namespace(DCAT_NS)
+    output = add_zenodo_url("https://doi.org/10.5281/zenodo.12345678")["hasOutput"]
+    doc = {"@context": {"dcat": str(dcat)}, **output}
+    graph = Graph().parse(data=json.dumps(doc), format="json-ld")
+
+    assert list(graph.objects(None, dcat.accessURL)) == [
+        URIRef("https://doi.org/10.5281/zenodo.12345678")
+    ]
+    assert list(graph.objects(None, dcat.endpointURL)) == [
+        URIRef("https://zenodo.org/api/records/12345678")
+    ]
+
+
 def test_template_every_single_hash_line_uncomments_to_valid_toml(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None: