# http://pyrocko.org - GPLv3
#
# The Pyrocko Developers, 21st Century
# ---|P------/S----------~Lg----------
'''
Waveform data storage utilities.
'''
import math
import os
from collections import defaultdict
from pyrocko import guts, util, trace, io, carpet
from pyrocko.io import rug
from pyrocko.plot import nice_time_tick_inc_approx_secs
from . import error
guts_prefix = 'squirrel'
nsamples_block = 100000
nsamples_segment = 1024**2
def _translate_path_template(s):
map_name = {
'year': '%(wmin_year)s',
'month': '%(wmin_month)s',
'day': '%(wmin_day)s',
'jday': '%(wmin_jday)s',
'hour': '%(wmin_hour)s',
'minute': '%(wmin_minute)s',
'second': '%(wmin_second)s',
'net': '%(network_dsafe)s',
'sta': '%(station_dsafe)s',
'loc': '%(location_dsafe)s',
'cha': '%(channel_dsafe)s',
'ext': '%(extra_dsafe)s'}
return os.path.join(*[
'.'.join(map_name[x] for x in entry.split('.'))
for entry in s.split('/')])
[docs]
def time_to_template_vars(prefix, t):
'''
Get dict of path-template substitution variables for a time instant.
:param prefix:
Prefix to prepend to each generated variable name (e.g.
``'wmin'`` or ``'wmax'``).
:type prefix:
str
:param t:
Time instant to be converted.
:type t:
:py:func:`~pyrocko.util.get_time_float`
:returns:
Dict with keys ``'<prefix>_year'``, ``'<prefix>_month'``,
``'<prefix>_day'``, ``'<prefix>_jday'``, ``'<prefix>_hour'``,
``'<prefix>_minute'``, ``'<prefix>_second'`` and (for backwards
compatibility) ``'<prefix>'``.
:rtype:
dict
'''
d = dict(zip(
[prefix + '_' + var
for var in [
'year', 'month', 'day', 'jday', 'hour', 'minute', 'second']],
util.time_to_str(t, '%Y.%m.%d.%j.%H.%M.%S').split('.')))
# needed for backwards compatibility
d[prefix] = util.time_to_str(t, '%Y-%m-%d_%H-%M-%S')
return d
[docs]
def iter_windows(tmin, tmax, tinc, tinc_nonuniform):
'''
Iterate over time windows covering a given time span.
:param tmin:
Start of the time span to be covered.
:type tmin:
:py:func:`~pyrocko.util.get_time_float`
:param tmax:
End of the time span to be covered.
:type tmax:
:py:func:`~pyrocko.util.get_time_float`
:param tinc:
Window length [s], or ``None`` to yield a single window covering
``[tmin, tmax]``.
:type tinc:
float
:param tinc_nonuniform:
If given, ``tinc`` is ignored and windows follow non-uniform
calendar boundaries instead: ``'month'`` or ``'year'``.
:type tinc_nonuniform:
str
:yields:
``(wmin, wmax)`` tuples.
:raises SquirrelError:
If ``tinc_nonuniform`` is set to an unsupported value.
'''
if tinc is None:
yield tmin, tmax
elif tinc_nonuniform is None:
tmin = math.floor(tmin / tinc) * tinc
t = tmin
while t < tmax:
yield t, t + tinc
t += tinc
elif tinc_nonuniform == 'month':
yield from util.iter_months(tmin, tmax)
elif tinc_nonuniform == 'year':
yield from util.iter_years(tmin, tmax)
else:
raise error.SquirrelError(
'Available non-uniform time intervals: month, year. '
'Invalid choice: %s' % tinc_nonuniform)
[docs]
class StorageSchemeLayout(guts.Object):
'''
Specific directory layout within a storage scheme.
'''
name = guts.String.T(
help='Name of the layout, informational.')
time_increment = guts.Float.T(
optional=True,
help='Time window length stored in each file[s]. Exact or '
'approximate, depending on '
':py:gattr:`time_increment_nonuniform`.')
time_increment_nonuniform = guts.String.T(
optional=True,
help='Identifier for non-uniform time windows. E.g. ``\'month\'`` or '
'``\'year\'``.')
path_template = guts.String.T(
help='Template for file paths.')
[docs]
def get_additional(self, wmin, wmax):
'''
Get path-template substitution variables for a time window.
:param wmin:
Start of the window.
:type wmin:
:py:func:`~pyrocko.util.get_time_float`
:param wmax:
End of the window.
:type wmax:
:py:func:`~pyrocko.util.get_time_float`
:returns:
Combined ``'wmin_*'`` and ``'wmax_*'`` variables, see
:py:func:`time_to_template_vars`.
:rtype:
dict
'''
d = {}
d.update(time_to_template_vars('wmin', wmin))
d.update(time_to_template_vars('wmax', wmax))
return d
[docs]
def describe(self, deltat):
'''
Get one-line summary of file sizing for a given sampling interval,
for display purposes (see :py:meth:`describe_header` for the
matching column header).
:param deltat:
Sampling interval [s] to describe file sizing for.
:type deltat:
float
:rtype:
str
'''
rate = 1.0 / deltat
tblock = nice_time_tick_inc_approx_secs(deltat * nsamples_block)
tsegment = deltat * nsamples_segment
segments_per_file = self.time_increment / tsegment \
if self.time_increment is not None else 0.0,
bytesize = (self.time_increment or 0.0) / deltat * 4
return '%8.1f %11.5f %6.0f %10s %12.0f %10s %6s %4.0f %8s %6i' % (
rate,
deltat,
tblock,
guts.str_duration(tblock),
tsegment,
guts.str_duration(tsegment),
self.name,
segments_per_file,
util.human_bytesize(bytesize),
len(self.path_template.split('/')))
[docs]
class StorageScheme(guts.Object):
'''
Storage scheme for waveform archive data.
'''
name = guts.String.T(
help='Storage scheme name.')
layouts = guts.List.T(
StorageSchemeLayout.T(),
help='Directory layouts supported by the scheme.')
min_segments_per_file = guts.Float.T(
default=1.0,
help='Target minimum number of segments to be stored in each file.')
format = guts.String.T(
default='mseed',
help='File format of waveform data files.')
description = guts.String.T(
default='',
help='Description of the storage scheme.')
mseed_record_length = guts.IntChoice.T(
optional=True,
choices=list(io.mseed.VALID_RECORD_LENGTHS),
help='MiniSEED record length [bytes] to use when saving. '
'Default: 4096.')
mseed_steim = guts.IntChoice.T(
optional=True,
choices=[1, 2],
help='MiniSEED STEIM compression variant to use when saving, '
'``1`` or ``2``. Default: 2.')
[docs]
def post_init(self):
'''
Guts post-initialization hook, setting up the (unset) base path.
'''
self._base_path = None
[docs]
def set_base_path(self, base_path):
'''
Set base directory prepended to layout path templates.
:param base_path:
Base directory path, or ``None`` to use path templates as-is.
:type base_path:
str
'''
self._base_path = base_path
[docs]
def select_layout(self, deltat):
'''
Select the most appropriate directory layout for a given sampling
interval.
:param deltat:
Sampling interval [s] to select a layout for.
:type deltat:
float
:returns:
The first of :py:attr:`layouts` whose
:py:attr:`~StorageSchemeLayout.time_increment` is large enough
to hold :py:attr:`min_segments_per_file` segments at the given
sampling interval (or has no time increment restriction),
falling back to the last layout.
:rtype:
:py:class:`StorageSchemeLayout`
'''
tsegment = deltat * nsamples_segment
twant = tsegment * self.min_segments_per_file
for layout in self.layouts:
if layout.time_increment is None or layout.time_increment > twant:
return layout
return layout
def save(
self, items, **kwargs):
if isinstance(items, (trace.Trace, carpet.Carpet)):
items = [items]
if self.format == 'rug':
if not all(isinstance(item, carpet.Carpet) for item in items):
raise error.SquirrelError(
'Items to be stored with StorageScheme.save must be '
'of type `Carpet` when "rug" format is selected.')
return self.save_carpets(
items, **kwargs)
else:
traces = []
for item in items:
if isinstance(item, carpet.Carpet):
traces.extend(item.get_traces())
else:
assert isinstance(item, trace.Trace)
traces.append(item)
return self.save_traces(
traces, **kwargs)
def save_traces(
self, traces, tmin=None, tmax=None, overwrite=False,
check_append_hook=None, check_append_merge=False):
if isinstance(traces, trace.Trace):
traces = [traces]
save_kwargs = dict(
append=self.format in io.g_formats_supporting_append,
check_append=True,
format=self.format,
overwrite=overwrite,
check_append_hook=check_append_hook,
check_append_merge=check_append_merge)
if self.format == 'mseed':
save_kwargs.update(
record_length=self.mseed_record_length or 4096,
steim=self.mseed_steim or 2)
by_deltat = defaultdict(list)
for tr in traces:
by_deltat[tr.deltat].append(tr)
file_names = set()
for deltat, traces_group in by_deltat.items():
layout = self.select_layout(deltat)
traces_group.sort(key=lambda tr: tr.full_id)
traces_group = trace.deoverlap(traces_group)
if tmin is None:
tmin = min(tr.tmin for tr in traces_group)
if tmax is None:
tmax = max(tr.tmax for tr in traces_group) + deltat
for wmin, wmax in iter_windows(
tmin,
tmax,
layout.time_increment,
layout.time_increment_nonuniform):
traces_window = []
for tr in traces_group:
try:
traces_window.append(
tr.chop(wmin, wmax, inplace=False))
except trace.NoData:
pass
additional = layout.get_additional(wmin, wmax)
file_names.update(io.save(
traces_window,
layout.path_template
if self._base_path is None
else os.path.join(self._base_path, layout.path_template),
additional=additional,
**save_kwargs))
return sorted(file_names)
def save_carpets(
self, carpets, tmin=None, tmax=None, overwrite=False,
check_append_hook=None, check_append_merge=False):
if isinstance(carpets, carpet.Carpet):
carpets = [carpets]
by_deltat = defaultdict(list)
for cp in carpets:
by_deltat[cp.deltat].append(cp)
save_kwargs = dict(
append=True,
check_append=True,
overwrite=overwrite,
check_append_merge=check_append_merge,
check_append_hook=check_append_hook)
file_names = set()
for deltat, carpet_group in by_deltat.items():
layout = self.select_layout(deltat)
carpet_group.sort(key=lambda cp: cp.codes)
carpet_group = carpet.deoverlap(carpet_group)
tmin_group = tmin if tmin is not None else \
min(cp.tmin for cp in carpet_group)
tmax_group = tmax if tmax is not None else \
max(cp.tmax for cp in carpet_group) + deltat
for wmin, wmax in iter_windows(
tmin_group,
tmax_group,
layout.time_increment,
layout.time_increment_nonuniform):
carpet_window = []
for cp in carpet_group:
try:
carpet_window.append(cp.chop(wmin, wmax))
except trace.NoData:
pass
additional = layout.get_additional(wmin, wmax)
file_names.update(rug.save(
carpet_window,
layout.path_template
if self._base_path is None
else os.path.join(self._base_path, layout.path_template),
additional=additional,
**save_kwargs))
return sorted(file_names)
_g_schemes_list = []
_g_schemes_list.append(StorageScheme(
name='default',
description='Dynamic storage scheme with balanced file sizes of '
'10 - 400 MB and a balanced directory hierarchy of 4-6 levels',
layouts=[
StorageSchemeLayout(
name=name,
time_increment=time_increment,
time_increment_nonuniform=time_increment_nonuniform,
path_template=_translate_path_template(path_template))
for (name, time_increment, time_increment_nonuniform, path_template)
in [
('second', 1.0, None, 'net/sta/loc.cha/year/month/day/hour/net.sta.loc.cha.ext.year.month.day.hour.minute.second'), # noqa
('minute', 60.0, None,'net/sta/loc.cha/year/month/day/net.sta.loc.cha.ext.year.month.day.hour.minute'), # noqa
('hour', 3600.0, None,'net/sta/loc.cha/year/month/net.sta.loc.cha.ext.year.month.day.hour'), # noqa
('day', 86400.0, None,'net/sta/loc.cha/year/net.sta.loc.cha.ext.year.month.day'), # noqa
('month', 2628000.0, 'month','net/sta/loc.cha/year/net.sta.loc.cha.ext.year.month'), # noqa
('year', 31536000.0, 'year', 'net/sta/loc.cha/net.sta.loc.cha.ext.year')]], # noqa
min_segments_per_file=1.5))
_g_schemes_list.extend(StorageScheme(
name='rug-store-%i' % factor,
description='Storage scheme used to store carpets with typically %i '
'components as rug files.',
format='rug',
layouts=[
StorageSchemeLayout(
name=name,
time_increment=time_increment,
time_increment_nonuniform=time_increment_nonuniform,
path_template=_translate_path_template(path_template))
for (name, time_increment, time_increment_nonuniform, path_template)
in [
('second', 1.0, None, 'net/sta/loc.cha/year/month/day/hour/net.sta.loc.cha.ext.year.month.day.hour.minute.second'), # noqa
('minute', 60.0, None,'net/sta/loc.cha/year/month/day/net.sta.loc.cha.ext.year.month.day.hour.minute'), # noqa
('hour', 3600.0, None,'net/sta/loc.cha/year/month/net.sta.loc.cha.ext.year.month.day.hour'), # noqa
('day', 86400.0, None,'net/sta/loc.cha/year/net.sta.loc.cha.ext.year.month.day'), # noqa
('month', 2628000.0, 'month','net/sta/loc.cha/year/net.sta.loc.cha.ext.year.month'), # noqa
('year', 31536000.0, 'year', 'net/sta/loc.cha/net.sta.loc.cha.ext.year')]], # noqa
min_segments_per_file=1.5/factor) for factor in [10, 100, 1000])
_g_schemes_list.append(StorageScheme(
name='sds',
description='Directory scheme conforming to SeisComP Data Structure (SDS) '
'archive format (https://www.seiscomp.de/seiscomp3/doc'
'/applications/slarchive/SDS.html). The scheme has a fixed '
'layout with day files.',
layouts=[
StorageSchemeLayout(
name='sds',
time_increment=24*3600.,
path_template=os.path.join(
'%(wmin_year)s',
'%(network_safe)s',
'%(station_safe)s',
'%(channel_safe)s.D',
'%(network_safe)s.%(station)s.%(location)s.%(channel)s.D'
'.%(wmin_year)s.%(wmin_jday)s'))]))
g_schemes = dict((scheme.name, scheme) for scheme in _g_schemes_list)
[docs]
def get_storage_scheme(name='default'):
'''
Get a fresh copy of a named built-in storage scheme.
:param name:
Name of the storage scheme, see :py:class:`StorageSchemeChoice`.
:type name:
str
:rtype:
:py:class:`StorageScheme`
'''
return guts.clone(g_schemes[name])
[docs]
class StorageSchemeChoice(guts.StringChoice):
'''
Name of a supported storage scheme.
'''
choices = list(g_schemes.keys())
__all__ = [
'get_storage_scheme',
'StorageScheme',
'StorageSchemeLayout',
'StorageSchemeChoice'
]
