Measure reaction times and classify responses from Epochs and reject based upon the result.

evaluate_manual_responses.py

def _check_resp(stim, resp, l_cor, r_cor):
    """ Helper function for evaluating responses
    Parameters
    """
    ret = 0
    if resp == 8 and stim in l_cor:
        ret = 1
    elif resp == 64 and stim in r_cor:
        ret = 1
    return ret


def evaluate_responses(raw, ev_merged, event_id, tmin, tmax, st_ch, resp_ch,
                       l_cor, r_cor, resp_events=True):
    """ Evaluate responses
 
    Parameters
    ==========
 
    raw : instance of Raw
        raw meg data
    ev_onsets : array like
        event onsets for all or certain conditions
    tmin : float
        post stimulus response window onset in seconds
    tmax : float
        post stimulus response window offset in seconds
    st_ch : string
        stimulus channel name
    resp_ch : string
        response channel name
    l_cor : array like
        event codes for which a left response is correct
    r_cor : array like
        event codes for which a right response is correct
 
    Return values
    =============
 
    epochs_evaluated : dictionary
        classification of the epoch with regard to id, type of response
        duration, latency, validity
    good_resps : array
        cleaned reponse channel
 
    """
    shape_error = ValueError('You must pass an 2d array as returned'
                                ' from mne.find_events')
 
    if not np.ndim(ev_merged) == 2:
        raise shape_error
    elif not ev_merged.shape[1] == 3:
        raise shape_error
 
    ev_condition = ev_merged[ev_merged[:, 2] == event_id]
    candidates = list(ev_condition)
    tsample = 1e3 / raw.info['sfreq']
    st_idx = raw.info['ch_names'].index(st_ch)
    rs_idx = raw.info['ch_names'].index(resp_ch)
    resp_ch = raw._data[rs_idx]
    epochs_evaluated = []  # get epochs for expected response
    approved = []
    for i, (ev) in enumerate(candidates):
        this_resp = {'id': i, 'response': None, 'off_set': None, 'latency': None,
                     'duration': None, 'valid': 0}
        imin = int(tmin * raw.info['sfreq'])
        imax = int(tmax * raw.info['sfreq'])
        win = np.arange(ev[0] + imin, ev[0] + imax)
        resp_data = resp_ch[win]
        resp = np.unique(resp_data[resp_data != 0])
        if 0 < len(resp) < 2:  # filter non and double responses
            if np.diff(resp_data).nonzero()[0].shape[0] < 3:
                this_resp['response'] = resp
            else:
                this_resp['response'] = 'repeat'
        elif len(resp) > 1:
            this_resp['response'] = 'non-unique'
        elif len(resp) < 1:
            this_resp['response'] = 'missing'
        if this_resp['response'] not in ('non-unique', 'repeat', 'missing'):
            duration = resp_data.nonzero()[0].shape[0] * tsample
            this_resp['duration'] = duration
            this_resp['off_set'] = (resp_data.nonzero()[0][0] + imin)
            this_resp['latency'] = this_resp['off_set'] * tsample
        if this_resp['latency']:
            stim = raw._data[st_idx][ev[0]]
            this_resp['valid'] = _check_resp(stim, resp, l_cor, r_cor)
        if this_resp['valid']:
            if resp_events:
                ev[0] += this_resp['off_set']
                ev[2] = resp_ch[ev[0]]
            approved.append(ev)
 
        epochs_evaluated.append(this_resp)
 
    return epochs_evaluated, np.array(approved)