<gh_stars>0
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
pyplr.utils
===========

Scripting tools for working with pupil data.

@author: jtm

'''

import os
import shutil
import os.path as op
from copy import deepcopy

import numpy as np
import pandas as pd

def new_subject(rec_dir, export='000', out_dir_nm='pyplr_analysis'):
    '''Get a handle on a new subject for data analysis.
    
    Parameters
    ----------
    rec_dir : str
        Full path to a Pupil Labs recording directory for a given subject.
    export : str
        The export folder in which to look for files (in case of multiple 
        exports). The default is '000'.
    out_dir_nm : str, optional
        Name for the folder where output will be saved. The default is
        'pyplr_analysis'.

    Raises
    ------
    FileNotFoundError
        If subjdir does not exist.

    Returns
    -------
    s : dict
        Dictionary of subject information.

    '''
    if not op.isdir(rec_dir):
        raise FileNotFoundError(
            '"{}" does not appear to exist.'.format(rec_dir))
    root = op.abspath(rec_dir)
    identifier = op.basename(rec_dir)
    data_dir = op.abspath(op.join(rec_dir, 'exports', '', export, ''))
    out_dir = op.abspath(op.join(rec_dir, out_dir_nm, ''))
    if os.path.exists(out_dir):
        shutil.rmtree(out_dir)
    os.mkdir(out_dir)
    print('{}\n{:*^60s}\n{}'.format('*'*60, ' ' + identifier + ' ', '*'*60))
    return {
        'root':root,
        'id':identifier,
        'data_dir':data_dir,
        'out_dir':out_dir
        }

def print_file_structure(rec_dir):
    '''Print the file structure of a recording directory.

    Returns
    -------
    None.

    '''
    for root, dirs, files in os.walk(rec_dir):
        level = root.replace(rec_dir, '').count(os.sep)
        indent = ' ' * 4 * (level)
        print(f'{indent}{os.path.basename(root)}/')
        subindent = ' ' * 4 * (level + 1)
        for f in sorted(files):
            print(f'{subindent}{f}')
            
def load_pupil(data_dir, eye_id='best', method='3d c++', cols=None):
    '''Loads 'pupil_positions.csv' data exported from Pupil Player.
    
    Parameters
    ----------
    data_dir : str
        Directory where the Pupil Labs 'pupil_positions.csv' data exists.
    eye : str
        Eye to load. Must be 'left' (1), 'right' (0), or 'best'. If 'best', 
        the eye with the highest average confidence will be loaded. The
        default is None.
    method : string, optional
        Whether to load pupil data generated by the 2d or 3d fitting method. 
        The default is '3d'.
    cols : list, optional
        Columns to load from the file (check file for options). Must include 
        ['pupil_timestap', 'method', 'confidence'] and The default is None 
        (loads all columns).
        
    Returns
    -------
    samps : pandas.DataFrame
        Pandas DataFrame containing requested samples.
        
    '''
    fname = op.join(data_dir, '', 'pupil_positions.csv')
    try:
        if cols is None:
            samples = pd.read_csv(fname, index_col='pupil_timestamp')
        else:
            samples = pd.read_csv(
                fname, usecols=cols, index_col='pupil_timestamp')
    except FileNotFoundError as fnf_error:
        print(fnf_error)
    else:
        samples = samples.loc[samples.method.str.contains(method)]  
        if eye_id == 'left':
            samples = samples[samples.eye_id==1]
        elif eye_id == 'right':
            samples = samples[samples.eye_id==0]
        elif eye_id == 'best':
            best = samples.groupby(['eye_id'])['confidence'].mean().idxmax()
            samples = samples[samples.eye_id==best]
        else:
            raise ValueError('Eye must be "left", "right" or "best".')
        print('Loaded {} samples'.format(len(samples)))
        return samples
    
def load_annotations(data_dir):
    '''Loads 'annotations' exported from Pupil Player.
    
    Parameters
    ----------
    data_dir : str
        Directory where the Pupil Labs 'annotations' data exists.
        
    Returns
    -------
    events : pandas.DataFrame
        Pandas DataFrame containing annotations / events.
        
    '''
    fname = op.join(data_dir, '', 'annotations.csv')
    try:
        events = pd.read_csv(fname, index_col='timestamp')
        print('Loaded {} events'.format(len(events)))
    except FileNotFoundError as fnf_error:
        print(fnf_error)
    else:
        return events
   
def load_blinks(data_dir):
    '''Loads 'blinks' data exported from Pupil Player.
    
    Parameters
    ----------
    data_dir : str
        Directory where the Pupil Labs 'blinks' data exists.
        
    Returns
    -------
    blinks : pandas.DataFrame
        Pandas DataFrame containing blink events.
        
    '''
    fname = op.join(data_dir, '', 'blinks.csv')
    try:
        blinks = pd.read_csv(fname, index_col='id')
        print('{} blinks detected by Pupil Labs (mean dur = {:.3f} s)'.format(
            len(blinks), blinks.duration.mean()))
    except FileNotFoundError as fnf_error:
        print(fnf_error)
    else:
        return blinks

#TODO: optimse and debug
def extract(samples, 
            events, 
            offset=0, 
            duration=0,
            borrow_attributes=[]):
    '''
    Extracts ranges from samples based on event timing and sample count.
    
    Parameters
    ----------
    samples : pandas.DataFrame
        The samples from which to extract events. Index must be timestamp.
    events : pandas.DataFrame
        The events to extract. Index must be timestamp.
    offset : int, optional
        Number of samples to offset from baseline. The default is 0.
    duration : int, optional
        Duration of all events in terms of the number of samples. Currently 
        this has to be the same for all events, but could use a 'duration' 
        column in events DataFrame if needs be. The default is 0.
    borrow_attributes : list of str, optional
        List of column names in the events DataFrame whose values should be
        copied to the respective ranges. For each item in the list, a
        column will be created in the ranges dataframe - if the column does
        not exist in the events dataframe, the values in the each
        corresponding range will be set to float('nan'). This is uesful for 
        marking conditions, grouping variables, etc. The default is [].
        
    Returns
    -------
    df : pandas.DataFrame
        Extracted events complete with hierarchical multi-index.
        
    '''
    # negative duration should raise an exception
    if duration <= 0:
        raise ValueError('Duration must be >0')

    # get the list of start time indices
    event_starts = events.index.to_series()

    # find the indexes of the event starts, and offset by sample count
    range_idxs = np.searchsorted(
        samples.index, event_starts.iloc[:], 'left') + offset
    range_duration = duration
    
    # make a hierarchical index
    samples['orig_idx'] = samples.index
    midx = pd.MultiIndex.from_product(
        [list(range(len(event_starts))), list(range(range_duration))],
        names=['event', 'onset'])
    
    # TODO: can add option for time index here
    
    # get the samples
    df = pd.DataFrame()
    idx = 0
    for start_idx in range_idxs:
        # get the start time and add the required number of indices
        end_idx = start_idx + range_duration - 1  # .loc indexing is inclusive
        new_df = deepcopy(
            samples.loc[samples.index[start_idx] : samples.index[end_idx]])
        for ba in borrow_attributes:
            new_df[ba] = events.iloc[idx].get(ba, float('nan'))
        df = pd.concat([df, new_df])
        idx += 1
    df.index = midx
    print('Extracted ranges for {} events'.format(len(events)))
    return df

def reject_bad_trials(ranges, interp_thresh=20, drop=False):
    '''Drop or markup trials which exceed a threshold of interpolated data.
    
    Parameters
    ----------
    ranges : pandas.DataFrame
        Extracted event ranges with hierarchical pd.MultiIndex.
    interp_thresh : int, optional
        Percentage of interpolated data permitted before trials are marked for
        rejection / dropped. The default is 20.
    drop : bool, optional
        Whether to drop the trials from the ranges. The default is False.
        
    Returns
    -------
    ranges : pandas.DataFrame
        Same as ranges but with a column identifying trials marked for
        rejection (drop = False) or with those trials dropped from the 
        DataFrame (drop = True).
        
    '''
    if not isinstance(ranges.index, pd.MultiIndex):
        raise ValueError('Index of ranges must be pd.MultiIndex')
        
    pct_interp = ranges.groupby(by='event').agg(
        {'interpolated':lambda x: float(x.sum())/len(x)*100})
    print('Percentage of data interpolated for each trial (mean = {:.2f}): \
          \n'.format(pct_interp.mean()[0]), pct_interp)
    reject_idxs = (pct_interp.loc[pct_interp['interpolated'] > interp_thresh]
                             .index.to_list())
    ranges['reject'] = 0
    if reject_idxs:
        ranges.loc[reject_idxs, 'reject'] = 1
    if drop:
        ranges = ranges.drop(index=reject_idxs)
        print('{} trials were dropped from the DataFrame'.format(
            len(reject_idxs)))
    else:
        print('{} trials were marked for rejection'.format(len(reject_idxs)))
    return ranges

def unpack_data_numpy(data, what):
    return np.array([entry[what] for entry in data])

def unpack_data_pandas(data, cols=['timestamp','diameter']):
    return (pd.DataFrame(data)
              .set_index('timestamp')) 
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
pyplr.utils
===========

Scripting tools for working with pupil data.

@author: jtm

'''

import os
import shutil
import os.path as op
from copy import deepcopy

import numpy as np
import pandas as pd

def new_subject(rec_dir, export='000', out_dir_nm='pyplr_analysis'):
    '''Get a handle on a new subject for data analysis.
    
    Parameters
    ----------
    rec_dir : str
        Full path to a Pupil Labs recording directory for a given subject.
    export : str
        The export folder in which to look for files (in case of multiple 
        exports). The default is '000'.
    out_dir_nm : str, optional
        Name for the folder where output will be saved. The default is
        'pyplr_analysis'.

    Raises
    ------
    FileNotFoundError
        If subjdir does not exist.

    Returns
    -------
    s : dict
        Dictionary of subject information.

    '''
    if not op.isdir(rec_dir):
        raise FileNotFoundError(
            '"{}" does not appear to exist.'.format(rec_dir))
    root = op.abspath(rec_dir)
    identifier = op.basename(rec_dir)
    data_dir = op.abspath(op.join(rec_dir, 'exports', '', export, ''))
    out_dir = op.abspath(op.join(rec_dir, out_dir_nm, ''))
    if os.path.exists(out_dir):
        shutil.rmtree(out_dir)
    os.mkdir(out_dir)
    print('{}\n{:*^60s}\n{}'.format('*'*60, ' ' + identifier + ' ', '*'*60))
    return {
        'root':root,
        'id':identifier,
        'data_dir':data_dir,
        'out_dir':out_dir
        }

def print_file_structure(rec_dir):
    '''Print the file structure of a recording directory.

    Returns
    -------
    None.

    '''
    for root, dirs, files in os.walk(rec_dir):
        level = root.replace(rec_dir, '').count(os.sep)
        indent = ' ' * 4 * (level)
        print(f'{indent}{os.path.basename(root)}/')
        subindent = ' ' * 4 * (level + 1)
        for f in sorted(files):
            print(f'{subindent}{f}')
            
def load_pupil(data_dir, eye_id='best', method='3d c++', cols=None):
    '''Loads 'pupil_positions.csv' data exported from Pupil Player.
    
    Parameters
    ----------
    data_dir : str
        Directory where the Pupil Labs 'pupil_positions.csv' data exists.
    eye : str
        Eye to load. Must be 'left' (1), 'right' (0), or 'best'. If 'best', 
        the eye with the highest average confidence will be loaded. The
        default is None.
    method : string, optional
        Whether to load pupil data generated by the 2d or 3d fitting method. 
        The default is '3d'.
    cols : list, optional
        Columns to load from the file (check file for options). Must include 
        ['pupil_timestap', 'method', 'confidence'] and The default is None 
        (loads all columns).
        
    Returns
    -------
    samps : pandas.DataFrame
        Pandas DataFrame containing requested samples.
        
    '''
    fname = op.join(data_dir, '', 'pupil_positions.csv')
    try:
        if cols is None:
            samples = pd.read_csv(fname, index_col='pupil_timestamp')
        else:
            samples = pd.read_csv(
                fname, usecols=cols, index_col='pupil_timestamp')
    except FileNotFoundError as fnf_error:
        print(fnf_error)
    else:
        samples = samples.loc[samples.method.str.contains(method)]  
        if eye_id == 'left':
            samples = samples[samples.eye_id==1]
        elif eye_id == 'right':
            samples = samples[samples.eye_id==0]
        elif eye_id == 'best':
            best = samples.groupby(['eye_id'])['confidence'].mean().idxmax()
            samples = samples[samples.eye_id==best]
        else:
            raise ValueError('Eye must be "left", "right" or "best".')
        print('Loaded {} samples'.format(len(samples)))
        return samples
    
def load_annotations(data_dir):
    '''Loads 'annotations' exported from Pupil Player.
    
    Parameters
    ----------
    data_dir : str
        Directory where the Pupil Labs 'annotations' data exists.
        
    Returns
    -------
    events : pandas.DataFrame
        Pandas DataFrame containing annotations / events.
        
    '''
    fname = op.join(data_dir, '', 'annotations.csv')
    try:
        events = pd.read_csv(fname, index_col='timestamp')
        print('Loaded {} events'.format(len(events)))
    except FileNotFoundError as fnf_error:
        print(fnf_error)
    else:
        return events
   
def load_blinks(data_dir):
    '''Loads 'blinks' data exported from Pupil Player.
    
    Parameters
    ----------
    data_dir : str
        Directory where the Pupil Labs 'blinks' data exists.
        
    Returns
    -------
    blinks : pandas.DataFrame
        Pandas DataFrame containing blink events.
        
    '''
    fname = op.join(data_dir, '', 'blinks.csv')
    try:
        blinks = pd.read_csv(fname, index_col='id')
        print('{} blinks detected by Pupil Labs (mean dur = {:.3f} s)'.format(
            len(blinks), blinks.duration.mean()))
    except FileNotFoundError as fnf_error:
        print(fnf_error)
    else:
        return blinks

#TODO: optimse and debug
def extract(samples, 
            events, 
            offset=0, 
            duration=0,
            borrow_attributes=[]):
    '''
    Extracts ranges from samples based on event timing and sample count.
    
    Parameters
    ----------
    samples : pandas.DataFrame
        The samples from which to extract events. Index must be timestamp.
    events : pandas.DataFrame
        The events to extract. Index must be timestamp.
    offset : int, optional
        Number of samples to offset from baseline. The default is 0.
    duration : int, optional
        Duration of all events in terms of the number of samples. Currently 
        this has to be the same for all events, but could use a 'duration' 
        column in events DataFrame if needs be. The default is 0.
    borrow_attributes : list of str, optional
        List of column names in the events DataFrame whose values should be
        copied to the respective ranges. For each item in the list, a
        column will be created in the ranges dataframe - if the column does
        not exist in the events dataframe, the values in the each
        corresponding range will be set to float('nan'). This is uesful for 
        marking conditions, grouping variables, etc. The default is [].
        
    Returns
    -------
    df : pandas.DataFrame
        Extracted events complete with hierarchical multi-index.
        
    '''
    # negative duration should raise an exception
    if duration <= 0:
        raise ValueError('Duration must be >0')

    # get the list of start time indices
    event_starts = events.index.to_series()

    # find the indexes of the event starts, and offset by sample count
    range_idxs = np.searchsorted(
        samples.index, event_starts.iloc[:], 'left') + offset
    range_duration = duration
    
    # make a hierarchical index
    samples['orig_idx'] = samples.index
    midx = pd.MultiIndex.from_product(
        [list(range(len(event_starts))), list(range(range_duration))],
        names=['event', 'onset'])
    
    # TODO: can add option for time index here
    
    # get the samples
    df = pd.DataFrame()
    idx = 0
    for start_idx in range_idxs:
        # get the start time and add the required number of indices
        end_idx = start_idx + range_duration - 1  # .loc indexing is inclusive
        new_df = deepcopy(
            samples.loc[samples.index[start_idx] : samples.index[end_idx]])
        for ba in borrow_attributes:
            new_df[ba] = events.iloc[idx].get(ba, float('nan'))
        df = pd.concat([df, new_df])
        idx += 1
    df.index = midx
    print('Extracted ranges for {} events'.format(len(events)))
    return df

def reject_bad_trials(ranges, interp_thresh=20, drop=False):
    '''Drop or markup trials which exceed a threshold of interpolated data.
    
    Parameters
    ----------
    ranges : pandas.DataFrame
        Extracted event ranges with hierarchical pd.MultiIndex.
    interp_thresh : int, optional
        Percentage of interpolated data permitted before trials are marked for
        rejection / dropped. The default is 20.
    drop : bool, optional
        Whether to drop the trials from the ranges. The default is False.
        
    Returns
    -------
    ranges : pandas.DataFrame
        Same as ranges but with a column identifying trials marked for
        rejection (drop = False) or with those trials dropped from the 
        DataFrame (drop = True).
        
    '''
    if not isinstance(ranges.index, pd.MultiIndex):
        raise ValueError('Index of ranges must be pd.MultiIndex')
        
    pct_interp = ranges.groupby(by='event').agg(
        {'interpolated':lambda x: float(x.sum())/len(x)*100})
    print('Percentage of data interpolated for each trial (mean = {:.2f}): \
          \n'.format(pct_interp.mean()[0]), pct_interp)
    reject_idxs = (pct_interp.loc[pct_interp['interpolated'] > interp_thresh]
                             .index.to_list())
    ranges['reject'] = 0
    if reject_idxs:
        ranges.loc[reject_idxs, 'reject'] = 1
    if drop:
        ranges = ranges.drop(index=reject_idxs)
        print('{} trials were dropped from the DataFrame'.format(
            len(reject_idxs)))
    else:
        print('{} trials were marked for rejection'.format(len(reject_idxs)))
    return ranges

def unpack_data_numpy(data, what):
    return np.array([entry[what] for entry in data])

def unpack_data_pandas(data, cols=['timestamp','diameter']):
    return (pd.DataFrame(data)
              .set_index('timestamp'))
#!/usr/bin/env python3 # -*- coding: utf-8 -*- pyplr.utils ===========  Scripting tools for working with pupil data.  @author: jtm Get a handle on a new subject for data analysis.          Parameters     ----------     rec_dir : str         Full path to a Pupil Labs recording directory for a given subject.     export : str         The export folder in which to look for files (in case of multiple          exports). The default is '000'.     out_dir_nm : str, optional         Name for the folder where output will be saved. The default is         'pyplr_analysis'.      Raises     ------     FileNotFoundError         If subjdir does not exist.      Returns     -------     s : dict         Dictionary of subject information. Print the file structure of a recording directory.      Returns     -------     None. Loads 'pupil_positions.csv' data exported from Pupil Player.          Parameters     ----------     data_dir : str         Directory where the Pupil Labs 'pupil_positions.csv' data exists.     eye : str         Eye to load. Must be 'left' (1), 'right' (0), or 'best'. If 'best',          the eye with the highest average confidence will be loaded. The         default is None.     method : string, optional         Whether to load pupil data generated by the 2d or 3d fitting method.          The default is '3d'.     cols : list, optional         Columns to load from the file (check file for options). Must include          ['pupil_timestap', 'method', 'confidence'] and The default is None          (loads all columns).              Returns     -------     samps : pandas.DataFrame         Pandas DataFrame containing requested samples. Loads 'annotations' exported from Pupil Player.          Parameters     ----------     data_dir : str         Directory where the Pupil Labs 'annotations' data exists.              Returns     -------     events : pandas.DataFrame         Pandas DataFrame containing annotations / events. Loads 'blinks' data exported from Pupil Player.          Parameters     ----------     data_dir : str         Directory where the Pupil Labs 'blinks' data exists.              Returns     -------     blinks : pandas.DataFrame         Pandas DataFrame containing blink events. #TODO: optimse and debug Extracts ranges from samples based on event timing and sample count.          Parameters     ----------     samples : pandas.DataFrame         The samples from which to extract events. Index must be timestamp.     events : pandas.DataFrame         The events to extract. Index must be timestamp.     offset : int, optional         Number of samples to offset from baseline. The default is 0.     duration : int, optional         Duration of all events in terms of the number of samples. Currently          this has to be the same for all events, but could use a 'duration'          column in events DataFrame if needs be. The default is 0.     borrow_attributes : list of str, optional         List of column names in the events DataFrame whose values should be         copied to the respective ranges. For each item in the list, a         column will be created in the ranges dataframe - if the column does         not exist in the events dataframe, the values in the each         corresponding range will be set to float('nan'). This is uesful for          marking conditions, grouping variables, etc. The default is [].              Returns     -------     df : pandas.DataFrame         Extracted events complete with hierarchical multi-index. # negative duration should raise an exception # get the list of start time indices # find the indexes of the event starts, and offset by sample count # make a hierarchical index # TODO: can add option for time index here # get the samples # get the start time and add the required number of indices # .loc indexing is inclusive Drop or markup trials which exceed a threshold of interpolated data.          Parameters     ----------     ranges : pandas.DataFrame         Extracted event ranges with hierarchical pd.MultiIndex.     interp_thresh : int, optional         Percentage of interpolated data permitted before trials are marked for         rejection / dropped. The default is 20.     drop : bool, optional         Whether to drop the trials from the ranges. The default is False.              Returns     -------     ranges : pandas.DataFrame         Same as ranges but with a column identifying trials marked for         rejection (drop = False) or with those trials dropped from the          DataFrame (drop = True).