Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
assert hypno.size == data.size, 'Hypno must have same size as data.'
unique_hypno = np.unique(hypno)
logger.info('Number of unique values in hypno = %i', unique_hypno.size)
# Check include
assert include is not None, 'include cannot be None if hypno is given'
include = np.atleast_1d(np.asarray(include))
assert include.size >= 1, '`include` must have at least one element.'
assert hypno.dtype.kind == include.dtype.kind, ('hypno and include '
'must have same dtype')
if not np.in1d(hypno, include).any():
logger.error('None of the stages specified in `include` '
'are present in hypno. Returning None.')
return None
# Check data amplitude
data_trimstd = trimbothstd(data, cut=0.10)
data_ptp = np.ptp(data)
logger.info('Number of samples in data = %i', data.size)
logger.info('Sampling frequency = %.2f Hz', sf)
logger.info('Data duration = %.2f seconds', data.size / sf)
logger.info('Trimmed standard deviation of data = %.4f uV', data_trimstd)
logger.info('Peak-to-peak amplitude of data = %.4f uV', data_ptp)
if not(1 < data_trimstd < 1e3 or 1 < data_ptp < 1e6):
logger.error('Wrong data amplitude. Unit must be uV. Returning None.')
return None
if 'rel_pow' not in thresh.keys():
thresh['rel_pow'] = 0.20
if 'corr' not in thresh.keys():
thresh['corr'] = 0.65
if 'rms' not in thresh.keys():
thresh['rms'] = 1.5
# Hilbert power (to define the instantaneous frequency / power)
n = data_sigma.size
nfast = next_fast_len(n)
analytic = signal.hilbert(data_sigma, N=nfast)[:n]
inst_phase = np.angle(analytic)
inst_pow = np.square(np.abs(analytic))
# inst_freq = sf / 2pi * 1st-derivative of the phase of the analytic signal
inst_freq = (sf / (2 * np.pi) * np.ediff1d(inst_phase))
# Let's define the thresholds
if hypno is None:
thresh_rms = mrms.mean() + thresh['rms'] * trimbothstd(mrms, cut=0.10)
else:
thresh_rms = mrms[mask].mean() + thresh['rms'] * \
trimbothstd(mrms[mask], cut=0.10)
# Avoid too high threshold caused by Artefacts / Motion during Wake.
thresh_rms = min(thresh_rms, 10)
idx_rel_pow = (rel_pow >= thresh['rel_pow']).astype(int)
idx_mcorr = (mcorr >= thresh['corr']).astype(int)
idx_mrms = (mrms >= thresh_rms).astype(int)
idx_sum = (idx_rel_pow + idx_mcorr + idx_mrms).astype(int)
# Make sure that we do not detect spindles in REM or Wake if hypno != None
if hypno is not None:
idx_sum[~mask] = 0
# For debugging
logger.info('Moving RMS threshold = %.3f', thresh_rms)
logger.info('Number of supra-theshold samples for relative power = %i',
idx_rel_pow.sum())
logger.info('Number of unique values in hypno = %i', unique_hypno.size)
# Check include
assert include is not None, 'include cannot be None if hypno is given'
include = np.atleast_1d(np.asarray(include))
assert include.size >= 1, '`include` must have at least one element.'
assert hypno.dtype.kind == include.dtype.kind, ('hypno and include '
'must have same dtype')
if not np.in1d(hypno, include).any():
logger.error('None of the stages specified in `include` '
'are present in hypno. Returning None.')
return None
# Check data amplitude
# times = np.arange(data.size) / sf
data = np.vstack((loc, roc))
loc_trimstd = trimbothstd(loc, cut=0.10)
roc_trimstd = trimbothstd(roc, cut=0.10)
loc_ptp, roc_ptp = np.ptp(loc), np.ptp(roc)
logger.info('Number of samples in data = %i', data.shape[1])
logger.info('Original sampling frequency = %.2f Hz', sf)
logger.info('Data duration = %.2f seconds', data.shape[1] / sf)
logger.info('Trimmed standard deviation of LOC = %.4f uV', loc_trimstd)
logger.info('Trimmed standard deviation of ROC = %.4f uV', roc_trimstd)
logger.info('Peak-to-peak amplitude of LOC = %.4f uV', loc_ptp)
logger.info('Peak-to-peak amplitude of ROC = %.4f uV', roc_ptp)
if not(1 < loc_trimstd < 1e3 or 1 < loc_ptp < 1e6):
logger.error('Wrong LOC amplitude. Unit must be uV. Returning None.')
return None
if not(1 < roc_trimstd < 1e3 or 1 < roc_ptp < 1e6):
logger.error('Wrong ROC amplitude. Unit must be uV. Returning None.')
return None
method='corr', interp=True)
_, mrms = moving_transform(data_sigma, data, sf, window=.3, step=.1,
method='rms', interp=True)
# Hilbert power (to define the instantaneous frequency / power)
n = data_sigma.size
nfast = next_fast_len(n)
analytic = signal.hilbert(data_sigma, N=nfast)[:n]
inst_phase = np.angle(analytic)
inst_pow = np.square(np.abs(analytic))
# inst_freq = sf / 2pi * 1st-derivative of the phase of the analytic signal
inst_freq = (sf / (2 * np.pi) * np.ediff1d(inst_phase))
# Let's define the thresholds
if hypno is None:
thresh_rms = mrms.mean() + thresh['rms'] * trimbothstd(mrms, cut=0.10)
else:
thresh_rms = mrms[mask].mean() + thresh['rms'] * \
trimbothstd(mrms[mask], cut=0.10)
# Avoid too high threshold caused by Artefacts / Motion during Wake.
thresh_rms = min(thresh_rms, 10)
idx_rel_pow = (rel_pow >= thresh['rel_pow']).astype(int)
idx_mcorr = (mcorr >= thresh['corr']).astype(int)
idx_mrms = (mrms >= thresh_rms).astype(int)
idx_sum = (idx_rel_pow + idx_mcorr + idx_mrms).astype(int)
# Make sure that we do not detect spindles in REM or Wake if hypno != None
if hypno is not None:
idx_sum[~mask] = 0
# For debugging
assert hypno.size == data.size, 'Hypno must have same size as data.'
unique_hypno = np.unique(hypno)
logger.info('Number of unique values in hypno = %i', unique_hypno.size)
# Check include
assert include is not None, 'include cannot be None if hypno is given'
include = np.atleast_1d(np.asarray(include))
assert include.size >= 1, '`include` must have at least one element.'
assert hypno.dtype.kind == include.dtype.kind, ('hypno and include '
'must have same dtype')
if not np.in1d(hypno, include).any():
logger.error('None of the stages specified in `include` '
'are present in hypno. Returning None.')
return None
# Check data amplitude
data_trimstd = trimbothstd(data, cut=0.10)
data_ptp = np.ptp(data)
logger.info('Number of samples in data = %i', data.size)
logger.info('Sampling frequency = %.2f Hz', sf)
logger.info('Data duration = %.2f seconds', data.size / sf)
logger.info('Trimmed standard deviation of data = %.4f uV', data_trimstd)
logger.info('Peak-to-peak amplitude of data = %.4f uV', data_ptp)
if not(1 < data_trimstd < 1e3 or 1 < data_ptp < 1e6):
logger.error('Wrong data amplitude. Unit must be uV. Returning None.')
return None
# Check if we can downsample to 100 or 128 Hz
if downsample is True and sf > 128:
if sf % 100 == 0 or sf % 128 == 0:
new_sf = 100 if sf % 100 == 0 else 128
fac = int(sf / new_sf)
sf = new_sf
# Check include
assert include is not None, 'include cannot be None if hypno is given'
include = np.atleast_1d(np.asarray(include))
assert include.size >= 1, '`include` must have at least one element.'
assert hypno.dtype.kind == include.dtype.kind, ('hypno and include '
'must have same dtype')
if not np.in1d(hypno, include).any():
logger.error('None of the stages specified in `include` '
'are present in hypno. Returning None.')
return None
# Check data amplitude
# times = np.arange(data.size) / sf
data = np.vstack((loc, roc))
loc_trimstd = trimbothstd(loc, cut=0.10)
roc_trimstd = trimbothstd(roc, cut=0.10)
loc_ptp, roc_ptp = np.ptp(loc), np.ptp(roc)
logger.info('Number of samples in data = %i', data.shape[1])
logger.info('Original sampling frequency = %.2f Hz', sf)
logger.info('Data duration = %.2f seconds', data.shape[1] / sf)
logger.info('Trimmed standard deviation of LOC = %.4f uV', loc_trimstd)
logger.info('Trimmed standard deviation of ROC = %.4f uV', roc_trimstd)
logger.info('Peak-to-peak amplitude of LOC = %.4f uV', loc_ptp)
logger.info('Peak-to-peak amplitude of ROC = %.4f uV', roc_ptp)
if not(1 < loc_trimstd < 1e3 or 1 < loc_ptp < 1e6):
logger.error('Wrong LOC amplitude. Unit must be uV. Returning None.')
return None
if not(1 < roc_trimstd < 1e3 or 1 < roc_ptp < 1e6):
logger.error('Wrong ROC amplitude. Unit must be uV. Returning None.')
return None
# Check if we can downsample to 100 or 128 Hz