Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
end_block(fid, FIFF.FIFFB_MNE_BAD_CHANNELS)
# General
if info.get('experimenter') is not None:
write_string(fid, FIFF.FIFF_EXPERIMENTER, info['experimenter'])
if info.get('description') is not None:
write_string(fid, FIFF.FIFF_DESCRIPTION, info['description'])
if info.get('proj_id') is not None:
write_int(fid, FIFF.FIFF_PROJ_ID, info['proj_id'])
if info.get('proj_name') is not None:
write_string(fid, FIFF.FIFF_PROJ_NAME, info['proj_name'])
if info.get('meas_date') is not None:
write_int(fid, FIFF.FIFF_MEAS_DATE, _dt_to_stamp(info['meas_date']))
if info.get('utc_offset') is not None:
write_string(fid, FIFF.FIFF_UTC_OFFSET, info['utc_offset'])
write_int(fid, FIFF.FIFF_NCHAN, info['nchan'])
write_float(fid, FIFF.FIFF_SFREQ, info['sfreq'])
if info['lowpass'] is not None:
write_float(fid, FIFF.FIFF_LOWPASS, info['lowpass'])
if info['highpass'] is not None:
write_float(fid, FIFF.FIFF_HIGHPASS, info['highpass'])
if info.get('line_freq') is not None:
write_float(fid, FIFF.FIFF_LINE_FREQ, info['line_freq'])
if info.get('gantry_angle') is not None:
write_float(fid, FIFF.FIFF_GANTRY_ANGLE, info['gantry_angle'])
if data_type is not None:
write_int(fid, FIFF.FIFF_DATA_PACK, data_type)
if info.get('custom_ref_applied'):
write_int(fid, FIFF.FIFF_MNE_CUSTOM_REF, info['custom_ref_applied'])
if info.get('xplotter_layout'):
write_string(fid, FIFF.FIFF_XPLOTTER_LAYOUT, info['xplotter_layout'])
data *= decal[np.newaxis, :, np.newaxis]
write_function(fid, FIFF.FIFF_EPOCH, data)
# undo modifications to data
data /= decal[np.newaxis, :, np.newaxis]
write_string(fid, FIFF.FIFF_MNE_EPOCHS_DROP_LOG,
json.dumps(epochs.drop_log))
reject_params = _pack_reject_params(epochs)
if reject_params:
write_string(fid, FIFF.FIFF_MNE_EPOCHS_REJECT_FLAT,
json.dumps(reject_params))
write_int(fid, FIFF.FIFF_MNE_EPOCHS_SELECTION,
epochs.selection)
# And now write the next file info in case epochs are split on disk
if next_fname is not None and n_parts > 1:
start_block(fid, FIFF.FIFFB_REF)
write_int(fid, FIFF.FIFF_REF_ROLE, FIFF.FIFFV_ROLE_NEXT_FILE)
write_string(fid, FIFF.FIFF_REF_FILE_NAME, op.basename(next_fname))
if meas_id is not None:
write_id(fid, FIFF.FIFF_REF_FILE_ID, meas_id)
write_int(fid, FIFF.FIFF_REF_FILE_NUM, next_idx)
end_block(fid, FIFF.FIFFB_REF)
end_block(fid, FIFF.FIFFB_MNE_EPOCHS)
end_block(fid, FIFF.FIFFB_PROCESSED_DATA)
end_block(fid, FIFF.FIFFB_MEAS)
end_file(fid)
for hpi_meas in info['hpi_meas']:
start_block(fid, FIFF.FIFFB_HPI_MEAS)
if hpi_meas.get('creator') is not None:
write_string(fid, FIFF.FIFF_CREATOR, hpi_meas['creator'])
if hpi_meas.get('sfreq') is not None:
write_float(fid, FIFF.FIFF_SFREQ, hpi_meas['sfreq'])
if hpi_meas.get('nchan') is not None:
write_int(fid, FIFF.FIFF_NCHAN, hpi_meas['nchan'])
if hpi_meas.get('nave') is not None:
write_int(fid, FIFF.FIFF_NAVE, hpi_meas['nave'])
if hpi_meas.get('ncoil') is not None:
write_int(fid, FIFF.FIFF_HPI_NCOIL, hpi_meas['ncoil'])
if hpi_meas.get('first_samp') is not None:
write_int(fid, FIFF.FIFF_FIRST_SAMPLE, hpi_meas['first_samp'])
if hpi_meas.get('last_samp') is not None:
write_int(fid, FIFF.FIFF_LAST_SAMPLE, hpi_meas['last_samp'])
for hpi_coil in hpi_meas['hpi_coils']:
start_block(fid, FIFF.FIFFB_HPI_COIL)
if hpi_coil.get('number') is not None:
write_int(fid, FIFF.FIFF_HPI_COIL_NO, hpi_coil['number'])
if hpi_coil.get('epoch') is not None:
write_float_matrix(fid, FIFF.FIFF_EPOCH, hpi_coil['epoch'])
if hpi_coil.get('slopes') is not None:
write_float(fid, FIFF.FIFF_HPI_SLOPES, hpi_coil['slopes'])
if hpi_coil.get('corr_coeff') is not None:
write_float(fid, FIFF.FIFF_HPI_CORR_COEFF,
hpi_coil['corr_coeff'])
if hpi_coil.get('coil_freq') is not None:
write_float(fid, FIFF.FIFF_HPI_COIL_FREQ,
hpi_coil['coil_freq'])
end_block(fid, FIFF.FIFFB_HPI_COIL)
end_block(fid, FIFF.FIFFB_HPI_MEAS)
"""Write a projection operator to a file.
Parameters
----------
fid : file
The file descriptor of the open file.
projs : dict
The projection operator.
"""
if len(projs) == 0:
return
start_block(fid, FIFF.FIFFB_PROJ)
for proj in projs:
start_block(fid, FIFF.FIFFB_PROJ_ITEM)
write_int(fid, FIFF.FIFF_NCHAN, proj['data']['ncol'])
write_name_list(fid, FIFF.FIFF_PROJ_ITEM_CH_NAME_LIST,
proj['data']['col_names'])
write_string(fid, FIFF.FIFF_NAME, proj['desc'])
write_int(fid, FIFF.FIFF_PROJ_ITEM_KIND, proj['kind'])
if proj['kind'] == FIFF.FIFFV_PROJ_ITEM_FIELD:
write_float(fid, FIFF.FIFF_PROJ_ITEM_TIME, 0.0)
write_int(fid, FIFF.FIFF_PROJ_ITEM_NVEC, proj['data']['nrow'])
write_int(fid, FIFF.FIFF_MNE_PROJ_ITEM_ACTIVE, proj['active'])
write_float_matrix(fid, FIFF.FIFF_PROJ_ITEM_VECTORS,
proj['data']['data'])
if proj['explained_var'] is not None:
write_float(fid, FIFF.FIFF_MNE_ICA_PCA_EXPLAINED_VAR,
proj['explained_var'])
end_block(fid, FIFF.FIFFB_PROJ_ITEM)
data = this.get('subject_his_id', None)
if data:
write_string(fid, FIFF.FIFF_SUBJ_HIS_ID, data)
write_int(fid, FIFF.FIFF_MNE_COORD_FRAME, this['coord_frame'])
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NPOINTS, this['np'])
write_float_matrix(fid, FIFF.FIFF_MNE_SOURCE_SPACE_POINTS, this['rr'])
write_float_matrix(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NORMALS, this['nn'])
# Which vertices are active
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_SELECTION, this['inuse'])
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NUSE, this['nuse'])
if this['ntri'] > 0:
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NTRI, this['ntri'])
write_int_matrix(fid, FIFF.FIFF_MNE_SOURCE_SPACE_TRIANGLES,
this['tris'] + 1)
if this['type'] != 'vol' and this['use_tris'] is not None:
# Use triangulation
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NUSE_TRI, this['nuse_tri'])
write_int_matrix(fid, FIFF.FIFF_MNE_SOURCE_SPACE_USE_TRIANGLES,
this['use_tris'] + 1)
if this['type'] == 'vol':
neighbor_vert = this.get('neighbor_vert', None)
if neighbor_vert is not None:
nneighbors = np.array([len(n) for n in neighbor_vert])
neighbors = np.concatenate(neighbor_vert)
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NNEIGHBORS, nneighbors)
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NEIGHBORS, neighbors)
write_string(fid, FIFF.FIFF_COMMENT, e.comment)
# First time, num. samples, first and last sample
write_float(fid, FIFF.FIFF_FIRST_TIME, e.times[0])
write_int(fid, FIFF.FIFF_NO_SAMPLES, len(e.times))
write_int(fid, FIFF.FIFF_FIRST_SAMPLE, e.first)
write_int(fid, FIFF.FIFF_LAST_SAMPLE, e.last)
# The epoch itself
if e.info.get('maxshield'):
aspect = FIFF.FIFFB_IAS_ASPECT
else:
aspect = FIFF.FIFFB_ASPECT
start_block(fid, aspect)
write_int(fid, FIFF.FIFF_ASPECT_KIND, e._aspect_kind)
# convert nave to integer to comply with FIFF spec
nave_int = int(round(e.nave))
if nave_int != e.nave and not warned:
warn('converting "nave" to integer before saving evoked; this '
'can have a minor effect on the scale of source '
'estimates that are computed using "nave".')
warned = True
write_int(fid, FIFF.FIFF_NAVE, nave_int)
del nave_int
decal = np.zeros((e.info['nchan'], 1))
for k in range(e.info['nchan']):
decal[k] = 1.0 / (e.info['chs'][k]['cal'] *
e.info['chs'][k].get('scale', 1.0))
write_float_matrix(fid, FIFF.FIFF_EPOCH, decal * e.data)
start_block(fid, FIFF.FIFFB_EVENTS)
if event.get('channels') is not None:
write_int(fid, FIFF.FIFF_EVENT_CHANNELS, event['channels'])
if event.get('list') is not None:
write_int(fid, FIFF.FIFF_EVENT_LIST, event['list'])
end_block(fid, FIFF.FIFFB_EVENTS)
# HPI Result
for hpi_result in info['hpi_results']:
start_block(fid, FIFF.FIFFB_HPI_RESULT)
write_dig_points(fid, hpi_result['dig_points'])
if 'order' in hpi_result:
write_int(fid, FIFF.FIFF_HPI_DIGITIZATION_ORDER,
hpi_result['order'])
if 'used' in hpi_result:
write_int(fid, FIFF.FIFF_HPI_COILS_USED, hpi_result['used'])
if 'moments' in hpi_result:
write_float_matrix(fid, FIFF.FIFF_HPI_COIL_MOMENTS,
hpi_result['moments'])
if 'goodness' in hpi_result:
write_float(fid, FIFF.FIFF_HPI_FIT_GOODNESS,
hpi_result['goodness'])
if 'good_limit' in hpi_result:
write_float(fid, FIFF.FIFF_HPI_FIT_GOOD_LIMIT,
hpi_result['good_limit'])
if 'dist_limit' in hpi_result:
write_float(fid, FIFF.FIFF_HPI_FIT_DIST_LIMIT,
hpi_result['dist_limit'])
if 'accept' in hpi_result:
write_int(fid, FIFF.FIFF_HPI_FIT_ACCEPT, hpi_result['accept'])
if 'coord_trans' in hpi_result:
write_coord_trans(fid, hpi_result['coord_trans'])
_check_option('fmt', fmt, ['single', 'double'])
if np.iscomplexobj(data):
if fmt == 'single':
write_function = write_complex_float_matrix
elif fmt == 'double':
write_function = write_complex_double_matrix
else:
if fmt == 'single':
write_function = write_float_matrix
elif fmt == 'double':
write_function = write_double_matrix
start_block(fid, FIFF.FIFFB_MNE_EVENTS)
write_int(fid, FIFF.FIFF_MNE_EVENT_LIST, epochs.events.T)
mapping_ = ';'.join([k + ':' + str(v) for k, v in
epochs.event_id.items()])
write_string(fid, FIFF.FIFF_DESCRIPTION, mapping_)
end_block(fid, FIFF.FIFFB_MNE_EVENTS)
# Metadata
if epochs.metadata is not None:
start_block(fid, FIFF.FIFFB_MNE_METADATA)
metadata = _prepare_write_metadata(epochs.metadata)
write_string(fid, FIFF.FIFF_DESCRIPTION, metadata)
end_block(fid, FIFF.FIFFB_MNE_METADATA)
# First and last sample
first = int(round(epochs.tmin * info['sfreq'])) # round just to be safe
last = first + len(epochs.times) - 1
if hs.get('hpi_coils') is not None:
for coil in hs['hpi_coils']:
start_block(fid, FIFF.FIFFB_HPI_COIL)
if coil.get('event_bits') is not None:
write_int(fid, FIFF.FIFF_EVENT_BITS,
coil['event_bits'])
end_block(fid, FIFF.FIFFB_HPI_COIL)
end_block(fid, FIFF.FIFFB_HPI_SUBSYSTEM)
del hs
# CTF compensation info
write_ctf_comp(fid, info['comps'])
# KIT system ID
if info.get('kit_system_id') is not None:
write_int(fid, FIFF.FIFF_MNE_KIT_SYSTEM_ID, info['kit_system_id'])
end_block(fid, FIFF.FIFFB_MEAS_INFO)
# Processing history
_write_proc_history(fid, info)
write_coord_trans(fid, this['mri_ras_t'])
write_coord_trans(fid, this['vox_mri_t'])
mri_volume_name = this.get('mri_volume_name', None)
if mri_volume_name is not None:
write_string(fid, FIFF.FIFF_MNE_FILE_NAME, mri_volume_name)
write_float_sparse_rcs(fid, FIFF.FIFF_MNE_SOURCE_SPACE_INTERPOLATOR,
this['interpolator'])
if 'mri_file' in this and this['mri_file'] is not None:
write_string(fid, FIFF.FIFF_MNE_SOURCE_SPACE_MRI_FILE,
this['mri_file'])
write_int(fid, FIFF.FIFF_MRI_WIDTH, this['mri_width'])
write_int(fid, FIFF.FIFF_MRI_HEIGHT, this['mri_height'])
write_int(fid, FIFF.FIFF_MRI_DEPTH, this['mri_depth'])
end_block(fid, FIFF.FIFFB_MNE_PARENT_MRI_FILE)
# Patch-related information
if this['nearest'] is not None:
write_int(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NEAREST, this['nearest'])
write_float_matrix(fid, FIFF.FIFF_MNE_SOURCE_SPACE_NEAREST_DIST,
this['nearest_dist'])
# Distances
if this['dist'] is not None:
# Save only upper triangular portion of the matrix
dists = this['dist'].copy()
dists = sparse.triu(dists, format=dists.format)
write_float_sparse_rcs(fid, FIFF.FIFF_MNE_SOURCE_SPACE_DIST, dists)