Source code for allensdk.brain_observatory.behavior.data_files.sync_file

import abc
import json
from typing import Union
from pathlib import Path

from cachetools import cached, LRUCache
from cachetools.keys import hashkey

from allensdk.internal.api import PostgresQueryMixin
from allensdk.internal.core.lims_utilities import safe_system_path
from allensdk.internal.core import DataFile
from allensdk.brain_observatory.behavior.sync import get_sync_data
from allensdk.core import DataObject


def _get_sync_file_query_template(behavior_session_id: int):

    """Query returns path to sync timing file associated with behavior session
    """
    SYNC_FILE_QUERY_TEMPLATE = f"""
        SELECT wkf.storage_directory || wkf.filename AS sync_file
        FROM behavior_sessions bs
        JOIN ophys_sessions os ON bs.ophys_session_id = os.id
        JOIN well_known_files wkf ON wkf.attachable_id = os.id
        JOIN well_known_file_types wkft
        ON wkft.id = wkf.well_known_file_type_id
        WHERE wkf.attachable_type = 'OphysSession'
        AND wkft.name = 'OphysRigSync'
        AND bs.id = {behavior_session_id}
    """
    return SYNC_FILE_QUERY_TEMPLATE




def from_json_cache_key(cls, dict_repr: dict, permissive: bool = False):
    return hashkey(json.dumps(dict_repr))





def from_lims_cache_key(cls, db, behavior_session_id: int):
    return hashkey(behavior_session_id)





class SyncFile(DataFile):
    """A DataFile which contains methods for accessing and loading visual
    behavior stimulus *.pkl files.

    This file type contains global timing information for different data
    streams collected during a behavior + ophys session.

    Attributes
    ----------
    filepath : str or Path
        Full path to sync file on disk.
    permissive : bool
        Permissively load from sync file. If True, do no raise when a given
        sync line is not present. Default False.
    """

    def __init__(self, filepath: Union[str, Path], permissive: bool = False):
        self._permissive = permissive
        super().__init__(filepath=filepath, permissive=permissive)

    @property
    def permissive(self) -> bool:  # pragma: no cover
        return self._permissive



    @classmethod
    @cached(cache=LRUCache(maxsize=10), key=from_json_cache_key)
    def from_json(cls,
                  dict_repr: dict,
                  permissive: bool = False) -> "SyncFile":
        filepath = dict_repr["sync_file"]
        return cls(filepath=filepath, permissive=permissive)




    @classmethod
    @cached(cache=LRUCache(maxsize=10), key=from_lims_cache_key)
    def from_lims(
        cls, db: PostgresQueryMixin,
        behavior_session_id: Union[int, str],
        permissive: bool = False
    ) -> "SyncFile":
        query = _get_sync_file_query_template(
            behavior_session_id=behavior_session_id)
        filepath = db.fetchone(query, strict=True)
        return cls(filepath=filepath, permissive=permissive)




    @staticmethod
    def load_data(filepath: Union[str, Path],
                  permissive: bool = False) -> dict:
        filepath = safe_system_path(file_name=filepath)
        return get_sync_data(sync_path=filepath, permissive=permissive)






class SyncFileReadableInterface(abc.ABC):
    """Marks a data object as readable from sync file"""


    @classmethod
    @abc.abstractmethod
    def from_sync_file(cls, *args) -> "DataObject":
        """Populate a DataObject from the sync file

        Returns
        -------
        DataObject:
            An instantiated DataObject which has `name` and `value` properties
        """
        raise NotImplementedError()