Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
else:
out.set(group, 'channels', clist + '\n%s' % entry)
if isinstance(fobj, FILE_LIKE):
close = False
else:
fobj = open(fobj, 'w')
close = True
out.write(fobj)
if close:
fobj.close()
registry.register_reader('ini', ChannelList, read_channel_list_file)
registry.register_identifier('ini', ChannelList,
identify_factory('.ini', '.clf'))
registry.register_writer('ini', ChannelList, write_channel_list_file)
write_timeseriesdict
for available arguments and keyword arguments
"""
return write_timeseriesdict({None: series}, *args, **kwargs)
# -- register -------------------------------
# register specific format
register_reader(fmt, TimeSeriesDict, read_timeseriesdict)
register_reader(fmt, TimeSeries, read_timeseries)
register_reader(fmt, StateVectorDict, read_statevectordict)
register_reader(fmt, StateVector, read_statevector)
register_writer(fmt, TimeSeriesDict, write_timeseriesdict)
register_writer(fmt, TimeSeries, write_timeseries)
register_writer(fmt, StateVectorDict, write_timeseriesdict)
register_writer(fmt, StateVector, write_timeseries)
def register_hdf5_array_io(array_type, format='hdf5', identify=True):
"""Registry read() and write() methods for the HDF5 format
"""
def from_hdf5(*args, **kwargs):
"""Read an array from HDF5
"""
kwargs.setdefault('array_type', array_type)
return read_hdf5_array(*args, **kwargs)
io_registry.register_reader(format, array_type, from_hdf5)
if issubclass(array_type, Series):
io_registry.register_writer(format, array_type, write_hdf5_series)
else:
io_registry.register_writer(format, array_type, write_hdf5_array)
if identify:
io_registry.register_identifier(format, array_type,
io_hdf5.identify_hdf5)
else:
h5g = h5f
# write each timeseries
kwargs.setdefault('format', 'hdf5')
for key, series in tsdict.items():
series.write(h5g, path=str(key), **kwargs)
# -- register -----------------------------------------------------------------
# series classes
for series_class in (TimeSeries, StateVector):
reader = read_hdf5_factory(series_class)
io_registry.register_reader('hdf5', series_class, reader)
io_registry.register_writer('hdf5', series_class, write_hdf5_series)
io_registry.register_identifier('hdf5', series_class, identify_hdf5)
# dict classes
for dict_class in (TimeSeriesDict, StateVectorDict):
reader = read_hdf5_factory(dict_class)
io_registry.register_reader('hdf5', dict_class, reader)
io_registry.register_writer('hdf5', dict_class, write_hdf5_dict)
io_registry.register_identifier('hdf5', dict_class, identify_hdf5)
# -- register -----------------------------------------------------------------
# series classes
for series_class in (TimeSeries, StateVector):
reader = read_hdf5_factory(series_class)
io_registry.register_reader('hdf5', series_class, reader)
io_registry.register_writer('hdf5', series_class, write_hdf5_series)
io_registry.register_identifier('hdf5', series_class, identify_hdf5)
# dict classes
for dict_class in (TimeSeriesDict, StateVectorDict):
reader = read_hdf5_factory(dict_class)
io_registry.register_reader('hdf5', dict_class, reader)
io_registry.register_writer('hdf5', dict_class, write_hdf5_dict)
io_registry.register_identifier('hdf5', dict_class, identify_hdf5)
frame number to write into frame header
"""
return write_timeseriesdict(
{timeseries.name: timeseries}, outfile, *args, **kwargs)
# -- registration -------------------------------------------------------------
# TODO: remove force=True when frameCPP reading is removed
register_reader('gwf', TimeSeries, read_timeseries, force=True)
register_reader('gwf', TimeSeriesDict, read_timeseriesdict, force=True)
register_reader('gwf', StateVector, read_statevector, force=True)
register_reader('gwf', StateVectorDict, read_statevectordict, force=True)
register_writer('gwf', TimeSeries, write_timeseries, force=True)
register_writer('gwf', TimeSeriesDict, write_timeseriesdict, force=True)
register_writer('gwf', StateVector, write_timeseries, force=True)
register_writer('gwf', StateVectorDict, write_timeseriesdict, force=True)
_identify = identify_factory('gwf') # pylint:disable=invalid-name
for cls in (TimeSeries, TimeSeriesDict, StateVector, StateVectorDict):
register_identifier('gwf', cls, _identify, force=True)
def register_ascii_series_io(array_type, format='txt', identify=True,
**defaults):
"""Register ASCII read/write/identify methods for the given array
"""
def _read(filepath, **kwargs):
kwgs = defaults.copy()
kwgs.update(kwargs)
return read_ascii_series(filepath, array_type=array_type, **kwgs)
def _write(series, output, **kwargs):
kwgs = defaults.copy()
kwgs.update(kwargs)
return write_ascii_series(series, output, **kwgs)
io_registry.register_reader(format, array_type, _read)
io_registry.register_writer(format, array_type, _write)
if identify:
io_registry.register_identifier(format, array_type,
identify_factory(format))
See also
--------
write_timeseriesdict
for available arguments and keyword arguments
"""
return write_timeseriesdict({None: series}, *args, **kwargs)
# -- register -------------------------------
# register specific format
register_reader(fmt, TimeSeriesDict, read_timeseriesdict)
register_reader(fmt, TimeSeries, read_timeseries)
register_reader(fmt, StateVectorDict, read_statevectordict)
register_reader(fmt, StateVector, read_statevector)
register_writer(fmt, TimeSeriesDict, write_timeseriesdict)
register_writer(fmt, TimeSeries, write_timeseries)
register_writer(fmt, StateVectorDict, write_timeseriesdict)
register_writer(fmt, StateVector, write_timeseries)
# write file object
if header:
print('# seg\tstart\tstop\tduration', file=target)
for i, seg in enumerate(segs):
print(
'\t'.join(map(
str, (i, coltype(seg[0]), coltype(seg[1]), coltype(abs(seg))),
)), file=target,
)
# -- register -----------------------------------------------------------------
registry.register_reader('segwizard', SegmentList, from_segwizard)
registry.register_writer('segwizard', SegmentList, to_segwizard)
registry.register_identifier('segwizard', SegmentList,
identify_factory('txt', 'dat'))