Skip to content

pysmo.functions

Building-block functions for pysmo types.

The pysmo.functions module provides low-level functions that perform common operations on pysmo types. They are intended as building blocks for constructing more complex processing workflows.

Many functions accept a replace argument. Without it they modify the seismogram in place and return None; with replace=True they leave the input untouched and return a new seismogram. For example:

>>> from pysmo.functions import resample
>>> from pysmo.classes import MSeed
>>> seis = MSeed.from_file("example.mseed")
>>> new_delta = seis.delta * 2
>>>
>>> # return a new seismogram, leaving seis untouched:
>>> new_seis = resample(seis, new_delta, replace=True)
>>>
>>> # modify data in seis directly:
>>> resample(seis, new_delta)
>>>

The new object is built from the input with copy.replace, substituting only the freshly computed data (and, where the operation moves or resamples the time axis, the corresponding begin_time or delta). Every other attribute is carried straight over from the input.

Attributes outside the Seismogram protocol

A concrete class often carries more than begin_time, delta and data: identity, provenance or acquisition metadata. replace=True keeps those values as they were, even where the operation has made them a poor description of the new data, carrying each straight over by reference. If such an attribute is itself mutable, the input and the returned seismogram share the same object, so mutating it through one is visible through the other.

Not every concrete type supports replace=True: rebuilding the object this way needs the substituted attributes to be constructor parameters. MiniSeismogram, MSeed and other value objects qualify; SacSeismogram does not, because its data is a property reading and writing through to the underlying SacIO instance rather than a stored field of its own.

>>> from pysmo.functions import clone_to_mini, detrend
>>> from pysmo import MiniSeismogram
>>> from pysmo.classes import SAC
>>> sac = SAC.from_file("example.sac")
>>>
>>> # replace=True cannot rebuild a SacSeismogram:
>>> detrend(sac.seismogram, replace=True)
Traceback (most recent call last):
...
TypeError: ...
>>>
>>> # convert to a value object first (or copy the whole SAC object):
>>> detrended = detrend(clone_to_mini(MiniSeismogram, sac.seismogram), replace=True)
>>> type(detrended).__name__
'MiniSeismogram'
>>>
Needless copy

Reassigning the result back to the same name (seis = resample(seis, new_delta, replace=True)) ends up equivalent to modifying seis in place, but pays for a copy to get there. Call resample(seis, new_delta) directly instead.

Three helpers work with a seismogram as JSON: seismogram_to_json encodes a value-object seismogram as a portable JSON document and seismogram_from_json reconstructs it, while seismogram_checksum fingerprints one for change detection.

More functions live in pysmo.tools

Additional functions may be found in pysmo.tools.

Functions:

Name Description
clone_to_mini

Create a Mini class instance from a compatible object.

copy_from_mini

Copy attributes from a Mini instance onto a compatible object.

crop

Shorten a seismogram to new begin and end times.

detrend

Remove linear and/or constant trends from a seismogram.

estimate_delta

Estimate a canonical sampling interval from a set of near-equal deltas.

merge

Merge contiguous seismograms into a single seismogram.

normalize

Normalise a seismogram with its absolute max value.

pad

Pad seismogram data.

resample

Resample Seismogram data using the Fourier method.

seismogram_checksum

Return a stable digest of a seismogram's samples and timing.

seismogram_from_json

Reconstruct a seismogram from a seismogram_to_json document.

seismogram_to_json

Encode a value-object seismogram as a portable JSON document.

taper

Apply a symmetric taper to the ends of a Seismogram.

time2index

Convert a timestamp to the corresponding data-array index.

window

Return an optionally padded and tapered window of a seismogram.

clone_to_mini

clone_to_mini(
    mini_cls: type[TMini],
    source: _AnyProto,
    update: dict[str, Any] | None = None,
) -> TMini

Create a Mini class instance from a compatible object.

Clones source by copying the attributes mini_cls defines from it onto a new mini_cls instance. Attributes present only on source are ignored, so the result can be smaller and faster to work with.

If the source instance is missing an attribute for which a default is defined in the target class, then that default value for that attribute is used.

Parameters:

Name Type Description Default
mini_cls type[TMini]

The type of Mini class to create.

required
source _AnyProto

The instance to clone (must contain all attributes present in mini_cls).

required
update dict[str, Any] | None

Update or add attributes in the returned mini_cls instance.

None

Returns:

Type Description
TMini

A new mini_cls instance.

Raises:

Type Description
AttributeError

If the source instance does not contain all attributes in mini_cls (unless they are provided with the update keyword argument).

Examples:

Create a MiniSeismogram from a SacSeismogram instance with a new begin_time.

>>> from pysmo.functions import clone_to_mini
>>> from pysmo import MiniSeismogram
>>> from pysmo.classes import SAC
>>> import pandas as pd
>>> from datetime import timezone
>>> now = pd.Timestamp.now(timezone.utc)
>>> sac_seismogram = SAC.from_file("example.sac").seismogram
>>> mini_seismogram = clone_to_mini(MiniSeismogram, sac_seismogram, update={"begin_time": now})
>>> all(sac_seismogram.data == mini_seismogram.data)
True
>>> mini_seismogram.begin_time == now
True
>>>
See Also

copy_from_mini: Copy attributes from a Mini instance onto a compatible object.

Source code in src/pysmo/functions/_utils.py
def clone_to_mini[TMini: _AnyMini](
    mini_cls: type[TMini], source: "_AnyProto", update: dict[str, Any] | None = None
) -> TMini:
    """Create a Mini class instance from a compatible object.

    Clones `source` by [copying][copy.copy] the attributes `mini_cls` defines
    from it onto a new `mini_cls` instance. Attributes present only on
    `source` are ignored, so the result can be smaller and faster to work
    with.

    If the source instance is missing an attribute for which a default is
    defined in the target class, then that default value for that attribute is
    used.

    Args:
        mini_cls: The type of Mini class to create.
        source: The instance to clone (must contain all attributes present
            in `mini_cls`).
        update: Update or add attributes in the returned `mini_cls` instance.

    Returns:
        A new `mini_cls` instance.

    Raises:
        AttributeError: If the `source` instance does not contain all
            attributes in `mini_cls` (unless they are provided with the
            `update` keyword argument).

    Examples:
        Create a [`MiniSeismogram`][pysmo.MiniSeismogram] from a
        [`SacSeismogram`][pysmo.classes.SacSeismogram] instance with
        a new `begin_time`.

        ```python
        >>> from pysmo.functions import clone_to_mini
        >>> from pysmo import MiniSeismogram
        >>> from pysmo.classes import SAC
        >>> import pandas as pd
        >>> from datetime import timezone
        >>> now = pd.Timestamp.now(timezone.utc)
        >>> sac_seismogram = SAC.from_file("example.sac").seismogram
        >>> mini_seismogram = clone_to_mini(MiniSeismogram, sac_seismogram, update={"begin_time": now})
        >>> all(sac_seismogram.data == mini_seismogram.data)
        True
        >>> mini_seismogram.begin_time == now
        True
        >>>
        ```

    Tip: See Also
        [`copy_from_mini`][pysmo.functions.copy_from_mini]: Copy attributes
        from a Mini instance onto a compatible object.
    """

    update = update or {}

    if not all(
        (hasattr(source, x.name) or x.name in update or x.default is not NOTHING)
        for x in fields(mini_cls)
    ):
        raise AttributeError(
            f"Unable to create clone: {source} not compatible with {mini_cls}."
        )

    # Omit a field the source lacks so mini_cls applies its own default:
    # passing attr.default would hand a Factory object to the constructor.
    clone_dict: dict[str, Any] = {}
    for attr in fields(mini_cls):
        if attr.name in update:
            clone_dict[attr.name] = update[attr.name]
        elif hasattr(source, attr.name):
            clone_dict[attr.name] = copy(getattr(source, attr.name))
    return mini_cls(**clone_dict)

copy_from_mini

copy_from_mini(
    source: _AnyMini,
    target: _AnyProto,
    update: dict[str, Any] | None = None,
) -> None

Copy attributes from a Mini instance onto a compatible object.

Copies every attribute of the source Mini instance onto a compatible target instance.

Parameters:

Name Type Description Default
source _AnyMini

The Mini instance to copy attributes from.

required
target _AnyProto

Compatible target instance.

required
update dict[str, Any] | None

Update or add attributes in the target instance.

None

Raises:

Type Description
AttributeError

If the target instance does not contain all attributes in the source instance (unless they are provided with the update keyword argument).

See Also

clone_to_mini: Create a Mini instance from a compatible object.

Source code in src/pysmo/functions/_utils.py
def copy_from_mini(
    source: "_AnyMini", target: "_AnyProto", update: dict[str, Any] | None = None
) -> None:
    """Copy attributes from a Mini instance onto a compatible object.

    [Copies][copy.copy] every attribute of the `source` Mini instance onto a
    compatible `target` instance.

    Args:
        source: The Mini instance to copy attributes from.
        target: Compatible target instance.
        update: Update or add attributes in the target instance.

    Raises:
        AttributeError: If the `target` instance does not contain all
            attributes in the `source` instance (unless they are
            provided with the `update` keyword argument).

    Tip: See Also
        [`clone_to_mini`][pysmo.functions.clone_to_mini]: Create a Mini
        instance from a compatible object.
    """

    update = update or {}

    attributes = set(unstructure(source).keys())
    attributes.update(update.keys())

    if not all(hasattr(target, x) for x in attributes):
        raise AttributeError(
            f"Unable to copy to target: {type(target)} not compatible with {type(source)}."
        )

    for attribute in attributes:
        if attribute in update:
            setattr(target, attribute, update[attribute])
        else:
            setattr(target, attribute, copy(getattr(source, attribute)))

crop

crop(
    seismogram: T,
    begin_time: Timestamp,
    end_time: Timestamp,
    *,
    replace: bool = False
) -> T | None

Shorten a seismogram to new begin and end times.

This function calculates the indices corresponding to the provided new begin and end times using time2index, then slices the seismogram data array accordingly and updates the begin_time.

Parameters:

Name Type Description Default
seismogram T

Seismogram object.

required
begin_time Timestamp

New begin time.

required
end_time Timestamp

New end time.

required
replace bool

Return a new seismogram and leave the input untouched, instead of modifying it in place. Not supported by every concrete type (see pysmo.functions).

False

Returns:

Type Description
T | None

Cropped Seismogram object if called with replace=True.

Raises:

Type Description
ValueError

If new begin time is after new end time.

Examples:

>>> from pysmo.functions import crop
>>> from pysmo.classes import SAC
>>> import pandas as pd
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> new_begin_time = sac_seis.begin_time + pd.Timedelta(seconds=10)
>>> new_end_time = sac_seis.end_time - pd.Timedelta(seconds=10)
>>> crop(sac_seis, new_begin_time, new_end_time)
>>>
Source code in src/pysmo/functions/_seismogram.py
def crop[T: Seismogram](
    seismogram: T,
    begin_time: pd.Timestamp,
    end_time: pd.Timestamp,
    *,
    replace: bool = False,
) -> T | None:
    """Shorten a seismogram to new begin and end times.

    This function calculates the indices corresponding to the provided new
    begin and end times using [`time2index`][pysmo.functions.time2index], then
    slices the seismogram `data` array accordingly and updates the
    `begin_time`.

    Args:
        seismogram: [`Seismogram`][pysmo.Seismogram] object.
        begin_time: New begin time.
        end_time: New end time.
        replace: Return a new seismogram and leave the input untouched,
            instead of modifying it in place. Not supported by every
            concrete type (see [`pysmo.functions`][]).

    Returns:
        Cropped [`Seismogram`][pysmo.Seismogram] object if called with `replace=True`.

    Raises:
        ValueError: If new begin time is after new end time.

    Examples:
        ```python
        >>> from pysmo.functions import crop
        >>> from pysmo.classes import SAC
        >>> import pandas as pd
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> new_begin_time = sac_seis.begin_time + pd.Timedelta(seconds=10)
        >>> new_end_time = sac_seis.end_time - pd.Timedelta(seconds=10)
        >>> crop(sac_seis, new_begin_time, new_end_time)
        >>>
        ```
    """

    if begin_time > end_time:
        raise ValueError("New begin_time cannot be after new end_time")

    start_index = time2index(seismogram, begin_time)
    end_index = time2index(seismogram, end_time)

    new_data = seismogram.data[start_index : end_index + 1]
    new_begin_time = seismogram.begin_time + seismogram.delta * start_index

    if replace:
        return copy.replace(
            seismogram,  # type: ignore[arg-type]
            data=new_data.copy(),
            begin_time=new_begin_time,
        )

    seismogram.data = new_data
    seismogram.begin_time = new_begin_time
    return None

detrend

detrend(
    seismogram: T, *, replace: bool = False
) -> T | None

Remove linear and/or constant trends from a seismogram.

Parameters:

Name Type Description Default
seismogram T

Seismogram object.

required
replace bool

Return a new seismogram and leave the input untouched, instead of modifying it in place. Not supported by every concrete type (see pysmo.functions).

False

Returns:

Type Description
T | None

Detrended Seismogram object if called with replace=True.

Examples:

>>> import numpy as np
>>> import pytest
>>> from pysmo.functions import detrend
>>> from pysmo.classes import SAC
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> 0 == pytest.approx(np.mean(sac_seis.data), abs=1e-8)
np.False_
>>> detrend(sac_seis)
>>> 0 == pytest.approx(np.mean(sac_seis.data), abs=1e-8)
np.True_
>>>
Source code in src/pysmo/functions/_seismogram.py
def detrend[T: Seismogram](seismogram: T, *, replace: bool = False) -> T | None:
    """Remove linear and/or constant trends from a seismogram.

    Args:
        seismogram: Seismogram object.
        replace: Return a new seismogram and leave the input untouched,
            instead of modifying it in place. Not supported by every
            concrete type (see [`pysmo.functions`][]).

    Returns:
        Detrended [`Seismogram`][pysmo.Seismogram] object if called with `replace=True`.

    Examples:
        ```python
        >>> import numpy as np
        >>> import pytest
        >>> from pysmo.functions import detrend
        >>> from pysmo.classes import SAC
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> 0 == pytest.approx(np.mean(sac_seis.data), abs=1e-8)
        np.False_
        >>> detrend(sac_seis)
        >>> 0 == pytest.approx(np.mean(sac_seis.data), abs=1e-8)
        np.True_
        >>>
        ```
    """
    detrended = scipy.signal.detrend(seismogram.data)

    if replace:
        return copy.replace(seismogram, data=detrended)  # type: ignore[arg-type]

    seismogram.data = detrended
    return None

estimate_delta

estimate_delta(
    deltas: Sequence[PositiveTimedelta],
) -> PositiveTimedelta

Estimate a canonical sampling interval from a set of near-equal deltas.

Useful when several seismograms nominally share a sampling interval but report values that differ only by measurement noise (e.g. clock drift reflected in a reported sample rate) or floating-point noise. Returns the low median of deltas: an order-independent choice that is always one of the input values, rather than a synthetic average that none of the seismograms actually have.

This does not check how close the given deltas are to each other; for a set of genuinely different sampling intervals it simply returns the low median of the sorted values.

Parameters:

Name Type Description Default
deltas Sequence[PositiveTimedelta]

Sampling intervals to estimate a canonical value from.

required

Returns:

Type Description
PositiveTimedelta

The estimated canonical sampling interval.

Raises:

Type Description
ValueError

If deltas is empty.

Examples:

>>> import pandas as pd
>>> from pysmo.functions import estimate_delta
>>> deltas = [
...     pd.Timedelta(seconds=0.01),
...     pd.Timedelta(seconds=0.010000000000001),
...     pd.Timedelta(seconds=0.01),
... ]
>>> estimate_delta(deltas)
Timedelta('0 days 00:00:00.010000')
>>>
Source code in src/pysmo/functions/_seismogram.py
def estimate_delta(deltas: Sequence[PositiveTimedelta]) -> PositiveTimedelta:
    """Estimate a canonical sampling interval from a set of near-equal deltas.

    Useful when several seismograms nominally share a sampling interval but
    report values that differ only by measurement noise (e.g. clock drift
    reflected in a reported sample rate) or floating-point noise. Returns the low
    median of `deltas`: an order-independent choice that is always one of
    the input values, rather than a synthetic average that none of the
    seismograms actually have.

    This does not check how close the given deltas are to each other; for a
    set of genuinely different sampling intervals it simply returns the low
    median of the sorted values.

    Args:
        deltas: Sampling intervals to estimate a canonical value from.

    Returns:
        The estimated canonical sampling interval.

    Raises:
        ValueError: If `deltas` is empty.

    Examples:
        ```python
        >>> import pandas as pd
        >>> from pysmo.functions import estimate_delta
        >>> deltas = [
        ...     pd.Timedelta(seconds=0.01),
        ...     pd.Timedelta(seconds=0.010000000000001),
        ...     pd.Timedelta(seconds=0.01),
        ... ]
        >>> estimate_delta(deltas)
        Timedelta('0 days 00:00:00.010000')
        >>>
        ```
    """
    if not deltas:
        raise ValueError("No deltas to estimate from.")
    return statistics.median_low(deltas)

merge

merge(
    seismograms: Sequence[Seismogram],
    *,
    delta: PositiveTimedelta | None = None,
    auto_delta: bool = False,
    gap_tolerance_factor: NonNegativeNumber = 0.5,
    replace: bool = False
) -> T | None

Merge contiguous seismograms into a single seismogram.

Empty seismograms take no part in the merge arithmetic (they never contribute data and never constrain sampling-interval or gap/overlap checks) and are absent from the result if there are non-empty seismograms to merge with. The remaining, non-empty seismograms are merged in chronological order of begin_time, regardless of the order they are given in, and must lie on a single regular sampling grid. By default, this requires equal sampling intervals; when delta is provided, each non-empty seismogram is first resampled to that common interval using resample. If delta is None and auto_delta is True, a common interval is estimated with estimate_delta instead of requiring an exact match; useful when sampling intervals only disagree by measurement or floating-point noise.

A small amount of boundary timestamp jitter is allowed, bounded by gap_tolerance_factor sampling intervals, so metadata rounding noise does not block otherwise valid merges. If consecutive seismograms overlap within this tolerance, the overlapping samples must match (compared with allclose and its default tolerances, to accommodate floating-point noise from e.g. prior resampling); they are verified and the duplicates are discarded rather than concatenated. A sub-tolerance positive gap is closed by concatenating the later seismogram's samples straight onto the preceding grid, shifting them earlier by up to gap_tolerance_factor of a sampling interval; no samples are inserted to span the gap.

When replace=False, the first seismogram in seismograms (as given, not necessarily the chronologically first, and regardless of whether it is itself empty) is modified in place and becomes the merged result: its begin_time and data are overwritten to reflect the full, chronologically-ordered merge of the non-empty seismograms. Other input seismograms are never modified. When replace=True, no input seismogram is modified and a new merged seismogram is returned instead (not supported by every concrete type; see pysmo.functions).

Parameters:

Name Type Description Default
seismograms Sequence[Seismogram]

Seismograms to merge. May be given in any order; any mix of types satisfying the Seismogram protocol works at runtime. When replace=True, the return type is inferred from seismograms; for a bare list/tuple literal mixing concrete types, annotate it as Sequence[Seismogram] to keep the call type-checked (see Examples).

required
delta PositiveTimedelta | None

Sampling interval to resample all non-empty seismograms to before merging. If None, all non-empty input seismograms must already share the same sampling interval, unless auto_delta is True.

None
auto_delta bool

Estimate a common sampling interval from the non-empty seismograms with estimate_delta instead of requiring an exact match. Mutually exclusive with delta; type checkers reject passing both, and passing both raises at runtime too.

False
gap_tolerance_factor NonNegativeNumber

Maximum allowed boundary timestamp jitter between consecutive seismograms, as a fraction of the sampling interval.

0.5
replace bool

Return a new merged seismogram and leave every input untouched, instead of modifying the first input in place. Not supported by every concrete type (see pysmo.functions).

False

Returns:

Type Description
T | None

Merged Seismogram object if called with

T | None

replace=True.

Raises:

Type Description
ValueError

If both delta and auto_delta are given, seismograms is empty, contains no non-empty seismograms, the sampling intervals of the non-empty seismograms differ and neither delta nor auto_delta is provided, the boundary between consecutive non-empty seismograms contains a gap or overlap exceeding the allowed tolerance, overlapping samples do not match, or gap_tolerance_factor is negative.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from pysmo import MiniSeismogram
>>> from pysmo.functions import merge
>>> first = MiniSeismogram(
...     begin_time=pd.Timestamp("2010-02-27T06:30:00Z"),
...     delta=pd.Timedelta(seconds=1),
...     data=np.array([1.0, 2.0, 3.0]),
... )
>>> second = MiniSeismogram(
...     begin_time=pd.Timestamp("2010-02-27T06:30:03Z"),
...     delta=pd.Timedelta(seconds=1),
...     data=np.array([4.0, 5.0]),
... )
>>> merged = merge([first, second], replace=True)
>>> merged.data
array([1., 2., 3., 4., 5.])
>>> merged.begin_time
Timestamp('2010-02-27 06:30:00+0000', tz='UTC')

Merging seismograms of different concrete types works the same way at runtime. A bare list literal's inferred type comes from its elements, though, and for a mix of concrete types that inferred type may not satisfy the Seismogram bound at all, making the call fail to type-check. Annotate the list as Sequence[Seismogram] to keep the result type-checked:

>>> from collections.abc import Sequence
>>> from pysmo import Seismogram
>>> from pysmo.classes import GeoCsvSeismogram
>>> geocsv_seis = GeoCsvSeismogram(
...     begin_time=pd.Timestamp("2010-02-27T06:30:05Z"),
...     delta=pd.Timedelta(seconds=1),
...     data=np.array([6.0, 7.0]),
...     sourceid="IU_ANMO_00_LHZ",
... )
>>> mixed: Sequence[Seismogram] = [merged, geocsv_seis]
>>> merged_mixed = merge(mixed, replace=True)
>>> merged_mixed.data
array([1., 2., 3., 4., 5., 6., 7.])
>>>

The merged object's actual class is always seismograms[0]'s class, regardless of what a type checker can infer; this is purely a static-typing concern. If downstream code depends on the concrete type, merging a single concrete type (the common case) lets it be inferred automatically, without needing the annotation above.

Seismograms whose sampling intervals only disagree by measurement or floating-point noise (see estimate_delta) can be merged with auto_delta=True instead of requiring an exact match:

>>> steady = MiniSeismogram(
...     begin_time=pd.Timestamp("2010-02-27T06:30:00Z"),
...     delta=pd.Timedelta(seconds=1),
...     data=np.array([1.0, 2.0, 3.0]),
... )
>>> jittery = MiniSeismogram(
...     begin_time=pd.Timestamp("2010-02-27T06:30:03Z"),
...     delta=pd.Timedelta(seconds=1) + pd.Timedelta(nanoseconds=1),
...     data=np.array([4.0, 5.0, 6.0]),
... )
>>> auto_merged = merge(
...     [steady, jittery], auto_delta=True, replace=True
... )
>>> auto_merged.delta
Timedelta('0 days 00:00:01')
>>> auto_merged.data
array([1., 2., 3., 4., 5., 6.])
>>>

auto_delta estimates a canonical interval with estimate_delta; it does not verify that the seismograms genuinely belong on the same sampling grid. Users are encouraged to inspect the resulting delta (as above), or call estimate_delta directly beforehand, to confirm the estimate is the value expected rather than assuming it silently.

Source code in src/pysmo/functions/_seismogram.py
def merge[T: Seismogram](
    seismograms: Sequence[Seismogram],
    *,
    delta: PositiveTimedelta | None = None,
    auto_delta: bool = False,
    gap_tolerance_factor: NonNegativeNumber = 0.5,
    replace: bool = False,
) -> T | None:
    """Merge contiguous seismograms into a single seismogram.

    Empty seismograms take no part in the merge arithmetic (they never
    contribute data and never constrain sampling-interval or gap/overlap
    checks) and are absent from the result if there are non-empty
    seismograms to merge with. The remaining, non-empty seismograms are
    merged in chronological order of `begin_time`, regardless of the order
    they are given in, and must lie on a single regular sampling grid. By
    default, this requires equal sampling intervals; when `delta` is
    provided, each non-empty seismogram is first resampled to that common
    interval using [`resample`][pysmo.functions.resample]. If `delta` is
    `None` and `auto_delta` is `True`, a common interval is estimated with
    [`estimate_delta`][pysmo.functions.estimate_delta] instead of requiring
    an exact match; useful when sampling intervals only disagree by
    measurement or floating-point noise.

    A small amount of boundary timestamp jitter is allowed, bounded by
    `gap_tolerance_factor` sampling intervals, so metadata rounding noise does
    not block otherwise valid merges. If consecutive seismograms overlap
    within this tolerance, the overlapping samples must match (compared with
    [`allclose`][numpy.allclose] and its default tolerances, to accommodate
    floating-point noise from e.g. prior resampling); they are verified and
    the duplicates are discarded rather than concatenated. A sub-tolerance
    positive gap is closed by concatenating the later seismogram's samples
    straight onto the preceding grid, shifting them earlier by up to
    `gap_tolerance_factor` of a sampling interval; no samples are inserted to
    span the gap.

    When `replace=False`, the first seismogram in `seismograms` (as given,
    not necessarily the chronologically first, and regardless of whether it
    is itself empty) is modified in place and becomes the merged result: its
    `begin_time` and `data` are overwritten to reflect the full,
    chronologically-ordered merge of the non-empty seismograms. Other input
    seismograms are never modified. When `replace=True`, no input seismogram
    is modified and a new merged seismogram is returned instead (not
    supported by every concrete type; see [`pysmo.functions`][]).

    Args:
        seismograms: Seismograms to merge. May be given in any order; any mix
            of types satisfying the [`Seismogram`][pysmo.Seismogram] protocol
            works at runtime. When `replace=True`, the return type is inferred
            from `seismograms`; for a bare list/tuple literal mixing concrete
            types, annotate it as `Sequence[Seismogram]` to keep the call
            type-checked (see Examples).
        delta: Sampling interval to resample all non-empty seismograms to
            before merging. If `None`, all non-empty input seismograms must
            already share the same sampling interval, unless `auto_delta`
            is `True`.
        auto_delta: Estimate a common sampling interval from the non-empty
            seismograms with
            [`estimate_delta`][pysmo.functions.estimate_delta] instead of
            requiring an exact match. Mutually exclusive with `delta`; type
            checkers reject passing both, and passing both raises at
            runtime too.
        gap_tolerance_factor: Maximum allowed boundary timestamp jitter between
            consecutive seismograms, as a fraction of the sampling interval.
        replace: Return a new merged seismogram and leave every input
            untouched, instead of modifying the first input in place. Not
            supported by every concrete type (see [`pysmo.functions`][]).

    Returns:
        Merged [`Seismogram`][pysmo.Seismogram] object if called with
        `replace=True`.

    Raises:
        ValueError: If both `delta` and `auto_delta` are given, `seismograms`
            is empty, contains no non-empty seismograms, the sampling
            intervals of the non-empty seismograms differ and neither
            `delta` nor `auto_delta` is provided, the boundary between
            consecutive non-empty seismograms contains a gap or overlap
            exceeding the allowed tolerance, overlapping samples do not
            match, or `gap_tolerance_factor` is negative.

    Examples:
        ```python
        >>> import numpy as np
        >>> import pandas as pd
        >>> from pysmo import MiniSeismogram
        >>> from pysmo.functions import merge
        >>> first = MiniSeismogram(
        ...     begin_time=pd.Timestamp("2010-02-27T06:30:00Z"),
        ...     delta=pd.Timedelta(seconds=1),
        ...     data=np.array([1.0, 2.0, 3.0]),
        ... )
        >>> second = MiniSeismogram(
        ...     begin_time=pd.Timestamp("2010-02-27T06:30:03Z"),
        ...     delta=pd.Timedelta(seconds=1),
        ...     data=np.array([4.0, 5.0]),
        ... )
        >>> merged = merge([first, second], replace=True)
        >>> merged.data
        array([1., 2., 3., 4., 5.])
        >>> merged.begin_time
        Timestamp('2010-02-27 06:30:00+0000', tz='UTC')

        ```

        Merging seismograms of different concrete types works the same way
        at runtime. A bare list literal's inferred type comes from its
        elements, though, and for a mix of concrete types that inferred type
        may not satisfy the `Seismogram` bound at all, making the call fail
        to type-check. Annotate the list as `Sequence[Seismogram]` to keep
        the result type-checked:

        ```python
        >>> from collections.abc import Sequence
        >>> from pysmo import Seismogram
        >>> from pysmo.classes import GeoCsvSeismogram
        >>> geocsv_seis = GeoCsvSeismogram(
        ...     begin_time=pd.Timestamp("2010-02-27T06:30:05Z"),
        ...     delta=pd.Timedelta(seconds=1),
        ...     data=np.array([6.0, 7.0]),
        ...     sourceid="IU_ANMO_00_LHZ",
        ... )
        >>> mixed: Sequence[Seismogram] = [merged, geocsv_seis]
        >>> merged_mixed = merge(mixed, replace=True)
        >>> merged_mixed.data
        array([1., 2., 3., 4., 5., 6., 7.])
        >>>
        ```

        The merged object's actual class is always `seismograms[0]`'s class,
        regardless of what a type checker can infer; this is purely a
        static-typing concern. If downstream code depends on the concrete
        type, merging a single concrete type (the common case) lets it be inferred
        automatically, without needing the annotation above.

        Seismograms whose sampling intervals only disagree by measurement or
        floating-point noise (see
        [`estimate_delta`][pysmo.functions.estimate_delta]) can be merged
        with `auto_delta=True` instead of requiring an exact match:

        ```python
        >>> steady = MiniSeismogram(
        ...     begin_time=pd.Timestamp("2010-02-27T06:30:00Z"),
        ...     delta=pd.Timedelta(seconds=1),
        ...     data=np.array([1.0, 2.0, 3.0]),
        ... )
        >>> jittery = MiniSeismogram(
        ...     begin_time=pd.Timestamp("2010-02-27T06:30:03Z"),
        ...     delta=pd.Timedelta(seconds=1) + pd.Timedelta(nanoseconds=1),
        ...     data=np.array([4.0, 5.0, 6.0]),
        ... )
        >>> auto_merged = merge(
        ...     [steady, jittery], auto_delta=True, replace=True
        ... )
        >>> auto_merged.delta
        Timedelta('0 days 00:00:01')
        >>> auto_merged.data
        array([1., 2., 3., 4., 5., 6.])
        >>>
        ```

        `auto_delta` estimates a canonical interval with
        [`estimate_delta`][pysmo.functions.estimate_delta]; it does not
        verify that the seismograms genuinely belong on the same sampling
        grid. Users are encouraged to inspect the resulting `delta` (as
        above), or call
        [`estimate_delta`][pysmo.functions.estimate_delta] directly
        beforehand, to confirm the estimate is the value expected rather
        than assuming it silently.
    """
    if delta is not None and auto_delta:
        raise ValueError("delta and auto_delta are mutually exclusive.")

    if gap_tolerance_factor < 0:
        raise ValueError("gap_tolerance_factor must be non-negative.")

    if not seismograms:
        raise ValueError("No seismograms to merge.")

    working = list(seismograms)

    non_empty = [seismogram for seismogram in working if len(seismogram.data)]
    if not non_empty:
        raise ValueError("No non-empty seismograms to merge.")

    if delta is None and auto_delta:
        delta = estimate_delta([seismogram.delta for seismogram in non_empty])

    if delta is None:
        reference_delta = non_empty[0].delta
        for seismogram in non_empty[1:]:
            if seismogram.delta != reference_delta:
                raise ValueError(
                    "Cannot merge seismograms with different sampling intervals "
                    + f"without resampling: {reference_delta} vs {seismogram.delta}."
                )
    else:
        reference_delta = delta
        for index, seismogram in enumerate(working):
            if len(seismogram.data) == 0:
                continue
            if seismogram.delta == delta:
                continue
            if replace or index > 0:
                working[index] = resample(seismogram, delta, replace=True)
            else:
                resample(seismogram, delta)
        non_empty = [seismogram for seismogram in working if len(seismogram.data)]

    ordered = sorted(non_empty, key=lambda seismogram: seismogram.begin_time)

    overlap_samples = [0] * len(ordered)
    for index, (prev, curr) in enumerate(pairwise(ordered), start=1):
        expected_next = prev.end_time + prev.delta
        gap = curr.begin_time - expected_next
        tolerance = prev.delta * gap_tolerance_factor
        if abs(gap) > tolerance:
            description = (
                f"gap of {gap.total_seconds():.6f} s"
                if gap > pd.Timedelta(0)
                else f"overlap of {-gap.total_seconds():.6f} s"
            )
            raise ValueError(
                f"Data {description} detected between seismogram ending at "
                + f"{prev.end_time} and seismogram starting at {curr.begin_time}."
            )
        if gap < pd.Timedelta(0):
            samples = min(round(-gap / prev.delta), len(prev.data), len(curr.data))
            if samples > 0 and not np.allclose(
                prev.data[-samples:], curr.data[:samples]
            ):
                raise ValueError(
                    "Overlapping samples between seismogram ending at "
                    + f"{prev.end_time} and seismogram starting at "
                    + f"{curr.begin_time} do not match; cannot merge."
                )
            overlap_samples[index] = samples

    merged_begin_time = ordered[0].begin_time
    merged_data = np.concatenate(
        [ordered[0].data]
        + [
            seismogram.data[samples:]
            for seismogram, samples in zip(ordered[1:], overlap_samples[1:])
        ]
    )

    if replace:
        return copy.replace(
            working[0],  # type: ignore[arg-type]
            data=merged_data,
            begin_time=merged_begin_time,
            delta=reference_delta,
        )

    merged = cast(T, working[0])
    merged.begin_time = merged_begin_time
    merged.delta = reference_delta
    merged.data = merged_data
    return None

normalize

normalize(
    seismogram: T,
    t1: Timestamp | None = None,
    t2: Timestamp | None = None,
    *,
    replace: bool = False
) -> T | None

Normalise a seismogram with its absolute max value.

Parameters:

Name Type Description Default
seismogram T

Seismogram object.

required
t1 Timestamp | None

Start of the window used to find the maximum. If None, the search starts from the beginning of the seismogram.

None
t2 Timestamp | None

End of the window used to find the maximum. If None, the search continues to the end of the seismogram.

None
replace bool

Return a new seismogram and leave the input untouched, instead of modifying it in place. Not supported by every concrete type (see pysmo.functions).

False

Returns:

Type Description
T | None

Normalised Seismogram object if replace=True.

Raises:

Type Description
ValueError

If the absolute maximum of the data (within the optional time window) is zero, as normalisation would produce undefined results.

Examples:

>>> import numpy as np
>>> from pysmo.functions import normalize
>>> from pysmo.classes import SAC
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> normalize(sac_seis)
>>> -1 <= np.max(sac_seis.data) <= 1
np.True_
>>>
Source code in src/pysmo/functions/_seismogram.py
def normalize[T: Seismogram](
    seismogram: T,
    t1: pd.Timestamp | None = None,
    t2: pd.Timestamp | None = None,
    *,
    replace: bool = False,
) -> T | None:
    """Normalise a seismogram with its absolute max value.

    Args:
        seismogram: Seismogram object.
        t1: Start of the window used to find the maximum. If `None`, the search
            starts from the beginning of the seismogram.
        t2: End of the window used to find the maximum. If `None`, the search
            continues to the end of the seismogram.
        replace: Return a new seismogram and leave the input untouched,
            instead of modifying it in place. Not supported by every
            concrete type (see [`pysmo.functions`][]).

    Returns:
        Normalised [`Seismogram`][pysmo.Seismogram] object if `replace=True`.

    Raises:
        ValueError: If the absolute maximum of the data (within the optional
            time window) is zero, as normalisation would produce undefined results.

    Examples:
        ```python
        >>> import numpy as np
        >>> from pysmo.functions import normalize
        >>> from pysmo.classes import SAC
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> normalize(sac_seis)
        >>> -1 <= np.max(sac_seis.data) <= 1
        np.True_
        >>>
        ```
    """

    start_index, end_index = None, None

    if t1 is not None:
        start_index = time2index(seismogram, t1)

    if t2 is not None:
        # +1 because time2index returns the sample nearest t2, which should
        # be included in the search window (matches crop()'s convention).
        end_index = time2index(seismogram, t2) + 1

    abs_max = np.max(np.abs(seismogram.data[start_index:end_index]))
    if abs_max == 0:
        raise ValueError(
            "Cannot normalise a seismogram because the absolute maximum "
            + "within the selected time window (or entire trace if no window "
            + "is given) is zero."
        )

    if replace:
        return copy.replace(
            seismogram,  # type: ignore[arg-type]
            data=seismogram.data / abs_max,
        )

    seismogram.data /= abs_max
    return None

pad

pad(
    seismogram: T,
    begin_time: Timestamp,
    end_time: Timestamp,
    mode: _ModeKind | _ModeFunc = "constant",
    *,
    replace: bool = False,
    **kwargs: Any
) -> T | None

Pad seismogram data.

This function calculates the indices corresponding to the provided new begin and end times using time2index, then pads the data array using numpy.pad and updates the begin_time. Note that the actual begin and end times are set by indexing, so they may be slightly different than the provided input begin and end times.

Parameters:

Name Type Description Default
seismogram T

Seismogram object.

required
begin_time Timestamp

New begin time.

required
end_time Timestamp

New end time.

required
mode _ModeKind | _ModeFunc

Pad mode to use (see numpy.pad for all modes).

'constant'
replace bool

Return a new seismogram and leave the input untouched, instead of modifying it in place. Not supported by every concrete type (see pysmo.functions).

False
kwargs Any

Keyword arguments to pass to numpy.pad.

{}

Returns:

Type Description
T | None

Padded Seismogram object if called with replace=True.

Raises:

Type Description
ValueError

If new begin time is after new end time.

Examples:

>>> from pysmo.functions import pad
>>> from pysmo.classes import SAC
>>> import pandas as pd
>>> import numpy as np
>>>
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> original_length = len(sac_seis.data)
>>> sac_seis.data
array([-47201., -47361., -47511., ..., -82144., -71072., -59960.],
      shape=(57465,))
>>> new_begin_time = sac_seis.begin_time - pd.Timedelta(seconds=10)
>>> new_end_time = sac_seis.end_time + pd.Timedelta(seconds=10)
>>> pad(sac_seis, new_begin_time, new_end_time)
>>> np.isclose(len(sac_seis.data), original_length + 20 / sac_seis.delta.total_seconds())
np.True_
>>> sac_seis.data
array([0., 0., 0., ..., 0., 0., 0.], shape=(57865,))
>>>
Source code in src/pysmo/functions/_seismogram.py
def pad[T: Seismogram](
    seismogram: T,
    begin_time: pd.Timestamp,
    end_time: pd.Timestamp,
    mode: "_ModeKind | _ModeFunc" = "constant",
    *,
    replace: bool = False,
    **kwargs: Any,
) -> T | None:
    """Pad seismogram data.

    This function calculates the indices corresponding to the provided new
    begin and end times using [`time2index`][pysmo.functions.time2index], then
    pads the [`data`][pysmo.Seismogram.data] array using [`numpy.pad`][] and
    updates the [`begin_time`][pysmo.Seismogram.begin_time]. Note that the
    actual begin and end times are set by indexing, so they may be slightly
    different than the provided input begin and end times.

    Args:
        seismogram: [`Seismogram`][pysmo.Seismogram] object.
        begin_time: New begin time.
        end_time: New end time.
        mode: Pad mode to use (see [`numpy.pad`][] for all modes).
        replace: Return a new seismogram and leave the input untouched,
            instead of modifying it in place. Not supported by every
            concrete type (see [`pysmo.functions`][]).
        kwargs: Keyword arguments to pass to [`numpy.pad`][].

    Returns:
        Padded [`Seismogram`][pysmo.Seismogram] object if called with `replace=True`.

    Raises:
        ValueError: If new begin time is after new end time.

    Examples:
        ```python
        >>> from pysmo.functions import pad
        >>> from pysmo.classes import SAC
        >>> import pandas as pd
        >>> import numpy as np
        >>>
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> original_length = len(sac_seis.data)
        >>> sac_seis.data
        array([-47201., -47361., -47511., ..., -82144., -71072., -59960.],
              shape=(57465,))
        >>> new_begin_time = sac_seis.begin_time - pd.Timedelta(seconds=10)
        >>> new_end_time = sac_seis.end_time + pd.Timedelta(seconds=10)
        >>> pad(sac_seis, new_begin_time, new_end_time)
        >>> np.isclose(len(sac_seis.data), original_length + 20 / sac_seis.delta.total_seconds())
        np.True_
        >>> sac_seis.data
        array([0., 0., 0., ..., 0., 0., 0.], shape=(57865,))
        >>>
        ```
    """

    if begin_time > end_time:
        raise ValueError("New begin_time cannot be after new end_time")

    start_index = time2index(seismogram, begin_time, allow_out_of_bounds=True)
    end_index = time2index(seismogram, end_time, allow_out_of_bounds=True)

    pad_before = max(0, -start_index)
    pad_after = max(0, end_index - (len(seismogram.data) - 1))
    padding_needed = pad_before > 0 or pad_after > 0

    if replace:
        if padding_needed:
            new_data = np.pad(
                seismogram.data,
                pad_width=(pad_before, pad_after),
                mode=mode,
                **kwargs,
            )
            new_begin_time = seismogram.begin_time + seismogram.delta * min(
                0, start_index
            )
        else:
            # replace=True must always hand back a fully independent object,
            # even when there is nothing to pad.
            new_data = seismogram.data.copy()
            new_begin_time = seismogram.begin_time
        return copy.replace(
            seismogram,  # type: ignore[arg-type]
            data=new_data,
            begin_time=new_begin_time,
        )

    if padding_needed:
        seismogram.data = np.pad(
            seismogram.data,
            pad_width=(pad_before, pad_after),
            mode=mode,
            **kwargs,
        )
        seismogram.begin_time += seismogram.delta * min(0, start_index)

    return None

resample

resample(
    seismogram: T,
    delta: PositiveTimedelta,
    *,
    replace: bool = False
) -> T | None

Resample Seismogram data using the Fourier method.

This function uses scipy.signal.resample to resample the data to a new sampling interval. If the new sampling interval is identical to the current one, no action is taken.

Parameters:

Name Type Description Default
seismogram T

Seismogram object.

required
delta PositiveTimedelta

New sampling interval.

required
replace bool

Return a new seismogram and leave the input untouched, instead of modifying it in place. Not supported by every concrete type (see pysmo.functions).

False

Returns:

Type Description
T | None

Resampled Seismogram object if called with replace=True.

Examples:

>>> from pysmo.functions import resample
>>> from pysmo.classes import SAC
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> len(sac_seis.data)
57465
>>> original_delta = sac_seis.delta
>>> new_delta = original_delta * 2
>>> resample(sac_seis, new_delta)
>>> len(sac_seis.data)
28732
>>>
Source code in src/pysmo/functions/_seismogram.py
def resample[T: Seismogram](
    seismogram: T, delta: PositiveTimedelta, *, replace: bool = False
) -> T | None:
    """Resample Seismogram data using the Fourier method.

    This function uses [`scipy.signal.resample`][] to resample the data to a
    new sampling interval. If the new sampling interval is identical to the
    current one, no action is taken.

    Args:
        seismogram: Seismogram object.
        delta: New sampling interval.
        replace: Return a new seismogram and leave the input untouched,
            instead of modifying it in place. Not supported by every
            concrete type (see [`pysmo.functions`][]).

    Returns:
        Resampled [`Seismogram`][pysmo.Seismogram] object if called with `replace=True`.

    Examples:
        ```python
        >>> from pysmo.functions import resample
        >>> from pysmo.classes import SAC
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> len(sac_seis.data)
        57465
        >>> original_delta = sac_seis.delta
        >>> new_delta = original_delta * 2
        >>> resample(sac_seis, new_delta)
        >>> len(sac_seis.data)
        28732
        >>>
        ```
    """
    if replace:
        if delta != seismogram.delta:
            npts = int(len(seismogram.data) * seismogram.delta / delta)
            new_data = scipy.signal.resample(seismogram.data, npts)
            new_delta = delta
        else:
            # replace=True must always hand back a fully independent object,
            # even when the sampling interval is unchanged.
            new_data = seismogram.data.copy()
            new_delta = seismogram.delta
        return copy.replace(
            seismogram,  # type: ignore[arg-type]
            data=new_data,
            delta=new_delta,
        )

    if delta != seismogram.delta:
        npts = int(len(seismogram.data) * seismogram.delta / delta)
        seismogram.data = scipy.signal.resample(seismogram.data, npts)
        seismogram.delta = delta

    return None

seismogram_checksum

seismogram_checksum(seismogram: Seismogram) -> str

Return a stable digest of a seismogram's samples and timing.

Covers data, begin_time, and delta, the three members of the Seismogram protocol, and nothing else: two seismograms with equal values for those hash the same regardless of their concrete type or any extra attributes it carries. The result is prefixed with the hash name (sha256:).

Examples:

>>> import pandas as pd
>>> from pysmo import MiniSeismogram
>>> from pysmo.functions import seismogram_checksum
>>>
>>> seismogram = MiniSeismogram(
...     begin_time=pd.Timestamp("2024-01-01T00:00:00Z"),
...     delta=pd.Timedelta(seconds=1),
...     data=[1.0, 2.0, 3.0],
... )
>>> seismogram_checksum(seismogram)
'sha256:...'
>>> rebuilt = MiniSeismogram(
...     begin_time=seismogram.begin_time,
...     delta=seismogram.delta,
...     data=[1.0, 2.0, 3.0],
... )
>>> seismogram_checksum(rebuilt) == seismogram_checksum(seismogram)
True
>>>
Source code in src/pysmo/functions/_serialize.py
def seismogram_checksum(seismogram: Seismogram) -> str:
    """Return a stable digest of a seismogram's samples and timing.

    Covers `data`, `begin_time`, and `delta`, the three members of the
    [`Seismogram`][pysmo.Seismogram] protocol, and nothing else: two
    seismograms with equal values for those hash the same regardless of their
    concrete type or any extra attributes it carries. The result is prefixed
    with the hash name (`sha256:`).

    Examples:
        >>> import pandas as pd
        >>> from pysmo import MiniSeismogram
        >>> from pysmo.functions import seismogram_checksum
        >>>
        >>> seismogram = MiniSeismogram(
        ...     begin_time=pd.Timestamp("2024-01-01T00:00:00Z"),
        ...     delta=pd.Timedelta(seconds=1),
        ...     data=[1.0, 2.0, 3.0],
        ... )
        >>> seismogram_checksum(seismogram)
        'sha256:...'
        >>> rebuilt = MiniSeismogram(
        ...     begin_time=seismogram.begin_time,
        ...     delta=seismogram.delta,
        ...     data=[1.0, 2.0, 3.0],
        ... )
        >>> seismogram_checksum(rebuilt) == seismogram_checksum(seismogram)
        True
        >>>
    """
    h = hashlib.sha256()
    data = np.ascontiguousarray(seismogram.data)
    _hash_field(h, b"dtype", data.dtype.str.encode())
    _hash_field(h, b"shape", repr(tuple(int(n) for n in data.shape)).encode())
    _hash_field(h, b"data", data.tobytes())
    # `.value` (integer nanoseconds) rather than `str()`: a fixed
    # representation that does not shift with the pandas version.
    _hash_field(h, b"begin_time", _int64(seismogram.begin_time.value))
    _hash_field(h, b"delta", _int64(seismogram.delta.value))
    return f"sha256:{h.hexdigest()}"

seismogram_from_json

seismogram_from_json(
    blob: bytes,
    cls: type[Seismogram] | None = None,
    *,
    trusted_modules: tuple[
        str, ...
    ] = _DEFAULT_TRUSTED_MODULES
) -> Seismogram

Reconstruct a seismogram from a seismogram_to_json document.

The document is data, not code: no part of it is executed. When cls is None the recorded module:qualname is imported to rebuild the type, but only from a package in trusted_modules — a tampered document cannot name an arbitrary importable module to trigger its import side effects.

Parameters:

Name Type Description Default
blob bytes

The document produced by seismogram_to_json.

required
cls type[Seismogram] | None

The attrs class to rebuild, returned as its own type. When None, the module:qualname recorded in the document is imported and the result is typed as Seismogram; pass cls explicitly when the type may have moved since it was encoded, to keep the concrete return type, or to rebuild a type from outside trusted_modules.

None
trusted_modules tuple[str, ...]

Top-level packages the recorded class may be imported from when cls is None. Defaults to pysmo's own types only.

_DEFAULT_TRUSTED_MODULES

Returns:

Type Description
Seismogram

A new instance of cls, or of the recorded type.

Raises:

Type Description
TypeError

If the document is malformed or an unsupported version, if cls is None and the recorded module is not trusted or can no longer be imported, if the resolved type is not an attrs class, or if the payload does not fit the resolved type.

Examples:

>>> import pandas as pd
>>> from pysmo import MiniSeismogram
>>> from pysmo.functions import seismogram_from_json, seismogram_to_json
>>>
>>> blob = seismogram_to_json(
...     MiniSeismogram(
...         begin_time=pd.Timestamp("2024-01-01T00:00:00Z"),
...         delta=pd.Timedelta(seconds=1),
...         data=[1.0, 2.0, 3.0],
...     )
... )
>>> seismogram_from_json(blob, cls=MiniSeismogram).data.tolist()
[1.0, 2.0, 3.0]
>>>
Source code in src/pysmo/functions/_serialize.py
def seismogram_from_json(
    blob: bytes,
    cls: type[Seismogram] | None = None,
    *,
    trusted_modules: tuple[str, ...] = _DEFAULT_TRUSTED_MODULES,
) -> Seismogram:
    """Reconstruct a seismogram from a `seismogram_to_json` document.

    The document is data, not code: no part of it is executed. When `cls` is
    `None` the recorded `module:qualname` is imported to rebuild the type, but
    only from a package in `trusted_modules` — a tampered document cannot name
    an arbitrary importable module to trigger its import side effects.

    Args:
        blob: The document produced by
            [`seismogram_to_json`][pysmo.functions.seismogram_to_json].
        cls: The attrs class to rebuild, returned as its own type. When
            `None`, the `module:qualname` recorded in the document is
            imported and the result is typed as
            [`Seismogram`][pysmo.Seismogram]; pass `cls` explicitly when the
            type may have moved since it was encoded, to keep the concrete
            return type, or to rebuild a type from outside `trusted_modules`.
        trusted_modules: Top-level packages the recorded class may be imported
            from when `cls` is `None`. Defaults to pysmo's own types only.

    Returns:
        A new instance of `cls`, or of the recorded type.

    Raises:
        TypeError: If the document is malformed or an unsupported version, if
            `cls` is `None` and the recorded module is not trusted or can no
            longer be imported, if the resolved type is not an attrs class, or
            if the payload does not fit the resolved type.

    Examples:
        >>> import pandas as pd
        >>> from pysmo import MiniSeismogram
        >>> from pysmo.functions import seismogram_from_json, seismogram_to_json
        >>>
        >>> blob = seismogram_to_json(
        ...     MiniSeismogram(
        ...         begin_time=pd.Timestamp("2024-01-01T00:00:00Z"),
        ...         delta=pd.Timedelta(seconds=1),
        ...         data=[1.0, 2.0, 3.0],
        ...     )
        ... )
        >>> seismogram_from_json(blob, cls=MiniSeismogram).data.tolist()
        [1.0, 2.0, 3.0]
        >>>
    """
    try:
        envelope = json.loads(blob)
    except (ValueError, TypeError) as exc:
        raise TypeError(f"not a valid seismogram JSON document ({exc}).") from exc
    if not isinstance(envelope, dict) or not {"cls", "v", "payload"} <= envelope.keys():
        raise TypeError(
            "seismogram JSON document is missing its cls/v/payload envelope."
        )
    version = envelope["v"]
    if not isinstance(version, int) or version not in _SUPPORTED_JSON_VERSIONS:
        raise TypeError(
            f"seismogram JSON document is version {version!r}; this pysmo reads "
            + f"{sorted(_SUPPORTED_JSON_VERSIONS)}."
        )

    resolved: type[Seismogram]
    if cls is None:
        resolved = _resolve_encoded_class(envelope["cls"], trusted_modules)
    else:
        resolved = cls
    if not attrs.has(resolved):
        raise TypeError(f"{resolved!r} is not an attrs class.")
    try:
        return cast(Seismogram, _converter.structure(envelope["payload"], resolved))
    except MemoryError:
        raise
    except Exception as exc:
        raise TypeError(
            f"seismogram JSON payload does not fit {resolved!r} ({exc})."
        ) from exc

seismogram_to_json

seismogram_to_json(
    seismogram: Seismogram, *, verify: bool = False
) -> bytes

Encode a value-object seismogram as a portable JSON document.

The document is a {"cls", "v", "payload"} envelope: cls records the seismogram's module:qualname so seismogram_from_json can rebuild the same type, v is the format version, and payload holds the seismogram's fields with pd.Timestamp and pd.Timedelta as integer nanoseconds and np.ndarray as a base64 dtype/shape/b64 triple.

Not every seismogram can be encoded

Only an attrs value object declaring begin_time, delta and data as real fields round-trips: MiniSeismogram, MiniIccsSeismogram, GeoCsvSeismogram, and user types built the same way. A live view such as SacSeismogram, or a type carrying a field the converter has no hook for, raises TypeError; convert it with clone_to_mini first.

Parameters:

Name Type Description Default
seismogram Seismogram

The seismogram to encode.

required
verify bool

Decode the fresh document and compare it back to seismogram, raising TypeError on any mismatch. Catches a round trip that silently drops information on a rich field (a non-primitive value in MiniIccsSeismogram.extra, say). data is compared with equal_nan, so a genuine NaN sample (a data gap, a masked window) is not reported as a lossy round trip.

False

Returns:

Type Description
bytes

The UTF-8 JSON document.

Raises:

Type Description
TypeError

If seismogram is not a serialisable attrs seismogram, a field has no cattrs hook, or verify is set and the document does not round-trip.

Examples:

>>> import json
>>> import pandas as pd
>>> from pysmo import MiniSeismogram
>>> from pysmo.functions import seismogram_from_json, seismogram_to_json
>>>
>>> seismogram = MiniSeismogram(
...     begin_time=pd.Timestamp("2024-01-01T00:00:00Z"),
...     delta=pd.Timedelta(seconds=1),
...     data=[1.0, 2.0, 3.0],
... )
>>> blob = seismogram_to_json(seismogram)
>>> json.loads(blob)
{'cls': 'pysmo:MiniSeismogram', 'v': 1,
 'payload': {'begin_time': 1704067200000000000, 'delta': 1000000000,
             'data': {'dtype': 'float64', 'shape': [3],
                      'b64': 'AAAAAAAA8D8AAAAAAAAAQAAAAAAAAAhA'}}}
>>> seismogram_from_json(blob) == seismogram
True
>>>
Source code in src/pysmo/functions/_serialize.py
def seismogram_to_json(seismogram: Seismogram, *, verify: bool = False) -> bytes:
    """Encode a value-object seismogram as a portable JSON document.

    The document is a `{"cls", "v", "payload"}` envelope: `cls` records the
    seismogram's `module:qualname` so
    [`seismogram_from_json`][pysmo.functions.seismogram_from_json] can rebuild
    the same type, `v` is the format version, and `payload` holds the
    seismogram's fields with `pd.Timestamp` and `pd.Timedelta` as integer
    nanoseconds and `np.ndarray` as a base64 `dtype`/`shape`/`b64` triple.

    Note: Not every seismogram can be encoded
        Only an `attrs` value object declaring `begin_time`, `delta` and
        `data` as real fields round-trips:
        [`MiniSeismogram`][pysmo.MiniSeismogram],
        [`MiniIccsSeismogram`][pysmo.tools.iccs.MiniIccsSeismogram],
        [`GeoCsvSeismogram`][pysmo.classes.GeoCsvSeismogram], and user types
        built the same way. A live view such as
        [`SacSeismogram`][pysmo.classes.SacSeismogram], or a type carrying a
        field the converter has no hook for, raises `TypeError`; convert it with
        [`clone_to_mini`][pysmo.functions.clone_to_mini] first.

    Args:
        seismogram: The seismogram to encode.
        verify: Decode the fresh document and compare it back to `seismogram`,
            raising `TypeError` on any mismatch. Catches a round trip that
            silently drops information on a rich field (a non-primitive value in
            [`MiniIccsSeismogram.extra`][pysmo.tools.iccs.MiniIccsSeismogram],
            say). `data` is compared with `equal_nan`, so a genuine `NaN`
            sample (a data gap, a masked window) is not reported as a lossy
            round trip.

    Returns:
        The UTF-8 JSON document.

    Raises:
        TypeError: If `seismogram` is not a serialisable attrs seismogram, a
            field has no `cattrs` hook, or `verify` is set and the document
            does not round-trip.

    Examples:
        >>> import json
        >>> import pandas as pd
        >>> from pysmo import MiniSeismogram
        >>> from pysmo.functions import seismogram_from_json, seismogram_to_json
        >>>
        >>> seismogram = MiniSeismogram(
        ...     begin_time=pd.Timestamp("2024-01-01T00:00:00Z"),
        ...     delta=pd.Timedelta(seconds=1),
        ...     data=[1.0, 2.0, 3.0],
        ... )
        >>> blob = seismogram_to_json(seismogram)
        >>> json.loads(blob)
        {'cls': 'pysmo:MiniSeismogram', 'v': 1,
         'payload': {'begin_time': 1704067200000000000, 'delta': 1000000000,
                     'data': {'dtype': 'float64', 'shape': [3],
                              'b64': 'AAAAAAAA8D8AAAAAAAAAQAAAAAAAAAhA'}}}
        >>> seismogram_from_json(blob) == seismogram
        True
        >>>
    """
    if not _is_serializable_seismogram(seismogram):
        raise TypeError(
            f"{type(seismogram).__name__} cannot be serialised: it is not an "
            + "attrs seismogram declaring begin_time/delta/data as real fields. "
            + "Convert it with clone_to_mini first."
        )
    try:
        envelope = {
            "cls": f"{type(seismogram).__module__}:{type(seismogram).__qualname__}",
            "v": _SEISMOGRAM_JSON_VERSION,
            "payload": _converter.unstructure(seismogram),
        }
        blob = json.dumps(envelope).encode("utf-8")
    except MemoryError:
        raise
    except Exception as exc:
        raise TypeError(
            f"{type(seismogram).__name__} has a field that cannot be serialised "
            + f"({exc}); register a hook on a custom converter, "
            + "or convert it with clone_to_mini first."
        ) from exc
    if verify:
        try:
            restored = seismogram_from_json(blob, cls=type(seismogram))
        except MemoryError:
            raise
        except Exception as exc:
            raise TypeError(
                f"{type(seismogram).__name__} does not survive a JSON round trip "
                + f"({exc}); a conversion hook is lossy for one of its fields."
            ) from exc
        if not _round_trip_faithful(restored, seismogram):
            raise TypeError(
                f"{type(seismogram).__name__} does not survive a JSON round trip "
                + "(decoded value differs); a conversion hook is lossy for one of "
                + "its fields."
            )
    return blob

taper

taper(
    seismogram: T,
    taper_width: NonNegativeTimedelta | UnitFloat,
    window_type: _WindowType = "hann",
    *,
    replace: bool = False
) -> T | None

Apply a symmetric taper to the ends of a Seismogram.

The taper width is understood as the portion of the seismogram affected by the taper window function. It can be provided as an absolute duration (non-negative Timedelta), or as a fraction of seismogram length (float between 0 and 1). Internally, absolute durations are converted to fractions by dividing by the total seismogram duration, and absolute durations should therefore not exceed the total seismogram duration.

The shape of the windowing function is calculated by calling the scipy get_window() function using the number of samples corresponding to the fraction specified above, then it is split in half and applied to the beginning and end of the seismogram data. Thus taper_width=0 corresponds to a rectangular window (i.e. no tapering), and taper_width=1 to a symmetric taper applied to the entire length of the seismogram. A value of e.g. 0.5 applies the "ramp up" portion of the window to the first quarter of the seismogram, while the "ramp down" portion of the window is applied to the last quarter.

Window-shape compatibility

The scipy get_window() function is a helper function that calculates a large variety of window shapes, which do not all make sense in this application (e.g. boxcar or tukey). Users are encouraged to read the documentation of the actual window functions available via get_window() to see if they can be split in the middle and used as "ramp up" and "ramp down" functions.

Parameters:

Name Type Description Default
seismogram T

Seismogram object.

required
taper_width NonNegativeTimedelta | UnitFloat

Width of the taper to use.

required
window_type _WindowType

Function to calculate taper shape (see get_window for valid inputs).

'hann'
replace bool

Return a new seismogram and leave the input untouched, instead of modifying it in place. Not supported by every concrete type (see pysmo.functions).

False

Returns:

Type Description
T | None

Tapered Seismogram object if called with replace=True.

No taper below 2 samples

If taper_width resolves to fewer than 2 samples, no taper is applied. This can occur when a very small Timedelta is provided.

Examples:

>>> from pysmo.functions import taper, detrend
>>> from pysmo.classes import SAC
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> detrend(sac_seis)
>>> sac_seis.data
array([   821.53861155,    661.55931267,    511.58001379, ...,
       -32931.93353333, -21859.9128322 , -10747.89213108], shape=(57465,))
>>> taper(sac_seis, 0.2)
>>> sac_seis.data
array([ 0.00000000e+00,  4.94398663e-05,  1.52926246e-04, ...,
       -9.84431924e-03, -1.63364213e-03, -0.00000000e+00], shape=(57465,))
>>>
Source code in src/pysmo/functions/_seismogram.py
def taper[T: Seismogram](
    seismogram: T,
    taper_width: NonNegativeTimedelta | UnitFloat,
    window_type: _WindowType = "hann",
    *,
    replace: bool = False,
) -> T | None:
    """Apply a symmetric taper to the ends of a Seismogram.

    The taper width is understood as the portion of the seismogram affected
    by the taper window function. It can be provided as an absolute duration
    (non-negative [`Timedelta`][pandas.Timedelta]), or as a fraction of
    seismogram length ([`float`][] between `0` and `1`). Internally, absolute
    durations are converted to fractions by dividing by the total seismogram
    duration, and absolute durations should therefore not exceed the total
    seismogram duration.

    The shape of the windowing function is calculated by calling the scipy
    [`get_window()`][scipy.signal.windows.get_window] function using the number
    of samples corresponding to the fraction specified above, then it is split
    in half and applied to the beginning and end of the seismogram data. Thus
    `taper_width=0` corresponds to a rectangular window (i.e. no tapering), and
    `taper_width=1` to a symmetric taper applied to the entire length of the
    seismogram. A value of e.g. `0.5` applies the "ramp up" portion of the
    window to the first quarter of the seismogram, while the "ramp down" portion
    of the window is applied to the last quarter.

    Warning: Window-shape compatibility
        The scipy [`get_window()`][scipy.signal.windows.get_window] function
        is a helper function that calculates a large variety of window shapes,
        which do not all make sense in this application (e.g. boxcar or tukey).
        Users are encouraged to read the documentation of the actual window
        functions available via
        [`get_window()`][scipy.signal.windows.get_window] to see if they can be
        split in the middle and used as "ramp up" and "ramp down" functions.

    Args:
        seismogram: Seismogram object.
        taper_width: Width of the taper to use.
        window_type: Function to calculate taper shape (see
            [`get_window`][scipy.signal.windows.get_window] for valid inputs).
        replace: Return a new seismogram and leave the input untouched,
            instead of modifying it in place. Not supported by every
            concrete type (see [`pysmo.functions`][]).

    Returns:
        Tapered [`Seismogram`][pysmo.Seismogram] object if called with `replace=True`.

    Note: No taper below 2 samples
        If `taper_width` resolves to fewer than 2 samples, no taper is applied.
        This can occur when a very small [`Timedelta`][pandas.Timedelta] is
        provided.

    Examples:
        ```python
        >>> from pysmo.functions import taper, detrend
        >>> from pysmo.classes import SAC
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> detrend(sac_seis)
        >>> sac_seis.data
        array([   821.53861155,    661.55931267,    511.58001379, ...,
               -32931.93353333, -21859.9128322 , -10747.89213108], shape=(57465,))
        >>> taper(sac_seis, 0.2)
        >>> sac_seis.data
        array([ 0.00000000e+00,  4.94398663e-05,  1.52926246e-04, ...,
               -9.84431924e-03, -1.63364213e-03, -0.00000000e+00], shape=(57465,))
        >>>
        ```
    """

    nsamples: int
    if isinstance(taper_width, pd.Timedelta):
        nsamples = taper_width // seismogram.delta
    else:
        nsamples = floor(len(seismogram.data) * taper_width)

    if nsamples > len(seismogram.data):
        raise ValueError(
            "'taper_width' is too large. Total taper width exceeds the duration of the seismogram."
        )

    data = seismogram.data.copy() if replace else seismogram.data

    # Need at least 2 samples to apply a taper
    if nsamples >= 2:
        window = scipy.signal.windows.get_window(window_type, nsamples, fftbins=False)
        ramp_samples = nsamples // 2
        data[:ramp_samples] *= window[:ramp_samples]
        data[-ramp_samples:] *= window[-ramp_samples:]

    if replace:
        return copy.replace(seismogram, data=data)  # type: ignore[arg-type]
    return None

time2index

time2index(
    seismogram: Seismogram,
    time: Timestamp,
    allow_out_of_bounds: bool = False,
) -> int

Convert a timestamp to the corresponding data-array index.

Seismic data are sampled at discrete intervals. When a requested time does not align perfectly with a sample, this function selects the nearest index using the following rules:

  1. If the time is within 0.1% of a sample interval of an integer, it "snaps" to that integer to account for floating-point jitter.
  2. Use standard rounding (0.5 rounds up to the next index) otherwise.

Parameters:

Name Type Description Default
seismogram Seismogram

Seismogram object.

required
time Timestamp

The absolute time to convert.

required
allow_out_of_bounds bool

If True, returns the calculated index even if it falls outside the seismogram's data range [0, len-1].

False

Returns:

Type Description
int

The index of the sample closest to the provided time.

Raises:

Type Description
ValueError

If the calculated index is outside the data array and allow_out_of_bounds is False.

Source code in src/pysmo/functions/_seismogram.py
def time2index(
    seismogram: Seismogram,
    time: pd.Timestamp,
    allow_out_of_bounds: bool = False,
) -> int:
    """Convert a timestamp to the corresponding data-array index.

    Seismic data are sampled at discrete intervals. When a requested time does
    not align perfectly with a sample, this function selects the nearest
    index using the following rules:

    1. If the time is within 0.1% of a sample interval of an integer, it
       "snaps" to that integer to account for floating-point jitter.
    2. Use standard rounding (0.5 rounds up to the next index) otherwise.

    Args:
        seismogram: Seismogram object.
        time: The absolute time to convert.
        allow_out_of_bounds: If True, returns the calculated index even if it
            falls outside the seismogram's data range [0, len-1].

    Returns:
        The index of the sample closest to the provided time.

    Raises:
        ValueError: If the calculated index is outside the data array and
            `allow_out_of_bounds` is False.
    """
    # Calculate the fractional index position
    index_float = (time - seismogram.begin_time) / seismogram.delta

    # Snap to nearest integer if within a tiny tolerance (1e-3 samples).
    # This prevents 2.999999999 from being floored to 2 instead of 3.
    # rtol=0 is required: np.isclose's default rtol scales the tolerance
    # with the rounded index's magnitude, so beyond ~50,000 samples the
    # tolerance alone would exceed 0.5 and this branch would always fire,
    # silently overriding the "0.5 rounds up" rule below at exact ties.
    if np.isclose(index_float, np.round(index_float), atol=1e-3, rtol=0):
        index = int(np.round(index_float))
    # Standard rounding within the trace (0.5 rounds up)
    else:
        index = int(np.floor(index_float + 0.5))

    # Validation
    if 0 <= index < len(seismogram.data) or allow_out_of_bounds:
        return index

    raise ValueError(
        f"Calculated index {index} is out of bounds for seismogram with "
        + f"{len(seismogram.data)} samples. (Target time: {time})"
    )

window

window(
    seismogram: T,
    window_begin_time: Timestamp,
    window_end_time: Timestamp,
    ramp_width: NonNegativeTimedelta | NonNegativeNumber,
    window_type: _WindowType = "hann",
    same_shape: bool = False,
    *,
    replace: bool = False
) -> T | None

Return an optionally padded and tapered window of a seismogram.

This function combines the crop, detrend, taper, and optionally pad functions to return a 'windowed' seismogram. Its purpose is to focus on a specific time window of interest, while also (optionally) preserving the original seismogram length and tapering the signal before and after the window.

Total length exceeds the requested window

Note that the window defined by window_begin_time and window_end_time excludes the tapered sections, so the total length of the window will be the provided window length plus the tapered sections of the signal. This behaviour is a bit different from taper(), where the taper is applied to the entire signal. In a sense the tapering here is applied to the 'outside' of the region of interest rather than the 'inside'.

Parameters:

Name Type Description Default
seismogram T

Seismogram object.

required
window_begin_time Timestamp

Begin time of the window.

required
window_end_time Timestamp

End time of the window.

required
ramp_width NonNegativeTimedelta | NonNegativeNumber

Duration of the taper on each side.

  • If float: calculated as a fraction of the window length.
  • If pd.Timedelta: used as absolute duration.

Note: Total duration = window length + (2 * ramp_width).

required
window_type _WindowType

Taper method to use (see taper).

'hann'
same_shape bool

If True, pad the seismogram to its original length after windowing.

False
replace bool

Return a new seismogram and leave the input untouched, instead of modifying it in place. Not supported by every concrete type (see pysmo.functions).

False

Returns:

Type Description
T | None

Windowed Seismogram object if called with replace=True.

Raises:

Type Description
ValueError

If window_end_time is not after window_begin_time, or the ramp extends beyond the seismogram on either side.

Examples:

In this example we focus on a window starting 600 seconds after the begin_time of the seismogram and lasting for 1200 seconds. Setting the ramp_width to 300 seconds means that the actual window will start 300 seconds earlier and end 300 seconds later than the specified window begin and end times.

>>> from pysmo.functions import window, detrend
>>> from pysmo.classes import MSeed
>>> from pysmo.tools.plotutils import plotseis
>>> import pandas as pd
>>>
>>> seis = MSeed.from_file("example.mseed")
>>> ramp_width = pd.Timedelta(seconds=300)
>>> window_begin_time = seis.begin_time + pd.Timedelta(seconds=600)
>>> window_end_time = window_begin_time + pd.Timedelta(seconds=1200)
>>> windowed_seis = window(seis, window_begin_time, window_end_time, ramp_width, same_shape=True, replace=True)
>>> detrend(seis)
>>> fig = plotseis(seis, windowed_seis)
>>>
A tapered window applied to a seismogram, shown over the original trace. A tapered window applied to a seismogram, shown over the original trace.
Source code in src/pysmo/functions/_seismogram.py
def window[T: Seismogram](
    seismogram: T,
    window_begin_time: pd.Timestamp,
    window_end_time: pd.Timestamp,
    ramp_width: NonNegativeTimedelta | NonNegativeNumber,
    window_type: _WindowType = "hann",
    same_shape: bool = False,
    *,
    replace: bool = False,
) -> T | None:
    """Return an optionally padded and tapered window of a seismogram.

    This function combines the [`crop`][pysmo.functions.crop],
    [`detrend`][pysmo.functions.detrend], [`taper`][pysmo.functions.taper], and
    optionally [`pad`][pysmo.functions.pad] functions to return a 'windowed'
    seismogram. Its purpose is to focus on a specific time window of interest,
    while also (optionally) preserving the original seismogram length and
    tapering the signal before and after the window.

    Tip: Total length exceeds the requested window
        Note that the window defined by `window_begin_time` and
        `window_end_time` *excludes* the tapered sections, so the total length
        of the window will be the provided window length plus the tapered
        sections of the signal. This behaviour is a bit different from
        [`taper()`][pysmo.functions.taper], where the taper is applied to the
        entire signal. In a sense the tapering here is applied to the 'outside'
        of the region of interest rather than the 'inside'.

    Args:
        seismogram: Seismogram object.
        window_begin_time: Begin time of the window.
        window_end_time: End time of the window.
        ramp_width: Duration of the taper on *each side*.

            - If `float`: calculated as a fraction of the window length.
            - If `pd.Timedelta`: used as absolute duration.

            Note: Total duration = window length + (2 * `ramp_width`).
        window_type: Taper method to use (see [`taper`][pysmo.functions.taper]).
        same_shape: If True, pad the seismogram to its original length after
            windowing.
        replace: Return a new seismogram and leave the input untouched,
            instead of modifying it in place. Not supported by every
            concrete type (see [`pysmo.functions`][]).

    Returns:
        Windowed [`Seismogram`][pysmo.Seismogram] object if called with `replace=True`.

    Raises:
        ValueError: If `window_end_time` is not after `window_begin_time`, or
            the ramp extends beyond the seismogram on either side.

    Examples:
        In this example we focus on a window starting 600 seconds after the
        `begin_time` of the seismogram and lasting for 1200 seconds. Setting the
        `ramp_width` to 300 seconds means that the actual window will start 300
        seconds earlier and end 300 seconds later than the specified window
        begin and end times.

        ```python
        >>> from pysmo.functions import window, detrend
        >>> from pysmo.classes import MSeed
        >>> from pysmo.tools.plotutils import plotseis
        >>> import pandas as pd
        >>>
        >>> seis = MSeed.from_file("example.mseed")
        >>> ramp_width = pd.Timedelta(seconds=300)
        >>> window_begin_time = seis.begin_time + pd.Timedelta(seconds=600)
        >>> window_end_time = window_begin_time + pd.Timedelta(seconds=1200)
        >>> windowed_seis = window(seis, window_begin_time, window_end_time, ramp_width, same_shape=True, replace=True)
        >>> detrend(seis)
        >>> fig = plotseis(seis, windowed_seis)
        >>>
        ```

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

        <figure markdown="span">
        ![A tapered window applied to a seismogram, shown over the original trace.](../../images/sybil/functions_window.png#only-light){ loading=lazy }
        ![A tapered window applied to a seismogram, shown over the original trace.](../../images/sybil/functions_window-dark.png#only-dark){ loading=lazy }
        </figure>
    """

    begin_time, end_time = seismogram.begin_time, seismogram.end_time

    if window_end_time <= window_begin_time:
        raise ValueError("window_end_time must be after window_begin_time.")

    window_duration = window_end_time - window_begin_time
    ramp_duration = (
        ramp_width
        if isinstance(ramp_width, pd.Timedelta)
        else ramp_width * window_duration  # ty: ignore[unsupported-operator]
    )

    if window_begin_time - ramp_duration < seismogram.begin_time:
        raise ValueError(
            f"ramp_width={ramp_width} requires data before {window_begin_time - ramp_duration}, "
            + f"but seismogram only starts at {seismogram.begin_time}."
        )
    if window_end_time + ramp_duration > seismogram.end_time:
        raise ValueError(
            f"ramp_width={ramp_width} requires data after {window_end_time + ramp_duration}, "
            + f"but seismogram only ends at {seismogram.end_time}."
        )

    window_begin_time -= ramp_duration
    window_end_time += ramp_duration

    # replace=True is threaded through every step so the chain never mutates an
    # object in place; a step run without it would fail on an immutable type.
    if replace:
        seismogram = crop(seismogram, window_begin_time, window_end_time, replace=True)
        seismogram = detrend(seismogram, replace=True)
        seismogram = taper(
            seismogram,
            taper_width=ramp_duration * 2,
            window_type=window_type,
            replace=True,
        )
        if same_shape is True:
            seismogram = pad(seismogram, begin_time, end_time, replace=True)
        return seismogram

    crop(seismogram, window_begin_time, window_end_time)
    detrend(seismogram)
    taper(seismogram, taper_width=ramp_duration * 2, window_type=window_type)
    if same_shape is True:
        pad(seismogram, begin_time, end_time)
    return None