Source code for allensdk.internal.brain_observatory.mouse

from typing import Optional, List

from allensdk.brain_observatory.behavior.data_objects.metadata\
    .behavior_metadata.behavior_metadata import \
    BehaviorMetadata
from allensdk.brain_observatory.behavior.data_objects.metadata\
    .subject_metadata.mouse_id import \
    MouseId
from allensdk.core.auth_config import LIMS_DB_CREDENTIAL_MAP
from allensdk.internal.api import db_connection_creator
from allensdk.internal.brain_observatory.util.multi_session_utils import \
    get_session_metadata_multiprocessing, get_images_shown, \
    remove_invalid_sessions




class Mouse:
    """A mouse"""
    def __init__(self, mouse_id: str):
        self._mouse_id = mouse_id

    @property
    def mouse_id(self) -> str:
        return self._mouse_id



    def get_behavior_sessions(
            self,
            exclude_invalid_sessions: bool = True
    ) -> List[BehaviorMetadata]:
        """
        Gets all behavior sessions for mouse

        Parameters
        ----------
        exclude_invalid_sessions:
            Whether to exclude invalid sessions

        Returns
        -------
        List[BehaviorMetadata]
        """
        lims_db = db_connection_creator(
            fallback_credentials=LIMS_DB_CREDENTIAL_MAP
        )

        query = f"""
            SELECT bs.id
            FROM behavior_sessions bs
            JOIN donors on donors.id = bs.donor_id
            WHERE external_donor_name = '{self.mouse_id}'
        """
        behavior_session_ids = lims_db.fetchall(query=query)
        behavior_sessions = get_session_metadata_multiprocessing(
            behavior_session_ids=behavior_session_ids,
            lims_engine=lims_db
        )
        if exclude_invalid_sessions:
            behavior_sessions = remove_invalid_sessions(
                behavior_sessions=behavior_sessions
            )
        return behavior_sessions




    @classmethod
    def from_behavior_session_id(
            cls,
            behavior_session_id: int
    ) -> "Mouse":
        """Instantiates `Mouse` from `behavior_session_id`"""
        lims_db = db_connection_creator(
            fallback_credentials=LIMS_DB_CREDENTIAL_MAP
        )
        mouse_id = MouseId.from_lims(
            behavior_session_id=behavior_session_id,
            lims_db=lims_db
        )
        return Mouse(mouse_id=mouse_id.value)




    def get_images_shown(
            self,
            up_to_behavior_session_id: Optional[int] = None,
            n_workers: Optional[int] = None
    ):
        """Gets all images presented to mouse up to (not including)
        `up_to_behavior_session_id` if provided

        Parameters
        ----------
        up_to_behavior_session_id
            Filters stimulus presentations to all those up to (not including)
            this behavior session, if provided
        n_workers
            Number of processes to spawn for reading image names from
            stimulus files
        """
        lims_db = db_connection_creator(
            fallback_credentials=LIMS_DB_CREDENTIAL_MAP
        )

        behavior_sessions = self.get_behavior_sessions()

        if up_to_behavior_session_id is not None:
            this_date_of_acquisition = [
                x.date_of_acquisition for x in behavior_sessions
                if x.behavior_session_id == up_to_behavior_session_id][0]

            prior_behavior_session_ids = set([
                x.behavior_session_id for x in behavior_sessions
                if x.date_of_acquisition < this_date_of_acquisition])
            behavior_sessions = [
                x for x in behavior_sessions
                if x.behavior_session_id in prior_behavior_session_ids]
        images_shown = get_images_shown(
            behavior_session_ids=(
                [x.behavior_session_id for x in behavior_sessions]),
            lims_engine=lims_db,
            n_workers=n_workers
        )
        return images_shown