Source code for allensdk.internal.api.queries.ecephys_lims_queries

from typing import List
import pandas as pd
import numpy as np
from allensdk.internal.api import PostgresQueryMixin
from allensdk.internal.api.queries.utils import build_in_list_selector_query


def donor_id_lookup_from_ecephys_session_ids(
        lims_connection: PostgresQueryMixin,
        session_id_list: List[int]) -> pd.DataFrame:
    """
    Return a dataframe with columns
    ecephys_session_id
    donor_id
    from a specified list of ecephys_session_ids
    """
    query = f"""
    SELECT
      donors.id as donor_id
      ,ecephys_sessions.id as ecephys_session_id
    FROM donors
    JOIN specimens ON
      specimens.donor_id = donors.id
    JOIN ecephys_sessions ON
      ecephys_sessions.specimen_id = specimens.id
    {build_in_list_selector_query(
        col='ecephys_sessions.id',
        valid_list=session_id_list
    )}
    """
    result = lims_connection.select(query)
    return result


def donor_id_list_from_ecephys_session_ids(
        lims_connection: PostgresQueryMixin,
        session_id_list: List[int]) -> List[int]:
    """
    Get the list of donor IDs associated with a list
    of ecephys_session_ids
    """
    lookup = donor_id_lookup_from_ecephys_session_ids(
                lims_connection=lims_connection,
                session_id_list=session_id_list)

    return list(np.unique(lookup.donor_id))