diff --git a/.github/workflows/link-issues-by-milestone.yml b/.github/workflows/link-issues-by-milestone.yml new file mode 100644 index 00000000..6301e549 --- /dev/null +++ b/.github/workflows/link-issues-by-milestone.yml @@ -0,0 +1,16 @@ +name: Link Issue to Milestone Parent + +on: + issues: + types: [milestoned] + +jobs: + link-issue: + if: github.event.issue.milestone != null + uses: AllenNeuralDynamics/.github/.github/workflows/util-link-issues-by-milestone.yml@main + with: + issue-number: ${{ github.event.issue.number }} + issue-id: ${{ github.event.issue.id }} + milestone-description: ${{ github.event.issue.milestone.description }} + secrets: + service-token: ${{ secrets.SERVICE_TOKEN }} diff --git a/README.md b/README.md index 4bb1aa94..6d5a0be5 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ Repository to contain code that will parse source files into aind-data-schema mo ## Usage -The `GatherMetadataJob` is used to create the `data_description.json` and pull the `subject.json` and `procedures.json` from `aind-metadata-service`. Users are expected to provide the `instrument.json` and the `acquisition.json` as well as optional `processing.json`, `quality_control.json` and `model.json`. The job will attempt to validate all of the metadata files, displaying errors, and then will save all metadata fields into the selected folder. +The `GatherMetadataJob` is used to create the `data_description.json` and pull the `subject.json` and `procedures.json` from `aind-metadata-service`. Users are expected to provide the `instrument.json` and the `acquisition.json` as well as optional `processing.json`, `quality_control.json` and `model.json`. If a user provides `procedures.json`, it will be merged with the procedures fetched from the service (subject and specimen procedures are deduplicated). The job will attempt to validate all of the metadata files, displaying errors, and then will save all metadata fields into the selected folder. ### Using the GatherMetadataJob @@ -51,8 +51,8 @@ If no exact match exists, it will construct, fetch, merge or run mappers to gene | File | Method 1 | Method 2 | Method 3 | |------|----------|----------|----------| | data_description.json | Exact match in input directory | Construct from settings / fetch from metadata-service | | -| subject.json | Exact match in input directory | Fetch from metadata-service (requires subject_id) | | -| procedures.json | Exact match in input directory | Fetch from metadata-service (requires subject_id) | | +| subject.json | Exact match in input directory | Fetch from metadata-service (requires subject_id) | Constructed locally when subject_id is "calibration" | +| procedures.json | Fetch from metadata-service (requires subject_id) | Merge with user-provided procedures but default to user-provided if there are any duplicates | Constructed locally (empty) when subject_id is "calibration" | | acquisition.json | Exact match in input directory | Run mappers on `.json` files (and merge) | Merge all `acquisition*.json` files | | instrument.json | Exact match in input directory | Fetch from metadata-service (requires instrument_id) | Merge all `instrument*.json` files | | processing.json | Exact match in input directory | | | @@ -71,6 +71,9 @@ When mappers are developed from the `BaseMapper` class and registered in `mapper - **`acquisition_start_time`** (datetime, optional): Acquisition start time in ISO 8601 format. This setting should only be used when an `acquisition.json` is not available. +- **`subject_settings`** (optional): Settings for subject metadata. Only used when `subject_id` is `"calibration"`. + - **`calibration_object`** ([CalibrationObject](https://aind-data-schema.readthedocs.io/en/latest/), optional): A `CalibrationObject` from `aind_data_schema.components.subjects`. When `subject_id` is `"calibration"`, the metadata service is not contacted — instead a `Subject` is constructed locally using this object and an empty `Procedures` (no subject or specimen procedures). If omitted, a default empty `CalibrationObject` is used. + - **`instrument_settings`**: - **`instrument_id`** (str): ID for the instrument used in data collection. When set, the instrument.json will attempt to be fetched from the metadata-service and saved as `instrument_.json`. If multiple `instrument*.json` files exist after fetching they will be merged. @@ -113,6 +116,35 @@ job = GatherMetadataJob(job_settings=job_settings) job.run_job() ``` +#### Calibration sessions + +When collecting data with a calibration object rather than a live subject, set `subject_id` to `"calibration"`. The job will skip the metadata service entirely and construct `subject.json` and `procedures.json` locally. + +```python +from aind_data_schema.components.subjects import CalibrationObject +from aind_data_schema_models.modalities import Modality +from aind_metadata_mapper.gather_metadata import GatherMetadataJob +from aind_metadata_mapper.models import JobSettings, DataDescriptionSettings, SubjectSettings + +job_settings = JobSettings( + output_dir="/path/to/output", + subject_id="calibration", + data_description_settings=DataDescriptionSettings( + project_name="", + modalities=[Modality.ECEPHYS], + ), + subject_settings=SubjectSettings( + calibration_object=CalibrationObject( + description="Neuropixels dummy probe", + empty=False, + ) + ), +) + +job = GatherMetadataJob(job_settings=job_settings) +job.run_job() +``` + #### Validation settings - **`raise_if_invalid`** (bool, default=False): Controls validation behavior: @@ -134,6 +166,30 @@ You probably shouldn't be modifying these. - `metadata_service_procedures_endpoint` (default="/api/v2/procedures/") - `metadata_service_instrument_endpoint` (default="/api/v2/instrument/") +### Instrument CLI + +The `aind-instrument` command lets you upload and retrieve instruments from the metadata service without writing code. + +```bash +# Upload an instrument +aind-instrument upload instrument.json + +# Upload and overwrite an existing record +aind-instrument upload instrument.json --replace + +# Keep the modification date from the file instead of updating to today +aind-instrument upload instrument.json --no-update-modification-date + +# Get the latest record for an instrument +aind-instrument get 422_MESO2_20241017 + +# Get a specific version by modification date +aind-instrument get 422_MESO2_20241017 --modification-date 2024-10-28 + +# Save to a file instead of printing to stdout +aind-instrument get 422_MESO2_20241017 --output-directory ./output +``` + ### Developing Mappers Each MapperJob class should inherit from `BaseMapper` in `base.py`. The only parameter should be the `MapperJobSettings` from `base.py`. You cannot add additional parameters to your job or it will not be possible for it to be run automatically on the data-transfer-service. GatherMetadataJob will then run your mappers automatically when it detects the extracted metadata output. diff --git a/pyproject.toml b/pyproject.toml index 17d2f324..750b8337 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,8 +21,8 @@ dependencies = [ "pyyaml", "pydantic-settings>=2.8.0", "jsonschema", - "aind-data-schema>=2.6.0,<3", - "aind-metadata-extractor==0.3.9", + "aind-data-schema>=2.7.1,<3", + "aind-metadata-extractor==0.3.11", ] [project.optional-dependencies] @@ -40,6 +40,9 @@ doc = [ "furo" ] +[project.scripts] +aind-instrument = "aind_metadata_mapper.cli.instrument:main" + [tool.setuptools.packages.find] where = ["src"] diff --git a/src/aind_metadata_mapper/__init__.py b/src/aind_metadata_mapper/__init__.py index 75c920e5..b512bb2b 100644 --- a/src/aind_metadata_mapper/__init__.py +++ b/src/aind_metadata_mapper/__init__.py @@ -1,3 +1,3 @@ """Init package""" -__version__ = "1.2.0" +__version__ = "1.3.0" diff --git a/src/aind_metadata_mapper/cli/__init__.py b/src/aind_metadata_mapper/cli/__init__.py new file mode 100644 index 00000000..af1209fc --- /dev/null +++ b/src/aind_metadata_mapper/cli/__init__.py @@ -0,0 +1 @@ +"""CLI tools for AIND metadata mapper.""" diff --git a/src/aind_metadata_mapper/cli/instrument.py b/src/aind_metadata_mapper/cli/instrument.py new file mode 100644 index 00000000..22c116bd --- /dev/null +++ b/src/aind_metadata_mapper/cli/instrument.py @@ -0,0 +1,115 @@ +"""CLI for instrument database operations. + +Provides ``upload`` and ``get`` subcommands for working with the AIND +instrument metadata service. + +Usage:: + + aind-instrument upload instrument.json + aind-instrument upload instrument.json --replace + aind-instrument upload instrument.json --no-update-modification-date # keep date from file + aind-instrument get 422_MESO2_20241017 + aind-instrument get 422_MESO2_20241017 --modification-date 2024-10-28 + aind-instrument get 422_MESO2_20241017 --output-directory ./output +""" + +import logging +import sys +from pathlib import Path +from typing import Optional + +from pydantic import BaseModel, Field +from pydantic_settings import CliApp, CliPositionalArg, CliSubCommand + +from aind_metadata_mapper.utils import INSTRUMENT_BASE_URL, get_instrument, save_instrument + +logger = logging.getLogger(__name__) + +BASE_URL_FIELD = Field( + default=INSTRUMENT_BASE_URL, + description="Base URL for the instrument metadata service.", +) + + +class Upload(BaseModel): + """Upload an instrument JSON file to the metadata service database.""" + + file: CliPositionalArg[Path] = Field( + description="Path to an instrument JSON file to upload.", + ) + replace: bool = Field( + default=False, + description=("If set, overwrite an existing record with the same " "instrument_id and modification_date."), + ) + update_modification_date: bool = Field( + default=True, + description=( + "Update modification_date to today's date." + " Use --no-update-modification-date to keep the date from the file." + ), + ) + base_url: str = BASE_URL_FIELD + + def cli_cmd(self) -> None: + """Upload an instrument JSON to the database.""" + try: + save_instrument( + self.file, + replace=self.replace, + update_modification_date=self.update_modification_date, + base_url=self.base_url, + ) + except Exception as e: + logger.error(f"Upload failed: {e}") + sys.exit(1) + + +class Get(BaseModel): + """Retrieve an instrument record from the metadata service database.""" + + instrument_id: CliPositionalArg[str] = Field( + description="Instrument identifier to look up.", + ) + modification_date: Optional[str] = Field( + default=None, + description=("Specific modification date (YYYY-MM-DD) to retrieve. " "If omitted, returns the latest record."), + ) + output_directory: Optional[Path] = Field( + default=None, + description=("Directory to save the instrument JSON file. " "If omitted, prints the JSON to stdout."), + ) + base_url: str = BASE_URL_FIELD + + def cli_cmd(self) -> None: + """Get an instrument record and display or save it.""" + result = get_instrument( + self.instrument_id, + modification_date=self.modification_date, + output_directory=self.output_directory, + base_url=self.base_url, + ) + if result is None: + logger.error(f"Instrument '{self.instrument_id}' not found.") + sys.exit(1) + if self.output_directory is None: + print(result.model_dump_json(indent=3)) + + +class InstrumentCLI(BaseModel): + """CLI for instrument database operations.""" + + upload: CliSubCommand[Upload] + get: CliSubCommand[Get] + + def cli_cmd(self) -> None: + """Dispatch to the active subcommand.""" + CliApp.run_subcommand(self) + + +def main() -> None: + """Entry point for the aind-instrument CLI.""" + CliApp.run(InstrumentCLI) + + +if __name__ == "__main__": + main() diff --git a/src/aind_metadata_mapper/gather_metadata.py b/src/aind_metadata_mapper/gather_metadata.py index 5f0175b8..d1f4140f 100644 --- a/src/aind_metadata_mapper/gather_metadata.py +++ b/src/aind_metadata_mapper/gather_metadata.py @@ -10,6 +10,7 @@ from typing import Any, Dict, Optional from urllib.parse import urljoin +from aind_data_schema.components.subjects import CalibrationObject from aind_data_schema.core.acquisition import Acquisition from aind_data_schema.core.data_description import DataDescription from aind_data_schema.core.instrument import Instrument @@ -246,6 +247,16 @@ def get_subject(self, subject_id: Optional[str] = None) -> Optional[dict]: logging.warning("No subject_id provided.") return None + if subject_id == "calibration": + logging.info("subject_id is 'calibration'; constructing Subject locally.") + calibration_object = ( + self.settings.subject_settings.calibration_object + if self.settings.subject_settings and self.settings.subject_settings.calibration_object + else CalibrationObject(description="", empty=True) + ) + subject = Subject(subject_id=subject_id, subject_details=calibration_object) + return json.loads(subject.model_dump_json()) + if not self._does_file_exist_in_user_defined_dir(file_name=file_name): logging.debug( f"No subject file found in directory. Downloading " @@ -277,21 +288,34 @@ def get_procedures(self, subject_id: str) -> Optional[dict]: logging.warning("No subject_id provided.") return None - if not self._does_file_exist_in_user_defined_dir(file_name=file_name): - logging.debug( - f"No procedures file found in directory. Downloading " - f"{self.settings.subject_id} from " - f"{self.settings.metadata_service_url}" - ) - base_url = urljoin( - self.settings.metadata_service_url, - self.settings.metadata_service_procedures_endpoint, - ) - contents = get_procedures(subject_id, base_url=base_url) + if subject_id == "calibration": + logging.info("subject_id is 'calibration'; constructing empty Procedures locally.") + procedures = Procedures(subject_id=subject_id) + return json.loads(procedures.model_dump_json()) + + base_url = urljoin( + self.settings.metadata_service_url, + self.settings.metadata_service_procedures_endpoint, + ) + service_procedures = get_procedures(subject_id, base_url=base_url) + + user_procedures = None + if self._does_file_exist_in_user_defined_dir(file_name=file_name): + logging.debug(f"Found user-provided {file_name}.") + user_procedures = self._get_file_from_user_defined_directory(file_name=file_name) + + if user_procedures and service_procedures: + logging.info("Merging user-provided and service procedures.") + return self._merge_procedures(user_procedures, service_procedures) + elif user_procedures: + logging.debug(f"Using user-provided {file_name}.") + return user_procedures + elif service_procedures: + logging.debug("Using procedures from metadata service.") + return service_procedures else: - logging.debug(f"Using existing {file_name}.") - contents = self._get_file_from_user_defined_directory(file_name=file_name) - return contents + logging.debug("No procedures metadata found.") + return None def _run_mappers_for_acquisition(self): """ @@ -357,6 +381,81 @@ def _merge_models(self, model_class, models: list[dict]) -> dict: return merged_model.model_dump(mode="json") + def _merge_procedures(self, user_procedures: dict, service_procedures: dict) -> dict: + """Merge user and service procedures, checking for duplicates. + + If duplicates are found, either raises an error or defaults to user procedures + based on raise_if_invalid setting. + + Parameters + ---------- + user_procedures : dict + User-provided procedures + service_procedures : dict + Service-fetched procedures + + Returns + ------- + dict + Merged procedures using the __add__ operator + """ + logging.info("Merging user and service procedures.") + user_proc_obj = Procedures.model_validate(user_procedures) + service_proc_obj = Procedures.model_validate(service_procedures) + + # Check for duplicate subject procedures + duplicate_subject_procs = self._find_duplicate_procedures( + user_proc_obj.subject_procedures or [], + service_proc_obj.subject_procedures or [], + ) + + # Check for duplicate specimen procedures + duplicate_specimen_procs = self._find_duplicate_procedures( + user_proc_obj.specimen_procedures or [], + service_proc_obj.specimen_procedures or [], + ) + + if duplicate_subject_procs or duplicate_specimen_procs: + error_msg = "Found duplicate procedures between user-provided and service procedures" + if duplicate_subject_procs or duplicate_specimen_procs: + error_msg += f"\n Duplicate procedures: {len(duplicate_subject_procs) + len(duplicate_specimen_procs)}" + + if self.settings.raise_if_invalid: + raise ValueError(error_msg) + else: + logging.warning(error_msg) + logging.info("Defaulting to user-provided procedures") + return user_procedures + + # No duplicates, merge using the __add__ operator + merged_procedures = user_proc_obj + service_proc_obj + return merged_procedures.model_dump(mode="json") + + def _find_duplicate_procedures(self, procedures_1: list, procedures_2: list) -> list: + """Find procedures that appear in both lists by comparing objects. + + Parameters + ---------- + procedures_1 : list + First list of procedures + procedures_2 : list + Second list of procedures + + Returns + ------- + list + List of procedures that appear in both lists + """ + duplicates = [] + for proc_1 in procedures_1: + proc_1_dict = proc_1.model_dump(mode="json") + for proc_2 in procedures_2: + proc_2_dict = proc_2.model_dump(mode="json") + if proc_1_dict == proc_2_dict: + duplicates.append(proc_1) + break + return duplicates + def get_instrument_from_service(self) -> Optional[dict]: """Get instrument metadata from service and write to the metadata directory""" if self.settings.instrument_settings and self.settings.instrument_settings.instrument_id: diff --git a/src/aind_metadata_mapper/models.py b/src/aind_metadata_mapper/models.py index 1f691507..6ba9100e 100644 --- a/src/aind_metadata_mapper/models.py +++ b/src/aind_metadata_mapper/models.py @@ -3,6 +3,7 @@ from typing import List, Optional from aind_data_schema.base import AwareDatetimeWithDefault +from aind_data_schema.components.subjects import CalibrationObject from aind_data_schema_models.data_name_patterns import Group from aind_data_schema_models.modalities import Modality from pydantic import Field, field_validator @@ -47,6 +48,18 @@ def convert_modalities_from_string(cls, v): return v +class SubjectSettings(BaseSettings): + """Settings specific to subject metadata""" + + calibration_object: Optional[CalibrationObject] = Field( + default=None, + description=( + "Optional calibration object. Used when subject_id is 'calibration' to construct " + "a Subject without contacting the metadata service." + ), + ) + + class InstrumentSettings(BaseSettings): """Settings specific to instrument metadata""" @@ -111,6 +124,10 @@ class JobSettings(BaseSettings, cli_parse_args=True, cli_ignore_unknown_args=Tru ..., description="Settings specific to data description metadata.", ) + subject_settings: Optional[SubjectSettings] = Field( + default=None, + description="Settings specific to subject metadata.", + ) instrument_settings: Optional[InstrumentSettings] = Field( default=None, description="Settings specific to instrument metadata.", diff --git a/src/aind_metadata_mapper/utils.py b/src/aind_metadata_mapper/utils.py index 75939f1c..0528c47e 100644 --- a/src/aind_metadata_mapper/utils.py +++ b/src/aind_metadata_mapper/utils.py @@ -349,6 +349,7 @@ def save_instrument( instrument_model: instrument.Instrument | dict | str | Path, replace: bool = False, update_modification_date: bool = True, + base_url: str = INSTRUMENT_BASE_URL, ) -> None: """Save instrument and validate round-trip. @@ -365,6 +366,8 @@ def save_instrument( update_modification_date : bool If True (default), set modification_date to today (YYYY-MM-DD). If False, keep the modification date as passed in the instrument. + base_url : str + Base URL for the instrument endpoint. Defaults to INSTRUMENT_BASE_URL. Raises ------ @@ -386,9 +389,9 @@ def save_instrument( # Use model_dump_json() and parse to ensure dates are properly serialized source_dict = json.loads(instrument_model.model_dump_json()) - logger.info(f"POSTing instrument to {INSTRUMENT_BASE_URL}") + logger.info(f"POSTing instrument to {base_url}") params = {"replace": "true"} if replace else {} - response = requests.post(INSTRUMENT_BASE_URL, json=source_dict, params=params) + response = requests.post(base_url, json=source_dict, params=params) # POST 400 is always an error (e.g., "Record already exists") if response.status_code == 400: error_msg = response.json().get("message", response.text) @@ -398,10 +401,9 @@ def save_instrument( # GET back and validate round-trip logger.info( - f"GETting instrument from {INSTRUMENT_BASE_URL}/{instrument_model.instrument_id} " - "to verify that save was successful" + f"GETting instrument from {base_url}/{instrument_model.instrument_id} " "to verify that save was successful" ) - read_back_instrument = get_instrument(instrument_model.instrument_id) + read_back_instrument = get_instrument(instrument_model.instrument_id, base_url=base_url) if read_back_instrument is None: raise ValueError(f"Instrument '{instrument_model.instrument_id}' not found in database") diff --git a/tests/cli/__init__.py b/tests/cli/__init__.py new file mode 100644 index 00000000..f3462c30 --- /dev/null +++ b/tests/cli/__init__.py @@ -0,0 +1 @@ +"""Tests for CLI module.""" diff --git a/tests/cli/test_instrument.py b/tests/cli/test_instrument.py new file mode 100644 index 00000000..a3b24da2 --- /dev/null +++ b/tests/cli/test_instrument.py @@ -0,0 +1,178 @@ +"""Tests for instrument CLI commands.""" + +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import aind_data_schema.core.instrument as instrument + +from aind_metadata_mapper.cli.instrument import Get, Upload +from aind_metadata_mapper.utils import INSTRUMENT_BASE_URL + +INSTRUMENT_JSON = Path(__file__).parent.parent / "resources" / "v2_metadata" / "instrument.json" + + +class TestUploadCmd(unittest.TestCase): + """Tests for the Upload subcommand.""" + + @patch("aind_metadata_mapper.cli.instrument.save_instrument") + def test_upload_default_flags(self, mock_save): + """Upload passes file path and default flags to save_instrument.""" + cmd = Upload(file=INSTRUMENT_JSON) + cmd.cli_cmd() + mock_save.assert_called_once_with( + INSTRUMENT_JSON, + replace=False, + update_modification_date=True, + base_url=INSTRUMENT_BASE_URL, + ) + + @patch("aind_metadata_mapper.cli.instrument.save_instrument") + def test_upload_with_replace(self, mock_save): + """Upload passes replace=True when flag is set.""" + cmd = Upload(file=INSTRUMENT_JSON, replace=True) + cmd.cli_cmd() + mock_save.assert_called_once_with( + INSTRUMENT_JSON, + replace=True, + update_modification_date=True, + base_url=INSTRUMENT_BASE_URL, + ) + + @patch("aind_metadata_mapper.cli.instrument.save_instrument") + def test_upload_no_update_modification_date(self, mock_save): + """Upload passes update_modification_date=False when flag is set.""" + cmd = Upload(file=INSTRUMENT_JSON, update_modification_date=False) + cmd.cli_cmd() + mock_save.assert_called_once_with( + INSTRUMENT_JSON, + replace=False, + update_modification_date=False, + base_url=INSTRUMENT_BASE_URL, + ) + + @patch("aind_metadata_mapper.cli.instrument.save_instrument") + def test_upload_exits_on_error(self, mock_save): + """Upload exits with code 1 when save_instrument raises.""" + mock_save.side_effect = ValueError("Record already exists") + cmd = Upload(file=INSTRUMENT_JSON) + with self.assertRaises(SystemExit) as cm: + cmd.cli_cmd() + self.assertEqual(cm.exception.code, 1) + + +class TestGetCmd(unittest.TestCase): + """Tests for the Get subcommand.""" + + @patch("aind_metadata_mapper.cli.instrument.get_instrument") + def test_get_prints_json_to_stdout(self, mock_get): + """Get prints instrument JSON to stdout when no output_directory.""" + with open(INSTRUMENT_JSON) as f: + data = json.load(f) + mock_get.return_value = instrument.Instrument.model_validate(data) + cmd = Get(instrument_id="422_MESO2_20241017") + with patch("builtins.print") as mock_print: + cmd.cli_cmd() + mock_get.assert_called_once_with( + "422_MESO2_20241017", + modification_date=None, + output_directory=None, + base_url=INSTRUMENT_BASE_URL, + ) + mock_print.assert_called_once() + # Verify it's valid JSON + json.loads(mock_print.call_args[0][0]) + + @patch("aind_metadata_mapper.cli.instrument.get_instrument") + def test_get_with_modification_date(self, mock_get): + """Get passes modification_date to get_instrument.""" + with open(INSTRUMENT_JSON) as f: + data = json.load(f) + mock_get.return_value = instrument.Instrument.model_validate(data) + cmd = Get(instrument_id="422_MESO2_20241017", modification_date="2024-10-28") + with patch("builtins.print"): + cmd.cli_cmd() + mock_get.assert_called_once_with( + "422_MESO2_20241017", + modification_date="2024-10-28", + output_directory=None, + base_url=INSTRUMENT_BASE_URL, + ) + + @patch("aind_metadata_mapper.cli.instrument.get_instrument") + def test_get_with_output_directory(self, mock_get): + """Get passes output_directory and does not print to stdout.""" + with open(INSTRUMENT_JSON) as f: + data = json.load(f) + mock_get.return_value = instrument.Instrument.model_validate(data) + with tempfile.TemporaryDirectory() as tmpdir: + cmd = Get(instrument_id="422_MESO2_20241017", output_directory=Path(tmpdir)) + with patch("builtins.print") as mock_print: + cmd.cli_cmd() + mock_get.assert_called_once_with( + "422_MESO2_20241017", + modification_date=None, + output_directory=Path(tmpdir), + base_url=INSTRUMENT_BASE_URL, + ) + mock_print.assert_not_called() + + @patch("aind_metadata_mapper.cli.instrument.get_instrument") + def test_get_exits_when_not_found(self, mock_get): + """Get exits with code 1 when instrument is not found.""" + mock_get.return_value = None + cmd = Get(instrument_id="nonexistent") + with self.assertRaises(SystemExit) as cm: + cmd.cli_cmd() + self.assertEqual(cm.exception.code, 1) + + +class TestInstrumentCLIDispatch(unittest.TestCase): + """Tests for CLI argument parsing and subcommand dispatch.""" + + @patch("aind_metadata_mapper.cli.instrument.save_instrument") + def test_cli_dispatches_upload(self, mock_save): + """CliApp.run dispatches upload subcommand from sys.argv.""" + from pydantic_settings import CliApp + + from aind_metadata_mapper.cli.instrument import InstrumentCLI + + cli = CliApp.run( + InstrumentCLI, + cli_args=["upload", str(INSTRUMENT_JSON)], + ) + mock_save.assert_called_once() + self.assertIsNotNone(cli.upload) + + @patch("aind_metadata_mapper.cli.instrument.CliApp") + def test_main_calls_cli_app_run(self, mock_cli_app): + """main() calls CliApp.run with InstrumentCLI.""" + from aind_metadata_mapper.cli.instrument import InstrumentCLI, main + + main() + mock_cli_app.run.assert_called_once_with(InstrumentCLI) + + @patch("aind_metadata_mapper.cli.instrument.get_instrument") + def test_cli_dispatches_get(self, mock_get): + """CliApp.run dispatches get subcommand from sys.argv.""" + with open(INSTRUMENT_JSON) as f: + data = json.load(f) + mock_get.return_value = instrument.Instrument.model_validate(data) + + from pydantic_settings import CliApp + + from aind_metadata_mapper.cli.instrument import InstrumentCLI + + with patch("builtins.print"): + cli = CliApp.run( + InstrumentCLI, + cli_args=["get", "422_MESO2_20241017"], + ) + mock_get.assert_called_once() + self.assertIsNotNone(cli.get) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gather_metadata.py b/tests/test_gather_metadata.py index caabf48b..496afaeb 100644 --- a/tests/test_gather_metadata.py +++ b/tests/test_gather_metadata.py @@ -12,12 +12,14 @@ from pathlib import Path from unittest.mock import MagicMock, mock_open, patch +from aind_data_schema.components.subjects import CalibrationObject from aind_data_schema.core.acquisition import Acquisition +from aind_data_schema.core.procedures import Procedures from aind_data_schema_models.modalities import Modality from aind_data_schema_models.organizations import Organization from aind_metadata_mapper.gather_metadata import GatherMetadataJob -from aind_metadata_mapper.models import DataDescriptionSettings, JobSettings +from aind_metadata_mapper.models import DataDescriptionSettings, JobSettings, SubjectSettings TEST_DIR = Path(os.path.dirname(os.path.realpath(__file__))) @@ -607,6 +609,52 @@ def test_get_subject_api_exception(self, mock_get_subject, mock_file_exists): self.assertIsNone(result) + @patch("os.makedirs") + def test_get_subject_calibration_with_object(self, mock_makedirs): + """Test get_subject when subject_id is 'calibration' with CalibrationObject""" + calibration_obj = CalibrationObject(description="Test calibration", empty=False) + job_settings = JobSettings( + metadata_dir="/test", + output_dir="/test/output", + subject_id="calibration", + data_description_settings=DataDescriptionSettings( + project_name="Test Project", + modalities=[Modality.ECEPHYS], + ), + subject_settings=SubjectSettings(calibration_object=calibration_obj), + acquisition_start_time=datetime(2023, 1, 1, 12, 0, 0), + ) + job = GatherMetadataJob(settings=job_settings) + + result = job.get_subject(subject_id="calibration") + + self.assertIsNotNone(result) + self.assertEqual(result["subject_id"], "calibration") + self.assertEqual(result["subject_details"]["description"], "Test calibration") + self.assertEqual(result["subject_details"]["empty"], False) + + @patch("os.makedirs") + def test_get_subject_calibration_without_object(self, mock_makedirs): + """Test get_subject when subject_id is 'calibration' without CalibrationObject""" + job_settings = JobSettings( + metadata_dir="/test", + output_dir="/test/output", + subject_id="calibration", + data_description_settings=DataDescriptionSettings( + project_name="Test Project", + modalities=[Modality.ECEPHYS], + ), + acquisition_start_time=datetime(2023, 1, 1, 12, 0, 0), + ) + job = GatherMetadataJob(settings=job_settings) + + result = job.get_subject(subject_id="calibration") + + self.assertIsNotNone(result) + self.assertEqual(result["subject_id"], "calibration") + self.assertEqual(result["subject_details"]["empty"], True) + self.assertEqual(result["subject_details"]["description"], "") + # Tests for get_procedures method @patch.object(GatherMetadataJob, "_does_file_exist_in_user_defined_dir") @patch("os.makedirs") @@ -688,6 +736,28 @@ def test_get_procedures_api_exception(self, mock_get_procedures, mock_file_exist self.assertIsNone(result) + @patch("os.makedirs") + def test_get_procedures_calibration(self, mock_makedirs): + """Test get_procedures when subject_id is 'calibration'""" + job_settings = JobSettings( + metadata_dir="/test", + output_dir="/test/output", + subject_id="calibration", + data_description_settings=DataDescriptionSettings( + project_name="Test Project", + modalities=[Modality.ECEPHYS], + ), + acquisition_start_time=datetime(2023, 1, 1, 12, 0, 0), + ) + job = GatherMetadataJob(settings=job_settings) + + result = job.get_procedures(subject_id="calibration") + + self.assertIsNotNone(result) + self.assertEqual(result["subject_id"], "calibration") + self.assertEqual(result["subject_procedures"], []) + self.assertEqual(result["specimen_procedures"], []) + # Tests for other metadata getter methods @patch.object(GatherMetadataJob, "_does_file_exist_in_user_defined_dir") def test_get_acquisition_no_file(self, mock_file_exists): @@ -1172,6 +1242,158 @@ def test_copy_original_metadata_files_copy_error(self, mock_copy, mock_makedirs, self.job.copy_original_metadata_files() mock_warning.assert_called_once_with("Failed to copy subject.json: Permission denied") + # Tests for procedures merging + @patch.object(GatherMetadataJob, "_does_file_exist_in_user_defined_dir") + @patch("aind_metadata_mapper.gather_metadata.get_procedures") + def test_get_procedures_from_service_only(self, mock_get_procedures, mock_file_exists): + """Test get_procedures when only service procedures are available""" + mock_file_exists.return_value = False + mock_get_procedures.return_value = {"subject_id": "123456", "subject_procedures": []} + + result = self.job.get_procedures(subject_id="123456") + + self.assertIsNotNone(result) + self.assertEqual(result["subject_id"], "123456") + mock_get_procedures.assert_called_once_with("123456", base_url="http://test-service.com/api/v2/procedures/") + + @patch.object(GatherMetadataJob, "_does_file_exist_in_user_defined_dir") + @patch.object(GatherMetadataJob, "_get_file_from_user_defined_directory") + @patch("aind_metadata_mapper.gather_metadata.get_procedures") + def test_get_procedures_from_user_only(self, mock_get_procedures, mock_get_file, mock_file_exists): + """Test get_procedures when only user procedures are available""" + mock_file_exists.return_value = True + mock_get_file.return_value = {"subject_id": "123456", "subject_procedures": []} + mock_get_procedures.return_value = None + + result = self.job.get_procedures(subject_id="123456") + + self.assertIsNotNone(result) + self.assertEqual(result["subject_id"], "123456") + mock_file_exists.assert_called_once_with(file_name="procedures.json") + + @patch.object(GatherMetadataJob, "_does_file_exist_in_user_defined_dir") + @patch.object(GatherMetadataJob, "_get_file_from_user_defined_directory") + @patch("aind_metadata_mapper.gather_metadata.get_procedures") + @patch.object(GatherMetadataJob, "_merge_procedures") + def test_get_procedures_merge_both_available( + self, mock_merge, mock_get_procedures, mock_get_file, mock_file_exists + ): + """Test get_procedures when both user and service procedures are available""" + mock_file_exists.return_value = True + user_procs = {"subject_id": "123456", "subject_procedures": []} + service_procs = {"subject_id": "123456", "subject_procedures": []} + mock_get_file.return_value = user_procs + mock_get_procedures.return_value = service_procs + mock_merge.return_value = user_procs + + result = self.job.get_procedures(subject_id="123456") + + self.assertIsNotNone(result) + mock_merge.assert_called_once_with(user_procs, service_procs) + + def test_find_duplicate_procedures_no_duplicates(self): + """Test _find_duplicate_procedures when no duplicates exist""" + with open(TEST_DIR / "resources" / "v2_metadata" / "procedures.json") as f: + base_procedures = json.load(f) + + procedures_obj = Procedures.model_validate(base_procedures) + self.assertTrue(procedures_obj.subject_procedures, "Test fixture must have at least one procedure") + + proc1 = procedures_obj.subject_procedures[0] + # Create a second different procedure by copying and modifying the first + proc2_dict = json.loads(proc1.model_dump_json()) + proc2_dict["start_date"] = "2025-07-11" # Make it different from proc1 + proc2 = type(proc1).model_validate(proc2_dict) + + duplicates = self.job._find_duplicate_procedures([proc1], [proc2]) + self.assertEqual(len(duplicates), 0) + + def test_find_duplicate_procedures_with_duplicates(self): + """Test _find_duplicate_procedures when duplicates exist""" + with open(TEST_DIR / "resources" / "v2_metadata" / "procedures.json") as f: + base_procedures = json.load(f) + + procedures_obj = Procedures.model_validate(base_procedures) + self.assertTrue(procedures_obj.subject_procedures, "Test fixture must have at least one procedure") + + # Use same procedure in both lists to create a duplicate + proc = procedures_obj.subject_procedures[0] + duplicates = self.job._find_duplicate_procedures([proc], [proc]) + self.assertEqual(len(duplicates), 1) + + @patch("os.makedirs") + def test_merge_procedures_no_duplicates(self, mock_makedirs): + """Test _merge_procedures with no duplicates merges successfully""" + with open(TEST_DIR / "resources" / "v2_metadata" / "procedures.json") as f: + base_procedures = json.load(f) + + user_procedures = base_procedures.copy() + service_procedures = base_procedures.copy() + + # Remove all procedures from service to avoid duplicates + service_procedures["subject_procedures"] = [] + service_procedures["specimen_procedures"] = [] + + result = self.job._merge_procedures(user_procedures, service_procedures) + + self.assertIsNotNone(result) + self.assertIsInstance(result, dict) + + @patch("os.makedirs") + def test_merge_procedures_with_duplicates_raises(self, mock_makedirs): + """Test _merge_procedures with duplicates raises error when raise_if_invalid is True""" + strict_settings = JobSettings( + metadata_dir="/test/metadata", + output_dir="/test/output", + subject_id="123456", + data_description_settings=DataDescriptionSettings( + project_name="Test Project", + modalities=[Modality.ECEPHYS], + ), + acquisition_start_time=datetime(2023, 1, 1, 12, 0, 0), + raise_if_invalid=True, + ) + strict_job = GatherMetadataJob(settings=strict_settings) + + with open(TEST_DIR / "resources" / "v2_metadata" / "procedures.json") as f: + base_procedures = json.load(f) + + user_procedures = base_procedures.copy() + service_procedures = base_procedures.copy() + + with self.assertRaises(ValueError) as context: + strict_job._merge_procedures(user_procedures, service_procedures) + + self.assertIn("duplicate procedures", str(context.exception).lower()) + + @patch("os.makedirs") + def test_merge_procedures_with_duplicates_defaults_to_user(self, mock_makedirs): + """Test _merge_procedures with duplicates defaults to user procedures when raise_if_invalid is False""" + lenient_settings = JobSettings( + metadata_dir="/test/metadata", + output_dir="/test/output", + subject_id="123456", + data_description_settings=DataDescriptionSettings( + project_name="Test Project", + modalities=[Modality.ECEPHYS], + ), + acquisition_start_time=datetime(2023, 1, 1, 12, 0, 0), + raise_if_invalid=False, + ) + lenient_job = GatherMetadataJob(settings=lenient_settings) + + with open(TEST_DIR / "resources" / "v2_metadata" / "procedures.json") as f: + base_procedures = json.load(f) + + user_procedures = base_procedures.copy() + service_procedures = base_procedures.copy() + + with patch("logging.warning"): + result = lenient_job._merge_procedures(user_procedures, service_procedures) + + # Should return user procedures + self.assertEqual(result, user_procedures) + if __name__ == "__main__": unittest.main()