Skip to content

Documents · Verb reference

timing_synchrony

Measure whether one-to-one matched events happen at similar times.

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

Usage

from cubedynamics import verbs as v
v.timing_synchrony(*, event_anchor='start', match_tolerance='7D', score='exponential', timescale='3D', spatial_mode='neighbors', radius_km=None, k_neighbors=None, reference=None)

Arguments

Argument Meaning Default
event_anchor See implementation docstring below; no parameter-specific description supplied. 'start'
match_tolerance See implementation docstring below; no parameter-specific description supplied. '7D'
score See implementation docstring below; no parameter-specific description supplied. 'exponential'
timescale See implementation docstring below; no parameter-specific description supplied. '3D'
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

Accepts

EventResult -> synchrony Dataset with lag and unmatched-event diagnostics.

Returns

EventResult -> synchrony Dataset with lag and unmatched-event diagnostics.

Order / grammar behavior

EventResult -> synchrony Dataset with lag and unmatched-event 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.timing_synchrony(spatial_mode="reference", reference="center")).unwrap()
result["timing_synchrony"].squeeze().plot()
plt.show()

Works with

EventResult -> synchrony Dataset with lag and unmatched-event 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.