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
[docs]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)())
[docs] def get_experiment_id(self):
return self.experiment_id
[docs] 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())
[docs] 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)