diff --git a/src/ndi/__init__.py b/src/ndi/__init__.py index 889785b..96446f0 100644 --- a/src/ndi/__init__.py +++ b/src/ndi/__init__.py @@ -65,6 +65,12 @@ from .neuron import ndi_neuron from .probe import ndi_probe from .query import ndi_query + +# Expose ndi_query as ``ndi.query`` so that MATLAB-style calling +# conventions (``ndi.query('','isa','base')``) work directly. +# The underlying ``ndi.query`` *module* remains importable via +# ``from ndi.query import ...``. +query = ndi_query from .session import empty_id, ndi_session, ndi_session_dir # Import Phase 8 classes diff --git a/src/ndi/daq/metadatareader/__init__.py b/src/ndi/daq/metadatareader/__init__.py index c8c6a79..d408c0b 100644 --- a/src/ndi/daq/metadatareader/__init__.py +++ b/src/ndi/daq/metadatareader/__init__.py @@ -62,7 +62,7 @@ def __init__( if document is not None: doc_props = getattr(document, "document_properties", document) if hasattr(doc_props, "base") and hasattr(doc_props.base, "id"): - self.identifier = doc_props.base.id + self._id = doc_props.base.id if hasattr(doc_props, "daqmetadatareader"): self._tab_separated_file_parameter = getattr( doc_props.daqmetadatareader, "tab_separated_file_parameter", "" diff --git a/src/ndi/daq/reader_base.py b/src/ndi/daq/reader_base.py index e047226..9b1650e 100644 --- a/src/ndi/daq/reader_base.py +++ b/src/ndi/daq/reader_base.py @@ -58,7 +58,7 @@ def __init__( if document is not None: doc_props = getattr(document, "document_properties", document) if hasattr(doc_props, "base") and hasattr(doc_props.base, "id"): - self.identifier = doc_props.base.id + self._id = doc_props.base.id def epochclock( self, diff --git a/src/ndi/file/navigator/__init__.py b/src/ndi/file/navigator/__init__.py index 35e6c85..dc09c3d 100644 --- a/src/ndi/file/navigator/__init__.py +++ b/src/ndi/file/navigator/__init__.py @@ -214,7 +214,7 @@ def _prop(obj: Any, key: str, default: Any = None) -> Any: if base is not None: base_id = _prop(base, "id") if base_id is not None: - self.identifier = base_id + self._id = base_id self._name = _prop(base, "name", "") or "unknown" filenavigator = _prop(doc_props, "filenavigator") diff --git a/src/ndi/subject.py b/src/ndi/subject.py index bc29f9d..918e8c0 100644 --- a/src/ndi/subject.py +++ b/src/ndi/subject.py @@ -106,7 +106,7 @@ def _load_from_session(self, session: Any, doc_or_id: Any) -> None: # Use document ID as our identifier base_id = props.get("base", {}).get("id", "") if base_id: - self.identifier = base_id + self._id = base_id @property def local_identifier(self) -> str: diff --git a/tests/symmetry/make_artifacts/session/_ingestion_helpers.py b/tests/symmetry/make_artifacts/session/_ingestion_helpers.py new file mode 100644 index 0000000..8c1c11b --- /dev/null +++ b/tests/symmetry/make_artifacts/session/_ingestion_helpers.py @@ -0,0 +1,197 @@ +"""Shared helpers for ingestion-based symmetry makeArtifacts tests. + +Provides functions to locate example data files and set up sessions +with DAQ systems for ingestion testing. +""" + +from __future__ import annotations + +import os +import shutil +from pathlib import Path + + +def _find_intan_rhd() -> Path | None: + """Locate an Intan .rhd example data file. + + Searches several well-known locations used by CI and local development. + + Returns: + Path to the directory containing .rhd + epochprobemap files, + or None if no example data can be found. + """ + candidates = [ + # Explicit env var + Path(os.environ.get("NDI_EXAMPLE_DATA", "")) / "exp1_eg_saved", + # CI layout: NDI-matlab checked out beside NDI-python + Path(__file__).resolve().parents[4] + / ".." + / "NDI-matlab" + / "example_data" + / "exp1_eg_saved", + # Home directory + Path.home() / "ndi_example_data" / "exp1_eg_saved", + # ndi_common within the package + Path(__file__).resolve().parents[4] + / "src" + / "ndi" + / "ndi_common" + / "example_sessions" + / "exp1_eg_saved", + ] + for d in candidates: + d = d.resolve() + if d.is_dir() and list(d.glob("*.rhd")): + return d + return None + + +def _find_axon_abf() -> Path | None: + """Locate an Axon .abf example data file. + + Searches several well-known locations used by CI and local development. + + Returns: + Path to the directory containing .abf file(s), + or None if no example data can be found. + """ + candidates = [ + # Explicit env var + Path(os.environ.get("NDI_EXAMPLE_DATA", "")) / "axon", + # CI layout: NDR library example data + Path(__file__).resolve().parents[4] + / ".." + / "NDI-matlab" + / "tools" + / "NDR" + / "example_data", + # Home directory + Path.home() / "ndi_example_data" / "axon", + ] + + # Also try to find NDR's example_data via the installed ndr package + try: + import ndr.fun + + ndr_path = Path(ndr.fun.ndrpath()) / "example_data" + candidates.insert(0, ndr_path) + except Exception: + pass + + for d in candidates: + d = d.resolve() + if d.is_dir() and list(d.glob("*.abf")): + return d + return None + + +def setup_intan_session(session_dir: Path, reader_class: str = "intan") -> dict: + """Set up a session directory with Intan data for ingestion testing. + + Copies example .rhd file and creates required epochprobemap file. + + Args: + session_dir: Directory to create the session in. + reader_class: Either ``"intan"`` for the native reader or + ``"ndr"`` for the NDR wrapper reader. + + Returns: + Dict with keys: ``data_dir`` (source dir), ``rhd_file`` (copied file name), + ``reader_class_str``, ``daqreader_class_str``. + + Raises: + FileNotFoundError: If no example data could be found. + """ + data_dir = _find_intan_rhd() + if data_dir is None: + raise FileNotFoundError("Intan example data (.rhd) not found") + + # Find the first .rhd file + rhd_files = sorted(data_dir.glob("*.rhd")) + rhd_file = rhd_files[0] + + # Copy .rhd to session dir + shutil.copy2(str(rhd_file), str(session_dir / rhd_file.name)) + + # Create epochprobemap file + stem = rhd_file.stem + probemap_path = session_dir / f"{stem}.epochprobemap.ndi" + if not probemap_path.exists(): + probemap_path.write_text( + "name\treference\ttype\tdevicestring\tsubjectstring\n" + "ctx\t1\tn-trode\tintan1:ai1\tanteater27@nosuchlab.org\n" + ) + + if reader_class == "intan": + return { + "data_dir": data_dir, + "rhd_file": rhd_file.name, + "daqreader_class_str": "ndi.daq.reader.mfdaq.intan", + "reader_class_str": "ndi_daq_reader_mfdaq_intan", + } + else: + return { + "data_dir": data_dir, + "rhd_file": rhd_file.name, + "daqreader_class_str": "ndi.daq.reader.mfdaq.ndr", + "reader_class_str": "ndi_daq_reader_mfdaq_ndr", + } + + +def setup_axon_session(session_dir: Path) -> dict: + """Set up a session directory with Axon ABF data for ingestion testing. + + Copies example .abf file and creates required epochprobemap file. + + Args: + session_dir: Directory to create the session in. + + Returns: + Dict with keys: ``data_dir``, ``abf_file``, ``daqreader_class_str``. + + Raises: + FileNotFoundError: If no example data could be found. + """ + data_dir = _find_axon_abf() + if data_dir is None: + raise FileNotFoundError("Axon example data (.abf) not found") + + # Find the first .abf file + abf_files = sorted(data_dir.glob("*.abf")) + abf_file = abf_files[0] + + # Copy .abf to session dir + shutil.copy2(str(abf_file), str(session_dir / abf_file.name)) + + # Create epochprobemap file + stem = abf_file.stem + probemap_path = session_dir / f"{stem}.epochprobemap.ndi" + if not probemap_path.exists(): + probemap_path.write_text( + "name\treference\ttype\tdevicestring\tsubjectstring\n" + "ctx\t1\tn-trode\taxon1:ai1\tanteateri27@nosuchlab.org\n" + ) + + return { + "data_dir": data_dir, + "abf_file": abf_file.name, + "daqreader_class_str": "ndi.daq.reader.mfdaq.ndr", + } + + +def delete_raw_files(session_path: Path) -> None: + """Delete all files in a session directory except the ``.ndi`` database. + + This mirrors the MATLAB ingestion test behavior of removing raw data + after ingestion so that only the ingested database remains. + + Args: + session_path: Root directory of the NDI session. + """ + for item in session_path.iterdir(): + if item.name == ".ndi": + continue + if item.is_dir(): + shutil.rmtree(item) + else: + item.unlink() diff --git a/tests/symmetry/make_artifacts/session/test_ingestion_axon_ndr.py b/tests/symmetry/make_artifacts/session/test_ingestion_axon_ndr.py new file mode 100644 index 0000000..d9a9f9e --- /dev/null +++ b/tests/symmetry/make_artifacts/session/test_ingestion_axon_ndr.py @@ -0,0 +1,173 @@ +"""Generate symmetry artifacts for an ingested Axon NDR session. + +Python equivalent of: + tests/+ndi/+symmetry/+makeArtifacts/+session/ingestionAxonNDR.m + +This test creates an NDI session with a real Axon .abf data file using +the NDR (Neuroscience Data Reader) wrapper reader, ingests the session, +deletes raw data files (keeping only the .ndi database), and exports the +resulting artifacts to: + + /NDI/symmetryTest/pythonArtifacts/session/ingestionAxonNDR/ + testIngestionAxonNDRArtefacts/ + +Note: The MATLAB test uses British spelling ``Artefacts`` in the test +method name; we match that exactly so both languages write to the same +artifact directory path. + +The artifacts are left on disk so that the MATLAB ``readArtifacts`` suite +(and the Python ``read_artifacts`` suite) can load and verify them. +""" + +import json +import shutil + +import pytest + +from ndi.query import ndi_query +from ndi.session.dir import ndi_session_dir +from ndi.subject import ndi_subject +from ndi.util import sessionSummary +from tests.symmetry.conftest import PYTHON_ARTIFACTS +from tests.symmetry.make_artifacts.session._ingestion_helpers import ( + delete_raw_files, + setup_axon_session, +) + +ARTIFACT_DIR = PYTHON_ARTIFACTS / "session" / "ingestionAxonNDR" / "testIngestionAxonNDRArtefacts" + + +def _have_axon_data() -> bool: + try: + from tests.symmetry.make_artifacts.session._ingestion_helpers import ( + _find_axon_abf, + ) + + return _find_axon_abf() is not None + except Exception: + return False + + +@pytest.mark.skipif(not _have_axon_data(), reason="Axon example data (.abf) not found") +class TestIngestionAxonNDR: + """Mirror of ndi.symmetry.makeArtifacts.session.ingestionAxonNDR.""" + + @pytest.fixture(autouse=True) + def _setup(self, tmp_path): + """Build an Axon NDR session in a temporary directory. + + Mirrors the MATLAB ``buildSessionNDRAxon`` setup with real .abf + data and the NDR wrapper reader. + """ + session_dir = tmp_path / "exp1" + session_dir.mkdir() + + session = ndi_session_dir("exp1", session_dir) + session.database_clear("yes") + session.cache.clear() + + # Copy Axon data file and create epochprobemap + setup_axon_session(session_dir) + + # Add DAQ system via documents (filenavigator + daqreader + daqsystem) + fn_doc = session.newdocument( + "daq/filenavigator", + **{ + "base.name": "unknown", + "filenavigator.ndi_filenavigator_class": "ndi.file.navigator", + "filenavigator.fileparameters": "{ '#.abf', '#.epochprobemap.ndi' }", + "filenavigator.epochprobemap_class": "ndi.epoch.epochprobemap_daqsystem", + "filenavigator.epochprobemap_fileparameters": "{ '(.*?)epochprobemap.ndi' }", + }, + ) + session.database_add(fn_doc) + + dr_doc = session.newdocument( + "daq/daqreader_ndr", + **{ + "base.name": "axon_ndr_reader", + "daqreader.ndi_daqreader_class": "ndi.daq.reader.mfdaq.ndr", + }, + ) + session.database_add(dr_doc) + + daq_doc = session.newdocument( + "daq/daqsystem", + **{ + "base.name": "axon1", + "daqsystem.ndi_daqsystem_class": "ndi.daq.system.mfdaq", + }, + ) + daq_doc = daq_doc.set_dependency_value( + "filenavigator_id", fn_doc.id, error_if_not_found=False + ) + daq_doc = daq_doc.set_dependency_value("daqreader_id", dr_doc.id, error_if_not_found=False) + session.database_add(daq_doc) + + # Add subject (note: Axon uses "anteateri27" — matches MATLAB buildSessionNDRAxon) + subject = ndi_subject("anteateri27@nosuchlab.org", "") + session.database_add(subject.newdocument()) + + # Add subjectmeasurement document + doc = session.newdocument( + "subjectmeasurement", + **{ + "base.name": "Animal statistics", + "subjectmeasurement.measurement": "age", + "subjectmeasurement.value": 30, + "subjectmeasurement.datestamp": "2017-03-17T19:53:57.066Z", + }, + ) + doc = doc.set_dependency_value("subject_id", subject.id, error_if_not_found=False) + session.database_add(doc) + + self.session = session + + def test_ingestion_axon_ndr_artefacts(self): + """Ingest via NDR reader, strip raw files, and export artifacts.""" + artifact_dir = ARTIFACT_DIR + + if artifact_dir.exists(): + shutil.rmtree(artifact_dir) + + # Call getprobes() BEFORE ingestion to generate internal documents. + self.session.getprobes() + + # Ingest the session + success, msg = self.session.ingest() + assert success, f"Ingestion failed: {msg}" + + # Delete raw data files (keep only .ndi database) + delete_raw_files(self.session.path) + + # Clear cache and re-open the session + self.session.cache.clear() + session_path = self.session.path + self.session = ndi_session_dir("exp1", session_path) + + # Create session summary + summary = sessionSummary(self.session) + summary_json = json.dumps(summary, indent=2, allow_nan=True) + + # Copy session to persistent artifact directory + shutil.copytree(str(session_path), str(artifact_dir)) + + # Write individual JSON documents + json_docs_dir = artifact_dir / "jsonDocuments" + json_docs_dir.mkdir(exist_ok=True) + + docs = self.session.database_search(ndi_query("base.id").match("(.*)")) + for doc in docs: + props = doc.document_properties + doc_path = json_docs_dir / f"{doc.id}.json" + doc_path.write_text(json.dumps(props, indent=2, allow_nan=True), encoding="utf-8") + + # Write sessionSummary.json + summary_path = artifact_dir / "sessionSummary.json" + summary_path.write_text(summary_json, encoding="utf-8") + + # Verify artifacts were created + assert artifact_dir.exists() + assert summary_path.exists() + assert json_docs_dir.exists() + assert len(list(json_docs_dir.glob("*.json"))) > 0 diff --git a/tests/symmetry/make_artifacts/session/test_ingestion_intan.py b/tests/symmetry/make_artifacts/session/test_ingestion_intan.py new file mode 100644 index 0000000..3aed06a --- /dev/null +++ b/tests/symmetry/make_artifacts/session/test_ingestion_intan.py @@ -0,0 +1,168 @@ +"""Generate symmetry artifacts for an ingested Intan session. + +Python equivalent of: + tests/+ndi/+symmetry/+makeArtifacts/+session/ingestionIntan.m + +This test creates an NDI session with a real Intan .rhd data file, +ingests the session, deletes raw data files (keeping only the .ndi +database), and exports the resulting artifacts to: + + /NDI/symmetryTest/pythonArtifacts/session/ingestionIntan/ + testIngestionIntanArtifacts/ + +The artifacts are left on disk so that the MATLAB ``readArtifacts`` suite +(and the Python ``read_artifacts`` suite) can load and verify them. +""" + +import json +import shutil + +import pytest + +from ndi.query import ndi_query +from ndi.session.dir import ndi_session_dir +from ndi.subject import ndi_subject +from ndi.util import sessionSummary +from tests.symmetry.conftest import PYTHON_ARTIFACTS +from tests.symmetry.make_artifacts.session._ingestion_helpers import ( + delete_raw_files, + setup_intan_session, +) + +ARTIFACT_DIR = PYTHON_ARTIFACTS / "session" / "ingestionIntan" / "testIngestionIntanArtifacts" + + +def _have_intan_data() -> bool: + try: + from tests.symmetry.make_artifacts.session._ingestion_helpers import ( + _find_intan_rhd, + ) + + return _find_intan_rhd() is not None + except Exception: + return False + + +@pytest.mark.skipif(not _have_intan_data(), reason="Intan example data (.rhd) not found") +class TestIngestionIntan: + """Mirror of ndi.symmetry.makeArtifacts.session.ingestionIntan.""" + + @pytest.fixture(autouse=True) + def _setup(self, tmp_path): + """Build an Intan session in a temporary directory. + + Mirrors the MATLAB ``buildSession`` setup with real .rhd data + and the native Intan DAQ reader. + """ + session_dir = tmp_path / "exp1" + session_dir.mkdir() + + session = ndi_session_dir("exp1", session_dir) + session.database_clear("yes") + session.cache.clear() + + # Copy Intan data file and create epochprobemap + setup_intan_session(session_dir, reader_class="intan") + + # Add DAQ system via documents (filenavigator + daqreader + daqsystem) + fn_doc = session.newdocument( + "daq/filenavigator", + **{ + "base.name": "unknown", + "filenavigator.ndi_filenavigator_class": "ndi.file.navigator", + "filenavigator.fileparameters": "{ '#.rhd', '#.epochprobemap.ndi' }", + "filenavigator.epochprobemap_class": "ndi.epoch.epochprobemap_daqsystem", + "filenavigator.epochprobemap_fileparameters": "{ '(.*?)epochprobemap.ndi' }", + }, + ) + session.database_add(fn_doc) + + dr_doc = session.newdocument( + "daq/daqreader", + **{ + "base.name": "intan_reader", + "daqreader.ndi_daqreader_class": "ndi.daq.reader.mfdaq.intan", + }, + ) + session.database_add(dr_doc) + + daq_doc = session.newdocument( + "daq/daqsystem", + **{ + "base.name": "intan1", + "daqsystem.ndi_daqsystem_class": "ndi.daq.system.mfdaq", + }, + ) + daq_doc = daq_doc.set_dependency_value( + "filenavigator_id", fn_doc.id, error_if_not_found=False + ) + daq_doc = daq_doc.set_dependency_value("daqreader_id", dr_doc.id, error_if_not_found=False) + session.database_add(daq_doc) + + # Add subject + subject = ndi_subject("anteater27@nosuchlab.org", "") + session.database_add(subject.newdocument()) + + # Add subjectmeasurement document + doc = session.newdocument( + "subjectmeasurement", + **{ + "base.name": "Animal statistics", + "subjectmeasurement.measurement": "age", + "subjectmeasurement.value": 30, + "subjectmeasurement.datestamp": "2017-03-17T19:53:57.066Z", + }, + ) + doc = doc.set_dependency_value("subject_id", subject.id, error_if_not_found=False) + session.database_add(doc) + + self.session = session + + def test_ingestion_intan_artifacts(self): + """Ingest, strip raw files, and export artifacts.""" + artifact_dir = ARTIFACT_DIR + + if artifact_dir.exists(): + shutil.rmtree(artifact_dir) + + # Call getprobes() BEFORE ingestion to generate internal documents. + self.session.getprobes() + + # Ingest the session + success, msg = self.session.ingest() + assert success, f"Ingestion failed: {msg}" + + # Delete raw data files (keep only .ndi database) + delete_raw_files(self.session.path) + + # Clear cache and re-open the session + self.session.cache.clear() + session_path = self.session.path + self.session = ndi_session_dir("exp1", session_path) + + # Create session summary + summary = sessionSummary(self.session) + summary_json = json.dumps(summary, indent=2, allow_nan=True) + + # Copy session to persistent artifact directory + shutil.copytree(str(session_path), str(artifact_dir)) + + # Write individual JSON documents + json_docs_dir = artifact_dir / "jsonDocuments" + json_docs_dir.mkdir(exist_ok=True) + + docs = self.session.database_search(ndi_query("base.id").match("(.*)")) + for doc in docs: + props = doc.document_properties + doc_path = json_docs_dir / f"{doc.id}.json" + doc_path.write_text(json.dumps(props, indent=2, allow_nan=True), encoding="utf-8") + + # Write sessionSummary.json + summary_path = artifact_dir / "sessionSummary.json" + summary_path.write_text(summary_json, encoding="utf-8") + + # Verify artifacts were created + assert artifact_dir.exists() + assert summary_path.exists() + assert json_docs_dir.exists() + assert len(list(json_docs_dir.glob("*.json"))) > 0 diff --git a/tests/symmetry/make_artifacts/session/test_ingestion_intan_ndr.py b/tests/symmetry/make_artifacts/session/test_ingestion_intan_ndr.py new file mode 100644 index 0000000..4147020 --- /dev/null +++ b/tests/symmetry/make_artifacts/session/test_ingestion_intan_ndr.py @@ -0,0 +1,170 @@ +"""Generate symmetry artifacts for an ingested Intan session using the NDR reader. + +Python equivalent of: + tests/+ndi/+symmetry/+makeArtifacts/+session/ingestionIntanNDR.m + +This test creates an NDI session with a real Intan .rhd data file using +the NDR (Neuroscience Data Reader) wrapper reader, ingests the session, +deletes raw data files (keeping only the .ndi database), and exports the +resulting artifacts to: + + /NDI/symmetryTest/pythonArtifacts/session/ingestionIntanNDR/ + testIngestionIntanNDRArtifacts/ + +The artifacts are left on disk so that the MATLAB ``readArtifacts`` suite +(and the Python ``read_artifacts`` suite) can load and verify them. +""" + +import json +import shutil + +import pytest + +from ndi.query import ndi_query +from ndi.session.dir import ndi_session_dir +from ndi.subject import ndi_subject +from ndi.util import sessionSummary +from tests.symmetry.conftest import PYTHON_ARTIFACTS +from tests.symmetry.make_artifacts.session._ingestion_helpers import ( + delete_raw_files, + setup_intan_session, +) + +ARTIFACT_DIR = PYTHON_ARTIFACTS / "session" / "ingestionIntanNDR" / "testIngestionIntanNDRArtifacts" + + +def _have_intan_data() -> bool: + try: + from tests.symmetry.make_artifacts.session._ingestion_helpers import ( + _find_intan_rhd, + ) + + return _find_intan_rhd() is not None + except Exception: + return False + + +@pytest.mark.skipif(not _have_intan_data(), reason="Intan example data (.rhd) not found") +class TestIngestionIntanNDR: + """Mirror of ndi.symmetry.makeArtifacts.session.ingestionIntanNDR.""" + + @pytest.fixture(autouse=True) + def _setup(self, tmp_path): + """Build an Intan NDR session in a temporary directory. + + Mirrors the MATLAB ``buildSessionNDRIntan`` setup with real .rhd + data and the NDR wrapper reader. + """ + session_dir = tmp_path / "exp1" + session_dir.mkdir() + + session = ndi_session_dir("exp1", session_dir) + session.database_clear("yes") + session.cache.clear() + + # Copy Intan data file and create epochprobemap + setup_intan_session(session_dir, reader_class="ndr") + + # Add DAQ system via documents (filenavigator + daqreader + daqsystem) + # NDR reader uses same file patterns as native Intan reader + fn_doc = session.newdocument( + "daq/filenavigator", + **{ + "base.name": "unknown", + "filenavigator.ndi_filenavigator_class": "ndi.file.navigator", + "filenavigator.fileparameters": "{ '#.rhd', '#.epochprobemap.ndi' }", + "filenavigator.epochprobemap_class": "ndi.epoch.epochprobemap_daqsystem", + "filenavigator.epochprobemap_fileparameters": "{ '(.*?)epochprobemap.ndi' }", + }, + ) + session.database_add(fn_doc) + + dr_doc = session.newdocument( + "daq/daqreader_ndr", + **{ + "base.name": "intan_ndr_reader", + "daqreader.ndi_daqreader_class": "ndi.daq.reader.mfdaq.ndr", + }, + ) + session.database_add(dr_doc) + + daq_doc = session.newdocument( + "daq/daqsystem", + **{ + "base.name": "intan1", + "daqsystem.ndi_daqsystem_class": "ndi.daq.system.mfdaq", + }, + ) + daq_doc = daq_doc.set_dependency_value( + "filenavigator_id", fn_doc.id, error_if_not_found=False + ) + daq_doc = daq_doc.set_dependency_value("daqreader_id", dr_doc.id, error_if_not_found=False) + session.database_add(daq_doc) + + # Add subject + subject = ndi_subject("anteater27@nosuchlab.org", "") + session.database_add(subject.newdocument()) + + # Add subjectmeasurement document + doc = session.newdocument( + "subjectmeasurement", + **{ + "base.name": "Animal statistics", + "subjectmeasurement.measurement": "age", + "subjectmeasurement.value": 30, + "subjectmeasurement.datestamp": "2017-03-17T19:53:57.066Z", + }, + ) + doc = doc.set_dependency_value("subject_id", subject.id, error_if_not_found=False) + session.database_add(doc) + + self.session = session + + def test_ingestion_intan_ndr_artifacts(self): + """Ingest via NDR reader, strip raw files, and export artifacts.""" + artifact_dir = ARTIFACT_DIR + + if artifact_dir.exists(): + shutil.rmtree(artifact_dir) + + # Call getprobes() BEFORE ingestion to generate internal documents. + self.session.getprobes() + + # Ingest the session + success, msg = self.session.ingest() + assert success, f"Ingestion failed: {msg}" + + # Delete raw data files (keep only .ndi database) + delete_raw_files(self.session.path) + + # Clear cache and re-open the session + self.session.cache.clear() + session_path = self.session.path + self.session = ndi_session_dir("exp1", session_path) + + # Create session summary + summary = sessionSummary(self.session) + summary_json = json.dumps(summary, indent=2, allow_nan=True) + + # Copy session to persistent artifact directory + shutil.copytree(str(session_path), str(artifact_dir)) + + # Write individual JSON documents + json_docs_dir = artifact_dir / "jsonDocuments" + json_docs_dir.mkdir(exist_ok=True) + + docs = self.session.database_search(ndi_query("base.id").match("(.*)")) + for doc in docs: + props = doc.document_properties + doc_path = json_docs_dir / f"{doc.id}.json" + doc_path.write_text(json.dumps(props, indent=2, allow_nan=True), encoding="utf-8") + + # Write sessionSummary.json + summary_path = artifact_dir / "sessionSummary.json" + summary_path.write_text(summary_json, encoding="utf-8") + + # Verify artifacts were created + assert artifact_dir.exists() + assert summary_path.exists() + assert json_docs_dir.exists() + assert len(list(json_docs_dir.glob("*.json"))) > 0 diff --git a/tests/symmetry/read_artifacts/session/test_ingestion_axon_ndr.py b/tests/symmetry/read_artifacts/session/test_ingestion_axon_ndr.py new file mode 100644 index 0000000..c9278ca --- /dev/null +++ b/tests/symmetry/read_artifacts/session/test_ingestion_axon_ndr.py @@ -0,0 +1,122 @@ +"""Read and verify symmetry artifacts for an ingested Axon NDR session. + +Python equivalent of: + tests/+ndi/+symmetry/+readArtifacts/+session/ingestionAxonNDR.m + +This test loads artifacts produced by *either* the MATLAB or the Python +``makeArtifacts`` suite and verifies that the Python NDI stack can: + +1. Open the ingested session database (raw files have been removed). +2. Generate a live session summary that matches the stored ``sessionSummary.json``. +3. Load every document whose JSON was exported to ``jsonDocuments/``. + +Note: The MATLAB test uses British spelling ``Artefacts`` in the test +method / artifact directory name; we match that exactly. + +The test is parameterized over ``source_type`` so that a single test class +covers both ``matlabArtifacts`` and ``pythonArtifacts``. +""" + +import json + +import pytest + +from ndi.query import ndi_query +from ndi.session.dir import ndi_session_dir +from ndi.util import compareSessionSummary, sessionSummary +from tests.symmetry.conftest import SOURCE_TYPES, SYMMETRY_BASE + + +@pytest.fixture(params=SOURCE_TYPES) +def source_type(request): + """Parameterize over matlabArtifacts / pythonArtifacts.""" + return request.param + + +class TestIngestionAxonNDR: + """Mirror of ndi.symmetry.readArtifacts.session.ingestionAxonNDR.""" + + def _artifact_dir(self, source_type: str): + return ( + SYMMETRY_BASE + / source_type + / "session" + / "ingestionAxonNDR" + / "testIngestionAxonNDRArtefacts" + ) + + def _open_session(self, source_type): + artifact_dir = self._artifact_dir(source_type) + if not artifact_dir.exists(): + pytest.skip( + f"Artifact directory from {source_type} does not exist. " + f"Run the corresponding makeArtifacts suite first." + ) + return artifact_dir, ndi_session_dir("exp1", artifact_dir) + + def test_ingestion_axon_ndr_summary(self, source_type): + """Verify that the live session summary matches sessionSummary.json.""" + artifact_dir, session = self._open_session(source_type) + + summary_path = artifact_dir / "sessionSummary.json" + if not summary_path.exists(): + pytest.skip(f"sessionSummary.json not found in {source_type} artifact directory.") + + expected_summary = json.loads(summary_path.read_text(encoding="utf-8")) + actual_summary = sessionSummary(session) + + exclude_fields = ["epochNodes_filenavigator", "epochNodes_daqsystem"] + + def _filter_epochid(files: list[str]) -> list[str]: + return [f for f in files if not f.endswith(".epochid.ndi")] + + actual_summary["files"] = _filter_epochid(actual_summary.get("files", [])) + expected_summary["files"] = _filter_epochid(expected_summary.get("files", [])) + + report = compareSessionSummary( + actual_summary, + expected_summary, + excludeFiles=["sessionSummary.json", "jsonDocuments"], + excludeFields=exclude_fields, + ) + + assert len(report) == 0, ( + f"ndi_session summary mismatch against {source_type} generated artifacts:\n" + + "\n".join(report) + ) + + def test_ingestion_axon_ndr_documents(self, source_type): + """Verify that every exported JSON document can be loaded from the session DB.""" + artifact_dir, session = self._open_session(source_type) + + json_docs_dir = artifact_dir / "jsonDocuments" + if not json_docs_dir.exists(): + pytest.skip(f"jsonDocuments directory not found in {source_type}.") + + json_files = list(json_docs_dir.glob("*.json")) + + actual_docs = session.database_search(ndi_query("base.id").match("(.*)")) + + assert len(actual_docs) == len(json_files), ( + f"Number of documents in session ({len(actual_docs)}) does not match " + f"{source_type} JSON artifacts ({len(json_files)})." + ) + + for jf in json_files: + expected_doc = json.loads(jf.read_text(encoding="utf-8")) + expected_id = expected_doc.get("base", {}).get("id", "") + + found = False + for actual in actual_docs: + if actual.id == expected_id: + found = True + actual_props = actual.document_properties + assert actual_props.get("document_class", {}).get( + "class_name" + ) == expected_doc.get("document_class", {}).get("class_name"), ( + f"ndi_document class mismatch for id: {expected_id} " f"in {source_type}" + ) + break + assert found, ( + f"ndi_document from {source_type} artifact not found in session: " f"{expected_id}" + ) diff --git a/tests/symmetry/read_artifacts/session/test_ingestion_intan.py b/tests/symmetry/read_artifacts/session/test_ingestion_intan.py new file mode 100644 index 0000000..b33ff9a --- /dev/null +++ b/tests/symmetry/read_artifacts/session/test_ingestion_intan.py @@ -0,0 +1,119 @@ +"""Read and verify symmetry artifacts for an ingested Intan session. + +Python equivalent of: + tests/+ndi/+symmetry/+readArtifacts/+session/ingestionIntan.m + +This test loads artifacts produced by *either* the MATLAB or the Python +``makeArtifacts`` suite and verifies that the Python NDI stack can: + +1. Open the ingested session database (raw files have been removed). +2. Generate a live session summary that matches the stored ``sessionSummary.json``. +3. Load every document whose JSON was exported to ``jsonDocuments/``. + +The test is parameterized over ``source_type`` so that a single test class +covers both ``matlabArtifacts`` and ``pythonArtifacts``. +""" + +import json + +import pytest + +from ndi.query import ndi_query +from ndi.session.dir import ndi_session_dir +from ndi.util import compareSessionSummary, sessionSummary +from tests.symmetry.conftest import SOURCE_TYPES, SYMMETRY_BASE + + +@pytest.fixture(params=SOURCE_TYPES) +def source_type(request): + """Parameterize over matlabArtifacts / pythonArtifacts.""" + return request.param + + +class TestIngestionIntan: + """Mirror of ndi.symmetry.readArtifacts.session.ingestionIntan.""" + + def _artifact_dir(self, source_type: str): + return ( + SYMMETRY_BASE + / source_type + / "session" + / "ingestionIntan" + / "testIngestionIntanArtifacts" + ) + + def _open_session(self, source_type): + artifact_dir = self._artifact_dir(source_type) + if not artifact_dir.exists(): + pytest.skip( + f"Artifact directory from {source_type} does not exist. " + f"Run the corresponding makeArtifacts suite first." + ) + return artifact_dir, ndi_session_dir("exp1", artifact_dir) + + def test_ingestion_intan_summary(self, source_type): + """Verify that the live session summary matches sessionSummary.json.""" + artifact_dir, session = self._open_session(source_type) + + summary_path = artifact_dir / "sessionSummary.json" + if not summary_path.exists(): + pytest.skip(f"sessionSummary.json not found in {source_type} artifact directory.") + + expected_summary = json.loads(summary_path.read_text(encoding="utf-8")) + actual_summary = sessionSummary(session) + + exclude_fields = ["epochNodes_filenavigator", "epochNodes_daqsystem"] + + def _filter_epochid(files: list[str]) -> list[str]: + return [f for f in files if not f.endswith(".epochid.ndi")] + + actual_summary["files"] = _filter_epochid(actual_summary.get("files", [])) + expected_summary["files"] = _filter_epochid(expected_summary.get("files", [])) + + report = compareSessionSummary( + actual_summary, + expected_summary, + excludeFiles=["sessionSummary.json", "jsonDocuments"], + excludeFields=exclude_fields, + ) + + assert len(report) == 0, ( + f"ndi_session summary mismatch against {source_type} generated artifacts:\n" + + "\n".join(report) + ) + + def test_ingestion_intan_documents(self, source_type): + """Verify that every exported JSON document can be loaded from the session DB.""" + artifact_dir, session = self._open_session(source_type) + + json_docs_dir = artifact_dir / "jsonDocuments" + if not json_docs_dir.exists(): + pytest.skip(f"jsonDocuments directory not found in {source_type}.") + + json_files = list(json_docs_dir.glob("*.json")) + + actual_docs = session.database_search(ndi_query("base.id").match("(.*)")) + + assert len(actual_docs) == len(json_files), ( + f"Number of documents in session ({len(actual_docs)}) does not match " + f"{source_type} JSON artifacts ({len(json_files)})." + ) + + for jf in json_files: + expected_doc = json.loads(jf.read_text(encoding="utf-8")) + expected_id = expected_doc.get("base", {}).get("id", "") + + found = False + for actual in actual_docs: + if actual.id == expected_id: + found = True + actual_props = actual.document_properties + assert actual_props.get("document_class", {}).get( + "class_name" + ) == expected_doc.get("document_class", {}).get("class_name"), ( + f"ndi_document class mismatch for id: {expected_id} " f"in {source_type}" + ) + break + assert found, ( + f"ndi_document from {source_type} artifact not found in session: " f"{expected_id}" + ) diff --git a/tests/symmetry/read_artifacts/session/test_ingestion_intan_ndr.py b/tests/symmetry/read_artifacts/session/test_ingestion_intan_ndr.py new file mode 100644 index 0000000..e18502b --- /dev/null +++ b/tests/symmetry/read_artifacts/session/test_ingestion_intan_ndr.py @@ -0,0 +1,119 @@ +"""Read and verify symmetry artifacts for an ingested Intan NDR session. + +Python equivalent of: + tests/+ndi/+symmetry/+readArtifacts/+session/ingestionIntanNDR.m + +This test loads artifacts produced by *either* the MATLAB or the Python +``makeArtifacts`` suite and verifies that the Python NDI stack can: + +1. Open the ingested session database (raw files have been removed). +2. Generate a live session summary that matches the stored ``sessionSummary.json``. +3. Load every document whose JSON was exported to ``jsonDocuments/``. + +The test is parameterized over ``source_type`` so that a single test class +covers both ``matlabArtifacts`` and ``pythonArtifacts``. +""" + +import json + +import pytest + +from ndi.query import ndi_query +from ndi.session.dir import ndi_session_dir +from ndi.util import compareSessionSummary, sessionSummary +from tests.symmetry.conftest import SOURCE_TYPES, SYMMETRY_BASE + + +@pytest.fixture(params=SOURCE_TYPES) +def source_type(request): + """Parameterize over matlabArtifacts / pythonArtifacts.""" + return request.param + + +class TestIngestionIntanNDR: + """Mirror of ndi.symmetry.readArtifacts.session.ingestionIntanNDR.""" + + def _artifact_dir(self, source_type: str): + return ( + SYMMETRY_BASE + / source_type + / "session" + / "ingestionIntanNDR" + / "testIngestionIntanNDRArtifacts" + ) + + def _open_session(self, source_type): + artifact_dir = self._artifact_dir(source_type) + if not artifact_dir.exists(): + pytest.skip( + f"Artifact directory from {source_type} does not exist. " + f"Run the corresponding makeArtifacts suite first." + ) + return artifact_dir, ndi_session_dir("exp1", artifact_dir) + + def test_ingestion_intan_ndr_summary(self, source_type): + """Verify that the live session summary matches sessionSummary.json.""" + artifact_dir, session = self._open_session(source_type) + + summary_path = artifact_dir / "sessionSummary.json" + if not summary_path.exists(): + pytest.skip(f"sessionSummary.json not found in {source_type} artifact directory.") + + expected_summary = json.loads(summary_path.read_text(encoding="utf-8")) + actual_summary = sessionSummary(session) + + exclude_fields = ["epochNodes_filenavigator", "epochNodes_daqsystem"] + + def _filter_epochid(files: list[str]) -> list[str]: + return [f for f in files if not f.endswith(".epochid.ndi")] + + actual_summary["files"] = _filter_epochid(actual_summary.get("files", [])) + expected_summary["files"] = _filter_epochid(expected_summary.get("files", [])) + + report = compareSessionSummary( + actual_summary, + expected_summary, + excludeFiles=["sessionSummary.json", "jsonDocuments"], + excludeFields=exclude_fields, + ) + + assert len(report) == 0, ( + f"ndi_session summary mismatch against {source_type} generated artifacts:\n" + + "\n".join(report) + ) + + def test_ingestion_intan_ndr_documents(self, source_type): + """Verify that every exported JSON document can be loaded from the session DB.""" + artifact_dir, session = self._open_session(source_type) + + json_docs_dir = artifact_dir / "jsonDocuments" + if not json_docs_dir.exists(): + pytest.skip(f"jsonDocuments directory not found in {source_type}.") + + json_files = list(json_docs_dir.glob("*.json")) + + actual_docs = session.database_search(ndi_query("base.id").match("(.*)")) + + assert len(actual_docs) == len(json_files), ( + f"Number of documents in session ({len(actual_docs)}) does not match " + f"{source_type} JSON artifacts ({len(json_files)})." + ) + + for jf in json_files: + expected_doc = json.loads(jf.read_text(encoding="utf-8")) + expected_id = expected_doc.get("base", {}).get("id", "") + + found = False + for actual in actual_docs: + if actual.id == expected_id: + found = True + actual_props = actual.document_properties + assert actual_props.get("document_class", {}).get( + "class_name" + ) == expected_doc.get("document_class", {}).get("class_name"), ( + f"ndi_document class mismatch for id: {expected_id} " f"in {source_type}" + ) + break + assert found, ( + f"ndi_document from {source_type} artifact not found in session: " f"{expected_id}" + )