Source code for allensdk.internal.api.lims_api

import pandas as pd
from typing import Optional

from allensdk.internal.api import PostgresQueryMixin
from allensdk.internal.core.lims_utilities import safe_system_path
from allensdk.core.auth_config import LIMS_DB_CREDENTIAL_MAP
from allensdk.core.authentication import credential_injector, DbCredentials


class LimsApi():

    def __init__(self, lims_credentials: Optional[DbCredentials] = None):
        if lims_credentials:
            self.lims_db = PostgresQueryMixin(
                dbname=lims_credentials.dbname, user=lims_credentials.user,
                host=lims_credentials.host, password=lims_credentials.password,
                port=lims_credentials.port)
        else:
            # Currying is equivalent to decorator syntactic sugar
            self.lims_db = (credential_injector(LIMS_DB_CREDENTIAL_MAP)
                            (PostgresQueryMixin)())

    def get_experiment_id(self):
        return self.experiment_id

    def get_behavior_tracking_video_filepath_df(self):
        query = '''
                SELECT wkf.storage_directory || wkf.filename AS raw_behavior_tracking_video_filepath, attachable_type 
                FROM well_known_files wkf WHERE wkf.well_known_file_type_id IN (SELECT id FROM well_known_file_types WHERE name = 'RawBehaviorTrackingVideo')
                '''
        return pd.read_sql(query, self.lims_db.get_connection())

    def get_eye_tracking_video_filepath_df(self):
        query = '''
                SELECT wkf.storage_directory || wkf.filename AS raw_behavior_tracking_video_filepath, attachable_type 
                FROM well_known_files wkf WHERE wkf.well_known_file_type_id IN (SELECT id FROM well_known_file_types WHERE name = 'RawEyeTrackingVideo')
                '''
        return pd.read_sql(query, self.lims_db.get_connection())


if __name__ == "__main__":

    api = LimsApi()
    for ii in range(5):
        print(api.get_eye_tracking_video_filepath_df().loc[ii].raw_behavior_tracking_video_filepath)