Skip to content

Documents · Verb reference

duration_synchrony

Compare durations of one-to-one matched events.

Callable type: Grammar verb / pipe stage · Browse: Synchrony and comparison

Usage

from cubedynamics import verbs as v
v.duration_synchrony(*, match_on='overlap', match_tolerance='7D', method='spearman', spatial_mode='neighbors', radius_km=None, k_neighbors=None, reference=None, min_matched_events=3)

Arguments

Argument Meaning Default
match_on See implementation docstring below; no parameter-specific description supplied. 'overlap'
match_tolerance See implementation docstring below; no parameter-specific description supplied. '7D'
method See implementation docstring below; no parameter-specific description supplied. 'spearman'
spatial_mode See implementation docstring below; no parameter-specific description supplied. 'neighbors'
radius_km See implementation docstring below; no parameter-specific description supplied. None
k_neighbors See implementation docstring below; no parameter-specific description supplied. None
reference See implementation docstring below; no parameter-specific description supplied. None
min_matched_events See implementation docstring below; no parameter-specific description supplied. 3

Accepts

EventResult -> synchrony Dataset with duration and match diagnostics.

Returns

EventResult -> synchrony Dataset with duration and match diagnostics.

Order / grammar behavior

EventResult -> synchrony Dataset with duration and match diagnostics.

Minimal example

Run from the repository root after python -m pip install -e '.[vignettes]'. Uses the checked observational PRISM fixture; no network is required.

from pathlib import Path
import xarray as xr
import matplotlib.pyplot as plt
from cubedynamics import pipe, verbs as v

# Frozen, reviewed PRISM observations; run from the repository root.
path = Path("tests/fixtures/real_data/prism_boulder_january_2024.nc")
with xr.open_dataset(path, engine="scipy") as observed:
    cube = observed["tmax"].load()
assert cube.attrs["units"] == "degC"

events = (pipe(cube) | v.threshold_state(threshold=0, direction="below") | v.detect_events()).unwrap()
result = (pipe(events) | v.duration_synchrony(spatial_mode="reference", reference="center", min_matched_events=1)).unwrap()
# A one-month sample can leave correlations undefined; inspect counts as well.
print(result)
result["duration_similarity"].squeeze().plot()
plt.show()

Works with

EventResult -> synchrony Dataset with duration and match diagnostics.

See also

Implementation notes

No additional implementation notes in the current docstring.

Implementation source. Signatures and descriptions on this page are generated from this checkout, not hand-maintained copies.