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 |
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 |
required |
update
|
dict[str, Any] | None
|
Update or add attributes in the returned |
None
|
Returns:
| Type | Description |
|---|---|
TMini
|
A new |
Raises:
| Type | Description |
|---|---|
AttributeError
|
If the |
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
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 |
See Also
clone_to_mini: Create a Mini
instance from a compatible object.
Source code in src/pysmo/functions/_utils.py
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
|
|
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 |
False
|
Returns:
| Type | Description |
|---|---|
T | None
|
Cropped |
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
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 |
False
|
Returns:
| Type | Description |
|---|---|
T | None
|
Detrended |
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
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 |
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
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 |
required |
delta
|
PositiveTimedelta | None
|
Sampling interval to resample all non-empty seismograms to
before merging. If |
None
|
auto_delta
|
bool
|
Estimate a common sampling interval from the non-empty
seismograms with
|
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 |
False
|
Returns:
| Type | Description |
|---|---|
T | None
|
Merged |
T | None
|
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If both |
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
534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 | |
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
|
t2
|
Timestamp | None
|
End of the window used to find the maximum. If |
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 |
False
|
Returns:
| Type | Description |
|---|---|
T | None
|
Normalised |
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
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
|
|
required |
begin_time
|
Timestamp
|
New begin time. |
required |
end_time
|
Timestamp
|
New end time. |
required |
mode
|
_ModeKind | _ModeFunc
|
Pad mode to use (see |
'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 |
False
|
kwargs
|
Any
|
Keyword arguments to pass to |
{}
|
Returns:
| Type | Description |
|---|---|
T | None
|
Padded |
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
277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 | |
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 |
False
|
Returns:
| Type | Description |
|---|---|
T | None
|
Resampled |
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
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
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
|
required |
cls
|
type[Seismogram] | None
|
The attrs class to rebuild, returned as its own type. When
|
None
|
trusted_modules
|
tuple[str, ...]
|
Top-level packages the recorded class may be imported
from when |
_DEFAULT_TRUSTED_MODULES
|
Returns:
| Type | Description |
|---|---|
Seismogram
|
A new instance of |
Raises:
| Type | Description |
|---|---|
TypeError
|
If the document is malformed or an unsupported version, if
|
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
309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 | |
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 |
False
|
Returns:
| Type | Description |
|---|---|
bytes
|
The UTF-8 JSON document. |
Raises:
| Type | Description |
|---|---|
TypeError
|
If |
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
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 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 | |
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
|
'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 |
False
|
Returns:
| Type | Description |
|---|---|
T | None
|
Tapered |
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
816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 | |
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:
- 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.
- 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
|
Source code in src/pysmo/functions/_seismogram.py
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.
Note: Total duration = window length + (2 * |
required |
window_type
|
_WindowType
|
Taper method to use (see |
'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 |
False
|
Returns:
| Type | Description |
|---|---|
T | None
|
Windowed |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
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)
>>>
Source code in src/pysmo/functions/_seismogram.py
990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 | |