# -*- coding: utf8 -*-
# SPDX-License-Identifier: GPL-3.0-or-later
"""
Uses ObsPy mass downloader to download event waveforms from FDSN web services.
:copyright:
2022-2026 Claudio Satriano <satriano@ipgp.fr>
:license:
GNU General Public License v3.0 or later
(https://www.gnu.org/licenses/gpl-3.0-standalone.html)
"""
import sys
import pathlib
import logging
from obspy.clients.fdsn.mass_downloader import (
CircularDomain, Restrictions, MassDownloader
)
from .event_waveforms_utils import prefer_high_sampling_rate
from ..utils import ExceptionExit
mdl_logger = logging.getLogger('obspy.clients.fdsn.mass_downloader')
def _parse_magnitude_bins(magnitude, station_radius_max_mag):
"""
Parse magnitude bins and calculate radius using discrete binning.
Uses the radius from the largest bin magnitude that doesn't exceed
the event magnitude.
:param magnitude: Event magnitude
:param station_radius_max_mag: String with format
"mag1: radius1, mag2: radius2, ..."
:return: Calculated radius
:raises ValueError: If parsing fails
:raises IndexError: If parsing fails
"""
# Parse magnitude bins
bins = [
(float(mag_str.strip()), float(rad_str.strip()))
for pair in station_radius_max_mag.split(',')
for mag_str, rad_str in [pair.split(':')]
]
# Sort bins by magnitude
bins.sort(key=lambda x: x[0])
# Find the appropriate radius using discrete binning
# Use the radius from the largest bin magnitude that doesn't exceed
# the event magnitude
radius = bins[0][1] # Default to first bin
for mag_bin, rad_bin in bins:
if magnitude >= mag_bin:
radius = rad_bin
else:
break
return radius
def _evaluate_magnitude_expression(magnitude, station_radius_max_mag):
"""
Evaluate a mathematical expression to calculate radius.
:param magnitude: Event magnitude
:param station_radius_max_mag: Mathematical expression string
(e.g., "2.0 * mag - 3.0")
:return: Calculated radius
:raises ValueError: If expression is invalid
:raises Exception: If evaluation fails
"""
# Create a safe namespace with only allowed operations
safe_dict = {
'mag': magnitude,
'__builtins__': {},
'abs': abs,
'min': min,
'max': max,
'pow': pow,
}
# Evaluate the expression
# pylint: disable=eval-used
radius = eval(station_radius_max_mag, safe_dict)
if not isinstance(radius, (int, float)):
raise ValueError('Expression must evaluate to a number')
return radius
def _calculate_station_radius_max(
magnitude, station_radius_max_mag, station_radius_min,
station_radius_max):
"""
Calculate the magnitude-dependent maximum station radius.
:param magnitude: Event magnitude
:param station_radius_max_mag: Configuration string for
magnitude-dependent radius. Can be either:
- Magnitude bins (discrete): "mag1: radius1, mag2: radius2, ..."
- Mathematical expression: "2.0 * mag - 3.0"
:param station_radius_min: Minimum station radius (lower bound)
:param station_radius_max: Maximum station radius (upper bound)
:return: Calculated radius, bounded by station_radius_min and
station_radius_max
"""
if station_radius_max_mag is None:
return station_radius_max
# Try to parse as magnitude bins first
# (format: "mag1: radius1, mag2: radius2, ...")
if ':' in station_radius_max_mag:
try:
radius = _parse_magnitude_bins(
magnitude, station_radius_max_mag)
except (ValueError, IndexError) as e:
mdl_logger.warning(
'Failed to parse station_radius_max_mag as bins: %s. '
'Using default station_radius_max.', e
)
return station_radius_max
else:
# Try to parse as mathematical expression
try:
radius = _evaluate_magnitude_expression(
magnitude, station_radius_max_mag)
except Exception as e: # pylint: disable=broad-except
mdl_logger.warning(
'Failed to evaluate station_radius_max_mag expression: '
'%s. Using default station_radius_max.', e
)
return station_radius_max
# Apply bounds
radius = max(station_radius_min, min(radius, station_radius_max))
return radius
def _check_fdsn_providers(fdsn_providers):
"""
Check if the user wants to use all known FDSN providers.
Exit if the user does not want to use any provider.
:param providers: list of FDSN providers or None
"""
if fdsn_providers is not None:
return
print(
'No FDSN providers set in config file. Do you want to use all '
'known providers? (y/N)', end=' '
)
while True:
answer = input().strip().lower()
if answer in ('y', 'yes'):
break
if answer in ('n', 'no', ''):
print('Exiting.')
sys.exit()
print('Please answer y or n:', end=' ')
def _set_mdl_logger(evid):
"""
Set up the ObsPy mass downloader logger.
Prepend the event ID to each log message.
:param evid: event ID
"""
mdl_logger.setLevel(logging.DEBUG)
# Prevent propagating to higher loggers.
mdl_logger.propagate = 0
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(f'{evid}: %(message)s'))
handler.setLevel(logging.INFO)
mdl_logger.addHandler(handler)
def _unset_mdl_logger():
"""
Unset the ObsPy mass downloader logger.
"""
mdl_logger.handlers = []