Extract asset anatomy from HERD ontology links

dandi/metadata/brain_areas.py

@@ -15,6 +15,9 @@
 
 MBAO_URI_TEMPLATE = "http://purl.obolibrary.org/obo/MBA_{}"
 
+#: Ontology prefixes (upper-cased) whose HERD entities are taken as anatomy
+ANATOMY_ONTOLOGY_PREFIXES = frozenset({"UBERON", "MBA"})
+
 # Values that should be treated as missing / uninformative
 _TRIVIAL_VALUES = frozenset(
     {
@@ -253,3 +256,57 @@ def locations_to_ccf_mouse_anatomy(locations: list[str]) -> list[models.Anatomy]
                     seen_ids.add(id_str)
                     results.append(anatomy)
     return results
+
+
+@lru_cache(maxsize=1)
+def _allen_structures_by_id() -> dict[str, dict[str, Any]]:
+    return {str(s["id"]): s for s in _load_allen_structures()}
+
+
+def _ontology_label(entity_id: str) -> str | None:
+    """Return a label for an ontology entity ID from bundled tables, if known."""
+    prefix, _, local_id = entity_id.partition(":")
+    if prefix.upper() == "MBA":
+        s = _allen_structures_by_id().get(local_id)
+        if s is not None:
+            return str(s["name"])
+    return None
+
+
+def herd_anatomy_to_models(entries: list[dict[str, Any]]) -> list[models.Anatomy]:
+    """Convert HERD anatomy entries to deduplicated ``Anatomy`` models.
+
+    HERD stores no labels, so the name comes from a bundled ontology table
+    when the term is in one, else from the longest value in the file linked to
+    the term (e.g. ``"trapezius muscle"`` over ``"EMG"``), else the entity ID.
+
+    Parameters
+    ----------
+    entries : list[dict]
+        As returned by :func:`dandi.pynwb_utils._get_herd_anatomy`.
+
+    Returns
+    -------
+    list[models.Anatomy]
+    """
+    results: list[models.Anatomy] = []
+    seen_ids: set[str] = set()
+    for entry in entries:
+        identifier = entry["identifier"]
+        if identifier in seen_ids:
+            continue
+        keys = [k.strip() for k in entry.get("keys") or [] if k and k.strip()]
+        name = (
+            _ontology_label(entry.get("entity_id") or "")
+            or max(keys, key=len, default="")
+            or entry.get("entity_id")
+            or identifier
+        )
+        try:
+            anatomy = models.Anatomy(identifier=identifier, name=name)
+        except Exception:
+            lgr.debug("Invalid HERD anatomy entry %r", entry, exc_info=True)
+            continue
+        seen_ids.add(identifier)
+        results.append(anatomy)
+    return results

dandi/metadata/util.py

@@ -14,7 +14,7 @@
 import tenacity
 from yarl import URL
 
-from .brain_areas import locations_to_ccf_mouse_anatomy
+from .brain_areas import herd_anatomy_to_models, locations_to_ccf_mouse_anatomy
 from .. import __version__
 from ..utils import ensure_datetime
 
@@ -774,7 +774,16 @@ def extract_wasDerivedFrom(metadata: dict) -> list[models.BioSample] | None:
 
 
 def _extract_brain_anatomy(metadata: dict) -> list[models.Anatomy]:
-    """Extract brain anatomy from metadata, if the species is mouse."""
+    """Extract anatomy from metadata.
+
+    Ontology terms linked in the file's HERD table take precedence, for any
+    species.  Otherwise, location strings are matched against the Allen CCF
+    if the species is mouse.
+    """
+    if herd_anatomy := metadata.get("herd_anatomy"):
+        if anatomy := herd_anatomy_to_models(herd_anatomy):
+            return anatomy
+
     if not (locations := metadata.get("brain_locations")):
         return []
 

dandi/pynwb_utils.py

@@ -285,6 +285,8 @@ def _get_pynwb_metadata(path: str | Path | Readable) -> dict[str, Any]:
 
         # Extract brain locations for anatomy metadata
         out["brain_locations"] = _get_brain_locations(nwb)
+        # Ontology-linked anatomy from the file's HERD table, if any
+        out["herd_anatomy"] = _get_herd_anatomy(nwb)
 
     return out
 
@@ -472,6 +474,84 @@ def _get_brain_locations(nwb: pynwb.NWBFile) -> list[str]:
     return locations
 
 
+def _get_herd_anatomy(nwb: pynwb.NWBFile) -> list[dict[str, Any]]:
+    """Extract anatomy ontology terms linked in the file's HERD table.
+
+    HERD (``/general/external_resources``) links values in the file to
+    ontology entities.  Every entity whose ID uses an anatomy ontology prefix
+    (see :data:`~dandi.metadata.brain_areas.ANATOMY_ONTOLOGY_PREFIXES`) is
+    collected, whatever object or field it is linked from, except entities
+    linked only from :class:`pynwb.file.Subject` objects: UBERON also holds
+    life-stage terms that may be linked from ``Subject.age``.
+
+    Parameters
+    ----------
+    nwb : pynwb.NWBFile
+        An open NWB file object.
+
+    Returns
+    -------
+    list[dict]
+        One dict per distinct entity, in order of first appearance, with keys
+        ``identifier`` (entity URI, or entity ID when the URI is empty),
+        ``entity_id`` and ``keys`` (the linked values in the file).
+    """
+    from .metadata.brain_areas import ANATOMY_ONTOLOGY_PREFIXES
+
+    herd = getattr(nwb, "external_resources", None)
+    if herd is None:
+        return []
+
+    def rows(table: Any) -> list[tuple]:
+        return [
+            tuple(
+                v.decode("utf-8", errors="replace") if isinstance(v, bytes) else v
+                for v in row
+            )
+            for row in table.data[:]
+        ]
+
+    try:
+        objects = rows(herd.objects)
+        keys = [r[0] for r in rows(herd.keys)]
+        entities = rows(herd.entities)
+        object_keys = rows(herd.object_keys)
+        entity_keys = rows(herd.entity_keys)
+    except Exception:
+        lgr.debug("Failed to read HERD table", exc_info=True)
+        return []
+
+    objects_by_key: dict[int, list[int]] = {}
+    for objects_idx, keys_idx in object_keys:
+        objects_by_key.setdefault(int(keys_idx), []).append(int(objects_idx))
+
+    def linked_from_subject(objects_idx: int) -> bool:
+        _, object_id, object_type = objects[objects_idx][:3]
+        obj = nwb.objects.get(object_id)
+        if obj is None:
+            return bool(object_type == "Subject")
+        return isinstance(obj, pynwb.file.Subject)
+
+    entries: dict[str, dict[str, Any]] = {}
+    for entities_idx, keys_idx in entity_keys:
+        entity_id, entity_uri = entities[int(entities_idx)][:2]
+        entity_id = str(entity_id or "").strip()
+        if entity_id.split(":", 1)[0].upper() not in ANATOMY_ONTOLOGY_PREFIXES:
+            continue
+        linked = objects_by_key.get(int(keys_idx), [])
+        if not linked or all(linked_from_subject(i) for i in linked):
+            continue
+        identifier = str(entity_uri or "").strip() or entity_id
+        entry = entries.setdefault(
+            identifier,
+            {"identifier": identifier, "entity_id": entity_id, "keys": []},
+        )
+        key = str(keys[int(keys_idx)])
+        if key not in entry["keys"]:
+            entry["keys"].append(key)
+    return list(entries.values())
+
+
 def _get_image_series(nwb: pynwb.NWBFile) -> list[dict]:
     """Retrieves all ImageSeries related metadata from an open nwb file.
 

dandi/tests/test_metadata.py

@@ -48,6 +48,7 @@
 from ..dandiapi import RemoteBlobAsset
 from ..files import NWBBIDSAsset, dandi_file, find_bids_dataset_description
 from ..metadata import nwb as metadata_nwb
+from ..metadata.brain_areas import herd_anatomy_to_models
 from ..metadata.core import prepare_metadata
 from ..metadata.nwb import get_metadata, nwb2asset
 from ..metadata.util import (
@@ -93,6 +94,7 @@ def test_get_metadata(simple1_nwb: Path, simple1_nwb_metadata: dict[str, Any]) -
     target_metadata["nd_types"] = []
     target_metadata["external_file_objects"] = []
     target_metadata["brain_locations"] = []
+    target_metadata["herd_anatomy"] = []
     # we do not populate any subject fields in our simple1_nwb
     for f in metadata_nwb_subject_fields:
         target_metadata[f] = None
@@ -1054,6 +1056,174 @@ def test_brain_anatomy_unmatched_tokens_only() -> None:
     assert result is None
 
 
+_skip_no_in_file_herd = pytest.mark.skipif(
+    not hasattr(NWBFile, "get_external_resources"),
+    reason="this pynwb cannot store HERD in the NWB file",
+)
+
+
+def _write_nwb_with_herd(
+    nwb_path: Path, species: str, group_location: str, electrode_locations: list[str]
+) -> dict[str, str]:
+    """Write an NWB file whose HERD links locations (and Subject fields) to UBERON.
+
+    Returns the mapping of electrode location value to the linked UBERON ID.
+    """
+    nwbfile = NWBFile(
+        session_description="test herd",
+        identifier="test_herd_123",
+        session_start_time=datetime(2020, 1, 1, 12, 0, 0, tzinfo=tzutc()),
+        subject=Subject(subject_id="sub01", species=species, sex="U", age="P90D"),
+    )
+    device = nwbfile.create_device(name="device")
+    group = nwbfile.create_electrode_group(
+        name="group", description="d", location=group_location, device=device
+    )
+    for loc in electrode_locations:
+        nwbfile.add_electrode(group=group, location=loc)
+    uberon = {
+        "S1_Tr": "UBERON:0008933",
+        "M1_anterior": "UBERON:0001384",
+        "VISp": "UBERON:0002436",
+        "EMG": "UBERON:0002380",
+    }
+    herd = nwbfile.get_external_resources()
+    # Subject links must not become anatomy, even though age uses UBERON
+    herd.add_ref(
+        container=nwbfile.subject,
+        attribute="age",
+        key="P90D",
+        entity_id="UBERON:0000113",
+        entity_uri="http://purl.obolibrary.org/obo/UBERON_0000113",
+    )
+    herd.add_ref(
+        container=nwbfile.subject,
+        attribute="species",
+        key=species,
+        entity_id="NCBITaxon:10116",
+        entity_uri="http://purl.obolibrary.org/obo/NCBITaxon_10116",
+    )
+    herd.add_ref(
+        container=group,
+        attribute="location",
+        key=group_location,
+        entity_id="UBERON:0000956",
+        entity_uri="http://purl.obolibrary.org/obo/UBERON_0000956",
+    )
+    for loc in dict.fromkeys(electrode_locations):
+        uid = uberon[loc]
+        herd.add_ref(
+            container=nwbfile.electrodes["location"],
+            key=loc,
+            entity_id=uid,
+            entity_uri="http://purl.obolibrary.org/obo/" + uid.replace(":", "_"),
+        )
+    with NWBHDF5IO(str(nwb_path), "w") as io:
+        io.write(nwbfile)
+    return uberon
+
+
+def _deepest_anatomy_ids(asset: Any) -> list[str]:
+    assert asset.wasDerivedFrom is not None
+    sample = asset.wasDerivedFrom[0]
+    while sample.wasDerivedFrom:
+        sample = sample.wasDerivedFrom[0]
+    assert sample.anatomy is not None
+    return [str(a.identifier) for a in sample.anatomy]
+
+
+@_skip_no_in_file_herd
+@pytest.mark.ai_generated
+def test_brain_anatomy_from_herd_non_mouse(tmp_path: Path) -> None:
+    """HERD-linked anatomy is used for any species and skips Subject links."""
+    nwb_path = tmp_path / "test_herd_rat.nwb"
+    _write_nwb_with_herd(
+        nwb_path,
+        species="Rattus norvegicus",
+        group_location="cortex",
+        electrode_locations=["S1_Tr", "M1_anterior", "S1_Tr"],
+    )
+    asset = nwb2asset(nwb_path, digest=DUMMY_DANDI_ETAG)
+    assert _deepest_anatomy_ids(asset) == [
+        "http://purl.obolibrary.org/obo/UBERON_0000956",
+        "http://purl.obolibrary.org/obo/UBERON_0008933",
+        "http://purl.obolibrary.org/obo/UBERON_0001384",
+    ]
+    assert asset.wasDerivedFrom is not None
+    sample = asset.wasDerivedFrom[0]
+    assert sample.identifier == "brain-region-sample"
+    assert sample.anatomy is not None
+    assert [a.name for a in sample.anatomy] == ["cortex", "S1_Tr", "M1_anterior"]
+
+
+@_skip_no_in_file_herd
+@pytest.mark.ai_generated
+def test_brain_anatomy_herd_takes_precedence_over_string_matching(
+    tmp_path: Path,
+) -> None:
+    """For a mouse, HERD links win over Allen CCF matching of location strings."""
+    nwb_path = tmp_path / "test_herd_mouse.nwb"
+    _write_nwb_with_herd(
+        nwb_path,
+        species="Mus musculus",
+        group_location="cortex",
+        electrode_locations=["VISp"],
+    )
+    asset = nwb2asset(nwb_path, digest=DUMMY_DANDI_ETAG)
+    ids = _deepest_anatomy_ids(asset)
+    assert "http://purl.obolibrary.org/obo/UBERON_0002436" in ids
+    assert not any("MBA_" in i for i in ids)
+
+
+@pytest.mark.ai_generated
+def test_herd_anatomy_to_models_labels_and_dedup() -> None:
+    """Names come from bundled tables, else the longest linked key."""
+    anatomy = herd_anatomy_to_models(
+        [
+            {
+                "identifier": "http://purl.obolibrary.org/obo/MBA_385",
+                "entity_id": "MBA:385",
+                "keys": ["V1"],
+            },
+            {
+                "identifier": "http://purl.obolibrary.org/obo/UBERON_0002380",
+                "entity_id": "UBERON:0002380",
+                "keys": ["EMG", "trapezius muscle"],
+            },
+            {
+                "identifier": "http://purl.obolibrary.org/obo/UBERON_0002380",
+                "entity_id": "UBERON:0002380",
+                "keys": ["EMG"],
+            },
+            {"identifier": "UBERON:0000956", "entity_id": "UBERON:0000956", "keys": []},
+        ]
+    )
+    assert [(str(a.identifier), a.name) for a in anatomy] == [
+        ("http://purl.obolibrary.org/obo/MBA_385", "Primary visual area"),
+        ("http://purl.obolibrary.org/obo/UBERON_0002380", "trapezius muscle"),
+        ("UBERON:0000956", "UBERON:0000956"),
+    ]
+
+
+@pytest.mark.ai_generated
+def test_brain_anatomy_herd_without_species() -> None:
+    """HERD anatomy needs no species and attaches to a synthetic BioSample."""
+    metadata = {
+        "herd_anatomy": [
+            {
+                "identifier": "http://purl.obolibrary.org/obo/UBERON_0001384",
+                "entity_id": "UBERON:0001384",
+                "keys": ["M1_anterior"],
+            }
+        ],
+    }
+    result = extract_wasDerivedFrom(metadata)
+    assert result is not None
+    assert result[0].identifier == "brain-region-sample"
+    assert result[0].anatomy is not None
+    assert [a.name for a in result[0].anatomy] == ["M1_anterior"]
+
+
 @mark_xfail_ontobee
 @mark.skipif_no_network
 @pytest.mark.obolibrary