# http://pyrocko.org - GPLv3
#
# The Pyrocko Developers, 21st Century
# ---|P------/S----------~Lg----------
'''
Squirrel client to access FDSN web services for seismic waveforms and metadata.
'''
import time
import os
import copy
import logging
import tempfile
import importlib.util
from collections import defaultdict
try:
import cPickle as pickle
except ImportError:
import pickle
import os.path as op
from .base import Source, Constraint
from ..model import make_waveform_promise_nut, ehash, InvalidWaveform, \
order_summary, WaveformOrder, g_tmin, g_tmax, g_tmin_queries, \
codes_to_str_abbreviated, CodesNSLCE
from .. import storage
from ..database import ExecuteGet1Error
from pyrocko.squirrel.error import SquirrelError
from pyrocko.client import fdsn
from pyrocko import util, trace, io
from pyrocko.io.io_common import FileLoadError
from pyrocko.io import stationxml
from pyrocko import progress
from pyrocko import has_paths
from pyrocko.guts import Object, String, Timestamp, List, Tuple, Int, Dict, \
Duration, Bool, clone, dump_all_spickle
guts_prefix = 'squirrel'
logger = logging.getLogger('psq.client.fdsn')
g_sites_not_supporting = {
'startbefore': ['geonet'],
'includerestricted': ['geonet', 'ncedc', 'scedc']}
g_keys_conflicting_post_codes = {
'network', 'station', 'location', 'channel', 'minlatitude', 'maxlatitude',
'minlongitude', 'maxlongitude', 'latitude', 'longitude', 'minradius',
'maxradius'}
[docs]
def make_task(*args):
'''
Create progress indicator task (see :py:mod:`pyrocko.progress`), logging
to this module's logger.
'''
return progress.task(*args, logger=logger)
[docs]
def diff(fn_a, fn_b):
'''
Check if two files differ in size or content.
'''
try:
if os.stat(fn_a).st_size != os.stat(fn_b).st_size:
return True
except OSError:
return True
with open(fn_a, 'rb') as fa:
with open(fn_b, 'rb') as fb:
while True:
a = fa.read(1024)
b = fb.read(1024)
if a != b:
return True
if len(a) == 0 or len(b) == 0:
return False
[docs]
def move_or_keep(fn_temp, fn):
'''
Replace a file with a freshly downloaded temporary one, unless their
contents are identical.
:param fn_temp:
Path of newly downloaded temporary file.
:type fn_temp:
str
:param fn:
Path of target file to be replaced.
:type fn:
str
:returns:
``'updated'``, ``'new'`` or ``'upstream unchanged'``, describing
the action taken.
:rtype:
str
'''
if op.exists(fn):
if diff(fn, fn_temp):
os.rename(fn_temp, fn)
status = 'updated'
else:
os.unlink(fn_temp)
status = 'upstream unchanged'
else:
os.rename(fn_temp, fn)
status = 'new'
return status
[docs]
def combine_selections(selection):
'''
Merge adjacent NSLC-plus-time-span entries into contiguous ones.
:param selection:
Entries of the form ``(net, sta, loc, cha, tmin, tmax)``, sorted by
codes then start time, where adjacent entries sharing the same
codes and touching time spans (end time of one equal to start time
of the next) are merged.
:type selection:
iterable of :py:class:`tuple`
:rtype:
:py:class:`list` of :py:class:`tuple`
'''
out = []
last = None
for this in selection:
if last and this[:4] == last[:4] and this[4] == last[5]:
last = last[:5] + (this[5],)
else:
if last:
out.append(last)
last = this
if last:
out.append(last)
return out
[docs]
def orders_sort_key(order):
'''
Get sort key for a :py:class:`~pyrocko.squirrel.model.WaveformOrder`,
for sorting by codes, then start time.
'''
return (order.codes, order.tmin)
[docs]
def orders_to_selection(orders, pad=1.0):
'''
Convert waveform orders to an FDSN dataselect-style selection.
:param orders:
Orders to be converted.
:type orders:
iterable of
:py:class:`~pyrocko.squirrel.model.WaveformOrder` objects
:param pad:
Number of samples of padding to add at each end of the requested
time spans.
:type pad:
float
:returns:
Combined (see :py:func:`combine_selections`), padded selection
entries of the form ``(net, sta, loc, cha, tmin, tmax)``.
:rtype:
:py:class:`list` of :py:class:`tuple`
'''
selection = []
nslc_to_deltat = {}
for order in sorted(orders, key=orders_sort_key):
selection.append(
order.codes.nslc + (order.tmin, order.tmax))
nslc_to_deltat[order.codes.nslc] = order.deltat
selection = combine_selections(selection)
selection_padded = []
for (net, sta, loc, cha, tmin, tmax) in selection:
deltat = nslc_to_deltat[net, sta, loc, cha]
selection_padded.append((
net, sta, loc, cha, tmin-pad*deltat, tmax+pad*deltat))
return selection_padded
[docs]
def codes_to_selection(codes_list, tmin, tmax):
'''
Convert codes patterns and a time span to an FDSN-style selection.
:param codes_list:
Codes patterns to be converted, or ``None``.
:type codes_list:
:py:class:`list` of
:py:class:`~pyrocko.model.codes.CodesNSLCE` objects, or ``None``
:param tmin:
Start time to be used for all entries.
:type tmin:
:py:func:`~pyrocko.util.get_time_float`
:param tmax:
End time to be used for all entries.
:type tmax:
:py:func:`~pyrocko.util.get_time_float`
:returns:
Selection entries of the form ``(net, sta, loc, cha, tmin, tmax)``,
or ``None`` if ``codes_list`` is ``None``.
:rtype:
:py:class:`list` of :py:class:`tuple` or ``None``
'''
if codes_list is None:
return None
selection = []
for codes in sorted(codes_list):
selection.append(
codes.nslc + (tmin, tmax))
return selection
[docs]
class ErrorEntry(Object):
'''
Single waveform download error occurrence.
'''
time = Timestamp.T(
help='Time at which the error occurred.')
order = WaveformOrder.T(
help='Waveform order which caused the error.')
kind = String.T(
help="Error category, e.g. ``'empty result'``, "
"``'invalid waveform'``, ``'http error'``, ``'aborted'`` or "
"``'partial result'``.")
details = String.T(
optional=True,
help='Additional free-form details about the error.')
[docs]
class ErrorAggregate(Object):
'''
Summary of several :py:class:`ErrorEntry` occurrences sharing the same
``kind``/``details``, see :py:meth:`ErrorLog.iter_aggregates`.
'''
site = String.T(
help='FDSN site url or alias name the errors occurred at.')
kind = String.T(
help='Common error category of the aggregated entries, see '
':py:attr:`ErrorEntry.kind`.')
details = String.T(
help='Common details of the aggregated entries, see '
':py:attr:`ErrorEntry.details`.')
entries = List.T(
ErrorEntry.T(),
help='Aggregated error entries.')
codes = List.T(
CodesNSLCE.T(),
help='Channel codes affected by the aggregated errors.')
time_spans = List.T(
Tuple.T(2, Timestamp.T()),
help='Distinct ``(tmin, tmax)`` time spans affected by the '
'aggregated errors.')
def __str__(self):
codes = [str(x) for x in self.codes]
scodes = '\n' + util.ewrap(codes, indent=' ') if codes else '<none>'
tss = self.time_spans
sspans = '\n' + util.ewrap(('%s - %s' % (
util.time_to_str(ts[0]), util.time_to_str(ts[1])) for ts in tss),
indent=' ')
return ('FDSN "%s": download error summary for "%s" (%i)\n%s '
'Codes:%s\n Time spans:%s') % (
self.site,
self.kind,
len(self.entries),
' Details: %s\n' % self.details if self.details else '',
scodes,
sspans)
[docs]
class ErrorLog(Object):
'''
Log of waveform download errors encountered by :py:class:`FDSNSource`.
'''
site = String.T(
help='FDSN site url or alias name the errors occurred at.')
entries = List.T(
ErrorEntry.T(),
help='Logged error entries, in chronological order.')
checkpoints = List.T(
Int.T(),
help='Entry-count marks recorded by :py:meth:`append_checkpoint`, '
'used to delimit "recent" entries for '
':py:meth:`summarize_recent`.')
[docs]
def append_checkpoint(self):
'''
Mark current end of log, for later use by
:py:meth:`summarize_recent`.
'''
self.checkpoints.append(len(self.entries))
[docs]
def append(self, time, order, kind, details=''):
'''
Append an error entry to the log.
:param time:
Time at which the error occurred.
:type time:
:py:func:`~pyrocko.util.get_time_float`
:param order:
Waveform order which caused the error.
:type order:
:py:class:`~pyrocko.squirrel.model.WaveformOrder`
:param kind:
Error category, see :py:attr:`ErrorEntry.kind`.
:type kind:
str
:param details:
Additional free-form details about the error.
:type details:
str
'''
entry = ErrorEntry(time=time, order=order, kind=kind, details=details)
self.entries.append(entry)
[docs]
def iter_aggregates(self):
'''
Iterate over :py:class:`ErrorAggregate` summaries, grouping logged
entries by ``(kind, details)``.
:yields:
:py:class:`ErrorAggregate` objects.
'''
by_kind_details = defaultdict(list)
for entry in self.entries:
by_kind_details[entry.kind, entry.details].append(entry)
kind_details = sorted(by_kind_details.keys())
for kind, details in kind_details:
entries = by_kind_details[kind, details]
codes = sorted(set(entry.order.codes for entry in entries))
selection = orders_to_selection(entry.order for entry in entries)
time_spans = sorted(set(row[-2:] for row in selection))
yield ErrorAggregate(
site=self.site,
kind=kind,
details=details,
entries=entries,
codes=codes,
time_spans=time_spans)
[docs]
def summarize_recent(self):
'''
Get one-line summary of errors logged since the last checkpoint
(see :py:meth:`append_checkpoint`), or of all entries if no
checkpoint was set.
:rtype:
str
'''
ioff = self.checkpoints[-1] if self.checkpoints else 0
recent = self.entries[ioff:]
kinds = sorted(set(entry.kind for entry in recent))
if recent:
return '%i error%s (%s)' % (
len(recent), util.plural_s(recent), '; '.join(kinds))
else:
return ''
[docs]
class Aborted(SquirrelError):
'''
Raised internally to abort an in-progress waveform download, e.g. when
unexpectedly much data is being received.
'''
[docs]
class FDSNSource(Source):
'''
Squirrel data-source to transparently get data from FDSN web services.
Attaching an :py:class:`FDSNSource` object to a
:py:class:`~pyrocko.squirrel.base.Squirrel` allows the latter to download
station and waveform data from an FDSN web service should the data not
already happen to be available locally.
'''
site = String.T(
help='FDSN site url or alias name (see '
':py:mod:`pyrocko.client.fdsn`).')
query_args = Dict.T(
String.T(), String.T(),
optional=True,
help='Common query arguments, which are appended to all queries.')
codes = List.T(
CodesNSLCE.T(),
optional=True,
help='List of codes patterns to query via POST parameters.')
expires = Duration.T(
optional=True,
help='Expiration time [s]. Information older than this will be '
'refreshed. This only applies to station-metadata. Waveforms do '
'not expire. If set to ``None`` neither type of data expires.')
anxious = Duration.T(
default=600.,
help='Anxiety period [s]. Missing waveforms will not be treated as '
'permanently missing for orders spanning into times later than '
'(current time - anxious). Default: 600.')
cache_path = has_paths.Path.T(
optional=True,
help='Directory path where any downloaded waveforms and station '
'meta-data are to be kept. By default the Squirrel '
"environment's cache directory is used.")
shared_waveforms = Bool.T(
default=False,
help='If ``True``, waveforms are shared with other FDSN sources in '
'the same Squirrel environment. If ``False``, they are kept '
'separate.')
storage_path = has_paths.Path.T(
optional=True,
help='If set, manage waveforms in the given directory rather than '
'in the Squirrel cache.')
storage_scheme = storage.StorageSchemeChoice.T(
default='default',
help="Set layout of waveform storage. Available: %s. "
"Default: ``'default'``." % ', '.join(
"``'%s'``" % name
for name in storage.StorageSchemeChoice.choices))
user_credentials = Tuple.T(
2, String.T(),
optional=True,
help='User and password for FDSN servers requiring password '
'authentication')
auth_token = String.T(
optional=True,
help='Authentication token to be presented to the FDSN server.')
auth_token_path = has_paths.Path.T(
optional=True,
help='Path to file containing the authentication token to be '
'presented to the FDSN server.')
hotfix_module_path = has_paths.Path.T(
optional=True,
help='Path to Python module to locally patch metadata errors.')
def __init__(self, site, query_args=None, codes=None, **kwargs):
'''
Create FDSN data source.
Arguments are as for the Guts fields above. ``codes`` and
``query_args`` are mutually exclusive for the network/station/
location/channel/geographic-area keys (see
``g_keys_conflicting_post_codes``).
:raises SquirrelError:
If both ``codes`` and conflicting ``query_args`` keys are
given.
'''
if codes:
codes = [CodesNSLCE(codes_) for codes_ in codes]
if codes is not None and query_args is not None:
conflicting = g_keys_conflicting_post_codes \
& set(query_args.keys())
if conflicting:
raise SquirrelError(
'Cannot use %s in `query_args` when `codes` are also '
'given.' % ' or '.join("'%s'" % k for k in conflicting))
Source.__init__(
self, site=site, query_args=query_args, codes=codes, **kwargs)
self._constraint = None
self._hash = self.make_hash()
self._source_id = 'client:fdsn:%s' % self._hash
self._error_infos = []
self._cache_path = None
[docs]
def get_cache_path(self):
'''
Get path to this source's on-disk cache directory.
:rtype:
str
'''
return op.join(self._cache_path, self._hash)
[docs]
def describe(self):
'''
Get short identifying description of the source.
:rtype:
str
'''
return self._source_id
[docs]
def make_hash(self):
'''
Compute hash identifying this source's configuration.
Used to distinguish between differently-configured
:py:class:`FDSNSource` instances pointing at the same
:py:attr:`site` when choosing on-disk cache locations, so e.g.
anonymous and authenticated access to the same site are cached
separately.
:rtype:
str
'''
s = self.site
s += 'notoken' \
if (self.auth_token is None and self.auth_token_path is None) \
else 'token'
if self.user_credentials is not None:
s += self.user_credentials[0]
else:
s += 'nocred'
if self.query_args is not None:
s += ','.join(
'%s:%s' % (k, self.query_args[k])
for k in sorted(self.query_args.keys()))
else:
s += 'noqueryargs'
if self.codes is not None:
s += 'post_codes:' + ','.join(
codes.safe_str for codes in self.codes)
if self.hotfix_module_path is not None:
s += 'hotfix:' + self.hotfix_module_path
return ehash(s)
[docs]
def get_hash(self):
'''
Get hash identifying this source's configuration (see
:py:meth:`make_hash`).
:rtype:
str
'''
return self._hash
[docs]
def get_auth_token(self):
'''
Get authentication token, reading it from :py:attr:`auth_token_path`
if :py:attr:`auth_token` is not set directly.
:rtype:
str
:raises FileLoadError:
If :py:attr:`auth_token_path` cannot be read.
'''
if self.auth_token:
return self.auth_token
elif self.auth_token_path is not None:
try:
with open(self.auth_token_path, 'rb') as f:
return f.read().decode('ascii')
except OSError as e:
raise FileLoadError(
'Cannot load auth token file (%s): %s'
% (str(e), self.auth_token_path))
else:
raise Exception(
'FDSNSource: auth_token and auth_token_path are mutually '
'exclusive.')
[docs]
def setup(self, squirrel, check=True, upgrade=False):
'''
Set up cache directories and add already-downloaded local content
(waveforms, channel and response meta-data) to the squirrel.
See :py:meth:`~pyrocko.squirrel.client.base.Source.setup` for the
parameters. If ``upgrade`` is ``True``, any waveform archive found
using the old (pre ``-v2``) on-disk storage layout is migrated in
place; otherwise it is added as-is and a warning is logged.
'''
self._cache_path = op.join(
self.expand_path(self.cache_path)
if self.cache_path
else squirrel._cache_path, 'fdsn')
util.ensuredir(self._cache_path)
self._load_constraint()
self._archive = storage.get_storage_scheme(self.storage_scheme)
waveforms_path = self._get_waveforms_path()
util.ensuredir(waveforms_path)
self._archive.set_base_path(waveforms_path)
for waveforms_path_compat in self._get_waveforms_paths_compat():
if os.path.exists(waveforms_path_compat):
if upgrade:
import shutil
from pyrocko.squirrel.tool.commands.jackseis \
import Converter
logger.info(
'Upgrading waveform archive.\n old: %s\n new: %s',
waveforms_path_compat,
waveforms_path)
converter = Converter(
in_path=waveforms_path_compat,
out_storage_path=waveforms_path,
tinc=3600.)
converter.set_basepath('.')
converter.convert(
squirrel_factory=lambda: squirrel,
append=True)
shutil.rmtree(waveforms_path_compat)
else:
logger.warning(
'Waveform archive with old layout: %s\n💡 Use '
'`squirrel ... --upgrade-storage` or '
'`.add_dataset(..., upgrade=True)` '
'to upgrade.' % waveforms_path_compat)
squirrel.add(waveforms_path_compat, check=check)
squirrel.add(
self._get_waveforms_path(),
check=check)
fn = self._get_channels_path()
if os.path.exists(fn):
squirrel.add(fn)
squirrel.add_virtual(
[], virtual_paths=[self._source_id])
responses_path = self._get_responses_path()
if os.path.exists(responses_path):
squirrel.add(
responses_path, kinds=['response'], exclude=r'\.temp$')
self._hotfix_module = None
def _hotfix(self, query_type, sx):
if self.hotfix_module_path is None:
return
if self._hotfix_module is None:
module_path = self.expand_path(self.hotfix_module_path)
spec = importlib.util.spec_from_file_location(
'hotfix_' + self._hash, module_path)
self._hotfix_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(self._hotfix_module)
hook = getattr(
self._hotfix_module, 'stationxml_' + query_type + '_hook')
return hook(sx)
def _get_constraint_path(self):
return op.join(self._cache_path, self._hash, 'constraint.pickle')
def _get_channels_path(self):
return op.join(self._cache_path, self._hash, 'channels.stationxml')
def _get_responses_path(self, nslc=None):
dirpath = op.join(
self._cache_path, self._hash, 'responses')
if nslc is None:
return dirpath
else:
return op.join(
dirpath, 'response_%s_%s_%s_%s.stationxml' % nslc)
def _get_waveforms_path(self):
if self.storage_path:
return self.expand_path(self.storage_path)
if self.shared_waveforms:
return op.join(self._cache_path, 'waveforms-v2')
else:
return op.join(self._cache_path, self._hash, 'waveforms-v2')
def _get_waveforms_paths_compat(self):
if self.shared_waveforms:
return [op.join(self._cache_path, 'waveforms')]
else:
return [op.join(self._cache_path, self._hash, 'waveforms')]
def _log_meta(self, message, target=logger.info):
log_prefix = 'FDSN "%s" metadata:' % self.site
target(' '.join((log_prefix, message)))
def _log_responses(self, message, target=logger.info):
log_prefix = 'FDSN "%s" responses:' % self.site
target(' '.join((log_prefix, message)))
def _log_info_data(self, *args):
log_prefix = 'FDSN "%s" waveforms:' % self.site
logger.info(' '.join((log_prefix,) + args))
def _str_expires(self, t, now):
if t is None:
return 'expires: never'
else:
expire = 'expires' if t > now else 'expired'
return '%s: %s' % (
expire,
util.time_to_str(t, format='%Y-%m-%d %H:%M:%S'))
[docs]
def update_channel_inventory(self, squirrel, constraint=None):
'''
Query and cache channel/station meta-data for a given constraint,
unless already cached and not yet expired (see
:py:attr:`expires`).
:param squirrel:
Squirrel instance the source is attached to.
:type squirrel:
:py:class:`~pyrocko.squirrel.base.Squirrel`
:param constraint:
Selection of times or areas to be brought up to date. If not
given, an unconstrained (global) query is made.
:type constraint:
:py:class:`~pyrocko.squirrel.client.base.Constraint`
'''
if constraint is None:
constraint = Constraint()
expiration_time = self._get_channels_expiration_time()
now = time.time()
log_target = logger.info
if self._constraint and self._constraint.contains(constraint) \
and (expiration_time is None or now < expiration_time):
status = 'using cached'
else:
if self._constraint:
constraint_temp = copy.deepcopy(self._constraint)
constraint_temp.expand(constraint)
constraint = constraint_temp
try:
channel_sx = self._do_channel_query(constraint)
channel_sx.created = None # timestamp would ruin diff
fn = self._get_channels_path()
util.ensuredirs(fn)
fn_temp = fn + '.%i.temp' % os.getpid()
dump_all_spickle([channel_sx], filename=fn_temp)
# channel_sx.dump_xml(filename=fn_temp)
status = move_or_keep(fn_temp, fn)
if status == 'upstream unchanged':
squirrel.get_database().silent_touch(fn)
self._constraint = constraint
self._dump_constraint()
except OSError as e:
status = 'update failed (%s)' % str(e)
log_target = logger.error
expiration_time = self._get_channels_expiration_time()
self._log_meta(
'%s (%s)' % (status, self._str_expires(expiration_time, now)),
target=log_target)
fn = self._get_channels_path()
if os.path.exists(fn):
squirrel.add(fn)
def _do_channel_query(self, constraint):
extra_args = {}
tmin = constraint.tmin \
if constraint.tmin is not None and constraint.tmin != g_tmin \
else g_tmin_queries
tmax = constraint.tmax \
if constraint.tmax is not None and constraint.tmax != g_tmax \
else g_tmax
if self.site in g_sites_not_supporting['startbefore']:
ktmin = 'starttime'
ktmax = 'endtime'
else:
ktmin = 'endafter'
ktmax = 'startbefore'
if self.codes is None:
extra_args[ktmin] = tmin
extra_args[ktmax] = tmax
if self.site not in g_sites_not_supporting['includerestricted']:
extra_args.update(
includerestricted=(
self.user_credentials is not None
or self.auth_token is not None
or self.auth_token_path is not None))
if self.query_args is not None:
extra_args.update(self.query_args)
self._log_meta('querying...')
try:
channel_sx = fdsn.station(
site=self.site,
format='text',
level='channel',
selection=codes_to_selection(self.codes, tmin, tmax),
**extra_args)
self._hotfix('channel', channel_sx)
return channel_sx
except fdsn.EmptyResult:
return stationxml.FDSNStationXML(source='dummy-empty-result')
except fdsn.DownloadError as e:
raise SquirrelError(str(e))
def _load_constraint(self):
fn = self._get_constraint_path()
if op.exists(fn):
with open(fn, 'rb') as f:
self._constraint = pickle.load(f)
else:
self._constraint = None
def _dump_constraint(self):
with open(self._get_constraint_path(), 'wb') as f:
pickle.dump(self._constraint, f, protocol=2)
def _get_expiration_time(self, path):
if self.expires is None:
return None
try:
t = os.stat(path)[8]
return t + self.expires
except OSError:
return 0.0
def _get_channels_expiration_time(self):
return self._get_expiration_time(self._get_channels_path())
def _get_user_credentials(self):
d = {}
if self.user_credentials is not None:
d['user'], d['passwd'] = self.user_credentials
if self.auth_token is not None or self.auth_token_path is not None:
d['token'] = self.get_auth_token()
return d
def _do_response_query(self, selection):
extra_args = {}
if self.site not in g_sites_not_supporting['includerestricted']:
extra_args.update(
includerestricted=(
self.user_credentials is not None
or self.auth_token is not None
or self.auth_token_path is not None))
self._log_responses('querying...')
try:
response_sx = fdsn.station(
site=self.site,
level='response',
selection=selection,
**extra_args)
self._hotfix('response', response_sx)
return response_sx
except fdsn.EmptyResult:
return stationxml.FDSNStationXML(source='dummy-empty-result')
except fdsn.DownloadError as e:
raise SquirrelError(str(e))
[docs]
def update_response_inventory(self, squirrel, constraint):
'''
Query and cache instrument responses for channels matching a given
constraint's codes, unless already cached and not yet expired (see
:py:attr:`expires`).
:param squirrel:
Squirrel instance the source is attached to.
:type squirrel:
:py:class:`~pyrocko.squirrel.base.Squirrel`
:param constraint:
Selection of codes to update responses for. Only
:py:attr:`~pyrocko.squirrel.client.base.Constraint.codes` is
used; responses are queried for the channels' entire
available time span.
:type constraint:
:py:class:`~pyrocko.squirrel.client.base.Constraint`
'''
cpath = os.path.abspath(self._get_channels_path())
nuts = squirrel.iter_nuts(
'channel', path=cpath, codes=constraint.codes)
tmin = g_tmin_queries
tmax = g_tmax
selection = []
now = time.time()
have = set()
status = defaultdict(list)
for nut in nuts:
nslc = nut.codes.nslc
if nslc in have:
continue
have.add(nslc)
fn = self._get_responses_path(nslc)
expiration_time = self._get_expiration_time(fn)
if os.path.exists(fn) \
and (expiration_time is None or now < expiration_time):
status['using cached'].append(nslc)
else:
selection.append(nslc + (tmin, tmax))
dummy = stationxml.FDSNStationXML(source='dummy-empty')
neach = 100
i = 0
fns = []
while i < len(selection):
selection_now = selection[i:i+neach]
i += neach
try:
sx = self._do_response_query(selection_now)
except Exception as e:
status['update failed (%s)' % str(e)].extend(
entry[:4] for entry in selection_now)
continue
sx.created = None # timestamp would ruin diff
by_nslc = dict(stationxml.split_channels(sx))
for entry in selection_now:
nslc = entry[:4]
response_sx = by_nslc.get(nslc, dummy)
try:
fn = self._get_responses_path(nslc)
fn_temp = fn + '.%i.temp' % os.getpid()
util.ensuredirs(fn_temp)
dump_all_spickle([response_sx], filename=fn_temp)
# response_sx.dump_xml(filename=fn_temp)
status_this = move_or_keep(fn_temp, fn)
if status_this == 'upstream unchanged':
try:
squirrel.get_database().silent_touch(fn)
except ExecuteGet1Error:
pass
status[status_this].append(nslc)
fns.append(fn)
except OSError as e:
status['update failed (%s)' % str(e)].append(nslc)
for k in sorted(status):
if k.find('failed') != -1:
log_target = logger.error
else:
log_target = logger.info
self._log_responses(
'%s: %s' % (
k, codes_to_str_abbreviated(
CodesNSLCE(tup) for tup in status[k])),
target=log_target)
if fns:
squirrel.add(fns, kinds=['response'])
__all__ = [
'FDSNSource',
]
