Skip to content

pysmo.functions #

Simple operations using pysmo types.

This module provides functions that perform common operations using pysmo types (mostly Seismogram). They are meant to be building blocks that can be used to construct more complex processing algorithms.

Many functions have a clone argument that controls whether the function should operate on the input directly, or first create a clone of it (using deepcopy) and return the clone after using it for the the function. For example:

>>> from pysmo.functions import resample
>>> from pysmo.classes import SAC
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> new_delta = sac_seis.delta * 2
>>>
>>> # create a clone and modify data in clone instead of sac_seis:
>>> new_sac_seis = resample(sac_seis, new_delta, clone=True)
>>>
>>> # modify data in sac_seis directly:
>>> resample(sac_seis, new_delta)
>>>
>>> # because the deepcopy operation can be computationaly expensive,
>>> # you should NOT use the following pattern:
>>> sac_seis = resample(sac_seis, new_delta, clone=True)
>>>
Hint

Additional functions may be found in pysmo.tools.

Functions:

  • clone_to_mini –

    Create a new instance of a Mini class from a matching other one.

  • copy_from_mini –

    Copy attributes from a Mini instance to matching other one.

  • crop –

    Shorten a seismogram by providing new begin and end times.

  • detrend –

    Remove linear and/or constant trends from a seismogram.

  • normalize –

    Normalize a seismogram with its absolute max value.

  • pad –

    Pad a seismogram with zeros to new begin and end times.

  • resample –

    Resample Seismogram data using the Fourier method.

  • taper –

    Apply a symetric taper to the ends of a Seismogram.

  • time2index –

    Retuns data index corresponding to a given time.

clone_to_mini #

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

Create a new instance of a Mini class from a matching other one.

This function is creates a clone of an exising class by copying the attributes defined in mini_cls from the source to the target. Attributes only present in the source object are ignored, potentially resulting in a smaller and more performant object.

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:

  • mini_cls (type[clone_to_mini[TMini]]) –

    The type of Mini class to create.

  • source (_AnyProto) –

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

  • update (dict | None, default: None ) –

    Update or add attributes in the returned mini_cls instance.

Returns:

  • clone_to_mini[TMini] –

    A new Mini instance type mini_cls.

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 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
>>> from datetime import datetime, timezone
>>> now = datetime.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 to matching other one.

Source code in pysmo/functions/_utils.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def clone_to_mini[TMini: _AnyMini](
    mini_cls: type[TMini], source: _AnyProto, update: dict | None = None
) -> TMini:
    """Create a new instance of a Mini class from a matching other one.

    This function is creates a clone of an exising class by
    [copying][copy.copy] the attributes defined in `mini_cls` from the source
    to the target. Attributes only present in the source object are ignored,
    potentially resulting in a smaller and more performant object.

    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:
        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 instance type mini_cls.

    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
        >>> from datetime import datetime, timezone
        >>> now = datetime.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 to matching other one.
    """

    update = update or dict()

    if all(
        map(
            lambda x: hasattr(source, x.name)
            or x.name in update
            or x.default is not NOTHING,
            fields(mini_cls),
        )
    ):
        clone_dict = {
            attr.name: (
                update[attr.name]
                if attr.name in update
                else copy(getattr(source, attr.name, attr.default))
            )
            for attr in fields(mini_cls)
        }
        # TODO: why do we need cast here for mypy?
        return cast(TMini, mini_cls(**clone_dict))

    raise AttributeError(
        f"Unable to create clone: {source} not compatible with {mini_cls}."
    )

copy_from_mini #

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

Copy attributes from a Mini instance to matching other one.

This function copies all attributes in the source Mini class instance to a compatible target instance.

Parameters:

  • source (_AnyMini) –

    The Mini instance to copy attributes from.

  • target (_AnyProto) –

    Compatible target instance.

  • update (dict | None, default: None ) –

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

See Also

clone_to_mini: Create a new instance of a Mini class from a matching other one.

Source code in pysmo/functions/_utils.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def copy_from_mini(
    source: _AnyMini, target: _AnyProto, update: dict | None = None
) -> None:
    """Copy attributes from a Mini instance to matching other one.

    This function [copies][copy.copy] all attributes in the `source` Mini class
    instance to a compatible `target` instance.

    Parameters:
        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 new
        instance of a Mini class from a matching other one.
    """

    update = update or dict()

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

    if all(map(lambda x: hasattr(target, x), attributes)):
        for attribute in attributes:
            if attribute in update:
                setattr(target, attribute, update[attribute])
            else:
                setattr(target, attribute, copy(getattr(source, attribute)))
    else:
        raise AttributeError(
            f"Unable to copy to target: {type(target)} not compatible with {type(source)}."
        )

crop #

crop(
    seismogram: crop[T],
    begin_time: datetime,
    end_time: datetime,
    clone: bool = False,
) -> None | crop[T]

Shorten a seismogram by providing 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:

  • seismogram (crop[T]) –

    Seismogram object.

  • begin_time (datetime) –

    New begin time.

  • end_time (datetime) –

    New end time.

  • clone (bool, default: False ) –

    Operate on a clone of the input seismogram.

Returns:

  • None | crop[T] –

    Cropped Seismogram object if called with clone=True.

Raises:

  • ValueError –

    If new begin time is after new end time.

Examples:

>>> from pysmo.functions import crop
>>> from pysmo.classes import SAC
>>> from datetime import timedelta
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> new_begin_time = sac_seis.begin_time + timedelta(seconds=10)
>>> new_end_time = sac_seis.end_time - timedelta(seconds=10)
>>> crop(sac_seis, new_begin_time, new_end_time)
>>>
Source code in pysmo/functions/_seismogram.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def crop[T: Seismogram](
    seismogram: T, begin_time: datetime, end_time: datetime, clone: bool = False
) -> None | T:
    """Shorten a seismogram by providing 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`.

    Parameters:
        seismogram: [`Seismogram`][pysmo.Seismogram] object.
        begin_time: New begin time.
        end_time: New end time.
        clone: Operate on a clone of the input seismogram.

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

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

    Examples:
        ```python
        >>> from pysmo.functions import crop
        >>> from pysmo.classes import SAC
        >>> from datetime import timedelta
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> new_begin_time = sac_seis.begin_time + timedelta(seconds=10)
        >>> new_end_time = sac_seis.end_time - 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)

    if clone is True:
        seismogram = deepcopy(seismogram)

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

    if clone is True:
        return seismogram

    return None

detrend #

detrend(
    seismogram: detrend[T], clone: bool = False
) -> None | detrend[T]

Remove linear and/or constant trends from a seismogram.

Parameters:

  • seismogram (detrend[T]) –

    Seismogram object.

  • clone (bool, default: False ) –

    Operate on a clone of the input seismogram.

Returns:

  • None | detrend[T] –

    Detrended Seismogram object if called with clone=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-11)
np.False_
>>> detrend(sac_seis)
>>> 0 == pytest.approx(np.mean(sac_seis.data), abs=1e-11)
np.True_
>>>
Source code in pysmo/functions/_seismogram.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def detrend[T: Seismogram](seismogram: T, clone: bool = False) -> None | T:
    """Remove linear and/or constant trends from a seismogram.

    Parameters:
        seismogram: Seismogram object.
        clone: Operate on a clone of the input seismogram.

    Returns:
        Detrended [`Seismogram`][pysmo.Seismogram] object if called with `clone=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-11)
        np.False_
        >>> detrend(sac_seis)
        >>> 0 == pytest.approx(np.mean(sac_seis.data), abs=1e-11)
        np.True_
        >>>
        ```
    """
    if clone is True:
        seismogram = deepcopy(seismogram)

    seismogram.data = scipy.signal.detrend(seismogram.data)

    if clone is True:
        return seismogram

    return None

normalize #

normalize(
    seismogram: normalize[T],
    t1: datetime | None = None,
    t2: datetime | None = None,
    clone: bool = False,
) -> None | normalize[T]

Normalize a seismogram with its absolute max value.

Parameters:

  • seismogram (normalize[T]) –

    Seismogram object.

  • t1 (datetime | None, default: None ) –

    Optionally restrict searching of maximum to time after this time.

  • t2 (datetime | None, default: None ) –

    Optionally restrict searching of maximum to time before this time.

  • clone (bool, default: False ) –

    Operate on a clone of the input seismogram.

Returns:

  • None | normalize[T] –

    Normalized Seismogram object if clone=True.

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 pysmo/functions/_seismogram.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def normalize[T: Seismogram](
    seismogram: T,
    t1: datetime | None = None,
    t2: datetime | None = None,
    clone: bool = False,
) -> None | T:
    """Normalize a seismogram with its absolute max value.

    Parameters:
        seismogram: Seismogram object.
        t1: Optionally restrict searching of maximum to time after this time.
        t2: Optionally restrict searching of maximum to time before this time.
        clone: Operate on a clone of the input seismogram.

    Returns:
        Normalized [`Seismogram`][pysmo.Seismogram] object if `clone=True`.

    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_
        >>>
        ```
    """

    if clone is True:
        seismogram = deepcopy(seismogram)

    start_index, end_index = None, None

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

    if t2 is not None:
        end_index = time2index(seismogram, t2)

    seismogram.data /= np.max(np.abs(seismogram.data[start_index:end_index]))

    if clone is True:
        return seismogram

    return None

pad #

pad(
    seismogram: pad[T],
    begin_time: datetime,
    end_time: datetime,
    clone: bool = False,
) -> None | pad[T]

Pad a seismogram with zeros to new begin and end times.

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:

  • seismogram (pad[T]) –

    Seismogram object.

  • begin_time (datetime) –

    New begin time.

  • end_time (datetime) –

    New end time.

  • clone (bool, default: False ) –

    Operate on a clone of the input seismogram.

Returns:

  • None | pad[T] –

    Padded Seismogram object if called with clone=True.

Raises:

  • ValueError –

    If new begin time is after new end time.

Examples:

>>> from pysmo.functions import pad
>>> from pysmo.classes import SAC
>>> from datetime import timedelta
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> original_length = len(sac_seis)
>>> sac_seis.data
array([2302., 2313., 2345., ..., 2836., 2772., 2723.], shape=(180000,))
>>> new_begin_time = sac_seis.begin_time - timedelta(seconds=10)
>>> new_end_time = sac_seis.end_time + timedelta(seconds=10)
>>> pad(sac_seis, new_begin_time, new_end_time)
>>> len(sac_seis) == original_length + 20 * (1 / sac_seis.delta.total_seconds())
True
>>> sac_seis.data
array([0., 0., 0., ..., 0., 0., 0.], shape=(181000,))
>>>
Source code in pysmo/functions/_seismogram.py 
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
def pad[T: Seismogram](
    seismogram: T, begin_time: datetime, end_time: datetime, clone: bool = False
) -> None | T:
    """Pad a seismogram with zeros 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
    pads the [`data`][pysmo.Seismogram.data] array using
    [`numpy.pad`][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.

    Parameters:
        seismogram: [`Seismogram`][pysmo.Seismogram] object.
        begin_time: New begin time.
        end_time: New end time.
        clone: Operate on a clone of the input seismogram.

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

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

    Examples:
        ```python
        >>> from pysmo.functions import pad
        >>> from pysmo.classes import SAC
        >>> from datetime import timedelta
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> original_length = len(sac_seis)
        >>> sac_seis.data
        array([2302., 2313., 2345., ..., 2836., 2772., 2723.], shape=(180000,))
        >>> new_begin_time = sac_seis.begin_time - timedelta(seconds=10)
        >>> new_end_time = sac_seis.end_time + timedelta(seconds=10)
        >>> pad(sac_seis, new_begin_time, new_end_time)
        >>> len(sac_seis) == original_length + 20 * (1 / sac_seis.delta.total_seconds())
        True
        >>> sac_seis.data
        array([0., 0., 0., ..., 0., 0., 0.], shape=(181000,))
        >>>
        ```
    """

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

    start_index = time2index(
        seismogram, begin_time, method="floor", allow_out_of_bounds=True
    )
    end_index = time2index(
        seismogram, end_time, method="ceil", allow_out_of_bounds=True
    )

    if clone is True:
        seismogram = deepcopy(seismogram)

    pad_before = max(0, -start_index)
    pad_after = max(0, end_index - (len(seismogram) - 1))

    if pad_before > 0 or pad_after > 0:
        seismogram.data = np.pad(
            seismogram.data, (pad_before, pad_after), mode="constant", constant_values=0
        )
        seismogram.begin_time += seismogram.delta * min(0, start_index)

    if clone is True:
        return seismogram

    return None

resample #

resample(
    seismogram: resample[T],
    delta: timedelta,
    clone: bool = False,
) -> None | resample[T]

Resample Seismogram data using the Fourier method.

This function uses scipy.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:

  • seismogram (resample[T]) –

    Seismogram object.

  • delta (timedelta) –

    New sampling interval.

  • clone (bool, default: False ) –

    Operate on a clone of the input seismogram.

Returns:

  • None | resample[T] –

    Resampled Seismogram object if called with clone=True.

Examples:

>>> from pysmo.functions import resample
>>> from pysmo.classes import SAC
>>> sac_seis = SAC.from_file("example.sac").seismogram
>>> len(sac_seis)
180000
>>> original_delta = sac_seis.delta
>>> new_delta = original_delta * 2
>>> resample(sac_seis, new_delta)
>>> len(sac_seis)
90000
>>>
Source code in pysmo/functions/_seismogram.py 
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
def resample[T: Seismogram](
    seismogram: T, delta: timedelta, clone: bool = False
) -> None | T:
    """Resample Seismogram data using the Fourier method.

    This function uses [`scipy.resample`][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:
        seismogram: Seismogram object.
        delta: New sampling interval.
        clone: Operate on a clone of the input seismogram.

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

    Examples:
        ```python
        >>> from pysmo.functions import resample
        >>> from pysmo.classes import SAC
        >>> sac_seis = SAC.from_file("example.sac").seismogram
        >>> len(sac_seis)
        180000
        >>> original_delta = sac_seis.delta
        >>> new_delta = original_delta * 2
        >>> resample(sac_seis, new_delta)
        >>> len(sac_seis)
        90000
        >>>
        ```
    """
    if clone is True:
        seismogram = deepcopy(seismogram)

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

    if clone is True:
        return seismogram

    return None

taper #

taper(
    seismogram: taper[T],
    taper_width: timedelta | float,
    taper_method: Literal[
        "bartlett",
        "blackman",
        "hamming",
        "hanning",
        "kaiser",
    ] = "hanning",
    beta: float = 14.0,
    left: bool = True,
    right: bool = True,
    clone: bool = False,
) -> None | taper[T]

Apply a symetric taper to the ends of a Seismogram.

The taper() function applies a taper to the data at one or both ends of a Seismogram object. The width of this taper can be provided as either positive timedelta or as a fraction of the total seismogram length. In both cases the total width of the taper (i.e. left and right side combined) should not exceed the length of the seismogram.

Different methods for calculating the shape of the taper may be specified. They are all derived from the corresponding numpy window functions:

Parameters:

  • seismogram (taper[T]) –

    Seismogram object.

  • taper_width (timedelta | float) –

    With of the taper to use.

  • taper_method (Literal['bartlett', 'blackman', 'hamming', 'hanning', 'kaiser'], default: 'hanning' ) –

    Taper method to use.

  • beta (float, default: 14.0 ) –

    beta value for the Kaiser window function (ignored for other methods).

  • left (bool, default: True ) –

    Apply taper to the left side of the seismogram.

  • right (bool, default: True ) –

    Apply taper to the right side of the seismogram.

  • clone (bool, default: False ) –

    Operate on a clone of the input seismogram.

Returns:

  • None | taper[T] –

    Tapered Seismogram object if called with clone=True.

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([ 95.59652208, 106.59521819, 138.59391429, ..., 394.90004126,
       330.89873737, 281.89743348], shape=(180000,))
>>> taper(sac_seis, 0.1)
>>> sac_seis.data
array([0.00000000e+00, 8.11814104e-07, 4.22204657e-06, ...,
       1.20300114e-05, 2.52007798e-06, 0.00000000e+00], shape=(180000,))
>>>
Source code in pysmo/functions/_seismogram.py 
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
def taper[T: Seismogram](
    seismogram: T,
    taper_width: timedelta | float,
    taper_method: Literal[
        "bartlett", "blackman", "hamming", "hanning", "kaiser"
    ] = "hanning",
    beta: float = 14.0,
    left: bool = True,
    right: bool = True,
    clone: bool = False,
) -> None | T:
    """Apply a symetric taper to the ends of a Seismogram.

    The [`taper()`][pysmo.functions.taper] function applies a taper to the data
    at one or both ends of a [`Seismogram`][pysmo.Seismogram] object. The width
    of this taper can be provided as either positive
    [`timedelta`][datetime.timedelta] or as a fraction of the total seismogram
    length. In both cases the total width of the taper (i.e. left and right
    side combined) should not exceed the length of the seismogram.

    Different methods for calculating the shape of the taper may be specified.
    They are all derived from the corresponding `numpy` window functions:

    - [`numpy.bartlett`][numpy.bartlett]
    - [`numpy.blackman`][numpy.blackman]
    - [`numpy.hamming`][numpy.hamming]
    - [`numpy.hanning`][numpy.hanning]
    - [`numpy.kaiser`][numpy.kaiser]

    Parameters:
        seismogram: Seismogram object.
        taper_width: With of the taper to use.
        taper_method: Taper method to use.
        beta: beta value for the Kaiser window function (ignored for other methods).
        left: Apply taper to the left side of the seismogram.
        right: Apply taper to the right side of the seismogram.
        clone: Operate on a clone of the input seismogram.

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

    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([ 95.59652208, 106.59521819, 138.59391429, ..., 394.90004126,
               330.89873737, 281.89743348], shape=(180000,))
        >>> taper(sac_seis, 0.1)
        >>> sac_seis.data
        array([0.00000000e+00, 8.11814104e-07, 4.22204657e-06, ...,
               1.20300114e-05, 2.52007798e-06, 0.00000000e+00], shape=(180000,))
        >>>
        ```
    """

    def calc_window_data(window_length: int) -> npt.NDArray:
        if taper_method == "bartlett":
            return np.bartlett(window_length)
        elif taper_method == "blackman":
            return np.blackman(window_length)
        elif taper_method == "hamming":
            return np.hamming(window_length)
        elif taper_method == "hanning":
            return np.hanning(window_length)
        elif taper_method == "kaiser":
            return np.kaiser(window_length, beta)

    @singledispatch
    def calc_samples(taper_width: Any) -> int:
        raise TypeError(f"Unsupported type for 'taper_width': {type(taper_width)}")

    @calc_samples.register(float)
    def _(taper_width: float) -> int:
        return floor(len(seismogram) * taper_width)

    @calc_samples.register(timedelta)
    def _(taper_width: timedelta) -> int:
        return floor(taper_width / seismogram.delta) + 1

    if clone is True:
        seismogram = deepcopy(seismogram)

    if left is False and right is False:
        return seismogram if clone is True else None

    nsamples = calc_samples(taper_width)

    if nsamples * (left + right) > len(seismogram):
        raise ValueError(
            "'taper_width' is too large. Total taper width may exceed the length of the seismogram."
        )

    if nsamples > 0:
        taper_data = np.ones(len(seismogram))
        window = calc_window_data(nsamples * 2)
        if left is True:
            taper_data[:nsamples] = window[:nsamples]
        if right is True:
            taper_data[-nsamples:] = window[-nsamples:]
        seismogram.data *= taper_data

    if clone is True:
        return seismogram

    return None

time2index #

time2index(
    seismogram: Seismogram,
    time: datetime,
    method: Literal["ceil", "floor", "round"] = "round",
    allow_out_of_bounds: bool = False,
) -> int

Retuns data index corresponding to a given time.

This function converts time to index of a seismogram's data array. In most cases the time will not have an exact match in the data array. This function allows choosing how to select the index to return when that is the case with the method parameter:

  • round: round to nearest index.
  • ceil: always round up to next higher index.
  • floor: always round down to next lower index.

Parameters:

  • seismogram (Seismogram) –

    Seismogram object.

  • time (datetime) –

    Time to convert to index.

  • method (Literal['ceil', 'floor', 'round'], default: 'round' ) –

    Method to use for selecting the index to return.

  • allow_out_of_bounds (bool, default: False ) –

    If True, allow returning an index that is outside has no corresponding data point in the seismogram.

Returns:

  • int –

    Index of the sample corresponding to the given time.

Raises:

  • ValueError –

    If the calculated index is out of bounds and allow_out_of_bounds is not set to True.

Source code in pysmo/functions/_seismogram.py 
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
def time2index(
    seismogram: Seismogram,
    time: datetime,
    method: Literal["ceil", "floor", "round"] = "round",
    allow_out_of_bounds: bool = False,
) -> int:
    """Retuns data index corresponding to a given time.

    This function converts time to index of a seismogram's data array. In most
    cases the time will not have an exact match in the data array. This
    function allows choosing how to select the index to return when that is
    the case with the method parameter:

    - round: round to nearest index.
    - ceil: always round up to next higher index.
    - floor: always round down to next lower index.

    Parameters:
        seismogram: Seismogram object.
        time: Time to convert to index.
        method: Method to use for selecting the index to return.
        allow_out_of_bounds: If True, allow returning an index that is outside
            has no corresponding data point in the seismogram.

    Returns:
        Index of the sample corresponding to the given time.

    Raises:
        ValueError: If the calculated index is out of bounds and
            `allow_out_of_bounds` is not set to True.
    """

    if method == "ceil":
        index = ceil((time - seismogram.begin_time) / seismogram.delta)

    elif method == "floor":
        index = floor((time - seismogram.begin_time) / seismogram.delta)

    elif method == "round":
        index = round((time - seismogram.begin_time) / seismogram.delta)

    else:
        raise ValueError(
            "Invalid method provided. Valid options are 'ceil', 'floor', and 'round'."
        )

    if 0 <= index < len(seismogram) or allow_out_of_bounds is True:
        return index

    raise ValueError(f"Invalid time provided, calculated {index=} is out of bounds.")