Merge pull request #122 from AllenNeuralDynamics/release-v0.16.0

Release v0.16.0

pyproject.toml

@@ -24,7 +24,7 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
-    "aind-metadata-mapper[all] ; python_version >= '3.9'",
+    "aind-metadata-mapper[all]",
     "black",
     "coverage",
     "flake8",
@@ -48,19 +48,19 @@ bergamo = [
 ]
 
 bruker = [
-    "bruker2nifti==1.0.4"
+    "bruker2nifti==1.0.4",
 ]
 
 mesoscope = [
     "aind-metadata-mapper[bergamo]",
     "pillow",
-    "tifffile==2024.2.12",
+    "tifffile==2024.2.12 ; python_version >= '3.9'",
 ]
 
 openephys = [
     "h5py",
     "np_session",
-    "npc_ephys",
+    "npc_ephys ; python_version >= '3.9'",
     "scipy",
     "pandas",
     "numpy",
@@ -98,7 +98,7 @@ exclude = '''
 '''
 
 [tool.coverage.run]
-omit = ["*__init__*"]
+omit = ["*__init__*", "tests/integration/*"]
 source = ["aind_metadata_mapper", "tests"]
 
 [tool.coverage.report]

src/aind_metadata_mapper/__init__.py

@@ -1,3 +1,3 @@
 """Init package"""
 
-__version__ = "0.15.0"
+__version__ = "0.16.0"

src/aind_metadata_mapper/bergamo/models.py

@@ -1,13 +1,15 @@
 """Module defining JobSettings for Bergamo ETL"""
+
 from decimal import Decimal
 from pathlib import Path
 from typing import List, Literal, Optional
 
 from pydantic import Field
-from pydantic_settings import BaseSettings
+
+from aind_metadata_mapper.core import BaseJobSettings
 
 
-class JobSettings(BaseSettings):
+class JobSettings(BaseJobSettings):
     """Data that needs to be input by user. Can be pulled from env vars with
     BERGAMO prefix or set explicitly."""
 

src/aind_metadata_mapper/bergamo/session.py

@@ -1,4 +1,5 @@
 """Module to map bergamo metadata into a session model."""
+
 import argparse
 import bisect
 import json

src/aind_metadata_mapper/bruker/models.py

@@ -8,10 +8,11 @@
     ScannerLocation,
 )
 from pydantic import Field
-from pydantic_settings import BaseSettings
 
+from aind_metadata_mapper.core import BaseJobSettings
 
-class JobSettings(BaseSettings):
+
+class JobSettings(BaseJobSettings):
     """Data that needs to be input by user."""
 
     job_settings_name: Literal["Bruker"] = "Bruker"

src/aind_metadata_mapper/core.py

@@ -1,20 +1,178 @@
 """Core abstract class that can be used as a template for etl jobs."""
 
 import argparse
+import json
 import logging
 from abc import ABC, abstractmethod
+from functools import cached_property
 from os import PathLike
 from pathlib import Path
-from typing import Any, Generic, Optional, TypeVar, Union
+from typing import Any, Dict, Generic, Optional, Tuple, Type, TypeVar, Union
 
 from aind_data_schema.base import AindCoreModel
-from pydantic import ValidationError
-from pydantic_settings import BaseSettings
-from aind_metadata_mapper.models import JobResponse
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+from pydantic.fields import FieldInfo
+from pydantic_settings import (
+    BaseSettings,
+    EnvSettingsSource,
+    InitSettingsSource,
+    PydanticBaseSettingsSource,
+)
 
 _T = TypeVar("_T", bound=BaseSettings)
 
 
+class JobResponse(BaseModel):
+    """Standard model of a JobResponse."""
+
+    model_config = ConfigDict(extra="forbid")
+    status_code: int
+    message: Optional[str] = Field(None)
+    data: Optional[str] = Field(None)
+
+
+class JsonConfigSettingsSource(PydanticBaseSettingsSource):
+    """Base class for settings that parse JSON from various sources."""
+
+    def __init__(self, settings_cls, config_file_location: Path):
+        """Class constructor."""
+        self.config_file_location = config_file_location
+        super().__init__(settings_cls)
+
+    def _retrieve_contents(self) -> Dict[str, Any]:
+        """Retrieve and parse the JSON contents from the config file."""
+        try:
+            with open(self.config_file_location, "r") as f:
+                return json.load(f)
+        except (json.JSONDecodeError, IOError) as e:
+            logging.warning(
+                f"Error loading config from {self.config_file_location}: {e}"
+            )
+            raise e
+
+    @cached_property
+    def _json_contents(self):
+        """Cache contents to a property to avoid re-downloading."""
+        contents = self._retrieve_contents()
+        return contents
+
+    def get_field_value(
+        self, field: FieldInfo, field_name: str
+    ) -> Tuple[Any, str, bool]:
+        """
+        Gets the value, the key for model creation, and a flag to determine
+        whether value is complex.
+        Parameters
+        ----------
+        field : FieldInfo
+          The field
+        field_name : str
+          The field name
+
+        Returns
+        -------
+        Tuple[Any, str, bool]
+          A tuple contains the key, value and a flag to determine whether
+          value is complex.
+
+        """
+        file_content_json = self._json_contents
+        field_value = file_content_json.get(field_name)
+        return field_value, field_name, False
+
+    def prepare_field_value(
+        self,
+        field_name: str,
+        field: FieldInfo,
+        value: Any,
+        value_is_complex: bool,
+    ) -> Any:
+        """
+        Prepares the value of a field.
+        Parameters
+        ----------
+        field_name : str
+          The field name
+        field : FieldInfo
+          The field
+        value : Any
+          The value of the field that has to be prepared
+        value_is_complex : bool
+          A flag to determine whether value is complex
+
+        Returns
+        -------
+        Any
+          The prepared value
+
+        """
+        return value
+
+    def __call__(self) -> Dict[str, Any]:
+        """
+        Run this when this class is called. Required to be implemented.
+
+        Returns
+        -------
+        Dict[str, Any]
+          The fields for the settings defined as a dict object.
+
+        """
+        d: Dict[str, Any] = {}
+
+        for field_name, field in self.settings_cls.model_fields.items():
+            field_value, field_key, value_is_complex = self.get_field_value(
+                field, field_name
+            )
+            field_value = self.prepare_field_value(
+                field_name, field, field_value, value_is_complex
+            )
+            if field_value is not None:
+                d[field_key] = field_value
+
+        return d
+
+
+class BaseJobSettings(BaseSettings):
+    """Parent class for generating settings from a config file."""
+
+    user_settings_config_file: Optional[Path] = Field(
+        default=None,
+        repr=False,
+        description="Optionally pull settings from a local config file.",
+    )
+
+    class Config:
+        """Pydantic config to exclude field from displaying"""
+
+        exclude = {"user_settings_config_file"}
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: Type[BaseSettings],
+        init_settings: InitSettingsSource,
+        env_settings: EnvSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> Tuple[PydanticBaseSettingsSource, ...]:
+        """
+        Customize the order of settings sources, including JSON file.
+        """
+        config_file = init_settings.init_kwargs.get(
+            "user_settings_config_file"
+        )
+        sources = [init_settings, env_settings]
+
+        if isinstance(config_file, str):
+            config_file = Path(config_file)
+
+        if config_file and config_file.is_file():
+            sources.append(JsonConfigSettingsSource(settings_cls, config_file))
+
+        return tuple(sources)
+
+
 class GenericEtl(ABC, Generic[_T]):
     """A generic etl class. Child classes will need to create a JobSettings
     object that is json serializable. Child class will also need to implement

src/aind_metadata_mapper/fip/models.py

@@ -5,10 +5,11 @@
 from typing import List, Literal, Optional
 
 from pydantic import Field
-from pydantic_settings import BaseSettings
 
+from aind_metadata_mapper.core import BaseJobSettings
 
-class JobSettings(BaseSettings):
+
+class JobSettings(BaseJobSettings):
     """Data that needs to be input by user."""
 
     job_settings_name: Literal["FIP"] = "FIP"

src/aind_metadata_mapper/gather_metadata.py

@@ -37,8 +37,8 @@
 )
 from aind_metadata_mapper.fip.session import FIBEtl
 from aind_metadata_mapper.mesoscope.session import MesoscopeEtl
-from aind_metadata_mapper.smartspim.acquisition import SmartspimETL
 from aind_metadata_mapper.models import JobSettings
+from aind_metadata_mapper.smartspim.acquisition import SmartspimETL
 
 
 class GatherMetadataJob:

src/aind_metadata_mapper/mesoscope/models.py

@@ -5,10 +5,11 @@
 from typing import List, Literal
 
 from pydantic import Field
-from pydantic_settings import BaseSettings
 
+from aind_metadata_mapper.core import BaseJobSettings
 
-class JobSettings(BaseSettings):
+
+class JobSettings(BaseJobSettings):
     """Data to be entered by the user."""
 
     job_settings_name: Literal["Mesoscope"] = "Mesoscope"

src/aind_metadata_mapper/models.py

@@ -2,21 +2,20 @@
 
 from pathlib import Path
 from typing import List, Optional, Union
-from typing_extensions import Annotated
 
 from aind_data_schema.core.processing import PipelineProcess
 from aind_data_schema_models.modalities import Modality
 from aind_data_schema_models.organizations import Organization
-
-from pydantic import Field, BaseModel, ConfigDict
-from pydantic_settings import BaseSettings
+from pydantic import BaseModel, ConfigDict, Field
+from typing_extensions import Annotated
 
 from aind_metadata_mapper.bergamo.models import (
     JobSettings as BergamoSessionJobSettings,
 )
 from aind_metadata_mapper.bruker.models import (
     JobSettings as BrukerSessionJobSettings,
 )
+from aind_metadata_mapper.core import BaseJobSettings
 from aind_metadata_mapper.fip.models import (
     JobSettings as FipSessionJobSettings,
 )
@@ -37,7 +36,7 @@ class JobResponse(BaseModel):
     data: Optional[str] = Field(None)
 
 
-class SessionSettings(BaseSettings):
+class SessionSettings(BaseJobSettings):
     """Settings needed to retrieve session metadata"""
 
     job_settings: Annotated[
@@ -51,29 +50,29 @@ class SessionSettings(BaseSettings):
     ]
 
 
-class AcquisitionSettings(BaseSettings):
+class AcquisitionSettings(BaseJobSettings):
     """Fields needed to retrieve acquisition metadata"""
 
     # TODO: we can change this to a tagged union once more acquisition settings
     #  are added
     job_settings: SmartSpimAcquisitionJobSettings
 
 
-class SubjectSettings(BaseSettings):
+class SubjectSettings(BaseJobSettings):
     """Fields needed to retrieve subject metadata"""
 
     subject_id: str
     metadata_service_path: str = "subject"
 
 
-class ProceduresSettings(BaseSettings):
+class ProceduresSettings(BaseJobSettings):
     """Fields needed to retrieve procedures metadata"""
 
     subject_id: str
     metadata_service_path: str = "procedures"
 
 
-class RawDataDescriptionSettings(BaseSettings):
+class RawDataDescriptionSettings(BaseJobSettings):
     """Fields needed to retrieve data description metadata"""
 
     name: str
@@ -83,13 +82,13 @@ class RawDataDescriptionSettings(BaseSettings):
     metadata_service_path: str = "funding"
 
 
-class ProcessingSettings(BaseSettings):
+class ProcessingSettings(BaseJobSettings):
     """Fields needed to retrieve processing metadata"""
 
     pipeline_process: PipelineProcess
 
 
-class MetadataSettings(BaseSettings):
+class MetadataSettings(BaseJobSettings):
     """Fields needed to retrieve main Metadata"""
 
     name: str
@@ -104,7 +103,7 @@ class MetadataSettings(BaseSettings):
     instrument_filepath: Optional[Path] = None
 
 
-class JobSettings(BaseSettings):
+class JobSettings(BaseJobSettings):
     """Fields needed to gather all metadata"""
 
     metadata_service_domain: Optional[str] = None