Source code for ibllib.misc.exp_ref

"""
A module for processing experiment references in a human readable way

Three pieces of information are required to uniquely identify an experimental session: subject
nickname, the date, and the sequence number (whether the session was the first, second,
etc. on that date).

Alyx and ONE use uuids (a.k.a. eids) to uniquely identify sessions, however these are not
readable.  This module converts between these uuids an readable references.

References may be strings in the form yyyy-mm-dd_n_subject, which may be easily sorted,
or as bunches (dicts) of the form {'subject': str, 'date': datetime.date, 'sequence', int}.
"""
import re
import functools
from pathlib import Path
from typing import Union, List, Iterable as Iter
from datetime import datetime
from collections.abc import Iterable, Mapping

from oneibl.one import ONE
from brainbox.core import Bunch

__all__ = [
    'ref2eid', 'ref2dict', 'ref2path', 'eid2path', 'eid2ref', 'path2ref', 'ref2dj', 'is_exp_ref'
]


def parse_values(func):
    @functools.wraps(func)
    def wrapper_decorator(*args, **kwargs):
        parse = kwargs.pop('parse', True)
        ref = func(*args, **kwargs)
        if parse:
            if isinstance(ref['date'], str):
                if len(ref['date']) == 10:
                    ref['date'] = datetime.strptime(ref['date'], '%Y-%m-%d').date()
                else:
                    ref['date'] = datetime.fromisoformat(ref['date']).date()
            ref['sequence'] = int(ref['sequence'])
        return ref
    return wrapper_decorator


def recurse(func):
    @functools.wraps(func)
    def wrapper_decorator(*args, **kwargs):
        first, *args = args
        if isinstance(first, Iterable) and not isinstance(first, (str, Mapping)):
            return [func(item, *args, **kwargs) for item in first]
        else:
            return func(first, *args, **kwargs)
    return wrapper_decorator


[docs]@recurse def ref2eid(ref: Union[Mapping, str, Iter], one=None) -> Union[str, List]: """ Returns experiment uuid, given one or more experiment references :param ref: One or more objects with keys ('subject', 'date', 'sequence'), or strings with the form yyyy-mm-dd_n_subject :param one: An instance of ONE :return: an experiment uuid string Examples: >>> base = 'https://test.alyx.internationalbrainlab.org' >>> one = ONE(username='test_user', password='TapetesBloc18', base_url=base) Connected to... >>> ref = {'date': datetime(2018, 7, 13).date(), 'sequence': 1, 'subject': 'flowers'} >>> ref2eid(ref, one=one) '4e0b3320-47b7-416e-b842-c34dc9004cf8' >>> ref2eid(['2018-07-13_1_flowers', '2019-04-11_1_KS005'], one=one) ['4e0b3320-47b7-416e-b842-c34dc9004cf8', '7dc3c44b-225f-4083-be3d-07b8562885f4'] """ if not one: one = ONE() ref = ref2dict(ref, parse=False) # Ensure dict session = one.search( subjects=ref['subject'], date_range=(str(ref['date']), str(ref['date'])), number=ref['sequence']) assert len(session) == 1, 'session not found' return session[0]
[docs]@recurse @parse_values def ref2dict(ref: Union[str, Mapping, Iter]) -> Union[Bunch, List]: """ Returns a Bunch (dict-like) from a reference string (or list thereof) :param ref: One or more objects with keys ('subject', 'date', 'sequence') :return: A Bunch in with keys ('subject', 'sequence', 'date') Examples: >>> ref2dict('2018-07-13_1_flowers') {'date': datetime.date(2018, 7, 13), 'sequence': 1, 'subject': 'flowers'} >>> ref2dict('2018-07-13_001_flowers', parse=False) {'date': '2018-07-13', 'sequence': '001', 'subject': 'flowers'} >>> ref2dict(['2018-07-13_1_flowers', '2020-01-23_002_ibl_witten_01']) [{'date': datetime.date(2018, 7, 13), 'sequence': 1, 'subject': 'flowers'}, {'date': datetime.date(2020, 1, 23), 'sequence': 2, 'subject': 'ibl_witten_01'}] """ if isinstance(ref, (Bunch, dict)): return Bunch(ref) # Short circuit ref = dict(zip(['date', 'sequence', 'subject'], ref.split('_', 2))) return Bunch(ref)
[docs]@recurse def ref2path(ref: Union[str, Mapping, Iter], one=None, offline: bool = False) -> Union[Path, List]: """ Convert one or more experiment references to session path(s) :param ref: One or more objects with keys ('subject', 'date', 'sequence'), or strings with the form yyyy-mm-dd_n_subject :param one: An instance of ONE :param offline: Return path without connecting to database (unimplemented) :return: a Path object for the experiment session Examples: >>> base = 'https://test.alyx.internationalbrainlab.org' >>> one = ONE(username='test_user', password='TapetesBloc18', base_url=base) Connected to... >>> ref = {'subject': 'flowers', 'date': datetime(2018, 7, 13).date(), 'sequence': 1} >>> ref2path(ref, one=one) WindowsPath('E:/FlatIron/zadorlab/Subjects/flowers/2018-07-13/001') >>> ref2path(['2018-07-13_1_flowers', '2019-04-11_1_KS005'], one=one) [WindowsPath('E:/FlatIron/zadorlab/Subjects/flowers/2018-07-13/001'), WindowsPath('E:/FlatIron/cortexlab/Subjects/KS005/2019-04-11/001')] """ if not one: one = ONE() if offline: raise NotImplementedError # Requires lab name :( # root = Path(one._get_cache_dir(None)) # path = root / ref.subject / str(ref.date) / ('%03d' % ref.sequence) else: ref = ref2dict(ref, parse=False) eid, (d,) = one.search( subjects=ref['subject'], date_range=(str(ref['date']), str(ref['date'])), number=ref['sequence'], details=True) path = d.get('local_path') if not path: root = Path(one._get_cache_dir(None)) / 'Subjects' / d['lab'] return root / d['subject'] / d['start_time'][:10] / ('%03d' % d['number']) else: return Path(path)
[docs]@recurse def eid2path(eid: Union[str, Iter], one=None, offline: bool = False) -> Union[Path, List]: """ Returns a local path from an eid, regardless of whether the path exists locally :param eid: An experiment uuid :param one: An instance of ONE :param offline: If True, do not connect to database (not implemented) :return: a Path instance Examples: >>> base = 'https://test.alyx.internationalbrainlab.org' >>> one = ONE(username='test_user', password='TapetesBloc18', base_url=base) Connected to... >>> eid = '4e0b3320-47b7-416e-b842-c34dc9004cf8' >>> eid2path(eid, one=one) WindowsPath('E:/FlatIron/zadorlab/Subjects/flowers/2018-07-13/001') >>> eid2path([eid, '7dc3c44b-225f-4083-be3d-07b8562885f4'], one=one) [WindowsPath('E:/FlatIron/zadorlab/Subjects/flowers/2018-07-13/001'), WindowsPath('E:/FlatIron/cortexlab/Subjects/KS005/2019-04-11/001')] """ if not one: one = ONE() if offline: raise NotImplementedError # path = one.path_from_eid(eid, offline=True) else: d = one.get_details(eid) path = d.get('local_path') if not path: root = Path(one._get_cache_dir(None)) / d['lab'] / 'Subjects' path = root / d['subject'] / d['start_time'][:10] / ('%03d' % d['number']) return path
[docs]@recurse def eid2ref(eid: Union[str, Iter], one=None, as_dict=True, parse=True) \ -> Union[str, Mapping, List]: """ Get human-readable session ref from path :param eid: The experiment uuid to find reference for :param one: An ONE instance :param as_dict: If false a string is returned in the form 'subject_sequence_yyyy-mm-dd' :param parse: If true, the reference date and sequence are parsed from strings to their respective data types :return: one or more objects with keys ('subject', 'date', 'sequence'), or strings with the form yyyy-mm-dd_n_subject Examples: >>> base = 'https://test.alyx.internationalbrainlab.org' >>> one = ONE(username='test_user', password='TapetesBloc18', base_url=base) Connected to... >>> eid = '4e0b3320-47b7-416e-b842-c34dc9004cf8' >>> eid2ref(eid, one=one) {'subject': 'flowers', 'date': datetime.date(2018, 7, 13), 'sequence': 1} >>> eid2ref(eid, parse=False, one=one) {'subject': 'flowers', 'date': '2018-07-13', 'sequence': '001'} >>> eid2ref(eid, as_dict=False, one=one) '2018-07-13_1_flowers' >>> eid2ref(eid, as_dict=False, parse=False, one=one) '2018-07-13_001_flowers' >>> eid2ref([eid, '7dc3c44b-225f-4083-be3d-07b8562885f4'], one=one) [{'subject': 'flowers', 'date': datetime.date(2018, 7, 13), 'sequence': 1}, {'subject': 'KS005', 'date': datetime.date(2019, 4, 11), 'sequence': 1}] """ if not one: one = ONE() d = one.get_details(eid) if parse: date = datetime.fromisoformat(d['start_time']).date() ref = {'subject': d['subject'], 'date': date, 'sequence': d['number']} format_str = '{date:%Y-%m-%d}_{sequence:d}_{subject:s}' else: date = d['start_time'][:10] ref = {'subject': d['subject'], 'date': date, 'sequence': '%03d' % d['number']} format_str = '{date:s}_{sequence:s}_{subject:s}' return Bunch(ref) if as_dict else format_str.format(**ref)
[docs]@recurse @parse_values def path2ref(path_str: Union[str, Path, Iter]) -> Union[Bunch, List]: """ Returns a human readable experiment reference, given a session path. The path need not exist. :param path_str: A path to a given session :return: one or more objects with keys ('subject', 'date', 'sequence') Examples: >>> path_str = Path('E:/FlatIron/Subjects/zadorlab/flowers/2018-07-13/001') >>> path2ref(path_str) {'subject': 'flowers', 'date': datetime.date(2018, 7, 13), 'sequence': 1} >>> path2ref(path_str, parse=False) {'subject': 'flowers', 'date': '2018-07-13', 'sequence': '001'} >>> path_str2 = Path('E:/FlatIron/Subjects/churchlandlab/CSHL046/2020-06-20/002') >>> path2ref([path_str, path_str2]) [{'subject': 'flowers', 'date': datetime.date(2018, 7, 13), 'sequence': 1}, {'subject': 'CSHL046', 'date': datetime.date(2020, 6, 20), 'sequence': 2}] """ pattern = r'(?P<subject>[\w-]+)([\\/])(?P<date>\d{4}-\d{2}-\d{2})(\2)(?P<sequence>\d{3})' match = re.search(pattern, str(path_str)).groupdict() return Bunch(match)
[docs]def ref2dj(ref: Union[str, Mapping, Iter]): """ Return an ibl-pipeline sessions table, restricted by experiment reference(s) :param ref: one or more objects with keys ('subject', 'date', 'sequence'), or strings with the form yyyy-mm-dd_n_subject :return: an acquisition.Session table Examples: >>> ref2dj('2020-06-20_2_CSHL046').fetch1() Connecting... {'subject_uuid': UUID('dffc24bc-bd97-4c2a-bef3-3e9320dc3dd7'), 'session_start_time': datetime.datetime(2020, 6, 20, 13, 31, 47), 'session_number': 2, 'session_date': datetime.date(2020, 6, 20), 'subject_nickname': 'CSHL046'} >>> len(ref2dj({'date':'2020-06-20', 'sequence':'002', 'subject':'CSHL046'})) 1 >>> len(ref2dj(['2020-06-20_2_CSHL046', '2019-11-01_1_ibl_witten_13'])) 2 """ from ibl_pipeline import subject, acquisition sessions = acquisition.Session.proj('session_number', session_date='date(session_start_time)') sessions = sessions * subject.Subject.proj('subject_nickname') ref = ref2dict(ref) # Ensure dict-like @recurse def restrict(r): date, sequence, subject = dict(sorted(r.items())).values() # Unpack sorted restriction = { 'subject_nickname': subject, 'session_number': sequence, 'session_date': date} return restriction return sessions & restrict(ref)
[docs]@recurse def is_exp_ref(ref: Union[str, Mapping, Iter]) -> Union[bool, List[bool]]: """ Returns True is ref is a valid experiment reference :param ref: one or more objects with keys ('subject', 'date', 'sequence'), or strings with the form yyyy-mm-dd_n_subject :return: True if ref is valid Examples: >>> ref = {'date': datetime(2018, 7, 13).date(), 'sequence': 1, 'subject': 'flowers'} >>> is_exp_ref(ref) True >>> is_exp_ref('2018-07-13_001_flowers') True >>> is_exp_ref('invalid_ref') False """ if isinstance(ref, (Bunch, dict)): if not {'subject', 'date', 'sequence'}.issubset(ref): return False ref = '{date}_{sequence}_{subject}'.format(**ref) elif not isinstance(ref, str): return False return re.compile(r'\d{4}(-\d{2}){2}_(\d{1}|\d{3})_\w+').match(ref) is not None
if __name__ == "__main__": import doctest doctest.testmod(optionflags=doctest.NORMALIZE_WHITESPACE | doctest.ELLIPSIS)