Skip to content

pysmo.tools.iccs

Iterative Cross-Correlation and Stack (ICCS).

Under active development

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

The ICCS1 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 further 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 aligns them with the stack and thus each other.

The results of ICCS are similar to those produced by the mccc algorithm, 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 prepare data for a successful MCCC run (e.g. if the initial picks are calculated rather than hand picked).

Data requirements

The iccs module requires that seismograms contain extra attributes specific to the ICCS method. Hence it provides a protocol class (IccsSeismogram) and corresponding Mini class (MiniIccsSeismogram). 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. data are multiplied with -1) when using it in the stack and cross-correlation. Can be automatically toggled when autoflip is True during a run.

Ephemeral seismograms

As the ICCS algorithm operates on a window around the targeted phase arrival, only a small portion of the input seismogram data are used. These smaller portions are generated on the fly in two ways, each with a causally-filtered counterpart used by picking-oriented tools. The ICCS algorithm itself always runs on the zero-phase variant, as does MCCC, regardless of which one is currently displayed — see ICCS.corners, which sets the zero-phase filter's order and from which the causal variant's own order is derived:

  • Cross-correlation seismograms are used for the execution of the ICCS algorithm. They consist of the windowed portion around the phase arrival and a tapered ramp up and down outside the window.
  • Context seismograms are used to provide extra context. They consist of a broader window around the phase arrival, and without any tapering applied.

Both share common processing steps, and are used to create a corresponding stack. As they are completely reproducible, they only exist for the lifetime of the ICCS instance that contains the input seismograms and parameters used in their creation. Changing any of the parameters will lead to automatic regeneration of the ephemeral seismograms, however, as mutations of a list cannot be detected, adding or removing seismograms from the list will not trigger regeneration. In that case, clearing the cache must be done manually by calling clear_cache.

Both types support interactive picking

Both types can be used for visualisation purposes. It is therefore possible to e.g. pick an updated arrival in the cross-correlation seismograms, and pick new time window boundaries in the context seismograms.

Execution flow

The diagram below shows execution flow, and how the above parameters are used when the ICCS algorithm is executed (see ICCS.__call__ 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 is no longer changing significantly between iterations. Typically this happens within a few iterations.

Operator involvement

Using ICCS involves two distinct kinds of iteration, easy to conflate but serving different purposes:

  • Algorithmic iteration, shown in the diagram above, is automatic and internal to a single call of an ICCS instance. Each iteration cross-correlates seismograms with the current stack, updates picks (and, optionally, flip/select), and recomputes the stack, stopping once the stack converges or max_iter is reached. The operator has no part in this process; max_iter only bounds it.
  • Operator iteration is the repeated refinement of the parameters given to the algorithm, across repeated calls. After a call, the operator inspects the resulting stack and individual seismograms visually, and may decide that the pick, time window, minimum correlation coefficient, or bandpass filter need adjusting — e.g. narrowing the time window once the phase arrival is clearly visible, or raising min_cc once obviously poor seismograms have been excluded. This module provides interactive functions for making exactly these adjustments — update_pick, update_timewindow, update_min_cc, and update_bandpass — after which the instance is called again. How many times this happens, and when to stop, is entirely up to the operator; nothing in the algorithm tracks or limits it.

AIMBAT

AIMBAT builds this operator loop into a full interactive application, managing parameter snapshots and the wider ICCS → QC → MCCC pipeline, rather than requiring it to be scripted by hand as in the example below.

Basic example

This example starts with six synthetic seismograms, each built from the same underlying pulse (an idealised phase arrival) buried in independent background noise, so the perturbations introduced below remain a large enough fraction of the signal to be clearly visible; with the dozens of stations a real array typically provides, the same perturbations would be a much smaller fraction of the total, and their effect correspondingly harder to see. Using synthetic data instead of a real recording keeps this walkthrough fully self-contained and its outcome exactly reproducible:

>>> import numpy as np
>>> import pandas as pd
>>> from pysmo.tools.iccs import MiniIccsSeismogram
>>>
>>> def ricker(points: int, width: float) -> np.ndarray:
...     # a "Mexican hat" wavelet, standing in for a real phase arrival
...     t = np.arange(points) - (points - 1) / 2
...     return (
...         2
...         / (np.sqrt(3 * width) * np.pi**0.25)
...         * (1 - (t / width) ** 2)
...         * np.exp(-(t**2) / (2 * width**2))
...     )
...
>>> pulse = ricker(200, 12)
>>> pulse /= np.abs(pulse).max()
>>>
>>> npts = 2400
>>> pick_index = 1200
>>> delta = pd.Timedelta(seconds=0.05)
>>> begin_time = pd.Timestamp("2024-01-01", tz="UTC")
>>> t0 = begin_time + pick_index * delta
>>>
>>> rng = np.random.default_rng(42)
>>> seismograms = []
>>> for _ in range(6):
...     data = rng.normal(scale=0.03, size=npts)
...     data[pick_index - 100 : pick_index + 100] += pulse
...     seismograms.append(
...         MiniIccsSeismogram(begin_time=begin_time, delta=delta, data=data, t0=t0)
...     )
...
>>>

To illustrate the different modes of running the ICCS algorithm, the data and picks are then degraded. Every seismogram but the first has its pick shifted by a few seconds, varied enough that no phase emergence survives naive stacking on the raw picks. The first seismogram has its polarity reversed instead, and a seventh, entirely synthetic seismogram of random noise (no pulse at all) is appended:

>>> from copy import deepcopy
>>>
>>> # change the sign of the data in the first seismogram
>>> seismograms[0].data *= -1
>>>
>>> # shift the remaining picks by varying amounts, in both directions
>>> shifts = [-4, 4, -6, 6, -3]
>>> for seismogram, shift in zip(seismograms[1:], shifts):
...     seismogram.t0 += pd.Timedelta(seconds=shift)
...
>>>
>>> # create a seismogram with completely random data
>>> iccs_random: MiniIccsSeismogram = deepcopy(seismograms[-1])
>>> iccs_random.data = np.random.default_rng(1).normal(scale=0.3, size=npts)
>>> seismograms.append(iccs_random)
>>>

An ICCS instance can now be created and used to plot the initial stack and cc_seismograms:

>>> from pysmo.tools.iccs import ICCS, plot_stack
>>> iccs = ICCS(seismograms)
>>> fig, ax = plot_stack(iccs, context=False)
>>>

Initial stack Initial stack

No phase emergence is visible in the stack yet. To run the ICCS algorithm, 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.
>>> fig, ax = plot_stack(iccs, context=False)
>>>

Stack after first run Stack after first run

Despite the random noise seismogram, the phase arrival is now visible in the stack, and most correlation coefficients are high. The noise seismogram's correlation is clearly the lowest, but the reversed-polarity seismogram's is only mediocre rather than obviously bad — precisely the case ICCS is designed to catch automatically, since a real dataset of hundreds of seismograms cannot be checked individually by eye. It is annotated above: the trace whose largest excursion points downward rather than upward, easy to miss among the others at a glance but a useful hint once you know to look for it. Running ICCS again with autoflip=True checks the reversed-polarity hypothesis for every seismogram and finds a substantially better fit for this one:

>>> _ = iccs(autoflip=True)
>>> fig, ax = plot_stack(iccs, context=False)
>>>

Stack after run with autoflip Stack after run with autoflip

The previously-mediocre seismogram is now among the best-fitting of all seven. The noise seismogram is unaffected — no polarity reversal fixes what is not a real signal — and remains the clear outlier. Running ICCS again with autoselect=True deselects seismograms whose fit is genuinely poor, rather than merely reversed:

>>> _ = iccs(autoselect=True)
>>> [seismogram.select for seismogram in iccs.seismograms]
[True, True, True, True, True, True, False]
>>> fig, ax = plot_stack(iccs, context=False)
>>>

Stack after run with autoselect Stack after run with autoselect

Only the noise seismogram is deselected; every real seismogram, including the one that needed flipping, now contributes to the stack.

The stack above still has room for improvement: the pick sits several seconds off the main pulse, and the default ±15 s time window extends well past the pulse's energy into background noise that only degrades the cross-correlation. See operator iteration above for how to refine these before, say, proceeding to MCCC.


  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

Modules:

Name Description
plot

Extra plotting functions for the ICCS module.

Classes:

Name Description
ICCS

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

IccsResult

Result returned by ICCS.__call__().

IccsSeismogram

Protocol class to define the IccsSeismogram type.

McccResult

Result returned by ICCS.run_mccc().

MiniIccsSeismogram

Minimal implementation of the IccsSeismogram type.

Functions:

Name Description
plot_matrix_image

Plot the selected ICCS seismograms as a matrix image.

plot_stack

Plot the ICCS stack.

update_bandpass

Interactively update the bandpass filter parameters.

update_min_cc

Interactively pick a new min_cc.

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).

Before use, seismograms are internally prepared into cc_seismograms and context_seismograms, cached and only recalculated when relevant parameters change.

See the module documentation for a worked example.

Methods:

Name Description
__call__

Run the ICCS algorithm.

clear_cache

Clear all cached ephemeral seismograms, stacks, and derived results.

run_mccc

Refine picks with the MCCC algorithm.

update_all_picks

Update t1 in all seismograms by the same amount.

validate_pick

Check whether a new pick is valid given all seismograms in the instance.

validate_time_window

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

Attributes:

Name Type Description
bandpass_apply bool

Filter seismograms with a bandpass filter before running ICCS.

bandpass_fmax float

Bandpass filter maximum frequency (Hz). Only used if bandpass_apply is True.

bandpass_fmin float

Bandpass filter minimum frequency (Hz). Only used if bandpass_apply is True.

cc_seismograms list[_EphemeralSeismogram]

Return the seismograms as used for the cross-correlation.

cc_seismograms_causal list[_EphemeralSeismogram]

Return the seismograms as used for cross-correlation, causally filtered.

ccs ndarray

Return an array of the normalised cross-correlation coefficients.

context_seismograms list[_EphemeralSeismogram]

Return the seismograms with extra context for plotting.

context_seismograms_causal list[_EphemeralSeismogram]

Return the seismograms with extra context for plotting, causally filtered.

context_stack MiniSeismogram

Return the stacked context_seismograms.

context_stack_causal MiniSeismogram

Return the stacked context_seismograms_causal.

context_width PositiveTimedelta

Context padding to apply before and after the time window.

corners PositiveInt

Number of corners (poles) for the zero-phase bandpass filter applied to

max_delta Timedelta

Maximum sampling interval across all seismograms.

max_td_pre Timedelta

Maximum negative time delta between pick and seismogram begin_time.

min_cc float

Minimum normalised cross-correlation coefficient for seismograms.

min_delta Timedelta

Minimum sampling interval across all seismograms.

min_td_post Timedelta

Minimum positive time delta between pick and seismogram end_time.

ramp_width NonNegativeTimedelta | NonNegativeNumber

Width of taper ramp up and down.

ramp_width_timedelta Timedelta

Ramp width as a pandas.Timedelta, computed from the current window.

seismograms Sequence[IccsSeismogram]

Input seismograms.

selected_cc_seismograms list[_EphemeralSeismogram]

Return the cc_seismograms with select set to True.

stack MiniSeismogram

Return the stacked cc_seismograms.

stack_causal MiniSeismogram

Return the stacked cc_seismograms_causal.

window_post PositiveTimedelta

End of the time window relative to the pick.

window_pre NegativeTimedelta

Beginning of the time window relative to the pick.

Source code in src/pysmo/tools/iccs/_iccs.py
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
@define
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).

    Before use, seismograms are internally prepared into
    [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] and
    [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms], cached
    and only recalculated when relevant parameters change.

    See the [module documentation][pysmo.tools.iccs] for a worked example.
    """

    seismograms: Sequence[IccsSeismogram] = field(
        factory=list[IccsSeismogram], on_setattr=_on_setattr_clear_cache
    )
    """Input seismograms.

    These are the source seismograms from which the ephemeral seismograms
    (cross-correlation and context seismograms, see the
    [module documentation][pysmo.tools.iccs]) are derived on demand.
    The ephemeral seismograms are cached and regenerated automatically
    whenever a controlling attribute such as
    [`window_pre`][pysmo.tools.iccs.ICCS.window_pre] or
    [`window_post`][pysmo.tools.iccs.ICCS.window_post] changes.

    Warning: In-place mutation bypasses the cache
        Assigning a new list to this attribute clears the cache automatically.
        *Mutating* the list in place (e.g. with `append`, `remove`, or direct
        index assignment) bypasses the setter and does **not** clear the cache.
        Call [`clear_cache`][pysmo.tools.iccs.ICCS.clear_cache] manually after
        any such in-place mutation.

    Tip: Remove poor-quality seismograms, don't just deselect them
        When a seismogram is of sufficiently poor quality that it should play no
        further role in the analysis, consider *removing* it from this list rather
        than simply setting its
        [`select`][pysmo.tools.iccs.IccsSeismogram.select] attribute to
        `False`. A deselected seismogram is excluded from the stack and
        correlation output, but its pick and data span still constrain the valid
        ranges for [`window_pre`][pysmo.tools.iccs.ICCS.window_pre],
        [`window_post`][pysmo.tools.iccs.ICCS.window_post], and pick updates —
        because all seismograms (selected or not) are used to generate the
        ephemeral seismograms. A badly drifting pick on a deselected seismogram
        can therefore make it impossible to set useful window or pick ranges for
        the remaining good seismograms.
    """

    window_pre: NegativeTimedelta = field(
        default=IccsDefaults.window_pre,
        validator=[validators.lt(pd.Timedelta(0)), _validate_window_pre],
        on_setattr=setters.pipe(setters.validate, _on_setattr_clear_cache),
    )
    """Beginning of the time window relative to the pick."""

    window_post: PositiveTimedelta = field(
        default=IccsDefaults.window_post,
        validator=[validators.gt(pd.Timedelta(0)), _validate_window_post],
        on_setattr=setters.pipe(setters.validate, _on_setattr_clear_cache),
    )
    """End of the time window relative to the pick."""

    ramp_width: NonNegativeTimedelta | NonNegativeNumber = field(
        default=IccsDefaults.ramp_width,
        validator=_validate_ramp_width,
        on_setattr=setters.pipe(setters.validate, _on_setattr_clear_cache),
    )
    """Width of taper ramp up and down.

    Warning: Interpretation depends on the type
        Can be either a timedelta or a float, but they mean slightly different
        things. A float is interpreted as a fraction of the window duration,
        while a timedelta is an absolute duration. See the documentation of
        [`pysmo.functions.window()`][pysmo.functions.window] for details.
    """

    context_width: PositiveTimedelta = field(
        default=IccsDefaults.context_width,
        converter=convert_to_timedelta,
        validator=validators.gt(pd.Timedelta(0)),
        on_setattr=setters.pipe(
            setters.convert, setters.validate, _on_setattr_clear_cache
        ),
    )
    """Context padding to apply before and after the time window.

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

    bandpass_apply: bool = field(
        default=IccsDefaults.bandpass_apply,
        converter=bool,
        validator=[validators.instance_of(bool), _validate_bandpass_apply],
        on_setattr=setters.pipe(
            setters.convert, setters.validate, _on_setattr_clear_cache
        ),
    )
    """Filter seismograms with a bandpass filter before running ICCS.

    Setting this to [`True`][] will apply a
    [`bandpass`][pysmo.tools.signal.bandpass] filter (with `zerophase` set to
    [`True`][]) to the [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms]
    and [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms].
    It also gates whether a causal (single-pass) counterpart is produced for
    [`cc_seismograms_causal`][pysmo.tools.iccs.ICCS.cc_seismograms_causal] and
    [`context_seismograms_causal`][pysmo.tools.iccs.ICCS.context_seismograms_causal] —
    see [`corners`][pysmo.tools.iccs.ICCS.corners] for how the two variants
    relate.

    As the [`seismograms`][pysmo.tools.iccs.ICCS.seismograms] may have already
    been pre-processed (i.e. already filtered) the default value for this
    parameter is [`False`][].
    """

    bandpass_fmin: float = field(
        default=IccsDefaults.bandpass_fmin,
        converter=float,
        validator=[validators.instance_of(float), _validate_bandpass_fmin],
        on_setattr=setters.pipe(
            setters.convert, setters.validate, _on_setattr_clear_cache
        ),
    )
    """Bandpass filter minimum frequency (Hz). Only used if [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is `True`.

    The filter is applied to each seismogram at its original sampling rate, before
    any resampling to a common grid. Valid frequencies are therefore bounded by the
    Nyquist frequency of the most coarsely sampled seismogram, i.e.
    `0.5 / max_delta.total_seconds()`.
    """

    bandpass_fmax: float = field(
        default=IccsDefaults.bandpass_fmax,
        converter=float,
        validator=[validators.instance_of(float), _validate_bandpass_fmax],
        on_setattr=setters.pipe(
            setters.convert, setters.validate, _on_setattr_clear_cache
        ),
    )
    """Bandpass filter maximum frequency (Hz). Only used if [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is `True`.

    The filter is applied to each seismogram at its original sampling rate, before
    any resampling to a common grid. Valid frequencies are therefore bounded by the
    Nyquist frequency of the most coarsely sampled seismogram, i.e.
    `0.5 / max_delta.total_seconds()`.
    """

    corners: PositiveInt = field(
        default=IccsDefaults.corners,
        converter=int,
        validator=[
            validators.instance_of(int),
            validators.gt(0),
            _validate_corners_causal_band,
        ],
        on_setattr=setters.pipe(
            setters.convert, setters.validate, _on_setattr_clear_cache
        ),
    )
    r"""Number of corners (poles) for the zero-phase bandpass filter applied to
    [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms]/
    [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] when
    [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is `True`.

    The causal counterparts
    ([`cc_seismograms_causal`][pysmo.tools.iccs.ICCS.cc_seismograms_causal],
    [`context_seismograms_causal`][pysmo.tools.iccs.ICCS.context_seismograms_causal])
    use `2 * corners` poles, matching the rolloff steepness of the zero-phase
    filter ([`sosfiltfilt`][scipy.signal.sosfiltfilt] effectively doubles
    filter order by applying it twice). Their design `freqmin`/`freqmax`
    passed to [`bandpass`][pysmo.tools.signal.bandpass] are corrected via
    [`causal_band`][pysmo.tools.signal.causal_band] so the causal variant's
    actual -3dB point matches the zero-phase variant's actual (inward-shifted)
    -3dB point closely — not exactly. See
    [`causal_band`][pysmo.tools.signal.causal_band] for the correction
    itself and its derivation; in short, the residual comes
    from real Butterworth bandpass edges interacting with each other (not
    behaving as two independent single-edge filters, which is what the
    correction formula assumes). At `ICCS`'s own defaults
    (`corners=2`, `freqmax/freqmin=40`) the residual settles at **≈1.33%**
    at typical broadband seismic sample rates, shrinking further with wider
    relative bandwidth or higher `corners` and growing with narrower
    bandwidth or lower `corners`; at low sample rates relative to
    `bandpass_fmax` (e.g. 20 Hz) a second, otherwise-negligible effect
    compounds with it and the residual grows to **≈2.37%** (worked example
    and exact figures: [`causal_band`][pysmo.tools.signal.causal_band]'s own
    docstring).

    Because the correction moves both corrected edges inward from the
    nominal band, `corners` is no longer independent of `bandpass_fmin`/
    `bandpass_fmax`: lowering `corners` while they stay fixed (or narrowing
    them while `corners` stays fixed) can raise `ValueError` if the corrected
    band would invert — the ratio `causal_band` applies is monotonically
    increasing in `corners`, so *raising* `corners` can only
    move a combination towards validity, never away from it; only lowering
    it (or narrowing the nominal band) can break a previously-valid one. See
    [`bandpass_fmin`][pysmo.tools.iccs.ICCS.bandpass_fmin]/
    [`bandpass_fmax`][pysmo.tools.iccs.ICCS.bandpass_fmax] for the coupled
    validation. This is a distinct, *frequency-validity* constraint from the
    numerical-stability finding below — there is still no cap on `corners`
    for numerical-stability reasons; the new `ValueError` is a different
    mechanism entirely.

    The causal variant also has non-zero, frequency-dependent group delay
    (unlike the zero-phase variant, which has none by construction): a
    causal filter only pushes energy forward in time, so a pick made on
    [`cc_seismograms_causal`][pysmo.tools.iccs.ICCS.cc_seismograms_causal] is
    systematically biased later than the true onset by an amount that
    depends on `corners`, `bandpass_fmin`, and `bandpass_fmax` — this is the
    trade-off for avoiding zero-phase's backward precursor smearing, not
    something corrected here. Compute
    [`group_delay`][scipy.signal.group_delay] on the same `corners`/frequency
    combination if an exact per-configuration correction is needed.

    Raising `corners` well above typical seismological values has no
    dedicated guard: filtering stays numerically well-behaved even at
    extreme orders, with no warning (from scipy or otherwise) on the code
    path [`bandpass`][pysmo.tools.signal.bandpass] actually uses.
    """

    min_cc: float = field(
        default=IccsDefaults.min_cc,
        converter=float,
        validator=validators.instance_of(float),
        on_setattr=setters.pipe(
            setters.convert, setters.validate, _on_setattr_clear_cache
        ),
    )
    """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 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_cache: list[_EphemeralSeismogram] | None = field(
        default=None, init=False
    )
    """Cached list of the prepared seismograms for cross-correlation."""
    _context_seismograms_cache: list[_EphemeralSeismogram] | None = field(
        default=None, init=False
    )
    """Cached list of the prepared seismograms with context padding."""
    _ccs_cache: np.ndarray | None = field(default=None, init=False)
    """Cached array of the normalised cross-correlation coefficients."""
    _cc_stack_cache: MiniSeismogram | None = field(default=None, init=False)
    """Cached stack of the prepared seismograms for cross-correlation."""
    _context_stack_cache: MiniSeismogram | None = field(default=None, init=False)
    """Cached stack of the prepared seismograms with context padding."""
    _cc_seismograms_causal_cache: list[_EphemeralSeismogram] | None = field(
        default=None, init=False
    )
    """Cached list of the causally-filtered prepared seismograms for cross-correlation."""
    _context_seismograms_causal_cache: list[_EphemeralSeismogram] | None = field(
        default=None, init=False
    )
    """Cached list of the causally-filtered prepared seismograms with context padding."""
    _cc_stack_causal_cache: MiniSeismogram | None = field(default=None, init=False)
    """Cached stack of the causally-filtered prepared seismograms for cross-correlation."""
    _context_stack_causal_cache: MiniSeismogram | None = field(default=None, init=False)
    """Cached stack of the causally-filtered prepared seismograms with context padding."""
    _max_td_pre_cache: pd.Timedelta | None = field(default=None, init=False)
    """Cached maximum negative time delta between pick and seismogram begin_time."""
    _min_td_post_cache: pd.Timedelta | None = field(default=None, init=False)
    """Cached minimum positive time delta between pick and seismogram end_time."""
    _valid_pick_range_cache: tuple[pd.Timedelta, pd.Timedelta] | None = field(
        default=None, init=False
    )

    def clear_cache(self) -> None:
        """Clear all cached ephemeral seismograms, stacks, and derived results.

        Ephemeral seismograms (cross-correlation, context, and their causal
        counterparts), their stacks, cross-correlation norms, and the valid
        pick and window ranges are all computed on demand and cached to
        avoid redundant work.
        The cache is invalidated automatically when a controlling attribute
        such as [`window_pre`][pysmo.tools.iccs.ICCS.window_pre],
        [`window_post`][pysmo.tools.iccs.ICCS.window_post], or
        [`seismograms`][pysmo.tools.iccs.ICCS.seismograms] is *reassigned*.

        Call this method manually after any in-place mutation of
        [`seismograms`][pysmo.tools.iccs.ICCS.seismograms] (e.g. `append`,
        `remove`, or index assignment) to ensure all ephemeral seismograms and
        derived results are regenerated from the updated input.
        """
        self._cc_seismograms_cache = None
        self._context_seismograms_cache = None
        self._ccs_cache = None
        self._cc_stack_cache = None
        self._context_stack_cache = None
        self._cc_seismograms_causal_cache = None
        self._context_seismograms_causal_cache = None
        self._cc_stack_causal_cache = None
        self._context_stack_causal_cache = None
        self._max_td_pre_cache = None
        self._min_td_post_cache = None
        self._valid_pick_range_cache = None

    @property
    def ramp_width_timedelta(self) -> pd.Timedelta:
        """Ramp width as a `pandas.Timedelta`, computed from the current window.

        For a float `ramp_width`, the duration is computed as a fraction of the
        window duration `(window_post - window_pre)`, consistent with the
        behaviour of `pysmo.functions.window`.
        """
        return _compute_ramp(self.ramp_width, self.window_pre, self.window_post)

    @property
    def min_delta(self) -> pd.Timedelta:
        """Minimum sampling interval across all seismograms."""
        if not self.seismograms:
            return pd.Timedelta(0)
        return min(s.delta for s in self.seismograms)

    @property
    def max_delta(self) -> pd.Timedelta:
        """Maximum sampling interval across all seismograms."""
        if not self.seismograms:
            return pd.Timedelta(0)
        return max(s.delta for s in self.seismograms)

    @property
    def max_td_pre(self) -> pd.Timedelta:
        """Maximum negative time delta between pick and seismogram begin_time.

        This property is used to calculate valid values for updating picks and
        attributes. Specifically:

        - new_pick > old_pick + max_td_pre - window_pre + ramp_width
        - window_pre >= max_td_pre + ramp_width
        - ramp_width <= window_pre - max_td_pre
        """
        if not self.seismograms:
            return pd.Timedelta(days=-365 * 100)

        if self._max_td_pre_cache is None:
            self._max_td_pre_cache = max(
                s.begin_time - (s.t0 if pd.isnull(s.t1) else s.t1)
                for s in self.seismograms
            )
        return self._max_td_pre_cache

    @property
    def min_td_post(self) -> pd.Timedelta:
        """Minimum positive time delta between pick and seismogram end_time.

        This property is used to calculate valid values for updating picks and
        attributes. Specifically:

        - new_pick < old_pick + min_td_post - window_post - ramp_width
        - window_post <= min_td_post - ramp_width
        - ramp_width <= min_td_post - window_post
        """
        if not self.seismograms:
            return pd.Timedelta(days=365 * 100)

        if self._min_td_post_cache is None:
            self._min_td_post_cache = min(
                s.end_time - (s.t0 if pd.isnull(s.t1) else s.t1)
                for s in self.seismograms
            )
        return self._min_td_post_cache

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

        These seismograms are derived from the input seismograms and used for
        the cross-correlation steps. Starting with the input seismograms, they
        are processed as follows:

        1. Bandpass filtered if
           [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is
           [`True`][].
        2. Resampled to the minimum sampling interval of all input seismograms
           (only if it is not equal in all seismograms).
        3. Cropped to `ramp_width` + current time window + `ramp_width`.
        4. Detrended.
        5. Tapered using [`ramp_width`][pysmo.tools.iccs.ICCS.ramp_width]
           (tapered sections are *outside* time window).
        6. Normalised based on the highest absolute value within the cropped
           window. This step is done slightly differently in
           [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms]
           — see the documentation of that property for details.
        """

        if self._cc_seismograms_cache is None:
            self._cc_seismograms_cache = _prepare_seismograms(self, add_context=False)
        return self._cc_seismograms_cache

    @property
    def selected_cc_seismograms(self) -> list[_EphemeralSeismogram]:
        """Return the [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] with [`select`][pysmo.tools.iccs.IccsSeismogram.select] set to `True`."""
        return [s for s in self.cc_seismograms if s.parent_seismogram.select]

    @property
    def context_seismograms(self) -> list[_EphemeralSeismogram]:
        """Return the seismograms with extra context for plotting.

        These seismograms are derived from the input seismograms and used
        primarily for plotting with extra context (e.g. when selecting new
        time window boundaries). Starting with the input seismograms, they are
        processed as follows:

        1. Bandpass filtered if
           [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is
           [`True`][].
        2. Resampled to the minimum sampling interval of all input seismograms
           (only if it is not equal in all seismograms).
        3. Cropped and/or padded to `context_width` + current time window +
           `context_width`.
        4. Detrended.
        5. Normalised based on the highest absolute value within the selected
           time window (i.e. without the context).
        """

        if self._context_seismograms_cache is None:
            self._context_seismograms_cache = _prepare_seismograms(
                self, add_context=True
            )
        return self._context_seismograms_cache

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

        if self._ccs_cache is None:
            matrix = np.array([s.data for s in self.cc_seismograms])
            self._ccs_cache = pearson_matrix_vector(matrix, self.stack.data)
        return self._ccs_cache

    @property
    def stack(self) -> MiniSeismogram:
        """Return 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`][]. 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_cache is None:
            self._cc_stack_cache = _create_stack(self.cc_seismograms)
        return self._cc_stack_cache

    @property
    def context_stack(self) -> MiniSeismogram:
        """Return the stacked [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms].

        Returns:
            Stacked input seismograms with context padding.
        """
        if self._context_stack_cache is None:
            self._context_stack_cache = _create_stack(self.context_seismograms)
        return self._context_stack_cache

    @property
    def cc_seismograms_causal(self) -> list[_EphemeralSeismogram]:
        """Return the seismograms as used for cross-correlation, causally filtered.

        Mirrors [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms],
        with a causal (single-pass) rather than zero-phase bandpass filter —
        see [`corners`][pysmo.tools.iccs.ICCS.corners] for how the filter
        order and passband of the two variants relate. Intended for
        picking-oriented tools, since a causal filter avoids the acausal
        precursor smearing a zero-phase filter introduces before the true
        onset. Never used by the ICCS algorithm itself (cross-correlation,
        stacking, MCCC), which continues to run on
        [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms]
        unconditionally.

        When [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is
        `False`, this returns the same object as
        [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] (not a copy).
        """

        if not self.bandpass_apply:
            return self.cc_seismograms
        if self._cc_seismograms_causal_cache is None:
            self._cc_seismograms_causal_cache = _prepare_seismograms(
                self, add_context=False, causal=True
            )
        return self._cc_seismograms_causal_cache

    @property
    def context_seismograms_causal(self) -> list[_EphemeralSeismogram]:
        """Return the seismograms with extra context for plotting, causally filtered.

        Mirrors [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms],
        with a causal (single-pass) rather than zero-phase bandpass filter —
        see [`corners`][pysmo.tools.iccs.ICCS.corners] for how the filter
        order and passband of the two variants relate.

        The context padding
        ([`context_width`][pysmo.tools.iccs.ICCS.context_width]) exists to
        show pre-arrival "quiet" alongside the time window, but a causal
        filter's group delay pushes energy from the true onset forward in
        time — some of what the padding shows as pre-arrival quiet may
        actually contain the filter's response to the onset itself, not the
        true unfiltered signal before it. This is expected behaviour (see
        [`corners`][pysmo.tools.iccs.ICCS.corners]), not a bug.

        When [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is
        `False`, this returns the same object as
        [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms]
        (not a copy).
        """

        if not self.bandpass_apply:
            return self.context_seismograms
        if self._context_seismograms_causal_cache is None:
            self._context_seismograms_causal_cache = _prepare_seismograms(
                self, add_context=True, causal=True
            )
        return self._context_seismograms_causal_cache

    @property
    def stack_causal(self) -> MiniSeismogram:
        """Return the stacked [`cc_seismograms_causal`][pysmo.tools.iccs.ICCS.cc_seismograms_causal].

        When [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is
        `False`, this returns the same object as
        [`stack`][pysmo.tools.iccs.ICCS.stack] (not a copy).

        Returns:
            Stacked causally-filtered input seismograms.
        """
        if not self.bandpass_apply:
            return self.stack
        if self._cc_stack_causal_cache is None:
            self._cc_stack_causal_cache = _create_stack(self.cc_seismograms_causal)
        return self._cc_stack_causal_cache

    @property
    def context_stack_causal(self) -> MiniSeismogram:
        """Return the stacked [`context_seismograms_causal`][pysmo.tools.iccs.ICCS.context_seismograms_causal].

        When [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply] is
        `False`, this returns the same object as
        [`context_stack`][pysmo.tools.iccs.ICCS.context_stack] (not a copy).

        Returns:
            Stacked causally-filtered input seismograms with context padding.
        """
        if not self.bandpass_apply:
            return self.context_stack
        if self._context_stack_causal_cache is None:
            self._context_stack_causal_cache = _create_stack(
                self.context_seismograms_causal
            )
        return self._context_stack_causal_cache

    def validate_pick(self, pick: pd.Timedelta) -> bool:
        """Check whether a new pick is valid given all seismograms in the instance.

        The valid pick range is computed from every seismogram in
        [`seismograms`][pysmo.tools.iccs.ICCS.seismograms], including those with
        [`select`][pysmo.tools.iccs.IccsSeismogram.select] set to
        [`False`][]. A pick is considered valid if it lies within this
        global range; selection only affects stacking, not the validity bounds.

        Args:
            pick: New pick to validate.

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

        if self._valid_pick_range_cache is None:
            self._valid_pick_range_cache = _calc_valid_pick_range(self)

        return (
            self._valid_pick_range_cache[0] <= pick <= self._valid_pick_range_cache[1]
        )

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

        Validates that the proposed window fits within every seismogram,
        accounting for the taper ramp. The ramp duration is computed from
        the proposed `window_pre` and `window_post` values, consistent
        with how `pysmo.functions.window` computes it for float
        `ramp_width`.

        Args:
            window_pre: Proposed window start time (negative, relative to pick).
            window_post: Proposed window end time (positive, relative to pick).

        Returns:
            Whether the new time window is valid for all seismograms.
        """

        if window_pre >= window_post:
            return False

        if window_pre > -self.min_delta:
            return False

        if window_post < self.min_delta:
            return False

        ramp = _compute_ramp(self.ramp_width, window_pre, window_post)
        for s in self.seismograms:
            pick = s.t0 if pd.isnull(s.t1) else s.t1
            if window_pre < s.begin_time - pick + ramp:
                return False
            if window_post > s.end_time - pick - ramp:
                return False
        return True

    def __call__(
        self,
        autoflip: bool = False,
        autoselect: bool = False,
        convergence_limit: float = IccsDefaults.convergence_limit,
        convergence_method: ConvergenceMethod = IccsDefaults.convergence_method,
        max_iter: int = IccsDefaults.max_iter,
        max_shift: NonNegativeTimedelta | None = None,
    ) -> IccsResult:
        """Run the ICCS algorithm.

        Args:
            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 (absolute) shift to consider in each iteration,
                relative to the stack at that iteration (see
                [`delay()`][pysmo.tools.signal.delay]). This is not a bound on
                the cumulative shift over the full run.

        Returns:
            An [`IccsResult`][pysmo.tools.iccs.IccsResult] containing the
            convergence history and whether the convergence limit was reached.
        """
        convergence_list = []

        for _ in range(max_iter):
            # Save the previous stack to calculate convergence criterion after updating the seismograms.
            prev_stack = clone_to_mini(MiniSeismogram, self.stack)

            # Get delays and correlation coefficients for all seismograms in one go
            delays, ccs = multi_delay(
                self.stack,
                self.cc_seismograms,
                abs_max=autoflip,
                max_shift=max_shift,
            )

            # Update seismograms based on results and settings.
            for delay, cc, cc_seismogram in zip(delays, ccs, self.cc_seismograms):
                _update_seismogram(
                    delay,
                    cc,
                    cc_seismogram.parent_seismogram,
                    autoflip,
                    autoselect,
                    self.min_cc,
                    (self.window_pre, self.window_post),
                )

            self.clear_cache()

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

        converged = bool(convergence_list and convergence_list[-1] <= convergence_limit)
        return IccsResult(convergence=np.array(convergence_list), converged=converged)

    def run_mccc(
        self,
        all_seismograms: bool = False,
        min_cc: float = IccsDefaults.mccc_min_cc,
        damping: float = IccsDefaults.mccc_damp,
        abs_max: bool = False,
    ) -> McccResult:
        """Refine picks with the MCCC algorithm.

        This updates the picks of the seismograms with
        [`mccc`][pysmo.tools.signal.mccc]. It can be executed at any point to
        update picks. However, it will not autoselect or autoflip seismograms.
        It is therefore recommended as final step to refine the results of
        [`ICCS()`][pysmo.tools.iccs.ICCS.__call__].

        Args:
            all_seismograms: Whether to run MCCC on all seismograms or only on
                those with `select` set to `True`. Default is `False` (only on
                selected seismograms).
            min_cc: Minimum correlation coefficient required to include a pair
                in the inversion.
            damping: Tikhonov regularisation strength. Set to 0 to disable.
            abs_max: If `True`, uses absolute max correlation (polarity insensitive)
                for the pairwise delays.

        Returns:
            A [`McccResult`][pysmo.tools.iccs.McccResult] containing the
            updated picks and MCCC diagnostic values.
        """
        seismograms = (
            self.cc_seismograms if all_seismograms else self.selected_cc_seismograms
        )

        delays, errors, rmse, cc_means, cc_stds = mccc(
            seismograms, min_cc=min_cc, damping=damping, abs_max=abs_max
        )

        picks: list[pd.Timestamp] = []

        for delay, cc_seis in zip(delays, seismograms):
            seis = cc_seis.parent_seismogram
            _update_seismogram(
                delay,
                cc=None,
                seismogram=seis,
                autoflip=False,
                autoselect=False,
                min_cc_for_autoselect=self.min_cc,
                current_window=(self.window_pre, self.window_post),
            )
            # After update (or attempted update), retrieve the pick.
            # Fallback to t0 if t1 is None (e.g. if update failed and was None).
            picks.append(seis.t0 if pd.isnull(seis.t1) else seis.t1)

        self.clear_cache()
        return McccResult(
            picks=picks, errors=errors, rmse=rmse, cc_means=cc_means, cc_stds=cc_stds
        )

    def update_all_picks(self, pickdelta: pd.Timedelta) -> None:
        """Update [`t1`][pysmo.tools.iccs.IccsSeismogram.t1] in all seismograms by the same amount.

        Args:
            pickdelta: Delta applied to all picks.

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

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

        for seismogram in self.seismograms:
            current_pick = seismogram.t0 if pd.isnull(seismogram.t1) else seismogram.t1
            seismogram.t1 = current_pick + pickdelta
        self.clear_cache()  # seismograms and stack need to be refreshed

bandpass_apply class-attribute instance-attribute

bandpass_apply: bool = field(
    default=IccsDefaults.bandpass_apply,
    converter=bool,
    validator=[
        validators.instance_of(bool),
        _validate_bandpass_apply,
    ],
    on_setattr=setters.pipe(
        setters.convert,
        setters.validate,
        _on_setattr_clear_cache,
    ),
)

Filter seismograms with a bandpass filter before running ICCS.

Setting this to True will apply a bandpass filter (with zerophase set to True) to the cc_seismograms and context_seismograms. It also gates whether a causal (single-pass) counterpart is produced for cc_seismograms_causal and context_seismograms_causal — see corners for how the two variants relate.

As the seismograms may have already been pre-processed (i.e. already filtered) the default value for this parameter is False.

bandpass_fmax class-attribute instance-attribute

bandpass_fmax: float = field(
    default=IccsDefaults.bandpass_fmax,
    converter=float,
    validator=[
        validators.instance_of(float),
        _validate_bandpass_fmax,
    ],
    on_setattr=setters.pipe(
        setters.convert,
        setters.validate,
        _on_setattr_clear_cache,
    ),
)

Bandpass filter maximum frequency (Hz). Only used if bandpass_apply is True.

The filter is applied to each seismogram at its original sampling rate, before any resampling to a common grid. Valid frequencies are therefore bounded by the Nyquist frequency of the most coarsely sampled seismogram, i.e. 0.5 / max_delta.total_seconds().

bandpass_fmin class-attribute instance-attribute

bandpass_fmin: float = field(
    default=IccsDefaults.bandpass_fmin,
    converter=float,
    validator=[
        validators.instance_of(float),
        _validate_bandpass_fmin,
    ],
    on_setattr=setters.pipe(
        setters.convert,
        setters.validate,
        _on_setattr_clear_cache,
    ),
)

Bandpass filter minimum frequency (Hz). Only used if bandpass_apply is True.

The filter is applied to each seismogram at its original sampling rate, before any resampling to a common grid. Valid frequencies are therefore bounded by the Nyquist frequency of the most coarsely sampled seismogram, i.e. 0.5 / max_delta.total_seconds().

cc_seismograms property

cc_seismograms: list[_EphemeralSeismogram]

Return the seismograms as used for the cross-correlation.

These seismograms are derived from the input seismograms and used for the cross-correlation steps. Starting with the input seismograms, they are processed as follows:

  1. Bandpass filtered if bandpass_apply is True.
  2. Resampled to the minimum sampling interval of all input seismograms (only if it is not equal in all seismograms).
  3. Cropped to ramp_width + current time window + ramp_width.
  4. Detrended.
  5. Tapered using ramp_width (tapered sections are outside time window).
  6. Normalised based on the highest absolute value within the cropped window. This step is done slightly differently in context_seismograms — see the documentation of that property for details.

cc_seismograms_causal property

cc_seismograms_causal: list[_EphemeralSeismogram]

Return the seismograms as used for cross-correlation, causally filtered.

Mirrors cc_seismograms, with a causal (single-pass) rather than zero-phase bandpass filter — see corners for how the filter order and passband of the two variants relate. Intended for picking-oriented tools, since a causal filter avoids the acausal precursor smearing a zero-phase filter introduces before the true onset. Never used by the ICCS algorithm itself (cross-correlation, stacking, MCCC), which continues to run on cc_seismograms unconditionally.

When bandpass_apply is False, this returns the same object as cc_seismograms (not a copy).

ccs property

ccs: ndarray

Return an array of the normalised cross-correlation coefficients.

context_seismograms property

context_seismograms: list[_EphemeralSeismogram]

Return the seismograms with extra context for plotting.

These seismograms are derived from the input seismograms and used primarily for plotting with extra context (e.g. when selecting new time window boundaries). Starting with the input seismograms, they are processed as follows:

  1. Bandpass filtered if bandpass_apply is True.
  2. Resampled to the minimum sampling interval of all input seismograms (only if it is not equal in all seismograms).
  3. Cropped and/or padded to context_width + current time window + context_width.
  4. Detrended.
  5. Normalised based on the highest absolute value within the selected time window (i.e. without the context).

context_seismograms_causal property

context_seismograms_causal: list[_EphemeralSeismogram]

Return the seismograms with extra context for plotting, causally filtered.

Mirrors context_seismograms, with a causal (single-pass) rather than zero-phase bandpass filter — see corners for how the filter order and passband of the two variants relate.

The context padding (context_width) exists to show pre-arrival "quiet" alongside the time window, but a causal filter's group delay pushes energy from the true onset forward in time — some of what the padding shows as pre-arrival quiet may actually contain the filter's response to the onset itself, not the true unfiltered signal before it. This is expected behaviour (see corners), not a bug.

When bandpass_apply is False, this returns the same object as context_seismograms (not a copy).

context_stack property

context_stack: MiniSeismogram

Return the stacked context_seismograms.

Returns:

Type Description
MiniSeismogram

Stacked input seismograms with context padding.

context_stack_causal property

context_stack_causal: MiniSeismogram

Return the stacked context_seismograms_causal.

When bandpass_apply is False, this returns the same object as context_stack (not a copy).

Returns:

Type Description
MiniSeismogram

Stacked causally-filtered input seismograms with context padding.

context_width class-attribute instance-attribute

context_width: PositiveTimedelta = field(
    default=IccsDefaults.context_width,
    converter=convert_to_timedelta,
    validator=validators.gt(pd.Timedelta(0)),
    on_setattr=setters.pipe(
        setters.convert,
        setters.validate,
        _on_setattr_clear_cache,
    ),
)

Context padding to apply before and after the time window.

This padding is not used for the cross-correlation.

corners class-attribute instance-attribute

corners: PositiveInt = field(
    default=IccsDefaults.corners,
    converter=int,
    validator=[
        validators.instance_of(int),
        validators.gt(0),
        _validate_corners_causal_band,
    ],
    on_setattr=setters.pipe(
        setters.convert,
        setters.validate,
        _on_setattr_clear_cache,
    ),
)

Number of corners (poles) for the zero-phase bandpass filter applied to cc_seismograms/ context_seismograms when bandpass_apply is True.

The causal counterparts (cc_seismograms_causal, context_seismograms_causal) use 2 * corners poles, matching the rolloff steepness of the zero-phase filter (sosfiltfilt effectively doubles filter order by applying it twice). Their design freqmin/freqmax passed to bandpass are corrected via causal_band so the causal variant's actual -3dB point matches the zero-phase variant's actual (inward-shifted) -3dB point closely — not exactly. See causal_band for the correction itself and its derivation; in short, the residual comes from real Butterworth bandpass edges interacting with each other (not behaving as two independent single-edge filters, which is what the correction formula assumes). At ICCS's own defaults (corners=2, freqmax/freqmin=40) the residual settles at ≈1.33% at typical broadband seismic sample rates, shrinking further with wider relative bandwidth or higher corners and growing with narrower bandwidth or lower corners; at low sample rates relative to bandpass_fmax (e.g. 20 Hz) a second, otherwise-negligible effect compounds with it and the residual grows to ≈2.37% (worked example and exact figures: causal_band's own docstring).

Because the correction moves both corrected edges inward from the nominal band, corners is no longer independent of bandpass_fmin/ bandpass_fmax: lowering corners while they stay fixed (or narrowing them while corners stays fixed) can raise ValueError if the corrected band would invert — the ratio causal_band applies is monotonically increasing in corners, so raising corners can only move a combination towards validity, never away from it; only lowering it (or narrowing the nominal band) can break a previously-valid one. See bandpass_fmin/ bandpass_fmax for the coupled validation. This is a distinct, frequency-validity constraint from the numerical-stability finding below — there is still no cap on corners for numerical-stability reasons; the new ValueError is a different mechanism entirely.

The causal variant also has non-zero, frequency-dependent group delay (unlike the zero-phase variant, which has none by construction): a causal filter only pushes energy forward in time, so a pick made on cc_seismograms_causal is systematically biased later than the true onset by an amount that depends on corners, bandpass_fmin, and bandpass_fmax — this is the trade-off for avoiding zero-phase's backward precursor smearing, not something corrected here. Compute group_delay on the same corners/frequency combination if an exact per-configuration correction is needed.

Raising corners well above typical seismological values has no dedicated guard: filtering stays numerically well-behaved even at extreme orders, with no warning (from scipy or otherwise) on the code path bandpass actually uses.

max_delta property

max_delta: Timedelta

Maximum sampling interval across all seismograms.

max_td_pre property

max_td_pre: Timedelta

Maximum negative time delta between pick and seismogram begin_time.

This property is used to calculate valid values for updating picks and attributes. Specifically:

  • new_pick > old_pick + max_td_pre - window_pre + ramp_width
  • window_pre >= max_td_pre + ramp_width
  • ramp_width <= window_pre - max_td_pre

min_cc class-attribute instance-attribute

min_cc: float = field(
    default=IccsDefaults.min_cc,
    converter=float,
    validator=validators.instance_of(float),
    on_setattr=setters.pipe(
        setters.convert,
        setters.validate,
        _on_setattr_clear_cache,
    ),
)

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 correlation coefficients below this value is set to False, and they are no longer used for the stack.

min_delta property

min_delta: Timedelta

Minimum sampling interval across all seismograms.

min_td_post property

min_td_post: Timedelta

Minimum positive time delta between pick and seismogram end_time.

This property is used to calculate valid values for updating picks and attributes. Specifically:

  • new_pick < old_pick + min_td_post - window_post - ramp_width
  • window_post <= min_td_post - ramp_width
  • ramp_width <= min_td_post - window_post

ramp_width class-attribute instance-attribute

ramp_width: NonNegativeTimedelta | NonNegativeNumber = (
    field(
        default=IccsDefaults.ramp_width,
        validator=_validate_ramp_width,
        on_setattr=setters.pipe(
            setters.validate, _on_setattr_clear_cache
        ),
    )
)

Width of taper ramp up and down.

Interpretation depends on the type

Can be either a timedelta or a float, but they mean slightly different things. A float is interpreted as a fraction of the window duration, while a timedelta is an absolute duration. See the documentation of pysmo.functions.window() for details.

ramp_width_timedelta property

ramp_width_timedelta: Timedelta

Ramp width as a pandas.Timedelta, computed from the current window.

For a float ramp_width, the duration is computed as a fraction of the window duration (window_post - window_pre), consistent with the behaviour of pysmo.functions.window.

seismograms class-attribute instance-attribute

seismograms: Sequence[IccsSeismogram] = field(
    factory=list[IccsSeismogram],
    on_setattr=_on_setattr_clear_cache,
)

Input seismograms.

These are the source seismograms from which the ephemeral seismograms (cross-correlation and context seismograms, see the module documentation) are derived on demand. The ephemeral seismograms are cached and regenerated automatically whenever a controlling attribute such as window_pre or window_post changes.

In-place mutation bypasses the cache

Assigning a new list to this attribute clears the cache automatically. Mutating the list in place (e.g. with append, remove, or direct index assignment) bypasses the setter and does not clear the cache. Call clear_cache manually after any such in-place mutation.

Remove poor-quality seismograms, don't just deselect them

When a seismogram is of sufficiently poor quality that it should play no further role in the analysis, consider removing it from this list rather than simply setting its select attribute to False. A deselected seismogram is excluded from the stack and correlation output, but its pick and data span still constrain the valid ranges for window_pre, window_post, and pick updates — because all seismograms (selected or not) are used to generate the ephemeral seismograms. A badly drifting pick on a deselected seismogram can therefore make it impossible to set useful window or pick ranges for the remaining good seismograms.

selected_cc_seismograms property

selected_cc_seismograms: list[_EphemeralSeismogram]

Return the cc_seismograms with select set to True.

stack property

Return 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:

Type Description
MiniSeismogram

Stacked input seismograms.

stack_causal property

stack_causal: MiniSeismogram

Return the stacked cc_seismograms_causal.

When bandpass_apply is False, this returns the same object as stack (not a copy).

Returns:

Type Description
MiniSeismogram

Stacked causally-filtered input seismograms.

window_post class-attribute instance-attribute

window_post: PositiveTimedelta = field(
    default=IccsDefaults.window_post,
    validator=[
        validators.gt(pd.Timedelta(0)),
        _validate_window_post,
    ],
    on_setattr=setters.pipe(
        setters.validate, _on_setattr_clear_cache
    ),
)

End of the time window relative to the pick.

window_pre class-attribute instance-attribute

window_pre: NegativeTimedelta = field(
    default=IccsDefaults.window_pre,
    validator=[
        validators.lt(pd.Timedelta(0)),
        _validate_window_pre,
    ],
    on_setattr=setters.pipe(
        setters.validate, _on_setattr_clear_cache
    ),
)

Beginning of the time window relative to the pick.

__call__

__call__(
    autoflip: bool = False,
    autoselect: bool = False,
    convergence_limit: float = convergence_limit,
    convergence_method: ConvergenceMethod = convergence_method,
    max_iter: int = max_iter,
    max_shift: NonNegativeTimedelta | None = None,
) -> IccsResult

Run the ICCS algorithm.

Parameters:

Name Type Description Default
autoflip bool

Automatically toggle flip attribute of seismograms.

False
autoselect bool

Automatically set select attribute to False for poor quality seismograms.

False
convergence_limit float

Convergence limit at which the algorithm stops.

convergence_limit
convergence_method ConvergenceMethod

Method to calculate convergence criterion.

convergence_method
max_iter int

Maximum number of iterations.

max_iter
max_shift NonNegativeTimedelta | None

Maximum (absolute) shift to consider in each iteration, relative to the stack at that iteration (see delay()). This is not a bound on the cumulative shift over the full run.

None

Returns:

Type Description
IccsResult

An IccsResult containing the

IccsResult

convergence history and whether the convergence limit was reached.

Source code in src/pysmo/tools/iccs/_iccs.py
def __call__(
    self,
    autoflip: bool = False,
    autoselect: bool = False,
    convergence_limit: float = IccsDefaults.convergence_limit,
    convergence_method: ConvergenceMethod = IccsDefaults.convergence_method,
    max_iter: int = IccsDefaults.max_iter,
    max_shift: NonNegativeTimedelta | None = None,
) -> IccsResult:
    """Run the ICCS algorithm.

    Args:
        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 (absolute) shift to consider in each iteration,
            relative to the stack at that iteration (see
            [`delay()`][pysmo.tools.signal.delay]). This is not a bound on
            the cumulative shift over the full run.

    Returns:
        An [`IccsResult`][pysmo.tools.iccs.IccsResult] containing the
        convergence history and whether the convergence limit was reached.
    """
    convergence_list = []

    for _ in range(max_iter):
        # Save the previous stack to calculate convergence criterion after updating the seismograms.
        prev_stack = clone_to_mini(MiniSeismogram, self.stack)

        # Get delays and correlation coefficients for all seismograms in one go
        delays, ccs = multi_delay(
            self.stack,
            self.cc_seismograms,
            abs_max=autoflip,
            max_shift=max_shift,
        )

        # Update seismograms based on results and settings.
        for delay, cc, cc_seismogram in zip(delays, ccs, self.cc_seismograms):
            _update_seismogram(
                delay,
                cc,
                cc_seismogram.parent_seismogram,
                autoflip,
                autoselect,
                self.min_cc,
                (self.window_pre, self.window_post),
            )

        self.clear_cache()

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

    converged = bool(convergence_list and convergence_list[-1] <= convergence_limit)
    return IccsResult(convergence=np.array(convergence_list), converged=converged)

clear_cache

clear_cache() -> None

Clear all cached ephemeral seismograms, stacks, and derived results.

Ephemeral seismograms (cross-correlation, context, and their causal counterparts), their stacks, cross-correlation norms, and the valid pick and window ranges are all computed on demand and cached to avoid redundant work. The cache is invalidated automatically when a controlling attribute such as window_pre, window_post, or seismograms is reassigned.

Call this method manually after any in-place mutation of seismograms (e.g. append, remove, or index assignment) to ensure all ephemeral seismograms and derived results are regenerated from the updated input.

Source code in src/pysmo/tools/iccs/_iccs.py
def clear_cache(self) -> None:
    """Clear all cached ephemeral seismograms, stacks, and derived results.

    Ephemeral seismograms (cross-correlation, context, and their causal
    counterparts), their stacks, cross-correlation norms, and the valid
    pick and window ranges are all computed on demand and cached to
    avoid redundant work.
    The cache is invalidated automatically when a controlling attribute
    such as [`window_pre`][pysmo.tools.iccs.ICCS.window_pre],
    [`window_post`][pysmo.tools.iccs.ICCS.window_post], or
    [`seismograms`][pysmo.tools.iccs.ICCS.seismograms] is *reassigned*.

    Call this method manually after any in-place mutation of
    [`seismograms`][pysmo.tools.iccs.ICCS.seismograms] (e.g. `append`,
    `remove`, or index assignment) to ensure all ephemeral seismograms and
    derived results are regenerated from the updated input.
    """
    self._cc_seismograms_cache = None
    self._context_seismograms_cache = None
    self._ccs_cache = None
    self._cc_stack_cache = None
    self._context_stack_cache = None
    self._cc_seismograms_causal_cache = None
    self._context_seismograms_causal_cache = None
    self._cc_stack_causal_cache = None
    self._context_stack_causal_cache = None
    self._max_td_pre_cache = None
    self._min_td_post_cache = None
    self._valid_pick_range_cache = None

run_mccc

run_mccc(
    all_seismograms: bool = False,
    min_cc: float = mccc_min_cc,
    damping: float = mccc_damp,
    abs_max: bool = False,
) -> McccResult

Refine picks with the MCCC algorithm.

This updates the picks of the seismograms with mccc. It can be executed at any point to update picks. However, it will not autoselect or autoflip seismograms. It is therefore recommended as final step to refine the results of ICCS().

Parameters:

Name Type Description Default
all_seismograms bool

Whether to run MCCC on all seismograms or only on those with select set to True. Default is False (only on selected seismograms).

False
min_cc float

Minimum correlation coefficient required to include a pair in the inversion.

mccc_min_cc
damping float

Tikhonov regularisation strength. Set to 0 to disable.

mccc_damp
abs_max bool

If True, uses absolute max correlation (polarity insensitive) for the pairwise delays.

False

Returns:

Type Description
McccResult

A McccResult containing the

McccResult

updated picks and MCCC diagnostic values.

Source code in src/pysmo/tools/iccs/_iccs.py
def run_mccc(
    self,
    all_seismograms: bool = False,
    min_cc: float = IccsDefaults.mccc_min_cc,
    damping: float = IccsDefaults.mccc_damp,
    abs_max: bool = False,
) -> McccResult:
    """Refine picks with the MCCC algorithm.

    This updates the picks of the seismograms with
    [`mccc`][pysmo.tools.signal.mccc]. It can be executed at any point to
    update picks. However, it will not autoselect or autoflip seismograms.
    It is therefore recommended as final step to refine the results of
    [`ICCS()`][pysmo.tools.iccs.ICCS.__call__].

    Args:
        all_seismograms: Whether to run MCCC on all seismograms or only on
            those with `select` set to `True`. Default is `False` (only on
            selected seismograms).
        min_cc: Minimum correlation coefficient required to include a pair
            in the inversion.
        damping: Tikhonov regularisation strength. Set to 0 to disable.
        abs_max: If `True`, uses absolute max correlation (polarity insensitive)
            for the pairwise delays.

    Returns:
        A [`McccResult`][pysmo.tools.iccs.McccResult] containing the
        updated picks and MCCC diagnostic values.
    """
    seismograms = (
        self.cc_seismograms if all_seismograms else self.selected_cc_seismograms
    )

    delays, errors, rmse, cc_means, cc_stds = mccc(
        seismograms, min_cc=min_cc, damping=damping, abs_max=abs_max
    )

    picks: list[pd.Timestamp] = []

    for delay, cc_seis in zip(delays, seismograms):
        seis = cc_seis.parent_seismogram
        _update_seismogram(
            delay,
            cc=None,
            seismogram=seis,
            autoflip=False,
            autoselect=False,
            min_cc_for_autoselect=self.min_cc,
            current_window=(self.window_pre, self.window_post),
        )
        # After update (or attempted update), retrieve the pick.
        # Fallback to t0 if t1 is None (e.g. if update failed and was None).
        picks.append(seis.t0 if pd.isnull(seis.t1) else seis.t1)

    self.clear_cache()
    return McccResult(
        picks=picks, errors=errors, rmse=rmse, cc_means=cc_means, cc_stds=cc_stds
    )

update_all_picks

update_all_picks(pickdelta: Timedelta) -> None

Update t1 in all seismograms by the same amount.

Parameters:

Name Type Description Default
pickdelta Timedelta

Delta applied to all picks.

required

Raises:

Type Description
ValueError

If the new t1 is outside the valid range.

Source code in src/pysmo/tools/iccs/_iccs.py
def update_all_picks(self, pickdelta: pd.Timedelta) -> None:
    """Update [`t1`][pysmo.tools.iccs.IccsSeismogram.t1] in all seismograms by the same amount.

    Args:
        pickdelta: Delta applied to all picks.

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

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

    for seismogram in self.seismograms:
        current_pick = seismogram.t0 if pd.isnull(seismogram.t1) else seismogram.t1
        seismogram.t1 = current_pick + pickdelta
    self.clear_cache()  # seismograms and stack need to be refreshed

validate_pick

validate_pick(pick: Timedelta) -> bool

Check whether a new pick is valid given all seismograms in the instance.

The valid pick range is computed from every seismogram in seismograms, including those with select set to False. A pick is considered valid if it lies within this global range; selection only affects stacking, not the validity bounds.

Parameters:

Name Type Description Default
pick Timedelta

New pick to validate.

required

Returns:

Type Description
bool

Whether the new pick is valid.

Source code in src/pysmo/tools/iccs/_iccs.py
def validate_pick(self, pick: pd.Timedelta) -> bool:
    """Check whether a new pick is valid given all seismograms in the instance.

    The valid pick range is computed from every seismogram in
    [`seismograms`][pysmo.tools.iccs.ICCS.seismograms], including those with
    [`select`][pysmo.tools.iccs.IccsSeismogram.select] set to
    [`False`][]. A pick is considered valid if it lies within this
    global range; selection only affects stacking, not the validity bounds.

    Args:
        pick: New pick to validate.

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

    if self._valid_pick_range_cache is None:
        self._valid_pick_range_cache = _calc_valid_pick_range(self)

    return (
        self._valid_pick_range_cache[0] <= pick <= self._valid_pick_range_cache[1]
    )

validate_time_window

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

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

Validates that the proposed window fits within every seismogram, accounting for the taper ramp. The ramp duration is computed from the proposed window_pre and window_post values, consistent with how pysmo.functions.window computes it for float ramp_width.

Parameters:

Name Type Description Default
window_pre Timedelta

Proposed window start time (negative, relative to pick).

required
window_post Timedelta

Proposed window end time (positive, relative to pick).

required

Returns:

Type Description
bool

Whether the new time window is valid for all seismograms.

Source code in src/pysmo/tools/iccs/_iccs.py
def validate_time_window(
    self, window_pre: pd.Timedelta, window_post: pd.Timedelta
) -> bool:
    """Check if a new time window (relative to pick) is valid.

    Validates that the proposed window fits within every seismogram,
    accounting for the taper ramp. The ramp duration is computed from
    the proposed `window_pre` and `window_post` values, consistent
    with how `pysmo.functions.window` computes it for float
    `ramp_width`.

    Args:
        window_pre: Proposed window start time (negative, relative to pick).
        window_post: Proposed window end time (positive, relative to pick).

    Returns:
        Whether the new time window is valid for all seismograms.
    """

    if window_pre >= window_post:
        return False

    if window_pre > -self.min_delta:
        return False

    if window_post < self.min_delta:
        return False

    ramp = _compute_ramp(self.ramp_width, window_pre, window_post)
    for s in self.seismograms:
        pick = s.t0 if pd.isnull(s.t1) else s.t1
        if window_pre < s.begin_time - pick + ramp:
            return False
        if window_post > s.end_time - pick - ramp:
            return False
    return True

IccsResult

Result returned by ICCS.__call__().

Attributes:

Name Type Description
converged bool

Whether the convergence limit was reached before max_iter iterations.

convergence ndarray

Convergence criterion value after each iteration.

Source code in src/pysmo/tools/iccs/_types.py
@define(frozen=True)
class IccsResult:
    """Result returned by [`ICCS.__call__()`][pysmo.tools.iccs.ICCS.__call__]."""

    convergence: np.ndarray
    """Convergence criterion value after each iteration."""

    converged: bool
    """Whether the convergence limit was reached before `max_iter` iterations."""

converged instance-attribute

converged: bool

Whether the convergence limit was reached before max_iter iterations.

convergence instance-attribute

convergence: ndarray

Convergence criterion value after each iteration.

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.

Attributes:

Name Type Description
begin_time Timestamp

Seismogram begin time.

data ndarray

Seismogram data.

delta Timedelta

The sampling interval.

end_time Timestamp

Seismogram end time.

extra dict[Hashable, Any]

Extra metadata to store alongside the seismogram.

flip bool

Whether the seismogram data should be flipped for ICCS.

select bool

Whether to use the seismogram in the stack.

t0 Timestamp

Initial pick.

t1 Timestamp | None

Updated pick.

Source code in src/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.
    """

    t0: pd.Timestamp
    """Initial pick."""

    t1: pd.Timestamp | None
    """Updated pick."""

    flip: bool
    """Whether the seismogram data should be flipped for ICCS."""

    select: bool
    """Whether to use the seismogram in the stack."""

    extra: dict[Hashable, Any]
    """Extra metadata to store alongside the seismogram."""

begin_time instance-attribute

begin_time: Timestamp

Seismogram begin time.

data instance-attribute

data: ndarray

Seismogram data.

delta instance-attribute

delta: Timedelta

The sampling interval.

Should be a positive pd.Timedelta instance.

end_time property

end_time: Timestamp

Seismogram end time.

extra instance-attribute

extra: dict[Hashable, Any]

Extra metadata to store alongside the seismogram.

flip instance-attribute

flip: bool

Whether the seismogram data should be flipped for ICCS.

select instance-attribute

select: bool

Whether to use the seismogram in the stack.

t0 instance-attribute

Initial pick.

t1 instance-attribute

t1: Timestamp | None

Updated pick.

McccResult

Result returned by ICCS.run_mccc().

These results include all seismograms if run_mccc() is called with all_seismograms=True, only the selected ones otherwise.

Attributes:

Name Type Description
cc_means list[float]

Per-seismogram mean cross-correlation coefficient (waveform quality).

cc_stds list[float]

Per-seismogram standard deviation of cross-correlation coefficients (waveform consistency).

errors list[Timedelta]

Per-seismogram timing precision (standard error from covariance matrix).

picks list[Timestamp]

Final absolute arrival times for each seismogram.

rmse Timedelta

Root-mean-square error of the inversion fit across the whole array.

Source code in src/pysmo/tools/iccs/_types.py
@define(frozen=True)
class McccResult:
    """Result returned by [`ICCS.run_mccc()`][pysmo.tools.iccs.ICCS.run_mccc].

    These results include all seismograms if `run_mccc()` is called with
    `all_seismograms=True`, only the selected ones otherwise.
    """

    picks: list[pd.Timestamp]
    """Final absolute arrival times for each seismogram."""

    errors: list[pd.Timedelta]
    """Per-seismogram timing precision (standard error from covariance matrix)."""

    rmse: pd.Timedelta
    """Root-mean-square error of the inversion fit across the whole array."""

    cc_means: list[float]
    """Per-seismogram mean cross-correlation coefficient (waveform quality)."""

    cc_stds: list[float]
    """Per-seismogram standard deviation of cross-correlation coefficients (waveform consistency)."""

cc_means instance-attribute

cc_means: list[float]

Per-seismogram mean cross-correlation coefficient (waveform quality).

cc_stds instance-attribute

cc_stds: list[float]

Per-seismogram standard deviation of cross-correlation coefficients (waveform consistency).

errors instance-attribute

errors: list[Timedelta]

Per-seismogram timing precision (standard error from covariance matrix).

picks instance-attribute

picks: list[Timestamp]

Final absolute arrival times for each seismogram.

rmse instance-attribute

rmse: Timedelta

Root-mean-square error of the inversion fit across the whole array.

MiniIccsSeismogram

Bases: SeismogramEndtimeMixin

Minimal implementation of the IccsSeismogram type.

Examples:

Because IccsSeismogram inherits from Seismogram, MiniIccsSeismogram instances can easily be created from existing seismograms using the clone_to_mini() function, with the update parameter providing the extra information needed:

>>> from pysmo.classes import SAC
>>> from pysmo.functions import clone_to_mini
>>> from pysmo.tools.iccs import MiniIccsSeismogram
>>> import pandas as pd
>>> sac = SAC.from_file("example.sac")
>>> sac_seis = sac.seismogram
>>> # Use existing pick or set a new one 10 seconds after begin time
>>> update = {"t0": sac_seis.begin_time + pd.Timedelta(seconds=10) if pd.isnull(sac.timestamps.t0) else sac.timestamps.t0}
>>> mini_iccs_seis = clone_to_mini(MiniIccsSeismogram, sac_seis, update=update)
>>>

Attributes:

Name Type Description
begin_time UtcTimestamp

Seismogram begin time.

data ndarray

Seismogram data.

delta PositiveTimedelta

Seismogram sampling interval.

end_time Timestamp

Seismogram end time.

extra dict[Hashable, Any]

Extra metadata to store alongside the seismogram.

flip bool

Whether the seismogram data should be flipped for ICCS.

select bool

Whether to use the seismogram in the stack.

t0 UtcTimestamp

Initial pick.

t1 UtcTimestamp | None

Updated pick.

Source code in src/pysmo/tools/iccs/_types.py
@define(kw_only=True)
class MiniIccsSeismogram(SeismogramEndtimeMixin):
    """Minimal implementation of the [`IccsSeismogram`][pysmo.tools.iccs.IccsSeismogram] type.

    Examples:
        Because [`IccsSeismogram`][pysmo.tools.iccs.IccsSeismogram] inherits
        from [`Seismogram`][pysmo.Seismogram],
        [`MiniIccsSeismogram`][pysmo.tools.iccs.MiniIccsSeismogram] instances
        can easily be created from existing seismograms using the
        [`clone_to_mini()`][pysmo.functions.clone_to_mini] function, with the
        `update` parameter providing the extra information needed:

        ```python
        >>> from pysmo.classes import SAC
        >>> from pysmo.functions import clone_to_mini
        >>> from pysmo.tools.iccs import MiniIccsSeismogram
        >>> import pandas as pd
        >>> sac = SAC.from_file("example.sac")
        >>> sac_seis = sac.seismogram
        >>> # Use existing pick or set a new one 10 seconds after begin time
        >>> update = {"t0": sac_seis.begin_time + pd.Timedelta(seconds=10) if pd.isnull(sac.timestamps.t0) else sac.timestamps.t0}
        >>> mini_iccs_seis = clone_to_mini(MiniIccsSeismogram, sac_seis, update=update)
        >>>
        ```
    """

    begin_time: UtcTimestamp = field(
        default=SeismogramDefaults.begin_time,
        converter=convert_to_utc_timestamp,
        on_setattr=setters.convert,
    )
    """Seismogram begin time."""

    delta: PositiveTimedelta = field(
        default=SeismogramDefaults.delta,
        converter=convert_to_timedelta,
        validator=[
            validators.instance_of(pd.Timedelta),
            validators.gt(pd.Timedelta(0)),
        ],
        on_setattr=setters.pipe(setters.convert, setters.validate),
    )
    """Seismogram sampling interval."""

    data: np.ndarray = field(
        factory=lambda: np.array([]),
        converter=convert_to_ndarray,
        validator=validators.instance_of(np.ndarray),
        on_setattr=setters.pipe(setters.convert, setters.validate),
    )
    """Seismogram data."""

    t0: UtcTimestamp = field(
        converter=convert_to_utc_timestamp, on_setattr=setters.convert
    )
    """Initial pick."""

    t1: UtcTimestamp | None = field(
        default=None,
        converter=converters.optional(convert_to_utc_timestamp),
        on_setattr=setters.convert,
    )
    """Updated pick."""

    flip: bool = field(
        default=False,
        converter=bool,
        validator=validators.instance_of(bool),
        on_setattr=setters.pipe(setters.convert, setters.validate),
    )
    """Whether the seismogram data should be flipped for ICCS."""

    select: bool = field(
        default=True,
        converter=bool,
        validator=validators.instance_of(bool),
        on_setattr=setters.pipe(setters.convert, setters.validate),
    )
    """Whether to use the seismogram in the stack."""

    extra: dict[Hashable, Any] = field(factory=dict)
    """Extra metadata to store alongside the seismogram."""

begin_time class-attribute instance-attribute

begin_time: UtcTimestamp = field(
    default=SeismogramDefaults.begin_time,
    converter=convert_to_utc_timestamp,
    on_setattr=setters.convert,
)

Seismogram begin time.

data class-attribute instance-attribute

data: ndarray = field(
    factory=lambda: np.array([]),
    converter=convert_to_ndarray,
    validator=validators.instance_of(np.ndarray),
    on_setattr=setters.pipe(
        setters.convert, setters.validate
    ),
)

Seismogram data.

delta class-attribute instance-attribute

delta: PositiveTimedelta = field(
    default=SeismogramDefaults.delta,
    converter=convert_to_timedelta,
    validator=[
        validators.instance_of(pd.Timedelta),
        validators.gt(pd.Timedelta(0)),
    ],
    on_setattr=setters.pipe(
        setters.convert, setters.validate
    ),
)

Seismogram sampling interval.

end_time property

end_time: Timestamp

Seismogram end time.

extra class-attribute instance-attribute

extra: dict[Hashable, Any] = field(factory=dict)

Extra metadata to store alongside the seismogram.

flip class-attribute instance-attribute

flip: bool = field(
    default=False,
    converter=bool,
    validator=validators.instance_of(bool),
    on_setattr=setters.pipe(
        setters.convert, setters.validate
    ),
)

Whether the seismogram data should be flipped for ICCS.

select class-attribute instance-attribute

select: bool = field(
    default=True,
    converter=bool,
    validator=validators.instance_of(bool),
    on_setattr=setters.pipe(
        setters.convert, setters.validate
    ),
)

Whether to use the seismogram in the stack.

t0 class-attribute instance-attribute

t0: UtcTimestamp = field(
    converter=convert_to_utc_timestamp,
    on_setattr=setters.convert,
)

Initial pick.

t1 class-attribute instance-attribute

t1: UtcTimestamp | None = field(
    default=None,
    converter=converters.optional(convert_to_utc_timestamp),
    on_setattr=setters.convert,
)

Updated pick.

plot_matrix_image

plot_matrix_image(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    causal: bool = False,
    return_fig: bool = True,
) -> tuple[Figure, Axes] | None

Plot the selected ICCS seismograms as a matrix image.

Parameters:

Name Type Description Default
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

True
all_seismograms bool

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

False
causal bool

Determines the filter phase behaviour: - True: causally-filtered (single-pass) seismograms are used. - False: zero-phase filtered seismograms are used.

False
return_fig bool

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

True

Returns:

Type Description
tuple[Figure, Axes] | None

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

Examples:

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

>>> from pysmo.tools.iccs import ICCS, plot_matrix_image
>>> iccs = ICCS(iccs_seismograms)
>>> _ = iccs(autoselect=True, autoflip=True)
>>>
>>> fig, ax = plot_matrix_image(iccs)
>>> # fig.show() # or integrate into your own application
>>>

Matrix image of context seismograms Matrix image of context seismograms

To view the matrix image composed of seismograms as used in the cross-correlations, set the context argument to False:

>>> fig, ax = plot_matrix_image(iccs, context=False)
>>> # fig.show() # or integrate into your own application
>>>

View the matrix image of cc seismograms View the matrix image of cc seismograms

To view the causally-filtered variant used by picking-oriented tools (avoiding the acausal precursor smearing a zero-phase filter introduces before the true onset), set causal to True:

>>> fig, ax = plot_matrix_image(iccs, context=False, causal=True)
>>> # fig.show() # or integrate into your own application
>>>
Source code in src/pysmo/tools/iccs/plot.py
def plot_matrix_image(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    causal: bool = False,
    return_fig: bool = True,
) -> tuple[Figure, Axes] | None:
    """Plot the selected ICCS seismograms as a matrix image.

    Args:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        causal: Determines the filter phase behaviour:
            - `True`: causally-filtered (single-pass) seismograms are used.
            - `False`: zero-phase filtered seismograms are used.
        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 a matrix image if `return_fig` is `True`.

    Examples:
        The default plotting mode is to pad the seismograms 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_matrix_image
        >>> iccs = ICCS(iccs_seismograms)
        >>> _ = iccs(autoselect=True, autoflip=True)
        >>>
        >>> fig, ax = plot_matrix_image(iccs)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

        <!-- invisible-code-block: python
        ```
        >>> if savedir:
        ...     fig.savefig(savedir / "iccs_context_image.png", transparent=True)
        ...     plt.style.use("dark_background")
        ...     fig, ax = plot_matrix_image(iccs)
        ...     fig.savefig(savedir / "iccs_context_image-dark.png", transparent=True)
        ...     plt.style.use("default")
        >>>
        ```
        -->

        ![Matrix image of context seismograms](../../../images/sybil/iccs_context_image.png#only-light){ loading=lazy }
        ![Matrix image of context seismograms](../../../images/sybil/iccs_context_image-dark.png#only-dark){ loading=lazy }

        To view the matrix image composed of seismograms as used in the
        cross-correlations, set the `context` argument to `False`:

        ```python
        >>> fig, ax = plot_matrix_image(iccs, context=False)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

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

        ![View the matrix image of cc seismograms](../../../images/sybil/iccs_cc_image.png#only-light){ loading=lazy }
        ![View the matrix image of cc seismograms](../../../images/sybil/iccs_cc_image-dark.png#only-dark){ loading=lazy }

        To view the causally-filtered variant used by picking-oriented tools
        (avoiding the acausal precursor smearing a zero-phase filter
        introduces before the true onset), set `causal` to `True`:

        ```python
        >>> fig, ax = plot_matrix_image(iccs, context=False, causal=True)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```
    """
    fig, ax = plt.subplots(figsize=(10, 5.4))
    fig.subplots_adjust(bottom=0.12, left=0.05, right=0.95, top=0.93)
    draw_common_matrix_image(ax, iccs, context, all_seismograms, causal)
    if return_fig:
        return fig, ax
    plt.show()
    return None

plot_stack

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

Plot the ICCS stack.

Parameters:

Name Type Description Default
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

True
all_seismograms bool

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

False
causal bool

Determines the filter phase behaviour: - True: causally-filtered (single-pass) seismograms are used. - False: zero-phase filtered seismograms are used.

False
return_fig bool

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

True

Returns:

Type Description
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 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)
>>>
>>> fig, ax = plot_stack(iccs)
>>> # fig.show() # or integrate into your own application
>>>

View the context stack View the context stack

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

>>> fig, ax = plot_stack(iccs, context=False)
>>> # fig.show() # or integrate into your own application
>>>

View the cc stack View the cc stack

To view the causally-filtered variant used by picking-oriented tools (avoiding the acausal precursor smearing a zero-phase filter introduces before the true onset), set causal to True:

>>> fig, ax = plot_stack(iccs, context=False, causal=True)
>>> # fig.show() # or integrate into your own application
>>>
Source code in src/pysmo/tools/iccs/plot.py
def plot_stack(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    causal: bool = False,
    return_fig: bool = True,
) -> tuple[Figure, Axes] | None:
    """Plot the ICCS stack.

    Args:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        causal: Determines the filter phase behaviour:
            - `True`: causally-filtered (single-pass) seismograms are used.
            - `False`: zero-phase filtered seismograms are used.
        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
        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)
        >>>
        >>> fig, ax = plot_stack(iccs)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

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

        ![View the context stack](../../../images/sybil/iccs_context_stack.png#only-light){ loading=lazy }
        ![View the context stack](../../../images/sybil/iccs_context_stack-dark.png#only-dark){ loading=lazy }

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

        ```python
        >>> fig, ax = plot_stack(iccs, context=False)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

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

        ![View the cc stack](../../../images/sybil/iccs_cc_stack.png#only-light){ loading=lazy }
        ![View the cc stack](../../../images/sybil/iccs_cc_stack-dark.png#only-dark){ loading=lazy }

        To view the causally-filtered variant used by picking-oriented tools
        (avoiding the acausal precursor smearing a zero-phase filter
        introduces before the true onset), set `causal` to `True`:

        ```python
        >>> fig, ax = plot_stack(iccs, context=False, causal=True)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```
    """
    fig, ax = plt.subplots(figsize=(10, 5.4))
    fig.subplots_adjust(bottom=0.12, left=0.09, right=0.95, top=0.93)
    draw_common_stack(ax, iccs, context, all_seismograms, causal)
    if return_fig:
        return fig, ax
    plt.show()
    return None

update_bandpass

update_bandpass(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    use_matrix_image: bool = False,
    return_fig: bool = True,
) -> (
    tuple[
        Figure,
        Axes,
        tuple[
            CheckButtons,
            Slider,
            Slider,
            Slider,
            RadioButtons,
            Button,
            Button,
        ],
    ]
    | None
)

Interactively update the bandpass filter parameters.

This function launches an interactive figure to adjust bandpass_apply, bandpass_fmin, bandpass_fmax, and corners with a live preview. A radio button toggles the preview between the causal and zero-phase variants — this is UI-only, not a saved parameter, since both renderings are relevant while tuning rather than there being a default to pick.

Parameters:

Name Type Description Default
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

True
all_seismograms bool

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

False
use_matrix_image bool

Use the matrix image instead of the stack.

False
return_fig bool

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

True

Returns:

Type Description
tuple[Figure, Axes, tuple[CheckButtons, Slider, Slider, Slider, RadioButtons, Button, Button]] | None

Figure with the filter widgets if return_fig is True.

Examples:

>>> from pysmo.tools.iccs import ICCS, update_bandpass
>>> iccs = ICCS(iccs_seismograms)
>>> iccs.bandpass_apply = True  # start with bandpass applied
>>> _ = iccs(autoselect=True, autoflip=True)
>>>
>>> fig, ax, widgets = update_bandpass(iccs)
>>> # fig.show() # or integrate into your own application
>>>

Updating bandpass filter parameters Updating bandpass filter parameters

Source code in src/pysmo/tools/iccs/plot.py
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
def update_bandpass(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    use_matrix_image: bool = False,
    return_fig: bool = True,
) -> (
    tuple[
        Figure,
        Axes,
        tuple[CheckButtons, Slider, Slider, Slider, RadioButtons, Button, Button],
    ]
    | None
):
    """Interactively update the bandpass filter parameters.

    This function launches an interactive figure to adjust
    [`bandpass_apply`][pysmo.tools.iccs.ICCS.bandpass_apply],
    [`bandpass_fmin`][pysmo.tools.iccs.ICCS.bandpass_fmin],
    [`bandpass_fmax`][pysmo.tools.iccs.ICCS.bandpass_fmax], and
    [`corners`][pysmo.tools.iccs.ICCS.corners] with a live preview. A radio
    button toggles the preview between the causal and zero-phase variants —
    this is UI-only, not a saved parameter, since both renderings are
    relevant while tuning rather than there being a default to pick.

    Args:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        use_matrix_image: Use the
            [matrix image][pysmo.tools.iccs.plot_matrix_image]
            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 with the filter widgets if `return_fig` is `True`.

    Examples:
        ```python
        >>> from pysmo.tools.iccs import ICCS, update_bandpass
        >>> iccs = ICCS(iccs_seismograms)
        >>> iccs.bandpass_apply = True  # start with bandpass applied
        >>> _ = iccs(autoselect=True, autoflip=True)
        >>>
        >>> fig, ax, widgets = update_bandpass(iccs)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

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

        ![Updating bandpass filter parameters](../../../images/sybil/iccs_update_bandpass.png#only-light){ loading=lazy }
        ![Updating bandpass filter parameters](../../../images/sybil/iccs_update_bandpass-dark.png#only-dark){ loading=lazy }
    """
    _orig_apply = iccs.bandpass_apply
    _orig_fmin = iccs.bandpass_fmin
    _orig_fmax = iccs.bandpass_fmax
    _orig_corners = iccs.corners

    _BANDPASS_CACHE_SIZE = 8

    def _quantise_freq(f: float, sig_figs: int = 4) -> float:
        """Round *f* to *sig_figs* significant figures for use as a cache key."""
        if f <= 0:
            return f
        magnitude = 10 ** (sig_figs - 1 - int(np.floor(np.log10(f))))
        return round(f * magnitude) / magnitude

    nyquist = 0.5 / iccs.max_delta.total_seconds()
    _freq_eps = nyquist * 1e-4  # open-bound approximation matching bandpass constraints
    _log_min = np.log(_freq_eps)
    # Use a slightly tighter upper bound near Nyquist to keep inter-slider
    # constraints safely within the global slider range.
    _log_max = np.log(nyquist - 2 * _freq_eps)

    fig, ax = plt.subplots(figsize=(10, 7.7))

    _debounce_timer: list[TimerBase | None] = [None]
    _updating: list[bool] = [False]
    update_fn: Callable[[], None]

    def _current_causal() -> bool:
        return radio.value_selected == "Causal"

    bottom_margin = 0.38 if use_matrix_image else 0.35

    if use_matrix_image:
        axes_image, _ = _draw_matrix_image_initial(
            ax, iccs, context, all_seismograms, False
        )
        fig.subplots_adjust(
            bottom=bottom_margin, left=_left_margin(True), right=0.95, top=0.93
        )
        _matrix_cache: OrderedDict[tuple[bool, float, float, bool, int], np.ndarray] = (
            OrderedDict()
        )

        def _update_matrix() -> None:
            apply = check.get_status()[0]
            fmin = max(_quantise_freq(float(np.exp(slider_fmin.val))), _freq_eps)
            fmax = min(
                _quantise_freq(float(np.exp(slider_fmax.val))), nyquist - _freq_eps
            )
            if fmin >= fmax:
                return
            causal = _current_causal()
            corners = int(slider_corners.val)
            key = (apply, fmin, fmax, causal, corners)
            if key not in _matrix_cache:
                if len(_matrix_cache) >= _BANDPASS_CACHE_SIZE:
                    _matrix_cache.popitem(last=False)
                if not _apply_bandpass_params(iccs, apply, fmin, fmax, corners):
                    return
                if context:
                    seismograms = (
                        iccs.context_seismograms_causal
                        if causal
                        else iccs.context_seismograms
                    )
                else:
                    seismograms = (
                        iccs.cc_seismograms_causal if causal else iccs.cc_seismograms
                    )
                mask = _make_mask(iccs, all_seismograms)
                ccs = np.abs(np.compress(mask, iccs.ccs))
                matrix = np.array([s.data for s, m in zip(seismograms, mask) if m])
                _matrix_cache[key] = matrix[np.argsort(ccs)[::-1]]
            else:
                _matrix_cache.move_to_end(key)
            axes_image.set_data(_matrix_cache[key])
            ax.set_xlabel(f"Time relative to pick [s]{_variant_suffix(apply, causal)}")
            ax.set_ylabel(
                "Seismograms sorted by correlation coefficient"
                f"{_variant_suffix(apply, False)}"
            )
            fig.canvas.draw_idle()

        update_fn = _update_matrix
    else:
        seis_lines, stack_line, scalar_mappable, colorbar = _draw_stack_initial(
            ax, iccs, context, all_seismograms, False
        )
        fig.subplots_adjust(
            bottom=bottom_margin, left=_left_margin(False), right=0.95, top=0.93
        )
        _stack_cache: OrderedDict[
            tuple[bool, float, float, bool, int],
            tuple[list[np.ndarray], np.ndarray, np.ndarray],
        ] = OrderedDict()

        def _update_stack() -> None:
            apply = check.get_status()[0]
            fmin = max(_quantise_freq(float(np.exp(slider_fmin.val))), _freq_eps)
            fmax = min(
                _quantise_freq(float(np.exp(slider_fmax.val))), nyquist - _freq_eps
            )
            if fmin >= fmax:
                return
            causal = _current_causal()
            corners = int(slider_corners.val)
            key = (apply, fmin, fmax, causal, corners)
            if key not in _stack_cache:
                if len(_stack_cache) >= _BANDPASS_CACHE_SIZE:
                    _stack_cache.popitem(last=False)
                if not _apply_bandpass_params(iccs, apply, fmin, fmax, corners):
                    return
                if context:
                    seismograms = (
                        iccs.context_seismograms_causal
                        if causal
                        else iccs.context_seismograms
                    )
                    stack = iccs.context_stack_causal if causal else iccs.context_stack
                else:
                    seismograms = (
                        iccs.cc_seismograms_causal if causal else iccs.cc_seismograms
                    )
                    stack = iccs.stack_causal if causal else iccs.stack
                mask = _make_mask(iccs, all_seismograms)
                ccs = np.abs(np.compress(mask, iccs.ccs))
                seis_data = [s.data.copy() for s, m in zip(seismograms, mask) if m]
                _stack_cache[key] = (seis_data, stack.data.copy(), ccs)
            else:
                _stack_cache.move_to_end(key)
            seis_data, stack_data, ccs = _stack_cache[key]
            new_norm = PowerNorm(vmin=np.min(ccs), vmax=np.max(ccs), gamma=2)
            new_colors = IccsDefaults.stack_cmap(new_norm(ccs))
            for line, data, color in zip(seis_lines, seis_data, new_colors):
                line.set_ydata(data)
                line.set_color(color)
            stack_line.set_ydata(stack_data)
            scalar_mappable.set_norm(new_norm)
            colorbar.update_normal(scalar_mappable)
            colorbar.set_label(
                f"|Correlation coefficient|{_variant_suffix(apply, False)}"
            )
            ax.set_ylabel(f"Normalised amplitude{_variant_suffix(apply, causal)}")
            fig.canvas.draw_idle()

        update_fn = _update_stack

    ax.set_title("Update bandpass filter parameters.")

    gs_widgets = _widget_gridspec(
        fig, height_ratios=[1, 1, 1, 1.8], top=bottom_margin - 0.07, bottom=0.02
    )
    ax_fmin = fig.add_subplot(gs_widgets[0, 1:11])
    ax_fmax = fig.add_subplot(gs_widgets[1, 1:11])
    ax_corners = fig.add_subplot(gs_widgets[2, 1:11])
    ax_check = fig.add_subplot(gs_widgets[3, 1:4])
    ax_radio = fig.add_subplot(gs_widgets[3, 4:8])
    ax_save = fig.add_subplot(gs_widgets[3, 8:10])
    ax_cancel = fig.add_subplot(gs_widgets[3, 10:12])

    # Align left edges with the main plot's y-axis, not the grid's column 1.
    check_pos = ax_check.get_position()
    shift = _left_margin(use_matrix_image) - check_pos.x0
    ax_check.set_position(check_pos.translated(shift, 0))
    radio_pos = ax_radio.get_position()
    ax_radio.set_position(radio_pos.translated(shift, 0))

    _fg = plt.rcParams.get("text.color", "black")
    ax_check.set_frame_on(True)
    for spine in ax_check.spines.values():
        spine.set_edgecolor(_fg)
    check = CheckButtons(
        ax_check,
        ["Apply bandpass"],
        [iccs.bandpass_apply],
        label_props={"color": [_fg], "fontsize": [11]},
        frame_props={"edgecolor": _fg, "s": 200},
        check_props={"color": _fg, "s": 200},
    )
    ax_radio.set_frame_on(True)
    ax_radio.set_xticks([])
    ax_radio.set_yticks([])
    for spine in ax_radio.spines.values():
        spine.set_edgecolor(_fg)
    ax_radio.text(
        0.5,
        0.85,
        "Preview as (not saved)",
        ha="center",
        va="top",
        fontsize=9,
        color=_fg,
        transform=ax_radio.transAxes,
    )
    # Give RadioButtons its own axes covering only the box's lower portion,
    # so it doesn't centre itself over the label above. fig.add_axes, not
    # ax_radio.inset_axes: an inset axes' locator recomputes its position
    # from the parent on every redraw, discarding set_position() below.
    radio_box_pos = ax_radio.get_position()
    ax_radio_toggles = fig.add_axes(
        (
            radio_box_pos.x0,
            radio_box_pos.y0,
            radio_box_pos.width,
            radio_box_pos.height * 0.55,
        )
    )
    ax_radio_toggles.set_frame_on(False)
    radio = RadioButtons(
        ax_radio_toggles,
        ["Zero-phase", "Causal"],
        active=0,
        layout="horizontal",
        label_props={"color": [_fg, _fg], "fontsize": [11, 11]},
    )
    # RadioButtons has no option to centre the group — measure how much
    # width it actually used, once rendered, and shift to centre it.
    fig.canvas.draw()
    content_right = radio.labels[-1].get_window_extent().x1
    toggles_bbox = ax_radio_toggles.get_window_extent()
    content_fraction = (content_right - toggles_bbox.x0) / toggles_bbox.width
    toggles_pos = ax_radio_toggles.get_position()
    offset = (1 - content_fraction) / 2 * toggles_pos.width
    ax_radio_toggles.set_position(toggles_pos.translated(offset, 0))
    slider_fmin = Slider(
        ax_fmin, "fmin [Hz]", _log_min, _log_max, valinit=np.log(iccs.bandpass_fmin)
    )
    slider_fmax = Slider(
        ax_fmax, "fmax [Hz]", _log_min, _log_max, valinit=np.log(iccs.bandpass_fmax)
    )
    slider_corners = Slider(
        ax_corners,
        "corners",
        valmin=1,
        valmax=max(8, iccs.corners),
        valinit=iccs.corners,
        valstep=1,
    )
    # Show Hz values rather than the internal log values
    slider_fmin.valtext.set_text(f"{iccs.bandpass_fmin:.3f}")
    slider_fmax.valtext.set_text(f"{iccs.bandpass_fmax:.3f}")
    # Set initial inter-slider bounds
    slider_fmax.valmin = np.log(iccs.bandpass_fmin + _freq_eps)
    slider_fmin.valmax = np.log(iccs.bandpass_fmax - _freq_eps)

    if not iccs.bandpass_apply:
        slider_fmin.set_active(False)
        slider_fmax.set_active(False)
        slider_corners.set_active(False)

    def _schedule_update() -> None:
        if _debounce_timer[0] is not None:
            _debounce_timer[0].stop()
        timer = fig.canvas.new_timer(interval=150)
        timer.single_shot = True
        timer.add_callback(update_fn)
        _debounce_timer[0] = timer
        timer.start()

    def _on_fmin_change(_: float) -> None:
        if _updating[0]:
            return
        _updating[0] = True
        fmin = float(np.exp(slider_fmin.val))
        slider_fmax.valmin = np.log(fmin + _freq_eps)
        if float(slider_fmax.val) <= np.log(fmin + _freq_eps):
            slider_fmax.set_val(np.log(fmin + _freq_eps))
        _updating[0] = False
        slider_fmin.valtext.set_text(f"{fmin:.3f}")
        slider_fmax.valtext.set_text(f"{np.exp(slider_fmax.val):.3f}")
        _schedule_update()

    def _on_fmax_change(_: float) -> None:
        if _updating[0]:
            return
        _updating[0] = True
        fmax = float(np.exp(slider_fmax.val))
        slider_fmin.valmax = np.log(fmax - _freq_eps)
        if float(slider_fmin.val) >= np.log(fmax - _freq_eps):
            slider_fmin.set_val(np.log(fmax - _freq_eps))
        _updating[0] = False
        slider_fmin.valtext.set_text(f"{np.exp(slider_fmin.val):.3f}")
        slider_fmax.valtext.set_text(f"{fmax:.3f}")
        _schedule_update()

    def _on_corners_change(_: float) -> None:
        _schedule_update()

    def _on_check(_label: str | None) -> None:
        apply = check.get_status()[0]
        slider_fmin.set_active(apply)
        slider_fmax.set_active(apply)
        slider_corners.set_active(apply)
        _schedule_update()

    def _on_radio_change(_label: str | None) -> None:
        _schedule_update()

    slider_fmin.on_changed(_on_fmin_change)
    slider_fmax.on_changed(_on_fmax_change)
    slider_corners.on_changed(_on_corners_change)
    check.on_clicked(_on_check)
    radio.on_clicked(_on_radio_change)

    def on_save(_: Event) -> None:
        if _debounce_timer[0] is not None:
            _debounce_timer[0].stop()
        fmin = float(np.exp(slider_fmin.val))
        fmax = float(np.exp(slider_fmax.val))
        corners = int(slider_corners.val)
        apply = check.get_status()[0]
        if not _apply_bandpass_params(iccs, apply, fmin, fmax, corners):
            return
        if not return_fig:
            plt.close(fig)

    def on_cancel(_: Event) -> None:
        if _debounce_timer[0] is not None:
            _debounce_timer[0].stop()
        iccs.bandpass_apply = _orig_apply
        iccs.bandpass_fmin = _orig_fmin
        iccs.bandpass_fmax = _orig_fmax
        iccs.corners = _orig_corners
        if not return_fig:
            plt.close(fig)

    b_save, b_cancel = _add_save_cancel_buttons(ax_save, ax_cancel, on_save, on_cancel)

    if return_fig:
        return (
            fig,
            ax,
            (check, slider_fmin, slider_fmax, slider_corners, radio, b_save, b_cancel),
        )
    plt.show()
    return None

update_min_cc

update_min_cc(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    causal: bool = False,
    return_fig: bool = True,
) -> (
    tuple[
        Figure,
        Axes,
        tuple[
            Cursor,
            Line2D,
            Button,
            Button,
            _ScrollIndexTracker,
        ],
    ]
    | None
)

Interactively pick a new min_cc.

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

Parameters:

Name Type Description Default
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

True
all_seismograms bool

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

False
causal bool

Determines the filter phase behaviour: - True: causally-filtered (single-pass) seismograms are used. - False: zero-phase filtered seismograms are used. This is the default for this function — it sorts/thresholds by correlation coefficient, and zero-phase is what's actually used for the correlation being thresholded, so it's the more representative view.

False
return_fig bool

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

True

Returns:

Type Description
tuple[Figure, Axes, tuple[Cursor, Line2D, Button, Button, _ScrollIndexTracker]] | None

Figure with the selector widgets if return_fig is True.

Examples:

>>> from pysmo.tools.iccs import ICCS, update_min_cc
>>> iccs = ICCS(iccs_seismograms)
>>> _ = iccs()
>>> fig, ax, widgets = update_min_cc(iccs)
>>> # fig.show() # or integrate into your own application
>>>

Picking a new min_cc in matrix image Picking a new min_cc in matrix image

Source code in src/pysmo/tools/iccs/plot.py
def update_min_cc(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    causal: bool = False,
    return_fig: bool = True,
) -> (
    tuple[
        Figure,
        Axes,
        tuple[Cursor, Line2D, Button, Button, _ScrollIndexTracker],
    ]
    | None
):
    """Interactively pick a new [`min_cc`][pysmo.tools.iccs.ICCS.min_cc].

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

    Args:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        causal: Determines the filter phase behaviour:
            - `True`: causally-filtered (single-pass) seismograms are used.
            - `False`: zero-phase filtered seismograms are used. This is the
              default for this function — it sorts/thresholds by correlation
              coefficient, and zero-phase is what's actually used for the
              correlation being thresholded, so it's the more representative
              view.
        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 widgets if `return_fig` is `True`.

    Examples:
        ```python
        >>> from pysmo.tools.iccs import ICCS, update_min_cc
        >>> iccs = ICCS(iccs_seismograms)
        >>> _ = iccs()
        >>> fig, ax, widgets = update_min_cc(iccs)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

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

        ![Picking a new min_cc in matrix image](../../../images/sybil/iccs_update_min_cc.png#only-light){ loading=lazy }
        ![Picking a new min_cc in matrix image](../../../images/sybil/iccs_update_min_cc-dark.png#only-dark){ loading=lazy }
    """
    fig, ax = plt.subplots(figsize=(10, 6))
    matrix = draw_common_matrix_image(ax, iccs, context, all_seismograms, causal)
    fig.subplots_adjust(bottom=0.2, left=0.05, right=0.95, top=0.93)

    gs_widgets = _widget_gridspec(fig, height_ratios=[1], top=0.125, bottom=0.05)
    ax_save = fig.add_subplot(gs_widgets[0, 8:10])
    ax_cancel = fig.add_subplot(gs_widgets[0, 10:12])

    ax.set_title("Pick a new minimal cross-correlation coefficient.")
    pending_val = [iccs.min_cc]

    def handle_valid_pick(new_val: float) -> None:
        pending_val[0] = new_val
        ax.set_title(f"Click save to set min_cc to {new_val:.4f}")

    current_ccs = sorted(
        i for i, s in zip(iccs.ccs, iccs.seismograms) if s.select or all_seismograms
    )
    start_index = int(np.searchsorted(current_ccs, iccs.min_cc))
    max_index = len(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_cc(line: Line2D) -> float:
        index = round(line.get_ydata()[0], 0)  # type: ignore
        if index == 0:
            return IccsDefaults.index_zero_multiplier * current_ccs[0]
        return float(np.mean(current_ccs[index - 1 : index + 1]))

    def onclick(event: Event) -> None:
        if not isinstance(event, MouseEvent):
            return
        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)
            handle_valid_pick(calc_cc(pick_line))
            if ax.figure:
                ax.figure.canvas.draw_idle()

    def on_mouse_move(event: Event) -> None:
        if not isinstance(event, MouseEvent):
            return
        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)
        else:
            pick_line_cursor.set_visible(False)
        if ax.figure:
            ax.figure.canvas.draw_idle()

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

    if isinstance(ax.figure, Figure):
        tracker = _ScrollIndexTracker(ax, ax.figure)
        ax.figure.canvas.mpl_connect("scroll_event", tracker.on_scroll)
        ax.figure.canvas.mpl_connect("button_press_event", onclick)
        ax.figure.canvas.mpl_connect("motion_notify_event", on_mouse_move)
    else:
        tracker = _ScrollIndexTracker(ax, Figure())

    def on_save(_: Event) -> None:
        iccs.min_cc = pending_val[0]
        if not return_fig:
            plt.close(fig)

    def on_cancel(_: Event) -> None:
        if not return_fig:
            plt.close(fig)

    b_save, b_cancel = _add_save_cancel_buttons(ax_save, ax_cancel, on_save, on_cancel)

    if return_fig:
        return fig, ax, (cursor, pick_line, b_save, b_cancel, tracker)
    plt.show()
    return None

update_pick

update_pick(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    use_matrix_image: bool = False,
    causal: bool = True,
    return_fig: bool = True,
) -> (
    tuple[
        Figure, Axes, tuple[Cursor, Line2D, Button, Button]
    ]
    | 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:

Name Type Description Default
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

True
all_seismograms bool

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

False
use_matrix_image bool

Use the matrix image instead of the stack.

False
causal bool

Determines the filter phase behaviour: - True: causally-filtered (single-pass) seismograms are used. This is the default for this function, since locating a phase onset by eye is exactly what zero-phase filtering's acausal precursor smearing distorts. - False: zero-phase filtered seismograms are used.

True
return_fig bool

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

True

Returns:

Type Description
tuple[Figure, Axes, tuple[Cursor, Line2D, Button, Button]] | 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)
>>>
>>> fig, ax, widgets = update_pick(iccs)
>>> # fig.show() # or integrate into your own application
>>>

Picking a new T1 Picking a new T1

Source code in src/pysmo/tools/iccs/plot.py
def update_pick(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    use_matrix_image: bool = False,
    causal: bool = True,
    return_fig: bool = True,
) -> tuple[Figure, Axes, tuple[Cursor, Line2D, Button, Button]] | 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.

    Args:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        use_matrix_image: Use the
            [matrix image][pysmo.tools.iccs.plot_matrix_image]
            instead of the [stack][pysmo.tools.iccs.plot_stack].
        causal: Determines the filter phase behaviour:
            - `True`: causally-filtered (single-pass) seismograms are used.
              This is the default for this function, since locating a phase
              onset by eye is exactly what zero-phase filtering's acausal
              precursor smearing distorts.
            - `False`: zero-phase filtered seismograms are used.
        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)
        >>>
        >>> fig, ax, widgets = update_pick(iccs)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

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

        ![Picking a new T1](../../../images/sybil/iccs_update_pick.png#only-light){ loading=lazy }
        ![Picking a new T1](../../../images/sybil/iccs_update_pick-dark.png#only-dark){ loading=lazy }
    """
    fig, ax = plt.subplots(figsize=(10, 6))
    _draw_stack_or_matrix(ax, iccs, context, all_seismograms, causal, use_matrix_image)
    fig.subplots_adjust(
        bottom=0.2, left=_left_margin(use_matrix_image), right=0.95, top=0.93
    )

    gs_widgets = _widget_gridspec(fig, height_ratios=[1], top=0.125, bottom=0.05)
    ax_save = fig.add_subplot(gs_widgets[0, 8:10])
    ax_cancel = fig.add_subplot(gs_widgets[0, 10:12])

    ax.set_title("Update t1 for all seismograms.")
    pending_pick = [0.0]

    def handle_valid_pick(xdata: float) -> None:
        pending_pick[0] = xdata
        ax.set_title(f"Click save to adjust t1 by {xdata:.3f} seconds.")

    pick_line = ax.axvline(0, color="g", linewidth=2)
    cursor = Cursor(
        ax, useblit=True, color="g", linewidth=2, horizOn=False, linestyle="--"
    )

    def onclick(event: Event) -> None:
        if not isinstance(event, MouseEvent):
            return
        if (
            event.inaxes is ax
            and event.xdata is not None
            and iccs.validate_pick(pd.Timedelta(seconds=event.xdata))
        ):
            pick_line.set_xdata(np.array((event.xdata, event.xdata)))
            handle_valid_pick(event.xdata)
            if ax.figure:
                ax.figure.canvas.draw()
                ax.figure.canvas.flush_events()

    def on_mouse_move(event: Event) -> None:
        if not isinstance(event, MouseEvent):
            return
        if event.inaxes == ax and event.xdata is not None:
            is_valid = iccs.validate_pick(pd.Timedelta(seconds=event.xdata))
            cursor.linev.set_color("g" if is_valid else "r")

    if isinstance(ax.figure, Figure):
        ax.figure.canvas.mpl_connect("button_press_event", onclick)
        ax.figure.canvas.mpl_connect("motion_notify_event", on_mouse_move)

    def on_save(_: Event) -> None:
        iccs.update_all_picks(pd.Timedelta(seconds=pending_pick[0]))
        if not return_fig:
            plt.close(fig)

    def on_cancel(_: Event) -> None:
        if not return_fig:
            plt.close(fig)

    b_save, b_cancel = _add_save_cancel_buttons(ax_save, ax_cancel, on_save, on_cancel)

    if return_fig:
        return fig, ax, (cursor, pick_line, b_save, b_cancel)
    plt.show()
    return None

update_timewindow

update_timewindow(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    use_matrix_image: bool = False,
    causal: bool = False,
    return_fig: bool = True,
) -> (
    tuple[Figure, Axes, tuple[SpanSelector, Button, Button]]
    | None
)

Pick new time window limits.

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

Parameters:

Name Type Description Default
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

True
all_seismograms bool

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

False
use_matrix_image bool

Use the matrix image instead of the stack.

False
causal bool

Determines the filter phase behaviour: - True: causally-filtered (single-pass) seismograms are used. - False: zero-phase filtered seismograms are used. This is the default for this function: window_pre/window_post crop cc_seismograms (zero-phase), which is what cross-correlation, stacking, and MCCC actually run on regardless of what's displayed here. Picking the window against the causal display risks misjudging the pre-arrival margin: the causal view's clean quiet period before the onset doesn't reflect that the zero-phase data being cropped may already carry acausal precursor energy inside that same interval.

False
return_fig bool

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

True

Returns:

Type Description
tuple[Figure, Axes, tuple[SpanSelector, Button, Button]] | None

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

Window is clamped around the pick

The new time window may not be chosen such that the pick lies outside 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)
>>>
>>> fig, ax, widgets = update_timewindow(iccs)
>>> # fig.show() # or integrate into your own application
>>>

Picking a new time window Picking a new time window

Source code in src/pysmo/tools/iccs/plot.py
def update_timewindow(
    iccs: ICCS,
    context: bool = True,
    all_seismograms: bool = False,
    use_matrix_image: bool = False,
    causal: bool = False,
    return_fig: bool = True,
) -> tuple[Figure, Axes, tuple[SpanSelector, Button, Button]] | 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].

    Args:
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        use_matrix_image: Use the
            [matrix image][pysmo.tools.iccs.plot_matrix_image]
            instead of the [stack][pysmo.tools.iccs.plot_stack].
        causal: Determines the filter phase behaviour:
            - `True`: causally-filtered (single-pass) seismograms are used.
            - `False`: zero-phase filtered seismograms are used. This is
              the default for this function: `window_pre`/`window_post`
              crop [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms]
              (zero-phase), which is what cross-correlation, stacking, and
              MCCC actually run on regardless of what's displayed here.
              Picking the window against the causal display risks
              misjudging the pre-arrival margin: the causal view's clean
              quiet period before the onset doesn't reflect that the
              zero-phase data being cropped may already carry acausal
              precursor energy inside that same interval.
        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: Window is clamped around the pick
        The new time window may not be chosen such that the pick lies
        outside 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)
        >>>
        >>> fig, ax, widgets = update_timewindow(iccs)
        >>> # fig.show() # or integrate into your own application
        >>>
        ```

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

        ![Picking a new time window](../../../images/sybil/iccs_update_timewindow.png#only-light){ loading=lazy }
        ![Picking a new time window](../../../images/sybil/iccs_update_timewindow-dark.png#only-dark){ loading=lazy }
    """
    fig, ax = plt.subplots(figsize=(10, 6))
    _draw_stack_or_matrix(ax, iccs, context, all_seismograms, causal, use_matrix_image)
    fig.subplots_adjust(
        bottom=0.2, left=_left_margin(use_matrix_image), right=0.95, top=0.93
    )

    gs_widgets = _widget_gridspec(fig, height_ratios=[1], top=0.125, bottom=0.05)
    ax_save = fig.add_subplot(gs_widgets[0, 8:10])
    ax_cancel = fig.add_subplot(gs_widgets[0, 10:12])

    ax.set_title("Pick a new time window.")
    pending_window = [iccs.window_pre.total_seconds(), iccs.window_post.total_seconds()]

    def handle_valid_selection(xmin: float, xmax: float) -> None:
        pending_window[0], pending_window[1] = xmin, xmax
        ax.set_title(f"Click save to set window at {xmin:.3f} to {xmax:.3f} seconds.")

    old_extents = (iccs.window_pre.total_seconds(), iccs.window_post.total_seconds())
    default_title_color = ax.title.get_color()

    def onselect(xmin: float, xmax: float) -> None:
        nonlocal old_extents
        if iccs.validate_time_window(
            pd.Timedelta(seconds=xmin), pd.Timedelta(seconds=xmax)
        ):
            old_extents = xmin, xmax
            ax.title.set_color(default_title_color)
            if ax.figure:
                ax.figure.canvas.draw_idle()
            handle_valid_selection(xmin, xmax)
        else:
            span.extents = old_extents
            ax.set_title("Invalid window choice.", color="red")
            if ax.figure:
                ax.figure.canvas.draw_idle()

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

    def on_save(_: Event) -> None:
        iccs.window_pre = pd.Timedelta(seconds=pending_window[0])
        iccs.window_post = pd.Timedelta(seconds=pending_window[1])
        if not return_fig:
            plt.close(fig)

    def on_cancel(_: Event) -> None:
        if not return_fig:
            plt.close(fig)

    b_save, b_cancel = _add_save_cancel_buttons(ax_save, ax_cancel, on_save, on_cancel)

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

pysmo.tools.iccs.plot

Extra plotting functions for the ICCS module.

Most of this module's contents are internal helpers for the higher-level plotting functions (e.g. plot_stack, update_pick) exposed from the pysmo.tools.iccs namespace, and are not meant to be used directly. The exceptions are draw_common_stack and draw_common_matrix_image, lower-level drawing primitives exposed for users who wish to customise their own plotting workflows.

Functions:

Name Description
draw_common_matrix_image

Return a basic matrix image plot for use in other plots.

draw_common_stack

Return a basic stack plot for use in other plots.

draw_common_matrix_image

draw_common_matrix_image(
    ax: Axes,
    iccs: ICCS,
    context: bool,
    all_seismograms: bool,
    causal: bool = False,
) -> ndarray

Return a basic matrix image plot for use in other plots.

Parameters:

Name Type Description Default
ax Axes

Axes to plot on.

required
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

required
all_seismograms bool

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

required
causal bool

Determines the filter phase behaviour: - True: causally-filtered (single-pass) seismograms are used. - False: zero-phase filtered seismograms are used.

False

Returns:

Type Description
ndarray

Sorted seismogram matrix used for the plot.

Source code in src/pysmo/tools/iccs/plot.py
def draw_common_matrix_image(
    ax: Axes, iccs: ICCS, context: bool, all_seismograms: bool, causal: bool = False
) -> np.ndarray:
    """Return a basic matrix image plot for use in other plots.

    Args:
        ax: Axes to plot on.
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        causal: Determines the filter phase behaviour:
            - `True`: causally-filtered (single-pass) seismograms are used.
            - `False`: zero-phase filtered seismograms are used.

    Returns:
        Sorted seismogram matrix used for the plot.
    """
    _, seismogram_matrix = _draw_matrix_image_initial(
        ax, iccs, context, all_seismograms, causal
    )
    return seismogram_matrix

draw_common_stack

draw_common_stack(
    ax: Axes,
    iccs: ICCS,
    context: bool,
    all_seismograms: bool,
    causal: bool = False,
) -> None

Return a basic stack plot for use in other plots.

Parameters:

Name Type Description Default
ax Axes

Axes to plot on.

required
iccs ICCS

Instance of the ICCS class.

required
context bool

Determines which seismograms are used: - True: context_seismograms are used. - False: cc_seismograms are used.

required
all_seismograms bool

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

required
causal bool

Determines the filter phase behaviour: - True: causally-filtered (single-pass) seismograms are used. - False: zero-phase filtered seismograms are used.

False
Source code in src/pysmo/tools/iccs/plot.py
def draw_common_stack(
    ax: Axes, iccs: ICCS, context: bool, all_seismograms: bool, causal: bool = False
) -> None:
    """Return a basic stack plot for use in other plots.

    Args:
        ax: Axes to plot on.
        iccs: Instance of the [`ICCS`][pysmo.tools.iccs.ICCS] class.
        context: Determines which seismograms are used:
            - `True`: [`context_seismograms`][pysmo.tools.iccs.ICCS.context_seismograms] are used.
            - `False`: [`cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms] are used.
        all_seismograms: If `True`, all seismograms are shown in the plot instead of the
            selected ones only.
        causal: Determines the filter phase behaviour:
            - `True`: causally-filtered (single-pass) seismograms are used.
            - `False`: zero-phase filtered seismograms are used.
    """
    _draw_stack_initial(ax, iccs, context, all_seismograms, causal)