Skip to content

Commit

Permalink
New config parameter clipping_min_amplitude_ratio
Browse files Browse the repository at this point in the history
To set a threshold for trace amplitude below which the trace is not
checked for clipping
  • Loading branch information
claudiodsf committed Oct 21, 2024
1 parent 7223a09 commit cf37cee
Show file tree
Hide file tree
Showing 4 changed files with 138 additions and 3 deletions.
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ Copyright (c) 2011-2024 Claudio Satriano <[email protected]>
- New config parameter `refine_theoretical_arrivals` to refine the
theoretical P and S arrival times using a simple autopicker based on the
smoothed envelope of the trace
- New config parameter `clipping_min_amplitude_ratio` to set a threshold for
trace amplitude below which the trace is not checked for clipping

### Plotting

Expand All @@ -29,6 +31,7 @@ Copyright (c) 2011-2024 Claudio Satriano <[email protected]>

- New config parameters: `refine_theoretical_arrivals`, `autopick_freqmin`,
`autopick_debug_plot`
- New config parameter `clipping_min_amplitude_ratio`
- Improved documentation for the `sn_min` and `spectral_sn_min` parameters

### Bugfixes
Expand Down
117 changes: 115 additions & 2 deletions sourcespec/clipping_detection.py
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,53 @@ def compute_clipping_score(trace, remove_baseline=False, debug=False):
return clipping_score


def check_min_amplitude(trace, min_amplitude_ratio):
"""
Check if trace amplitude is above a minimum threshold computed from the
instrument sensitivity.
Parameters
----------
trace : :class:`~obspy.core.trace.Trace`
Trace to check. Must have an attached inventory.
min_amplitude_fraction : float
Minimum trace amplitude, as a fraction of the overall sensitivity of
the instrument, to check for clipping. Valid values are between 0 and
1.
Returns
-------
bool
True if trace amplitude is above the minimum threshold,
False otherwise.
trace_max : float
Maximum amplitude of the trace, in counts
min_threshold : float
Minimum amplitude threshold, in counts
"""
trace_max = np.max(np.abs(trace.data))
if min_amplitude_ratio is None:
return True, trace_max, None
if min_amplitude_ratio <= 0:
raise ValueError(
'min_amplitude_ratio must be a strictly positive value')
if not getattr(trace.stats, 'inventory', None):
raise RuntimeError(
f'{trace.id}: cannot get instrtype from inventory: '
'inventory is empty: skipping trace')
try:
resp = trace.stats.inventory.get_response(
trace.id, trace.stats.starttime)
except Exception as e:
raise RuntimeError(
f'{trace.id}: cannot get response from inventory: {e}'
) from e
overall_sensitivity = resp.instrument_sensitivity.value
min_threshold = overall_sensitivity/min_amplitude_ratio
amplitude_test = trace_max > min_threshold
return amplitude_test, trace_max, min_threshold


def _get_plotting_axes():
"""Get matplotlib axes for plotting"""
# pylint: disable=import-outside-toplevel unused-import
Expand Down Expand Up @@ -408,6 +455,24 @@ def _parse_arguments():
'--remove_baseline', '-r', action='store_true',
help='Remove trace baseline before processing',
default=False)
common_parser.add_argument(
'--min-amplitude-ratio', '-m', type=float, default=None,
help='Minimum ratio between the overall sensitivity of the '
'instrument and the maximum amplitude of the trace (both in counts), '
'to actually perform the clipping check. '
'Traces with max_amp < (sensitivity / MIN_AMPLITUDE_RATIO) '
'will not be checked for clipping. '
'It must be a strictly positive value. Default is None, meaning no '
'amplitude check is performed. '
'This option requires providing a station metadata file or dir.')
common_parser.add_argument(
'--station-metadata', '-w', dest='station_metadata',
action='store', default=None,
help='get station metadata from FILE (directory or single file '
'name). Supported format: StationXML, dataless SEED, SEED '
'RESP, PAZ (SAC polezero format). Station metadata is used to '
'check clipping against the instrument response.',
metavar='FILE')
common_parser.add_argument(
'--debug', '-d', action='store_true',
help='Plot trace, samples histogram, kernel density, and clipping '
Expand All @@ -434,9 +499,50 @@ def _parse_arguments():
sys.stderr.write(
'Error: at least one positional argument is required.\n')
sys.exit(2)
if (
args.min_amplitude_ratio is not None and
args.station_metadata is None
):
parser.error(
'--min-amplitude-ratio requires --station-metadata option')
if (
args.min_amplitude_ratio is not None and
args.min_amplitude_ratio <= 0
):
parser.error(
'--min-amplitude-ratio must be a strictly positive value')
return args


def _add_inventory(trace, inventory):
"""Add inventory to trace."""
if not inventory:
return
net, sta, loc, chan = trace.id.split('.')
inv = inventory.select(
network=net, station=sta, location=loc, channel=chan)
if inv is None:
raise RuntimeError(
f'{trace.id}: cannot get instrtype from inventory: '
'inventory is empty: skipping trace')
trace.stats.inventory = inv


def _run_amplitude_check(trace, args):
"""
Run amplitude check method. Raise exception if amplitude is below
minimum threshold.
"""
_add_inventory(trace, args.inv)
min_ampl_test, tr_max, min_thresh = check_min_amplitude(
trace, args.min_amplitude_ratio)
if not min_ampl_test:
raise RuntimeError(
f'{trace.id}: max amplitude ({tr_max:.1f}) below minimum '
f'threshold ({min_thresh:.1f}): skipping clipping check'
)


# ainsi codes for fancy output
RESET = "\u001b[0m"
RED = "\u001b[31m"
Expand All @@ -449,7 +555,7 @@ def _run_clipping_peaks(trace, args):
trace_clipped, properties = clipping_peaks(
trace, args.sensitivity, args.clipping_percentile, args.debug)
msg = (
f'{trace.id} - '
f'{trace.id}: '
f'total peaks: {properties["npeaks"]}, '
f'clipped peaks: {properties["npeaks_clipped"]}'
)
Expand All @@ -468,13 +574,14 @@ def _run_clipping_score(trace, args):
color = YELLOW
else:
color = RED
print(f'{trace.id} - clipping score: {color}{score:.2f}%{RESET}')
print(f'{trace.id}: clipping score: {color}{score:.2f}%{RESET}')


def _command_line_interface():
"""Command line interface"""
# pylint: disable=import-outside-toplevel
from obspy import read, Stream
from sourcespec.ssp_read_station_metadata import read_station_metadata
args = _parse_arguments()
st = Stream()
for file in args.infile:
Expand All @@ -487,7 +594,13 @@ def _command_line_interface():
if not st:
print('No traces found')
return
args.inv = read_station_metadata(args.station_metadata)
for tr in st:
try:
_run_amplitude_check(tr, args)
except RuntimeError as msg:
print(msg)
continue
if args.command == 'clipping_peaks':
_run_clipping_peaks(tr, args)
elif args.command == 'clipping_score':
Expand Down
8 changes: 8 additions & 0 deletions sourcespec/config_files/configspec.conf
Original file line number Diff line number Diff line change
Expand Up @@ -269,6 +269,14 @@ sn_min = float(min=0, default=0)
# values. Kernel density peaks for each trace are printed on the terminal
# and in the log file.
clipping_detection_algorithm = option('none', 'clipping_score', 'clipping_peaks', default='clipping_score')
# Minimum ratio between the overall sensitivity of the instrument and the
# maximum amplitude of the trace (both in counts), to actually perform the
# clipping check.
# Traces with max_amp < (sensitivity / clipping_min_amplitude_ratio) will not
# be checked for clipping.
# It must be a strictly positive value. Default is None, meaning no amplitude
# check is performed.
clipping_min_amplitude_ratio = float(min=1e-10, default=None)
# Plot a debug figure for each trace with the results of the clipping algorithm
# Note: the figures are always shown, even if "plot_show" is False (see below)
clipping_debug_plot = boolean(default=False)
Expand Down
13 changes: 12 additions & 1 deletion sourcespec/ssp_process_traces.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
from sourcespec.ssp_wave_arrival import add_arrival_to_trace
from sourcespec.ssp_wave_picking import refine_trace_picks
from sourcespec.clipping_detection import (
compute_clipping_score, clipping_peaks)
check_min_amplitude, compute_clipping_score, clipping_peaks)
logger = logging.getLogger(__name__.rsplit('.', maxsplit=1)[-1])


Expand Down Expand Up @@ -108,7 +108,18 @@ def _check_clipping(config, trace):
t2 = trace.stats.arrivals['S2'][1]
elif config.wave_type[0] == 'P':
t2 = trace.stats.arrivals['P2'][1]
else:
# this should never happen
raise ValueError(f'Unknown wave type: {config.wave_type[0]}')
tr = trace.copy().trim(t1, t2)
min_ampl_test, tr_max, min_thresh = check_min_amplitude(
tr, config.clipping_min_amplitude_ratio)
if not min_ampl_test:
logger.info(
f'{trace.id}: max amplitude ({tr_max:.1f}) below minimum '
f'threshold ({min_thresh:.1f}): skipping clipping check'
)
return
if config.clipping_detection_algorithm == 'clipping_score':
score = compute_clipping_score(
tr, config.remove_baseline, config.clipping_debug_plot)
Expand Down

0 comments on commit cf37cee

Please sign in to comment.