pysmo.tools.iccs

Iterative Cross-Correlation and Stack (ICCS).

Warning

This module is being developed alongside a complete rewrite of AIMBAT. Expect major changes until the rewrite is complete.

The ICCS¹ method is an iterative algorithm to rapidly determine the best fitting delay times between an arbitrary number of seismograms with minimal involvement by a human operator. Instead of looking at individual seismograms, parameters are set that control the algorithm, which then iteratively aligns seismograms, or discards them from futher consideration if they are of poor quality.

The basic idea of ICCS, is that stacking all seismograms (aligned with respect to an initial, and later improved, phase arrival pick) will lead to the targeted phase arrival becoming visible in the stack. As the stack is generated from all input seismograms, the phase arrival in the stack may be considered a representation of the "best" mean arrival time. Each individual seismogram can then be cross-correlated with the stack to determine a time shift that best alligns them with the stack and thus each other.

The results of ICCS are are similar to those of the the MCCC² method, while also requiring fewer cross-correlations to be computed (each individual seismogram is only cross-correlated with the stack, whereas in MCCC all seismograms are cross-correlated with each other). ICCS is therefore particularly useful to perepare data for a successful MCCC run (e.g. if the initial picks are calculated rather than hand picked).

Data requirements

The iccs module requires seismograms containing extra attributes specific to the ICCS method. Hence it provides a protocol class (ICCSSeismogram) and corresponding Mini class (MiniICCSSeismogram) for that purpose. In addition to the common attributes of a Seismogram in pysmo, the following parameters are required:

Attribute	Description
t0	Initial pick (typically computed). Serves as input only when t1 is not set.
t1	Improved pick. Serves as both input (if not None) and output (always) when running the ICCS algorithm. It should be set to None initially.
select	Determines if a seismogram is used for the stack, and should therefore be True initially. It is set to False for poor quality seismograms automatically during a run if autoselect is True. Note that this flag does not exclude a seismogram from being cross-correlated with the stack. Recovery is therefore possible and previously de-selected seismograms may be selected again for the next iteration.
flip	Determines if the seismogram data should be flipped (i.e. multiplied with -1) before using it in the stack and cross-correlation. Can be automatically toggled when autoflipis True during a run.

Two things worth mentioning:

• Because ICCSSeismogram is a subclass of Seismogram, it can be used anywhere a basic Seismogram is expected.
• All actions performed only effect the attributes listed above. The orignal data, begin_time, etc. are never modified.

Execution flow

The diagram below shows execution flow, and how the above parameters are used when the ICCS algorithm is executed (see here for parameters and default values)':

flowchart TD
Start(["ICCSSeismograms with initial parameters."])
Stack0["Generate windowed seismograms and create stack from them."]
C["Cross-correlate windowed seismograms with stack to obtain updated picks and normalised correlation coefficients."]
FlipQ{"Is **autoflip**
True?"}
Flip["Toggle **flip** attribute of seismograms with negative correlation coefficients."]
QualQ{"Is **autoselect**
True?"}
Qual1["Toggle **select** attribute of seismograms based on correlation coefficient."]
Stack1["Recompute windowed seismograms and stack with updated parameters."]
H{"Convergence
criteria met?"}
I{"Maximum
iterations
reached?"}
End(["ICCSSeismograms with updated **t1**, **flip**, and **select** parameters."])
Start --> Stack0 --> C --> FlipQ -->|No| QualQ -->|No| Stack1 --> H -->|No| I -->|No| C
FlipQ -->|Yes| Flip --> QualQ
QualQ -->|Yes| Qual1 -->  Stack1
H -->|Yes| End
I -->|Yes| End

Convergence is reached when the stack iself is not changing significantly anymore between iteration. Typically this happens within a few iterations.

Operator involvement

The ICCS algorithm relies on a few parameters that need to be adjusted by the user. This module provides functions to visualise the stack and individual seismograms (all at the same time), and to update the parameters based on visual inspection.

---

1. Lou, X., et al. “AIMBAT: A Python/Matplotlib Tool for Measuring Teleseismic Arrival Times.” Seismological Research Letters, vol. 84, no. 1, Jan. 2013, pp. 85–93, https://doi.org/10.1785/0220120033. ↩

2. VanDecar, J. C., and R. S. Crosson. “Determination of Teleseismic Relative Phase Arrival Times Using Multi-Channel Cross-Correlation and Least Squares.” Bulletin of the Seismological Society of America, vol. 80, no. 1, Feb. 1990, pp. 150–69, https://doi.org/10.1785/BSSA0800010150. ↩

Classes:

• ICCS –

  Class to store a list of ICCSSeismograms and run the ICCS algorithm.

• ICCSSeismogram –

  Protocol class to define the ICCSSeismogram type.

• MiniICCSSeismogram –

  Minimal implementation of the ICCSSeismogram type.

Functions:

• plot_seismograms –

  Plot the selected ICCS seismograms as an image.

• plot_stack –

  Plot the ICCS stack.

• update_all_picks –

  Update t1 in all seismograms by the same amount.

• update_min_ccnorm –

  Interactively pick a new min_ccnorm.

• update_pick –

  Manually pick t1 and apply it to all seismograms.

• update_timewindow –

  Pick new time window limits.

ICCS

Class to store a list of ICCSSeismograms and run the ICCS algorithm.

The ICCS class serves as a container to store a list of seismograms (typically recordings of the same event at different stations), and to then run the ICCS algorithm when an instance of this class is called. Processing parameters that are common to all seismograms are stored as attributes (e.g. time window limits).

Examples:

We begin with a set of SAC files of the same event, recorded at different stations. All files have a preliminary phase arrival estimate saved in the T0 SAC header, so we can use these files to create instances of the MiniICCSSeismogram class for use with the ICCS class:

>>> from pysmo.classes import SAC
>>> from pysmo.functions import clone_to_mini
>>> from pysmo.tools.iccs import MiniICCSSeismogram
>>> from pathlib import Path
>>>
>>> sacfiles = sorted(Path("iccs-example/").glob("*.bhz"))
>>>
>>> seismograms = []
>>> for sacfile in sacfiles:
...     sac = SAC.from_file(sacfile)
...     update = {"t0": sac.timestamps.t0}
...     iccs_seismogram = clone_to_mini(MiniICCSSeismogram, sac.seismogram, update=update)
...     seismograms.append(iccs_seismogram)
...
>>>

To better illustrate the different modes of running the ICCS algorithm, we modify the data and picks in the seismograms to make them worse than they actually are:

>>> from datetime import timedelta
>>> from copy import deepcopy
>>> import numpy as np
>>>
>>> # change the sign of the data in the first seismogram
>>> seismograms[0].data *= -1
>>>
>>> # move the initial pick 2 seconds earlier in second seismogram
>>> seismograms[1].t0 += timedelta(seconds=-2)
>>>
>>> # move the initial pick 2 seconds later in third seismogram
>>> seismograms[2].t0 += timedelta(seconds=2)
>>>
>>> # create a seismogram with completely random data
>>> iccs_random: MiniICCSSeismogram = deepcopy(seismograms[-1])
>>> np.random.seed(42)  # set this for consistent results during testing
>>> iccs_random.data = np.random.rand(len(iccs_random))
>>> seismograms.append(iccs_random)
>>>

We can now create an instance of the ICCS class and plot the initial stack together with the input seismograms:

>>> from pysmo.tools.iccs import ICCS, plot_stack
>>> iccs = ICCS(seismograms)
>>> plot_stack(iccs, padded=False)
>>>

The phase emergance is not visible in the stack, and the (absolute) correlation coefficents of the the seismograms are not very high. This shows the initial picks are not very good and/or that the data are of low quality. To run the ICCS algorithm, we simply call (execute) the ICCS instance:

>>> convergence_list = iccs()  # this runs the ICCS algorithm and returns
>>>                            # a list of the convergence value after each
>>>                            # iteration.
>>> plot_stack(iccs, padded=False)
>>>

Despite of the random noise seismogram, the phase arrival is now visible in the stack. Seismograms with low correlation coefficients can automatically be deselected from the calculations by running ICCS again with autoselect=True:

>>> _ = iccs(autoselect=True)
>>> plot_stack(iccs, padded=False)
>>>

Seismograms that fit better with their polarity reversed can be flipped automatically by setting autoflip=True:

>>> _ = iccs(autoflip=True)
>>> plot_stack(iccs, padded=False)
>>>

To further improve results, you can interactively update the picks, time window, and minimum correlation coefficient using update_pick, update_timewindow, and update_min_ccnorm, respectively, and then run the ICCS algorithm again.

Methods:

• __call__ –

  Run the iccs algorithm.

• validate_pick –

  Check if a new pick is valid.

• validate_time_window –

  Check if a new time window (relative to pick) is valid.

Attributes:

• cc_seismograms (list[MiniSeismogram]) –

  Returns the seismograms as used for the cross-correlation.

• ccnorms (ndarray) –

  Returns an array of the normalised cross-correlation coefficients.

• min_ccnorm (floating | float) –

  Minimum normalised cross-correlation coefficient for seismograms.

• padded_seismograms (list[MiniSeismogram]) –

  Returns the seismograms with extra padding for plotting.

• padded_stack (MiniSeismogram) –

  Returns the stacked padded_seismograms.

• plot_padding (timedelta) –

  Padding to apply before and after the time window.

• seismograms (Sequence[ICCSSeismogram]) –

  Input seismograms.

• stack (MiniSeismogram) –

  Returns the stacked cc_seismograms).

• taper_width (timedelta | float) –

  Taper width.

• window_post (timedelta) –

  End of the time window relative to the pick.

• window_pre (timedelta) –

  Begining of the time window relative to the pick.

Source code in pysmo/tools/iccs/_iccs.py

@define(slots=True)
class ICCS:
    """Class to store a list of [`ICCSSeismograms`][pysmo.tools.iccs.ICCSSeismogram] and run the ICCS algorithm.

    The [`ICCS`][pysmo.tools.iccs.ICCS] class serves as a container to store a
    list of seismograms (typically recordings of the same event at different
    stations), and to then run the ICCS algorithm when an instance of this
    class is called. Processing parameters that are common to all seismograms
    are stored as attributes (e.g. time window limits).

    Examples:
        We begin with a set of SAC files of the same event, recorded at different
        stations. All files have a preliminary phase arrival estimate saved in the
        `T0` SAC header, so we can use these files to create instances of the
        [`MiniICCSSeismogram`][pysmo.tools.iccs.MiniICCSSeismogram] class for use
        with the [`ICCS`][pysmo.tools.iccs.ICCS] class:

        ```python
        >>> from pysmo.classes import SAC
        >>> from pysmo.functions import clone_to_mini
        >>> from pysmo.tools.iccs import MiniICCSSeismogram
        >>> from pathlib import Path
        >>>
        >>> sacfiles = sorted(Path("iccs-example/").glob("*.bhz"))
        >>>
        >>> seismograms = []
        >>> for sacfile in sacfiles:
        ...     sac = SAC.from_file(sacfile)
        ...     update = {"t0": sac.timestamps.t0}
        ...     iccs_seismogram = clone_to_mini(MiniICCSSeismogram, sac.seismogram, update=update)
        ...     seismograms.append(iccs_seismogram)
        ...
        >>>
        ```

        To better illustrate the different modes of running the ICCS algorithm, we
        modify the data and picks in the seismograms to make them **worse** than
        they actually are:

        ```python
        >>> from datetime import timedelta
        >>> from copy import deepcopy
        >>> import numpy as np
        >>>
        >>> # change the sign of the data in the first seismogram
        >>> seismograms[0].data *= -1
        >>>
        >>> # move the initial pick 2 seconds earlier in second seismogram
        >>> seismograms[1].t0 += timedelta(seconds=-2)
        >>>
        >>> # move the initial pick 2 seconds later in third seismogram
        >>> seismograms[2].t0 += timedelta(seconds=2)
        >>>
        >>> # create a seismogram with completely random data
        >>> iccs_random: MiniICCSSeismogram = deepcopy(seismograms[-1])
        >>> np.random.seed(42)  # set this for consistent results during testing
        >>> iccs_random.data = np.random.rand(len(iccs_random))
        >>> seismograms.append(iccs_random)
        >>>
        ```

        We can now create an instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class
        and plot the initial stack together with the input seismograms:

        ```python
        >>> from pysmo.tools.iccs import ICCS, plot_stack
        >>> iccs = ICCS(seismograms)
        >>> plot_stack(iccs, padded=False)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_initial-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_initial.png", transparent=True)
        >>>
        ```
        -->

        ![Initial stack](../../../examples/figures/iccs_stack_initial.png#only-light){ loading=lazy }
        ![Initial stack](../../../examples/figures/iccs_stack_initial-dark.png#only-dark){ loading=lazy }

        The phase emergance is not visible in the stack, and the (absolute)
        correlation coefficents of the the seismograms are not very high. This
        shows the initial picks are not very good and/or that the data are of low
        quality. To run the ICCS algorithm, we simply call (execute) the ICCS
        instance:

        ```python
        >>> convergence_list = iccs()  # this runs the ICCS algorithm and returns
        >>>                            # a list of the convergence value after each
        >>>                            # iteration.
        >>> plot_stack(iccs, padded=False)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_first_run-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_first_run.png", transparent=True)
        >>>
        ```
        -->

        ![Stack after first run](../../../examples/figures/iccs_stack_first_run.png#only-light){ loading=lazy }
        ![Stack after first run](../../../examples/figures/iccs_stack_first_run-dark.png#only-dark){ loading=lazy }

        Despite of the random noise seismogram, the phase arrival is now visible in
        the stack. Seismograms with low correlation coefficients can automatically
        be deselected from the calculations by running ICCS again with
        `autoselect=True`:


        ```python
        >>> _ = iccs(autoselect=True)
        >>> plot_stack(iccs, padded=False)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_autoselect-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_autoselect.png", transparent=True)
        >>>
        ```
        -->

        ![Stack after run with autoselect](../../../examples/figures/iccs_stack_autoselect.png#only-light){ loading=lazy }
        ![Stack after run with autoselect](../../../examples/figures/iccs_stack_autoselect-dark.png#only-dark){ loading=lazy }


        Seismograms that fit better with their polarity reversed can be flipped
        automatically by setting `autoflip=True`:

        ```python
        >>> _ = iccs(autoflip=True)
        >>> plot_stack(iccs, padded=False)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_autoflip-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_stack_autoflip.png", transparent=True)
        >>>
        ```
        -->

        ![Stack after run with autoflip](../../../examples/figures/iccs_stack_autoflip.png#only-light){ loading=lazy }
        ![Stack after run with autoflip](../../../examples/figures/iccs_stack_autoflip-dark.png#only-dark){ loading=lazy }


        To further improve results, you can interactively update the picks,
        time window, and minimum correlation coefficient using
        [`update_pick`][pysmo.tools.iccs.update_pick],
        [`update_timewindow`][pysmo.tools.iccs.update_timewindow], and
        [`update_min_ccnorm`][pysmo.tools.iccs.update_min_ccnorm],
        respectively, and then run the ICCS algorithm again.
    """

    seismograms: Sequence[ICCSSeismogram] = field(
        factory=lambda: list[ICCSSeismogram](), validator=_clear_cache_on_update
    )
    """Input seismograms.

    These seismograms provide the input data for ICCS. They are used to store
    processing parameters and create shorter seismograms (based on pick and
    time window) that are then used for cross-correlation. The shorter
    seismograms are created on the fly and then cached within an [`ICCS`]
    [pysmo.tools.iccs.ICCS] instance.
    """

    window_pre: timedelta = field(
        default=ICCS_DEFAULTS.WINDOW_PRE,
        validator=[
            validators.lt(timedelta(seconds=0)),
            _validate_window_pre,
            _clear_cache_on_update,
        ],
    )
    """Begining of the time window relative to the pick."""

    window_post: timedelta = field(
        default=ICCS_DEFAULTS.WINDOW_POST,
        validator=[
            validators.gt(timedelta(seconds=0)),
            _validate_window_post,
            _clear_cache_on_update,
        ],
    )
    """End of the time window relative to the pick."""

    plot_padding: timedelta = field(
        default=ICCS_DEFAULTS.PLOT_PADDING,
        validator=[validators.gt(timedelta(seconds=0)), _clear_cache_on_update],
    )
    """Padding to apply before and after the time window.

    This padding is *not* used for the cross-correlation."""

    taper_width: timedelta | float = field(
        default=ICCS_DEFAULTS.TAPER_WIDTH, validator=_clear_cache_on_update
    )
    """Taper width.

    Can be either a timedelta or a float - see [`pysmo.functions.taper()`][pysmo.functions.taper]
    for details.
    """

    min_ccnorm: np.floating | float = ICCS_DEFAULTS.MIN_CCNORM
    """Minimum normalised cross-correlation coefficient for seismograms.

    When the ICCS algorithm is [executed][pysmo.tools.iccs.ICCS.__call__],
    the cross-correlation coefficient for each seismogram is calculated after
    each iteration. If `autoselect` is set to `True`, the
    [`select`][pysmo.tools.iccs.ICCSSeismogram.select] attribute of seismograms
    with with correlation coefficients below this value is set to `False`, and
    they are no longer used for the [`stack`][pysmo.tools.iccs.ICCS.stack].
    """

    # The following attributes are cached to prevent unnecessary processing.
    # Setting the caches to None will force a new calculation when they are
    # requested.
    _cc_seismograms: list[MiniSeismogram] | None = field(init=False)
    _padded_seismograms: list[MiniSeismogram] | None = field(init=False)
    _ccnorms: np.ndarray | None = field(init=False)
    _cc_stack: MiniSeismogram | None = field(init=False)
    _padded_stack: MiniSeismogram | None = field(init=False)
    _valid_pick_range: tuple[timedelta, timedelta] | None = field(init=False)
    _valid_time_window_range: tuple[timedelta, timedelta] | None = field(init=False)

    def __attrs_post_init__(self) -> None:
        self._clear_caches()

    def _clear_caches(self) -> None:
        """Clear all cached attributes."""
        self._cc_seismograms = None
        self._padded_seismograms = None
        self._ccnorms = None
        self._cc_stack = None
        self._padded_stack = None
        self._valid_pick_range = None
        self._valid_time_window_range = None

    @property
    def cc_seismograms(self) -> list[MiniSeismogram]:
        """Returns the seismograms as used for the cross-correlation."""

        if self._cc_seismograms is None:
            self._cc_seismograms = _prepare_seismograms(
                self.seismograms,
                self.window_pre,
                self.window_post,
                self.taper_width,
            )

        return self._cc_seismograms

    @property
    def padded_seismograms(self) -> list[MiniSeismogram]:
        """Returns the seismograms with extra padding for plotting."""

        if self._padded_seismograms is None:
            self._padded_seismograms = _prepare_seismograms(
                self.seismograms,
                self.window_pre,
                self.window_post,
                self.taper_width,
                True,
                self.plot_padding,
            )

        return self._padded_seismograms

    @property
    def ccnorms(self) -> np.ndarray:
        """Returns an array of the normalised cross-correlation coefficients."""

        if self._ccnorms is None:
            self._ccnorms = _calc_ccnorms(self.cc_seismograms, self.stack)

        return self._ccnorms

    @property
    def stack(self) -> MiniSeismogram:
        """Returns the stacked [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms]).

        The stack is calculated as the average of all seismograms with the
        attribute [`select`][pysmo.tools.iccs.ICCSSeismogram.select] set to
        [`True`][True]. The [`begin_time`][pysmo.MiniSeismogram.begin_time] of
        the returned stack is the average of the [`begin_time`]
        [pysmo.tools.iccs.ICCSSeismogram.begin_time] of the input seismograms.

        Returns:
            Stacked input seismograms.
        """
        if self._cc_stack is not None:
            return self._cc_stack

        self._cc_stack = _create_stack(self.cc_seismograms, self.seismograms)
        return self._cc_stack

    @property
    def padded_stack(self) -> MiniSeismogram:
        """Returns the stacked [`padded_seismograms`][pysmo.tools.iccs.ICCS.padded_seismograms].

        Returns:
            Stacked input seismograms.
        """
        if self._padded_stack is not None:
            return self._padded_stack

        self._padded_stack = _create_stack(self.padded_seismograms, self.seismograms)
        return self._padded_stack

    def validate_pick(self, pick: timedelta) -> bool:
        """Check if a new pick is valid.

        This checks if a new manual pick is valid for all selected seismograms.

        Parameters:
            pick: New pick to validate.

        Returns:
            Whether the new pick is valid.
        """

        # first calculate valid pick range if not done yet
        if self._valid_pick_range is None:
            self._valid_pick_range = _calc_valid_pick_range(self)

        if self._valid_pick_range[0] <= pick <= self._valid_pick_range[1]:
            return True
        return False

    def validate_time_window(
        self, window_pre: timedelta, window_post: timedelta
    ) -> bool:
        """Check if a new time window (relative to pick) is valid.

        Parameters:
            window_pre: New window start time to validate.
            window_post: New window end time to validate.

        Returns:
            Whether the new time window is valid.
        """

        if window_pre >= window_post:
            return False

        if window_pre > -self.stack.delta:
            return False

        if window_post < self.stack.delta:
            return False

        # Calculate valid time window range if not done yet
        if self._valid_time_window_range is None:
            self._valid_time_window_range = _calc_valid_time_window_range(self)

        if (
            self._valid_time_window_range[0] <= window_pre
            and window_post <= self._valid_time_window_range[1]
        ):
            return True
        return False

    def __call__(
        self,
        autoflip: bool = False,
        autoselect: bool = False,
        convergence_limit: float = ICCS_DEFAULTS.CONVERGENCE_LIMIT,
        convergence_method: CONVERGENCE_METHOD = ICCS_DEFAULTS.CONVERGENCE_METHOD,
        max_iter: int = ICCS_DEFAULTS.MAX_ITER,
        max_shift: timedelta | None = None,
        parallel: bool = False,
    ) -> np.ndarray:
        """Run the iccs algorithm.

        Parameters:
            autoflip: Automatically toggle [`flip`][pysmo.tools.iccs.ICCSSeismogram.flip] attribute of seismograms.
            autoselect: Automatically set `select` attribute to `False` for poor quality seismograms.
            convergence_limit: Convergence limit at which the algorithm stops.
            convergence_method: Method to calculate convergence criterion.
            max_iter: Maximum number of iterations.
            max_shift: Maximum shift in seconds (see [`delay()`][pysmo.tools.signal.delay]).
            parallel: Whether to use parallel processing. Setting this to `True`
                will perform the cross-correlation calculations in parallel
                using [`ProcessPoolExecutor`][concurrent.futures.ProcessPoolExecutor].
                In most cases this will *not* be faster.

        Returns:
            convergence: Array of convergence criterion values.
        """
        convergence_list = []

        for _ in range(max_iter):
            prev_stack = clone_to_mini(MiniSeismogram, self.stack)

            if parallel:
                with ProcessPoolExecutor() as executor:
                    for prepared_seismogram, seismogram in zip(
                        self.cc_seismograms, self.seismograms
                    ):
                        future = executor.submit(
                            delay,
                            self.stack,
                            prepared_seismogram,
                            max_shift=max_shift,
                            abs_max=autoflip,
                        )
                        future.add_done_callback(
                            partial(
                                _update_seismogram_fn,
                                seismogram,
                                autoflip,
                                autoselect,
                                self.min_ccnorm,
                                (self.window_pre, self.window_post),
                            )
                        )
            else:
                for prepared_seismogram, seismogram in zip(
                    self.cc_seismograms, self.seismograms
                ):
                    _delay, _ccnorm = delay(
                        self.stack,
                        prepared_seismogram,
                        max_shift=max_shift,
                        abs_max=autoflip,
                    )

                    _update_seismogram(
                        _delay,
                        _ccnorm,
                        seismogram,
                        autoflip,
                        autoselect,
                        self.min_ccnorm,
                        (self.window_pre, self.window_post),
                    )

            self._clear_caches()

            convergence = _calc_convergence(self.stack, prev_stack, convergence_method)
            convergence_list.append(convergence)
            if convergence <= convergence_limit:
                break

        return np.array(convergence_list)

cc_seismograms property

cc_seismograms: list[MiniSeismogram]

Returns the seismograms as used for the cross-correlation.

ccnorms property

ccnorms: ndarray

Returns an array of the normalised cross-correlation coefficients.

min_ccnorm class-attribute instance-attribute

min_ccnorm: floating | float = MIN_CCNORM

Minimum normalised cross-correlation coefficient for seismograms.

When the ICCS algorithm is executed, the cross-correlation coefficient for each seismogram is calculated after each iteration. If autoselect is set to True, the select attribute of seismograms with with correlation coefficients below this value is set to False, and they are no longer used for the stack.

padded_seismograms property

padded_seismograms: list[MiniSeismogram]

Returns the seismograms with extra padding for plotting.

padded_stack property

padded_stack: MiniSeismogram

Returns the stacked padded_seismograms.

Returns:

• MiniSeismogram –

  Stacked input seismograms.

plot_padding class-attribute instance-attribute

plot_padding: timedelta = field(
    default=PLOT_PADDING,
    validator=[
        gt(timedelta(seconds=0)),
        _clear_cache_on_update,
    ],
)

Padding to apply before and after the time window.

This padding is not used for the cross-correlation.

seismograms class-attribute instance-attribute

seismograms: Sequence[ICCSSeismogram] = field(
    factory=lambda: list[ICCSSeismogram](),
    validator=_clear_cache_on_update,
)

Input seismograms.

These seismograms provide the input data for ICCS. They are used to store processing parameters and create shorter seismograms (based on pick and time window) that are then used for cross-correlation. The shorter seismograms are created on the fly and then cached within an ICCS instance.

stack property

stack: MiniSeismogram

Returns the stacked cc_seismograms).

The stack is calculated as the average of all seismograms with the attribute select set to True. The begin_time of the returned stack is the average of the begin_time of the input seismograms.

Returns:

• MiniSeismogram –

  Stacked input seismograms.

taper_width class-attribute instance-attribute

taper_width: timedelta | float = field(
    default=TAPER_WIDTH, validator=_clear_cache_on_update
)

Taper width.

Can be either a timedelta or a float - see pysmo.functions.taper() for details.

window_post class-attribute instance-attribute

window_post: timedelta = field(
    default=WINDOW_POST,
    validator=[
        gt(timedelta(seconds=0)),
        _validate_window_post,
        _clear_cache_on_update,
    ],
)

End of the time window relative to the pick.

window_pre class-attribute instance-attribute

window_pre: timedelta = field(
    default=WINDOW_PRE,
    validator=[
        lt(timedelta(seconds=0)),
        _validate_window_pre,
        _clear_cache_on_update,
    ],
)

Begining of the time window relative to the pick.

__call__

__call__(
    autoflip: bool = False,
    autoselect: bool = False,
    convergence_limit: float = CONVERGENCE_LIMIT,
    convergence_method: CONVERGENCE_METHOD = CONVERGENCE_METHOD,
    max_iter: int = MAX_ITER,
    max_shift: timedelta | None = None,
    parallel: bool = False,
) -> ndarray

Run the iccs algorithm.

Parameters:

• autoflip (bool, default: False ) –

  Automatically toggle flip attribute of seismograms.

• autoselect (bool, default: False ) –

  Automatically set select attribute to False for poor quality seismograms.

• convergence_limit (float, default: CONVERGENCE_LIMIT ) –

  Convergence limit at which the algorithm stops.

• convergence_method (CONVERGENCE_METHOD, default: CONVERGENCE_METHOD ) –

  Method to calculate convergence criterion.

• max_iter (int, default: MAX_ITER ) –

  Maximum number of iterations.

• max_shift (timedelta | None, default: None ) –

  Maximum shift in seconds (see delay()).

• parallel (bool, default: False ) –

  Whether to use parallel processing. Setting this to True will perform the cross-correlation calculations in parallel using ProcessPoolExecutor. In most cases this will not be faster.

Returns:

• convergence ( ndarray ) –

  Array of convergence criterion values.

Source code in pysmo/tools/iccs/_iccs.py

def __call__(
    self,
    autoflip: bool = False,
    autoselect: bool = False,
    convergence_limit: float = ICCS_DEFAULTS.CONVERGENCE_LIMIT,
    convergence_method: CONVERGENCE_METHOD = ICCS_DEFAULTS.CONVERGENCE_METHOD,
    max_iter: int = ICCS_DEFAULTS.MAX_ITER,
    max_shift: timedelta | None = None,
    parallel: bool = False,
) -> np.ndarray:
    """Run the iccs algorithm.

    Parameters:
        autoflip: Automatically toggle [`flip`][pysmo.tools.iccs.ICCSSeismogram.flip] attribute of seismograms.
        autoselect: Automatically set `select` attribute to `False` for poor quality seismograms.
        convergence_limit: Convergence limit at which the algorithm stops.
        convergence_method: Method to calculate convergence criterion.
        max_iter: Maximum number of iterations.
        max_shift: Maximum shift in seconds (see [`delay()`][pysmo.tools.signal.delay]).
        parallel: Whether to use parallel processing. Setting this to `True`
            will perform the cross-correlation calculations in parallel
            using [`ProcessPoolExecutor`][concurrent.futures.ProcessPoolExecutor].
            In most cases this will *not* be faster.

    Returns:
        convergence: Array of convergence criterion values.
    """
    convergence_list = []

    for _ in range(max_iter):
        prev_stack = clone_to_mini(MiniSeismogram, self.stack)

        if parallel:
            with ProcessPoolExecutor() as executor:
                for prepared_seismogram, seismogram in zip(
                    self.cc_seismograms, self.seismograms
                ):
                    future = executor.submit(
                        delay,
                        self.stack,
                        prepared_seismogram,
                        max_shift=max_shift,
                        abs_max=autoflip,
                    )
                    future.add_done_callback(
                        partial(
                            _update_seismogram_fn,
                            seismogram,
                            autoflip,
                            autoselect,
                            self.min_ccnorm,
                            (self.window_pre, self.window_post),
                        )
                    )
        else:
            for prepared_seismogram, seismogram in zip(
                self.cc_seismograms, self.seismograms
            ):
                _delay, _ccnorm = delay(
                    self.stack,
                    prepared_seismogram,
                    max_shift=max_shift,
                    abs_max=autoflip,
                )

                _update_seismogram(
                    _delay,
                    _ccnorm,
                    seismogram,
                    autoflip,
                    autoselect,
                    self.min_ccnorm,
                    (self.window_pre, self.window_post),
                )

        self._clear_caches()

        convergence = _calc_convergence(self.stack, prev_stack, convergence_method)
        convergence_list.append(convergence)
        if convergence <= convergence_limit:
            break

    return np.array(convergence_list)

validate_pick

validate_pick(pick: timedelta) -> bool

Check if a new pick is valid.

This checks if a new manual pick is valid for all selected seismograms.

Parameters:

• pick (timedelta) –

  New pick to validate.

Returns:

• bool –

  Whether the new pick is valid.

Source code in pysmo/tools/iccs/_iccs.py

def validate_pick(self, pick: timedelta) -> bool:
    """Check if a new pick is valid.

    This checks if a new manual pick is valid for all selected seismograms.

    Parameters:
        pick: New pick to validate.

    Returns:
        Whether the new pick is valid.
    """

    # first calculate valid pick range if not done yet
    if self._valid_pick_range is None:
        self._valid_pick_range = _calc_valid_pick_range(self)

    if self._valid_pick_range[0] <= pick <= self._valid_pick_range[1]:
        return True
    return False

validate_time_window

validate_time_window(
    window_pre: timedelta, window_post: timedelta
) -> bool

Check if a new time window (relative to pick) is valid.

Parameters:

• window_pre (timedelta) –

  New window start time to validate.

• window_post (timedelta) –

  New window end time to validate.

Returns:

• bool –

  Whether the new time window is valid.

Source code in pysmo/tools/iccs/_iccs.py

def validate_time_window(
    self, window_pre: timedelta, window_post: timedelta
) -> bool:
    """Check if a new time window (relative to pick) is valid.

    Parameters:
        window_pre: New window start time to validate.
        window_post: New window end time to validate.

    Returns:
        Whether the new time window is valid.
    """

    if window_pre >= window_post:
        return False

    if window_pre > -self.stack.delta:
        return False

    if window_post < self.stack.delta:
        return False

    # Calculate valid time window range if not done yet
    if self._valid_time_window_range is None:
        self._valid_time_window_range = _calc_valid_time_window_range(self)

    if (
        self._valid_time_window_range[0] <= window_pre
        and window_post <= self._valid_time_window_range[1]
    ):
        return True
    return False

ICCSSeismogram

Bases: Seismogram, Protocol

Protocol class to define the ICCSSeismogram type.

The ICCSSeismogram type extends the Seismogram type with the addition of parameters that are required for ICCS.

Methods:

• __len__ –

  The length of the Seismogram.

Attributes:

• begin_time (datetime) –

  Seismogram begin time.

• data (ndarray) –

  Seismogram data.

• delta (timedelta) –

  The sampling interval.

• end_time (datetime) –

  Seismogram end time.

• flip (bool) –

  Data in seismogram should be flipped for ICCS.

• select (bool) –

  Use seismogram to create stack.

• t0 (datetime) –

  Initial pick.

• t1 (datetime | None) –

  Updated pick.

Source code in pysmo/tools/iccs/_types.py

@runtime_checkable
class ICCSSeismogram(Seismogram, Protocol):
    """Protocol class to define the `ICCSSeismogram` type.

    The `ICCSSeismogram` type extends the [`Seismogram`][pysmo.Seismogram] type
    with the addition of parameters that are required for ICCS.
    """

    @property
    def flip(self) -> bool:
        """Data in seismogram should be flipped for ICCS."""
        ...

    @flip.setter
    def flip(self, value: bool) -> None: ...

    @property
    def select(self) -> bool:
        """Use seismogram to create stack."""
        ...

    @select.setter
    def select(self, value: bool) -> None: ...

    @property
    def t0(self) -> datetime:
        """Initial pick."""
        ...

    @t0.setter
    def t0(self, value: datetime) -> None: ...

    @property
    def t1(self) -> datetime | None:
        """Updated pick."""
        ...

    @t1.setter
    def t1(self, value: datetime | None) -> None: ...

begin_time property writable

begin_time: datetime

Seismogram begin time.

data property writable

data: ndarray

Seismogram data.

delta property writable

delta: timedelta

The sampling interval.

end_time property

end_time: datetime

Seismogram end time.

flip property writable

flip: bool

Data in seismogram should be flipped for ICCS.

select property writable

select: bool

Use seismogram to create stack.

t0 property writable

t0: datetime

Initial pick.

t1 property writable

t1: datetime | None

Updated pick.

__len__

__len__() -> int

The length of the Seismogram.

Returns:

• int –

  Number of samples in the data array.

Source code in pysmo/_types/_seismogram.py

def __len__(self) -> int:
    """The length of the Seismogram.

    Returns:
        Number of samples in the data array.
    """
    ...

MiniICCSSeismogram

Minimal implementation of the ICCSSeismogram type.

The MiniICCSSeismogram class provides a minimal implementation of class that is compatible with the ICCSSeismogram.

Examples:

Because ICCSSeismogram inherits from Seismogram, we can easily create MiniICCSSeismogram instances from exisiting seismograms using the clone_to_mini() function, whereby the update parameter is used to provide the extra information needed:

>>> from pysmo.classes import SAC
>>> from pysmo.functions import clone_to_mini
>>> from pysmo.tools.iccs import MiniICCSSeismogram
>>> sac = SAC.from_file("example.sac")
>>> sac_seis = sac.seismogram
>>> update = {"t0": sac.timestamps.t0}
>>> mini_iccs_seis = clone_to_mini(MiniICCSSeismogram, sac_seis, update=update)
>>>

Methods:

• __len__ –

  The length of the Seismogram.

Attributes:

• begin_time (datetime) –

  Seismogram begin time.

• data (ndarray) –

  Seismogram data.

• delta (timedelta) –

  Seismogram sampling interval.

• end_time (datetime) –

  Seismogram end time.

• flip (bool) –

  Data in seismogram should be flipped for ICCS.

• select (bool) –

  Use seismogram to create stack.

• t0 (datetime) –

  Initial pick.

• t1 (datetime | None) –

  Updated pick.

Source code in pysmo/tools/iccs/_types.py

@define(kw_only=True, slots=True)
class MiniICCSSeismogram:
    """Minimal implementation of the [`ICCSSeismogram`][pysmo.tools.iccs.ICCSSeismogram] type.

    The [`MiniICCSSeismogram`][pysmo.tools.iccs.ICCSSeismogram] class provides
    a minimal implementation of class that is compatible with the
    [`ICCSSeismogram`][pysmo.tools.iccs.ICCSSeismogram].

    Examples:
        Because [`ICCSSeismogram`][pysmo.tools.iccs.ICCSSeismogram] inherits
        from [`Seismogram`][pysmo.Seismogram], we can easily create
        [`MiniICCSSeismogram`][pysmo.tools.iccs.MiniICCSSeismogram] instances
        from exisiting seismograms using the
        [`clone_to_mini()`][pysmo.functions.clone_to_mini] function, whereby
        the `update` parameter is used to provide the extra information needed:

        ```python
        >>> from pysmo.classes import SAC
        >>> from pysmo.functions import clone_to_mini
        >>> from pysmo.tools.iccs import MiniICCSSeismogram
        >>> sac = SAC.from_file("example.sac")
        >>> sac_seis = sac.seismogram
        >>> update = {"t0": sac.timestamps.t0}
        >>> mini_iccs_seis = clone_to_mini(MiniICCSSeismogram, sac_seis, update=update)
        >>>
        ```
    """

    begin_time: datetime = field(
        default=SEISMOGRAM_DEFAULTS.begin_time.value, validator=datetime_is_utc
    )
    """Seismogram begin time."""

    delta: timedelta = SEISMOGRAM_DEFAULTS.delta.value
    """Seismogram sampling interval."""

    data: np.ndarray = field(factory=lambda: np.array([]))
    """Seismogram data."""

    t0: datetime = field(validator=datetime_is_utc)
    """Initial pick."""

    t1: datetime | None = field(
        default=None, validator=validators.optional(datetime_is_utc)
    )
    """Updated pick."""

    flip: bool = False
    """Data in seismogram should be flipped for ICCS."""

    select: bool = True
    """Use seismogram to create stack."""

    def __len__(self) -> int:
        """The length of the Seismogram.

        Returns:
            Number of samples in the data array.
        """
        return np.size(self.data)

    @property
    def end_time(self) -> datetime:
        """Seismogram end time."""
        if len(self) == 0:
            return self.begin_time
        return self.begin_time + self.delta * (len(self) - 1)

begin_time class-attribute instance-attribute

begin_time: datetime = field(
    default=begin_time.value, validator=datetime_is_utc
)

Seismogram begin time.

data class-attribute instance-attribute

data: ndarray = field(factory=lambda: array([]))

Seismogram data.

delta class-attribute instance-attribute

delta: timedelta = delta.value

Seismogram sampling interval.

end_time property

end_time: datetime

Seismogram end time.

flip class-attribute instance-attribute

flip: bool = False

Data in seismogram should be flipped for ICCS.

select class-attribute instance-attribute

select: bool = True

Use seismogram to create stack.

t0 class-attribute instance-attribute

t0: datetime = field(validator=datetime_is_utc)

Initial pick.

t1 class-attribute instance-attribute

t1: datetime | None = field(
    default=None, validator=optional(datetime_is_utc)
)

Updated pick.

__len__

__len__() -> int

The length of the Seismogram.

Returns:

• int –

  Number of samples in the data array.

Source code in pysmo/tools/iccs/_types.py

def __len__(self) -> int:
    """The length of the Seismogram.

    Returns:
        Number of samples in the data array.
    """
    return np.size(self.data)

plot_seismograms

plot_seismograms(
    iccs: ICCS,
    padded: bool = True,
    all: bool = False,
    return_fig: bool = False,
) -> tuple[Figure, Axes] | None

Plot the selected ICCS seismograms as an image.

Parameters:

• iccs (ICCS) –

  Instance of the ICCS class.

• padded (bool, default: True ) –

  If True, the plot is padded on both sides of the time window by the amount defined in ICCS.plot_padding.

• all (bool, default: False ) –

  If True, all seismograms are shown in the plot instead of the selected ones only.

• return_fig (bool, default: False ) –

  If True, the Figure and Axes objects are returned instead of shown.

Returns:

• tuple[Figure, Axes] | None –

  Figure of the selected seismograms as an image if return_fig is True.

Examples:

The default plotting mode is to pad the seismgorams beyond the time window used for the cross-correlations. This is particularly useful for narrow time windows.

>>> from pysmo.tools.iccs import ICCS, plot_seismograms
>>> iccs = ICCS(iccs_seismograms)
>>> _ = iccs(autoselect=True, autoflip=True)
>>>
>>> plot_seismograms(iccs)
>>>

To view the seismograms exactly as they are used in the cross-correlations, set the padded argument to False:

>>> plot_seismograms(iccs, padded=False)
>>>

Source code in pysmo/tools/iccs/_functions.py

def plot_seismograms(
    iccs: ICCS, padded: bool = True, all: bool = False, return_fig: bool = False
) -> tuple[Figure, Axes] | None:
    """Plot the selected ICCS seismograms as an image.

    Parameters:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        padded: If True, the plot is padded on both sides of the
            time window by the amount defined in
            [`ICCS.plot_padding`][pysmo.tools.iccs.ICCS.plot_padding].
        all: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        return_fig: If `True`, the [`Figure`][matplotlib.figure.Figure] and
            [`Axes`][matplotlib.axes.Axes] objects are returned instead of
            shown.

    Returns:
        Figure of the selected seismograms as an image if `return_fig` is `True`.

    Examples:
        The default plotting mode is to pad the seismgorams beyond the time
        window used for the cross-correlations. This is particularly useful
        for narrow time windows.

        ```python
        >>> from pysmo.tools.iccs import ICCS, plot_seismograms
        >>> iccs = ICCS(iccs_seismograms)
        >>> _ = iccs(autoselect=True, autoflip=True)
        >>>
        >>> plot_seismograms(iccs)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_seismograms(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_plot_seismograms_padded-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_seismograms(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_plot_seismograms_padded.png", transparent=True)
        >>>
        ```
        -->

        ![View the seismograms with padding](../../../examples/figures/iccs_plot_seismograms_padded.png#only-light){ loading=lazy }
        ![View the seismograms with padding](../../../examples/figures/iccs_plot_seismograms_padded-dark.png#only-dark){ loading=lazy }

        To view the seismograms exactly as they are used in the
        cross-correlations, set the `padded` argument to `False`:

        ```python
        >>> plot_seismograms(iccs, padded=False)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_seismograms(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_plot_seismograms-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_seismograms(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_plot_seismograms.png", transparent=True)
        >>>
        ```
        -->

        ![View the seismograms with padding](../../../examples/figures/iccs_plot_seismograms.png#only-light){ loading=lazy }
        ![View the seismograms with padding](../../../examples/figures/iccs_plot_seismograms-dark.png#only-dark){ loading=lazy }
    """

    fig, ax, _ = _plot_common_image(iccs, padded, all)
    if return_fig:
        return fig, ax

    plt.show()
    return None

plot_stack

plot_stack(
    iccs: ICCS,
    padded: bool = True,
    all: bool = False,
    return_fig: bool = False,
) -> tuple[Figure, Axes] | None

Plot the ICCS stack.

Parameters:

• iccs (ICCS) –

  Instance of the ICCS class.

• padded (bool, default: True ) –

  If True, the plot is padded on both sides of the time window by the amount defined in ICCS.plot_padding.

• all (bool, default: False ) –

  If True, all seismograms are shown in the plot instead of the selected ones only.

• return_fig (bool, default: False ) –

  If True, the Figure and Axes objects are returned instead of shown.

Returns:

• tuple[Figure, Axes] | None –

  Figure of the stack with the seismograms if return_fig is True.

Examples:

The default plotting mode is to pad the stack beyond the time window used for the cross-correlations (highlighted in light green). This is useful particularly useful for narrow time windows. Note that because of the padding, the displayed stack isn't exactly what is used for the cross-correlations.

>>> from pysmo.tools.iccs import ICCS, plot_stack
>>> iccs = ICCS(iccs_seismograms)
>>> _ = iccs(autoselect=True, autoflip=True)
>>>
>>> plot_stack(iccs)
>>>

To view the stack exactly as it is used in the cross-correlations, set the padded argument to False:

>>> plot_stack(iccs, padded=False)
>>>

Source code in pysmo/tools/iccs/_functions.py

def plot_stack(
    iccs: ICCS, padded: bool = True, all: bool = False, return_fig: bool = False
) -> tuple[Figure, Axes] | None:
    """Plot the ICCS stack.

    Parameters:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        padded: If True, the plot is padded on both sides of the
            time window by the amount defined in
            [`ICCS.plot_padding`][pysmo.tools.iccs.ICCS.plot_padding].
        all: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        return_fig: If `True`, the [`Figure`][matplotlib.figure.Figure] and
            [`Axes`][matplotlib.axes.Axes] objects are returned instead of
            shown.

    Returns:
        Figure of the stack with the seismograms if `return_fig` is `True`.

    Examples:
        The default plotting mode is to pad the stack beyond the time window
        used for the cross-correlations (highlighted in light green). This is
        useful particularly useful for narrow time windows. Note that because
        of the padding, the displayed stack isn't exactly what is used for the
        cross-correlations.

        ```python
        >>> from pysmo.tools.iccs import ICCS, plot_stack
        >>> iccs = ICCS(iccs_seismograms)
        >>> _ = iccs(autoselect=True, autoflip=True)
        >>>
        >>> plot_stack(iccs)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_stack(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_view_stack_padded-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_stack(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_view_stack_padded.png", transparent=True)
        >>>
        ```
        -->

        ![View the stack with padding](../../../examples/figures/iccs_view_stack_padded.png#only-light){ loading=lazy }
        ![View the stack with padding](../../../examples/figures/iccs_view_stack_padded-dark.png#only-dark){ loading=lazy }

        To view the stack exactly as it is used in the cross-correlations, set
        the `padded` argument to `False`:

        ```python
        >>> plot_stack(iccs, padded=False)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_view_stack-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = plot_stack(iccs, padded=False, return_fig=True)
        ...     fig.savefig(savedir / "iccs_view_stack.png", transparent=True)
        >>>
        ```
        -->

        ![View the stack with padding](../../../examples/figures/iccs_view_stack.png#only-light){ loading=lazy }
        ![View the stack with padding](../../../examples/figures/iccs_view_stack-dark.png#only-dark){ loading=lazy }
    """

    fig, ax = _plot_common_stack(iccs, padded, all)
    if return_fig:
        return fig, ax
    plt.show()
    return None

update_all_picks

update_all_picks(iccs: ICCS, pickdelta: timedelta) -> None

Update t1 in all seismograms by the same amount.

Parameters:

• iccs (ICCS) –

  Instance of the ICCS class.

• pickdelta (timedelta) –

  delta applied to all picks.

Raises:

• ValueError –

  If the new t1 is outside the valid range.

Source code in pysmo/tools/iccs/_functions.py

def update_all_picks(iccs: ICCS, pickdelta: timedelta) -> None:
    """Update [`t1`][pysmo.tools.iccs.ICCSSeismogram.t1] in all seismograms by the same amount.

    Parameters:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        pickdelta: delta applied to all picks.

    Raises:
        ValueError: If the new t1 is outside the valid range.
    """

    if not iccs.validate_pick(pickdelta):
        raise ValueError(
            "New t1 is outside the valid range. Consider reducing the time window."
        )

    for seismogram in iccs.seismograms:
        current_pick = seismogram.t1 or seismogram.t0
        seismogram.t1 = current_pick + pickdelta
    iccs._clear_caches()  # seismograms and stack need to be refreshed

update_min_ccnorm

update_min_ccnorm(
    iccs: ICCS,
    padded: bool = True,
    all: bool = False,
    return_fig: bool = False,
) -> tuple[Figure, Axes] | None

Interactively pick a new min_ccnorm.

This function launches an interactive figure to manually pick a new min_ccnorm, which is used when running the ICCS algorithm with autoselect set to True.

Parameters:

• iccs (ICCS) –

  Instance of the ICCS class.

• padded (bool, default: True ) –

  If True, the plot is padded on both sides of the time window by the amount defined in ICCS.plot_padding.

• all (bool, default: False ) –

  If True, all seismograms are shown in the plot instead of the selected ones only.

• return_fig (bool, default: False ) –

  If True, the Figure and Axes objects are returned instead of shown.

Returns:

• tuple[Figure, Axes] | None –

  Figure with the selector if return_fig is True.

Examples:

>>> from pysmo.tools.iccs import ICCS, update_min_ccnorm
>>> iccs = ICCS(iccs_seismograms)
>>> _ = iccs()
>>> update_min_ccnorm(iccs)
>>>

Source code in pysmo/tools/iccs/_functions.py

def update_min_ccnorm(
    iccs: ICCS, padded: bool = True, all: bool = False, return_fig: bool = False
) -> tuple[Figure, Axes] | None:
    """Interactively pick a new [`min_ccnorm`][pysmo.tools.iccs.ICCS.min_ccnorm].

    This function launches an interactive figure to manually pick a new
    [`min_ccnorm`][pysmo.tools.iccs.ICCS.min_ccnorm], which is used when
    [running][pysmo.tools.iccs.ICCS.__call__] the ICCS algorithm with
    `autoselect` set to `True`.

    Parameters:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        padded: If True, the plot is padded on both sides of the
            time window by the amount defined in
            [`ICCS.plot_padding`][pysmo.tools.iccs.ICCS.plot_padding].
        all: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        return_fig: If `True`, the [`Figure`][matplotlib.figure.Figure] and
            [`Axes`][matplotlib.axes.Axes] objects are returned instead of
            shown.

    Returns:
        Figure with the selector if `return_fig` is `True`.

    Examples:
        ```python
        >>> from pysmo.tools.iccs import ICCS, update_min_ccnorm
        >>> iccs = ICCS(iccs_seismograms)
        >>> _ = iccs()
        >>> update_min_ccnorm(iccs)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = update_min_ccnorm(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_update_min_ccnorm-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = update_min_ccnorm(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_update_min_ccnorm.png", transparent=True)
        ...     plt.close()
        >>>
        ```
        -->

        ![Picking a new time window in stack](../../../examples/figures/iccs_update_min_ccnorm.png#only-light){ loading=lazy }
        ![Picking a new time window in stack](../../../examples/figures/iccs_update_min_ccnorm-dark.png#only-dark){ loading=lazy }
    """

    # If the lowest index is chosen in the gui, set the new min_ccnorm to
    # the lowest value of all seismograms multiplied by a multiplier
    _INDEX_ZERO_MULTIPLIER = 0.9

    current_ccnorms = sorted(
        i for i, _ in zip(iccs.ccnorms, iccs.seismograms) if _.select or all
    )

    fig, ax, selected_seismogram_matrix = _plot_common_image(
        iccs, padded, all, figsize=(10, 6), constrained=False
    )
    fig.subplots_adjust(bottom=0.2, left=0.05, right=0.95, top=0.93)
    ax.set_title("Pick a new minimal cross-correlation coefficient.")

    start_index = int(np.searchsorted(current_ccnorms, iccs.min_ccnorm))
    max_index = len(selected_seismogram_matrix) - 1

    pick_line = ax.axhline(start_index, color="g", linewidth=2)
    pick_line_cursor = ax.axhline(start_index, color="g", linewidth=2, linestyle="--")

    def snap_ydata(ydata: float) -> int:
        return max(0, round(min(ydata, max_index)))

    def calc_ccnorm(pick_line: Line2D) -> float | np.floating:
        # NOTE: the pick_line ydata should already be a round number, but
        # mypy complains without it.
        index = round(pick_line.get_ydata()[0], 0)  # type: ignore
        if index == 0:
            return _INDEX_ZERO_MULTIPLIER * current_ccnorms[0]
        return np.mean(current_ccnorms[index - 1 : index + 1])

    def onclick(event: MouseEvent) -> Any:
        if event.inaxes is ax and event.ydata is not None:
            ydata = snap_ydata(event.ydata)
            pick_line.set_ydata((ydata, ydata))
            pick_line.set_visible(True)
            fig.canvas.draw()
            fig.canvas.flush_events()
            ax.set_title(
                f"Click save to set min_ccnorm to {calc_ccnorm(pick_line):.4f}"
            )
            fig.canvas.draw_idle()

    def on_mouse_move(event: MouseEvent) -> Any:
        if event.inaxes is ax and event.ydata is not None:
            ydata = snap_ydata(event.ydata)
            pick_line_cursor.set_ydata((ydata, ydata))
            pick_line_cursor.set_visible(True)
            fig.canvas.draw_idle()
        else:
            pick_line_cursor.set_visible(False)
            fig.canvas.draw_idle()

    class ScrollIndexTracker:
        def __init__(self, ax: Axes) -> None:
            self.scroll_index = ax.get_ylim()[1]
            self.max_scroll_index = ax.get_ylim()[1]
            self.ax = ax
            self.update()

        def on_scroll(self, event: MouseEvent) -> None:
            if event.inaxes is ax:
                increment = (
                    np.ceil(self.scroll_index / 10)
                    if event.button == "up"
                    else -np.ceil(self.scroll_index / 10)
                )
                self.scroll_index = max(
                    1, min(self.max_scroll_index, self.scroll_index + increment)
                )
                self.update()

        def update(self) -> None:
            self.ax.set_ylim(0, self.scroll_index)
            fig.canvas.draw_idle()

    class SaveOrCancel:
        def save(self, _: Event) -> None:
            iccs.min_ccnorm = calc_ccnorm(pick_line)
            iccs._clear_caches()
            plt.close()

        def cancel(self, _: Event) -> None:
            plt.close()

    _ = Cursor(
        ax,
        useblit=True,
        vertOn=False,
        horizOn=False,
    )

    callback = SaveOrCancel()
    ax_save = fig.add_axes((0.7, 0.05, 0.1, 0.075))
    ax_cancel = fig.add_axes((0.81, 0.05, 0.1, 0.075))
    b_save = Button(ax_save, "Save", color="darkgreen", hovercolor="green")
    b_save.on_clicked(callback.save)
    b_abort = Button(ax_cancel, "Cancel", color="darkred", hovercolor="red")
    b_abort.on_clicked(callback.cancel)
    tracker = ScrollIndexTracker(ax)

    _ = fig.canvas.mpl_connect("scroll_event", tracker.on_scroll)  # type: ignore
    _ = fig.canvas.mpl_connect("button_press_event", onclick)  # type: ignore
    _ = fig.canvas.mpl_connect("motion_notify_event", on_mouse_move)  # type: ignore

    if return_fig:
        return fig, ax
    plt.show()
    return None

update_pick

update_pick(
    iccs: ICCS,
    padded: bool = True,
    all: bool = False,
    use_seismogram_image: bool = False,
    return_fig: bool = False,
) -> tuple[Figure, Axes] | None

Manually pick t1 and apply it to all seismograms.

This function launches an interactive figure to manually pick a new phase arrival, and then apply it to all seismograms.

Parameters:

• iccs (ICCS) –

  Instance of the ICCS class.

• padded (bool, default: True ) –

  If True, the plot is padded on both sides of the time window by the amount defined in ICCS.plot_padding.

• all (bool, default: False ) –

  If True, all seismograms are shown in the plot instead of the selected ones only.

• use_seismogram_image (bool, default: False ) –

  Use the seismogram image instead of the stack).

• return_fig (bool, default: False ) –

  If True, the Figure and Axes objects are returned instead of shown.

Returns:

• tuple[Figure, Axes] | None –

  Figure of the stack with the picker if return_fig is True.

Examples:

>>> from pysmo.tools.iccs import ICCS, update_pick
>>> iccs = ICCS(iccs_seismograms)
>>> _ = iccs(autoselect=True, autoflip=True)
>>>
>>> update_pick(iccs)
>>>

Source code in pysmo/tools/iccs/_functions.py

def update_pick(
    iccs: ICCS,
    padded: bool = True,
    all: bool = False,
    use_seismogram_image: bool = False,
    return_fig: bool = False,
) -> tuple[Figure, Axes] | None:
    """Manually pick [`t1`][pysmo.tools.iccs.ICCSSeismogram.t1] and apply it to all seismograms.

    This function launches an interactive figure to manually pick a new phase
    arrival, and then apply it to all seismograms.

    Parameters:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        padded: If True, the plot is padded on both sides of the
            time window by the amount defined in
            [`ICCS.plot_padding`][pysmo.tools.iccs.ICCS.plot_padding].
        all: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        use_seismogram_image: Use the
            [seismogram image][pysmo.tools.iccs.plot_seismograms]
            instead of the [stack][pysmo.tools.iccs.plot_stack]).
        return_fig: If `True`, the [`Figure`][matplotlib.figure.Figure] and
            [`Axes`][matplotlib.axes.Axes] objects are returned instead of
            shown.

    Returns:
        Figure of the stack with the picker if `return_fig` is `True`.

    Examples:
        ```python
        >>> from pysmo.tools.iccs import ICCS, update_pick
        >>> iccs = ICCS(iccs_seismograms)
        >>> _ = iccs(autoselect=True, autoflip=True)
        >>>
        >>> update_pick(iccs)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = update_pick(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_update_pick-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = update_pick(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_update_pick.png", transparent=True)
        >>>
        ```
        -->

        ![Picking a new T1](../../../examples/figures/iccs_update_pick.png#only-light){ loading=lazy }
        ![Picking a new T1](../../../examples/figures/iccs_update_pick-dark.png#only-dark){ loading=lazy }
    """

    if use_seismogram_image:
        fig, ax, _ = _plot_common_image(
            iccs, padded, all, figsize=(10, 6), constrained=False
        )
        fig.subplots_adjust(bottom=0.2, left=0.05, right=0.95, top=0.93)
    else:
        fig, ax = _plot_common_stack(
            iccs, padded, all, figsize=(10, 6), constrained=False
        )
        fig.subplots_adjust(bottom=0.2, left=0.09, right=1.05, top=0.93)

    ax.set_title("Update t1 for all seismograms.")

    pick_line = ax.axvline(0, color="g", linewidth=2)

    cursor = Cursor(  # noqa: F841
        ax, useblit=True, color="g", linewidth=2, horizOn=False, linestyle="--"
    )

    def onclick(event: MouseEvent) -> Any:
        if (
            event.inaxes is ax
            and event.xdata is not None
            and iccs.validate_pick(timedelta(seconds=event.xdata))
        ):
            pick_line.set_xdata(np.array((event.xdata, event.xdata)))
            ax.set_title(f"Click save to adjust t1 by {event.xdata:.3f} seconds.")
            fig.canvas.draw()
            fig.canvas.flush_events()

    def on_mouse_move(event: MouseEvent) -> Any:
        if event.inaxes == ax and event.xdata is not None:
            if iccs.validate_pick(timedelta(seconds=event.xdata)):
                cursor.linev.set_color("g")
            else:
                cursor.linev.set_color("r")

    class SaveOrCancel:
        def save(self, _: Event) -> None:
            pickdelta = timedelta(seconds=pick_line.get_xdata(orig=True)[0])  # type: ignore
            plt.close()
            update_all_picks(iccs, pickdelta)

        def cancel(self, _: Event) -> None:
            plt.close()

    callback = SaveOrCancel()
    ax_save = fig.add_axes((0.7, 0.05, 0.1, 0.075))
    ax_cancel = fig.add_axes((0.81, 0.05, 0.1, 0.075))
    b_save = Button(ax_save, "Save", color="darkgreen", hovercolor="green")
    b_save.on_clicked(callback.save)
    b_abort = Button(ax_cancel, "Cancel", color="darkred", hovercolor="red")
    b_abort.on_clicked(callback.cancel)

    _ = fig.canvas.mpl_connect("button_press_event", onclick)  # type: ignore

    _ = fig.canvas.mpl_connect("motion_notify_event", on_mouse_move)  # type: ignore

    if return_fig:
        return fig, ax
    plt.show()
    return None

update_timewindow

update_timewindow(
    iccs: ICCS,
    padded: bool = True,
    all: bool = False,
    use_seismogram_image: bool = False,
    return_fig: bool = False,
) -> tuple[Figure, Axes] | None

Pick new time window limits.

This function launches an interactive figure to pick new values for window_pre and window_post.

Parameters:

• iccs (ICCS) –

  Instance of the ICCS class.

• padded (bool, default: True ) –

  If True, the plot is padded on both sides of the time window by the amount defined in ICCS.plot_padding.

• all (bool, default: False ) –

  If True, all seismograms are shown in the plot instead of the selected ones only.

• use_seismogram_image (bool, default: False ) –

  Use the seismogram image instead of the stack).

• return_fig (bool, default: False ) –

  If True, the Figure and Axes objects are returned instead of shown.

Returns:

• tuple[Figure, Axes] | None –

  Figure of the stack with the picker if return_fig is True.

Info

The new time window may not be chosen such that the pick lies outside the the window. The picker will therefore automatically correct itself for invalid window choices.

Examples:

>>> from pysmo.tools.iccs import ICCS, update_timewindow
>>> iccs = ICCS(iccs_seismograms)
>>> _ = iccs(autoselect=True, autoflip=True)
>>>
>>> update_timewindow(iccs)
>>>

Source code in pysmo/tools/iccs/_functions.py

def update_timewindow(
    iccs: ICCS,
    padded: bool = True,
    all: bool = False,
    use_seismogram_image: bool = False,
    return_fig: bool = False,
) -> tuple[Figure, Axes] | None:
    """Pick new time window limits.

    This function launches an interactive figure to pick new values for
    [`window_pre`][pysmo.tools.iccs.ICCS.window_pre] and
    [`window_post`][pysmo.tools.iccs.ICCS.window_post].

    Parameters:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        padded: If True, the plot is padded on both sides of the
            time window by the amount defined in
            [`ICCS.plot_padding`][pysmo.tools.iccs.ICCS.plot_padding].
        all: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        use_seismogram_image: Use the
            [seismogram image][pysmo.tools.iccs.plot_seismograms]
            instead of the [stack][pysmo.tools.iccs.plot_stack]).
        return_fig: If `True`, the [`Figure`][matplotlib.figure.Figure] and
            [`Axes`][matplotlib.axes.Axes] objects are returned instead of
            shown.

    Returns:
        Figure of the stack with the picker if `return_fig` is `True`.

    Info:
        The new time window may not be chosen such that the pick lies
        outside the the window. The picker will therefore automatically correct
        itself for invalid window choices.

    Examples:
        ```python
        >>> from pysmo.tools.iccs import ICCS, update_timewindow
        >>> iccs = ICCS(iccs_seismograms)
        >>> _ = iccs(autoselect=True, autoflip=True)
        >>>
        >>> update_timewindow(iccs)
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> import matplotlib.pyplot as plt
        >>> plt.close("all")
        >>> if savedir:
        ...     plt.style.use("dark_background")
        ...     fig, _ = update_timewindow(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_update_timewindow-dark.png", transparent=True)
        ...
        ...     plt.style.use("default")
        ...     fig, _ = update_timewindow(iccs, return_fig=True)
        ...     fig.savefig(savedir / "iccs_update_timewindow.png", transparent=True)
        >>>
        ```
        -->

        ![Picking a new time window](../../../examples/figures/iccs_update_timewindow.png#only-light){ loading=lazy }
        ![Picking a new time window](../../../examples/figures/iccs_update_timewindow-dark.png#only-dark){ loading=lazy }
    """

    if use_seismogram_image:
        fig, ax, _ = _plot_common_image(
            iccs, padded, all, figsize=(10, 6), constrained=False
        )
        fig.subplots_adjust(bottom=0.2, left=0.05, right=0.95, top=0.93)
    else:
        fig, ax = _plot_common_stack(
            iccs, padded, all, figsize=(10, 6), constrained=False
        )
        fig.subplots_adjust(bottom=0.2, left=0.09, right=1.05, top=0.93)

    ax.set_title("Pick a new time window.")

    # 'old_extents' is used to revert to the last valid extents
    old_extents = (iccs.window_pre.total_seconds(), iccs.window_post.total_seconds())

    def onselect(xmin: float, xmax: float) -> None:
        nonlocal old_extents
        if iccs.validate_time_window(timedelta(seconds=xmin), timedelta(seconds=xmax)):
            ax.set_title(
                f"Click save to set window at {xmin:.3f} to {xmax:.3f} seconds."
            )
            old_extents = xmin, xmax
            fig.canvas.draw()
            return
        span.extents = old_extents

    span = SpanSelector(
        ax,
        onselect,
        "horizontal",
        useblit=True,
        props=dict(alpha=0.5, facecolor="tab:blue"),
        interactive=True,
        drag_from_anywhere=True,
    )

    # Set the initial extents to the existing time window
    span.extents = old_extents

    class SaveOrCancel:
        def save(self, _: Event) -> None:
            iccs.window_pre = timedelta(seconds=span.extents[0])
            iccs.window_post = timedelta(seconds=span.extents[1])
            plt.close()

        def cancel(self, _: Event) -> None:
            plt.close()

    callback = SaveOrCancel()
    ax_save = fig.add_axes((0.7, 0.05, 0.1, 0.075))
    ax_cancel = fig.add_axes((0.81, 0.05, 0.1, 0.075))
    b_save = Button(ax_save, "Save", color="darkgreen", hovercolor="green")
    b_save.on_clicked(callback.save)
    b_abort = Button(ax_cancel, "Cancel", color="darkred", hovercolor="red")
    b_abort.on_clicked(callback.cancel)

    if return_fig:
        return fig, ax
    plt.show()
    return None