pysmo.tools.project

@define(kw_only=True, frozen=True)
class FetchContext[TStation: Station, TEvent: Event]:
    """Context passed to `seismogram_transform` with the downloaded seismogram.

    Bundles the originating [`ProjectEntry`][pysmo.tools.project.ProjectEntry]
    with what this specific fetch resolved but that doesn't belong on
    `ProjectEntry` itself. Recomputed fresh on every fetch, never persisted
    (unlike `entry.checksum`, which is deliberately pinned).

    Note the deliberate naming overlap with `entry.starttime`/`entry.endtime`:
    those are the entry's possibly-`None` *explicit override* (see
    [`ProjectEntry`][pysmo.tools.project.ProjectEntry]), while
    `starttime`/`endtime` here are always-populated and reflect the window
    that was *actually used*: identical to the entry's own when an explicit
    override was given, resolved by the project's `window` otherwise. A
    transform wanting "the window this fetch actually covered" should read
    `context.starttime`/`context.endtime`, not
    `context.entry.starttime`/`context.entry.endtime`.
    """

    entry: ProjectEntry[TStation, TEvent]
    """The entry this seismogram was fetched for."""

    starttime: pd.Timestamp
    """Absolute start of the window actually used for this fetch."""

    endtime: pd.Timestamp
    """Absolute end of the window actually used for this fetch."""

    reference: pd.Timestamp | None
    """Timestamp the window was placed around (a predicted phase arrival for
    the default `window`), or `None` if `entry.starttime`/`entry.endtime`
    were used directly."""

@define(kw_only=True, frozen=True)
class PhaseWindow:
    """Resolve a fetch window from an entry's event and a predicted phase arrival.

    The default [`WindowResolver`][pysmo.tools.project.WindowResolver] for
    [`PysmoProject`][pysmo.tools.project.PysmoProject]: predicts the `phase`
    arrival for the station/event geometry with `travel_time_backend`, then
    returns the window `[arrival + pre_pick, arrival + post_pick]`.

    Frozen and picklable by value (given a picklable `travel_time_backend`),
    so it travels with a pickled `PysmoProject`.
    """

    phase: str = field(default="P")
    """Seismic phase the window is placed around.

    With the default `travel_time_backend` this must be one of the phases in
    [`Phase`][pysmo.tools.traveltime.Phase]; any other name raises when a
    window is resolved. A custom backend may accept a wider set.
    """

    pre_pick: NonPositiveTimedelta = field(
        default=pd.Timedelta(minutes=-2),
        converter=to_timedelta,
        validator=[
            validators.instance_of(pd.Timedelta),
            validators.le(pd.Timedelta(0)),
        ],
    )
    """Offset from the predicted arrival to the window start; zero or negative."""

    post_pick: PositiveTimedelta = field(
        default=pd.Timedelta(minutes=8),
        converter=to_timedelta,
        validator=[
            validators.instance_of(pd.Timedelta),
            validators.gt(pd.Timedelta(0)),
        ],
    )
    """Offset from the predicted arrival to the window end. Must be positive."""

    travel_time_backend: TravelTimeBackend = field(default=builtin_backend)
    """Predicts the phase arrival the window is built around.

    Defaults to pysmo's built-in solver,
    [`travel_times`][pysmo.tools.traveltime.travel_times]. Any callable of
    the same shape
    ([`TravelTimeBackend`][pysmo.tools.traveltime.TravelTimeBackend]) can
    take its place, for another velocity model, a phase the built-in solver
    does not cover, or arrival times from an external source. Must be
    picklable: a top-level function, a [`functools.partial`][] of one, or an
    attrs instance with only picklable fields; not a lambda or closure.
    """

    def __call__[TS: Station, TE: Event](
        self, entry: ProjectEntry[TS, TE]
    ) -> WindowResult:
        """Resolve the window for `entry`.

        Raises:
            ValueError: If `entry` has no event, or no `phase` arrival is
                predicted for its station/event geometry.
        """
        if entry.event is None:
            raise ValueError(
                "PhaseWindow needs an entry with an event to derive a window."
            )
        distance = haversine(entry.event, entry.station)
        arrivals = self.travel_time_backend(
            depth=entry.event.depth, distance=distance, phases=[self.phase]
        )
        if self.phase not in arrivals:
            raise ValueError(
                f"No {self.phase!r} arrival predicted for "
                + f"{entry.station.network}.{entry.station.name} at this "
                + "distance/depth."
            )
        reference = entry.event.time + arrivals[self.phase]
        return WindowResult(
            starttime=reference + self.pre_pick,
            endtime=reference + self.post_pick,
            reference=reference,
        )

@define(kw_only=True)
class ProjectEntry[TStation: Station, TEvent: Event = Event]:
    """One station/event selection within a `PysmoProject`.

    Pairs a station with either an event (for a phase-arrival-relative
    window, resolved at fetch time) or an explicit absolute time window, or
    both; an explicit window always takes precedence over one derived from
    `event`. See
    [`PysmoProject`][pysmo.tools.project.PysmoProject] for how the window is
    actually resolved.

    At least one of (`event`, an explicit `starttime`/`endtime` pair) is
    required; a lone `starttime` or `endtime`, or a `starttime` at or after
    `endtime`, raises `ValueError` at construction.

    Generic over the station and event types it was built with, so a
    [`PysmoProject`][pysmo.tools.project.PysmoProject] built from a list of
    entries keeps those concrete types (e.g. `project.events` comes back as
    `list[QuakeML]`, not `list[Event]`). An event-less entry leaves `TEvent`
    at its default, [`Event`][pysmo.Event]. A list mixing event-bearing and
    event-less entries has no single inferred element type, so annotate it
    (`list[ProjectEntry[MyStation, MyEvent]]`) or reach for
    [`build_entries`][pysmo.tools.project.build_entries], which produces a
    homogeneous list.
    """

    station: TStation
    """Station to fetch waveform data for."""

    event: TEvent | None = None
    """Event for deriving a phase-arrival-relative window (if no explicit times)."""

    starttime: pd.Timestamp | None = field(
        default=None,
        converter=converters.optional(to_utc_timestamp),
        on_setattr=setters.convert,
    )
    """Explicit start of the fetch window (UTC).

    Overrides `event` when set together with `endtime`.
    """

    endtime: pd.Timestamp | None = field(
        default=None,
        converter=converters.optional(to_utc_timestamp),
        on_setattr=setters.convert,
    )
    """Explicit end of the fetch window (UTC).

    Overrides `event` when set together with `starttime`.
    """

    _identity_cache: tuple[dict[str, Any], str] | None = field(
        init=False, default=None, eq=False, repr=False
    )
    """Memoised (`identity_components`, `identity`) pair, reused only while the
    components still compare equal — so mutating a nested `station`/`event`
    field invalidates it too, not just reassigning the top-level field."""

    checksum: str | None = field(default=None)
    """Checksum of the fetched seismogram, set on first fetch; `None` until then.

    A mismatch on a later fetch means the underlying archive data changed
    since this entry was first fetched; see
    [`PysmoProject.on_checksum_mismatch`][pysmo.tools.project.PysmoProject.on_checksum_mismatch]
    for how that is reported. Deliberately mutated by
    [`PysmoProject`][pysmo.tools.project.PysmoProject] as a side effect of
    fetching, so it is captured the next time the containing project is
    pickled; this is what makes it a durable reproducibility pin rather
    than a one-session-only check.

    Because this field is mutated in place, a `ProjectEntry` shared across
    two different `PysmoProject` instances (e.g. reused deliberately in an
    iterative workflow, or accidentally via a shared `entries` list) has its
    checksum set/checked by *whichever* project fetches it first; the
    entry doesn't belong to one project. Sharing entries across projects is
    fine; sharing them without being aware their checksum state is joint,
    not per-project, is the surprise to avoid.
    """

    @property
    def identity_components(self) -> dict[str, Any]:
        """This entry's normalised natural key as a nested dict, before hashing.

        Examples:
            >>> import pandas as pd
            >>> from pysmo import MiniEvent, MiniStation
            >>> from pysmo.tools.project import ProjectEntry
            >>> station = MiniStation(
            ...     name="ANMO",
            ...     network="IU",
            ...     location="00",
            ...     channel="BHZ",
            ...     latitude=34.9459,
            ...     longitude=-106.4571,
            ... )
            >>> event = MiniEvent(
            ...     latitude=-36.122,
            ...     longitude=-72.898,
            ...     depth=22900.0,
            ...     time=pd.Timestamp("2010-02-27T06:34:11.53Z"),
            ... )
            >>> entry = ProjectEntry(station=station, event=event)
            >>> components = entry.identity_components
            >>> components["schema"]
            'v1'
            >>> components["station"]["name"]
            'ANMO'
        """
        return entry_identity_components(self)

    @property
    def identity(self) -> str:
        """Stable identity string for this entry, computed without fetching.

        Derived from the natural key: `'v1:'` and a sha256 hexdigest. Persisting
        it and passing it to
        [`PysmoProject.get`][pysmo.tools.project.PysmoProject.get] retrieves
        the entry's seismogram later.

        Examples:
            >>> import pandas as pd
            >>> from pysmo import MiniEvent, MiniStation
            >>> from pysmo.tools.project import ProjectEntry
            >>> station = MiniStation(
            ...     name="ANMO",
            ...     network="IU",
            ...     location="00",
            ...     channel="BHZ",
            ...     latitude=34.9459,
            ...     longitude=-106.4571,
            ... )
            >>> event = MiniEvent(
            ...     latitude=-36.122,
            ...     longitude=-72.898,
            ...     depth=22900.0,
            ...     time=pd.Timestamp("2010-02-27T06:34:11.53Z"),
            ... )
            >>> entry = ProjectEntry(station=station, event=event)
            >>> entry.identity.startswith("v1:")
            True
            >>> len(entry.identity)
            67
        """
        components = entry_identity_components(self)
        cached = self._identity_cache
        if cached is None or cached[0] != components:
            cached = (components, identity_digest(components))
            object.__setattr__(self, "_identity_cache", cached)
        return cached[1]

    def __attrs_post_init__(self) -> None:
        """Reject a half-specified, reversed, or (event-less) absent window."""
        if (self.starttime is None) != (self.endtime is None):
            raise ValueError(
                "ProjectEntry needs both starttime and endtime, or neither."
            )
        if (
            self.starttime is not None
            and self.endtime is not None
            and self.starttime >= self.endtime
        ):
            raise ValueError("ProjectEntry starttime must be before endtime.")
        if self.event is None and self.starttime is None:
            raise ValueError(
                "ProjectEntry needs an explicit starttime/endtime window or an "
                + "event to derive one from."
            )

@define(kw_only=True)
class PysmoProject[TStation: Station, TEvent: Event, TSeismogram = MiniSeismogram]:
    """Declares station/event data to fetch on demand and transform into `TSeismogram`.

    A `PysmoProject` holds a flat list of
    [`ProjectEntry`][pysmo.tools.project.ProjectEntry] objects plus three
    pluggable callables: `window` resolves each entry's fetch window,
    `fetch_seismogram` downloads the trace, and `seismogram_transform` turns
    it into the caller's target type `TSeismogram`. No waveform data are
    stored on the instance between calls beyond an in-memory cache of
    already-fetched-and-transformed results.

    An optional `name` can label the project for identification by tools
    that hold more than one project. It is free-text and carries no
    semantics: it is not a key, does not affect cache invalidation, and is
    not part of `resolution_context_digest` or any entry identity.

    Generic over the station and event types of its `entries` (matching
    [`ProjectEntry`][pysmo.tools.project.ProjectEntry]'s parameter order)
    and the return type of `seismogram_transform`, all three inferred at
    construction: `TStation` / `TEvent` from `entries` (build them with
    [`build_entries`][pysmo.tools.project.build_entries] from a list of, say,
    `StationXML` and `QuakeML`, and `project.stations` / `project.events`
    come back as `list[StationXML]` / `list[QuakeML]`), `TSeismogram` from
    `seismogram_transform`'s return type, defaulting to
    [`MiniSeismogram`][pysmo.MiniSeismogram] when the default transform is
    used.

    See the [module documentation][pysmo.tools.project] for a worked
    example.

    Note: Persistence
        A `PysmoProject` travels as a pickle. The in-memory cache and lock
        are dropped during pickling and reconstructed on unpickling. An
        optional `name` is preserved across pickles; unpickling a project
        serialised without a `name` defaults it to `None`.

        Unpickling executes code, so a project file is safe to load only
        when its origin is known and trusted (see the
        [module documentation][pysmo.tools.project]). The
        format-version check on load reports an incompatible pickle; it is
        not a security boundary and runs after unpickling. There is no
        cross-version migration: a pickle from an earlier state format must
        be rebuilt from its source.

    Note: Thread-safety
        The in-memory fetch cache is safe to touch from multiple threads
        calling [`seismogram`][pysmo.tools.project.PysmoProject.seismogram],
        [`seismograms_for`][pysmo.tools.project.PysmoProject.seismograms_for],
        or [`fetch_all`][pysmo.tools.project.PysmoProject.fetch_all] on the
        same instance concurrently. This does not parallelise fetching
        itself, though: `fetch_seismogram` is called outside the lock, so
        two threads racing the same not-yet-cached entry both still fetch
        before one result wins and is cached.

        Reassigning `window` (or any other cache-affecting field) on one
        thread while another is mid-fetch is also safe: the in-flight fetch
        still returns a result, it just isn't cached (the next call
        recomputes it with the current parameters).
    """

    # v3: seismogram_checksum's algorithm changed, so v2 checksums no longer
    # compare; there is no migration, so reject rather than mis-compare.
    _FORMAT_VERSION: ClassVar[int] = 3

    name: str | None = field(
        default=None,
        validator=validators.optional(validators.instance_of(str)),
    )
    """Optional free-text label for this project.

    Useful for identification by downstream tools that hold multiple
    projects. This is purely a label, not a key: no uniqueness is enforced,
    renaming does not invalidate the fetch cache, and it is not part of
    [`resolution_context_digest`][pysmo.tools.project.PysmoProject.resolution_context_digest]
    or any entry identity.
    """

    entries: list[ProjectEntry[TStation, TEvent]] = field(
        factory=list, on_setattr=setters.pipe(setters.convert, _on_setattr_clear_cache)
    )
    """Station/event/window selections making up this project.

    Build them with
    [`build_entries`][pysmo.tools.project.build_entries]; grow the project
    later with `project.entries.extend(build_entries(...))` (a plain
    in-place mutation; call
    [`clear_cache`][pysmo.tools.project.PysmoProject.clear_cache] afterwards
    only to free memory, never for correctness, since the fetch cache is
    keyed by entry content).
    """

    seismogram_transform: SeismogramTransform[TStation, TEvent, TSeismogram] = field(
        # The default returns `MiniSeismogram`, which is `TSeismogram`'s own
        # default, but mypy still can't match a concrete return against the
        # bare type parameter in the class body.
        default=_seismogram_to_mini_seismogram,  # type: ignore[assignment]
        on_setattr=setters.pipe(setters.convert, _on_setattr_clear_cache),
    )
    """Convert a freshly downloaded seismogram into the target type `TSeismogram`.

    Any [`SeismogramTransform`][pysmo.tools.project.SeismogramTransform] (see
    there for the call contract). Defaults to returning the raw trace as a
    [`MiniSeismogram`][pysmo.MiniSeismogram], with no processing: a custom
    transform is where response removal, detrending and resampling belong,
    and it may issue its own additional fetches, as the
    [module documentation][pysmo.tools.project]'s example does for instrument
    response metadata. A callable `attrs` class with only picklable fields is
    the way to give the transform its own configuration.
    """

    fetch_seismogram: SeismogramFetcher = field(
        default=_default_fetch_seismogram,
        on_setattr=setters.pipe(setters.convert, _on_setattr_clear_cache),
    )
    """Download a seismogram for a station and absolute time window.

    Any [`SeismogramFetcher`][pysmo.tools.project.SeismogramFetcher] (see
    there for the call contract). Defaults to a private helper wrapping
    [`MSeed.fetch`][pysmo.classes.MSeed.fetch], which goes to the network on
    every call.

    Data in a live archive can change between fetches. A
    [`FetchCache`][pysmo.tools.cache.FetchCache] stores each fetched waveform
    on disk, so later sessions work with the same data, as long as its
    `max_bytes` stays at the default (unlimited). A finite `max_bytes` evicts
    old waveforms, which are then fetched again. Fetches made by
    `seismogram_transform` itself are not covered; see
    [`TransformCache`][pysmo.tools.project.TransformCache] for those.
    """

    window: WindowResolver[TStation, TEvent] = field(
        default=PhaseWindow(),
        on_setattr=setters.pipe(setters.convert, _on_setattr_clear_cache),
    )
    """Resolve an entry's fetch window when it carries no explicit one.

    Any [`WindowResolver`][pysmo.tools.project.WindowResolver] (see there for
    the call contract). Defaults to
    [`PhaseWindow`][pysmo.tools.project.PhaseWindow], which places the window
    around a predicted phase arrival. An entry with an explicit
    `starttime`/`endtime` bypasses this entirely, so a custom resolver only
    ever handles the event-derived case.
    """

    on_checksum_mismatch: Literal["warn", "raise", "ignore"] = field(
        default="warn",
        validator=validators.in_(("warn", "raise", "ignore")),
    )
    """Behaviour when a fetched seismogram's checksum no longer matches the
    one recorded on `entry.checksum` from its first fetch.

    `"warn"` (default) emits a `UserWarning` and still returns the new data;
    `"raise"` raises `ValueError` instead of returning anything, for a
    pipeline that should hard-stop on detected drift; `"ignore"` returns the
    new data with no signal at all. In every case `entry.checksum` keeps the
    value from the *first* fetch; it is never overwritten by a mismatching
    value, so a mismatch is reported (or not) consistently on every
    subsequent fetch, not just the first time it's noticed.

    Deliberately not part of cache invalidation: changing this policy only
    affects how a *future* mismatch is handled, it doesn't change what data
    was fetched or would be re-fetched, so nothing about previously cached
    results becomes stale when it changes.
    """

    _cache: dict[_CacheKey, tuple[str, TSeismogram]] = field(
        init=False, factory=dict, repr=False, eq=False
    )
    _cache_generation: int = field(init=False, default=0, repr=False, eq=False)
    """Bumped by every `clear_cache()`. A `_fetch` in progress when the cache
    is cleared (e.g. a parameter reassigned on another thread mid-download)
    sees the mismatch and returns its result without caching it under a
    now-stale key."""
    _lock: threading.Lock = field(
        init=False, factory=threading.Lock, repr=False, eq=False
    )
    """Guards reads/writes of `_cache` against concurrent access from more
    than one thread (see the class docstring's thread-safety note)."""

    def __getstate__(self) -> dict[str, Any]:
        """Drop the fetch cache and lock, neither of which can survive pickling."""
        state = attrs_getstate(self, {"_cache": {}, "_cache_generation": 0})
        del state["_lock"]
        state["_format_version"] = self._FORMAT_VERSION
        state["_pysmo_version"] = __version__
        return state

    def __setstate__(self, state: dict[str, Any]) -> None:
        """Restore state without firing `on_setattr` hooks, then make a fresh lock."""
        # Format-compatibility gate only; pickle.load above has already run
        # any code in the file, so this is not a trust check.
        pickled_format = state.pop("_format_version", 0)
        pickled_pysmo = state.pop("_pysmo_version", "unknown")
        if pickled_format != self._FORMAT_VERSION:
            raise ValueError(
                f"This PysmoProject was pickled by pysmo {pickled_pysmo} in state "
                + f"format v{pickled_format}; this pysmo ({__version__}) uses "
                + f"v{self._FORMAT_VERSION}. Re-create the project."
            )
        state.setdefault("name", None)
        attrs_setstate(self, state)
        object.__setattr__(self, "_lock", threading.Lock())

    def clear_cache(self) -> None:
        """Clear the in-memory fetch cache.

        Cleared automatically whenever
        [`entries`][pysmo.tools.project.PysmoProject.entries],
        [`window`][pysmo.tools.project.PysmoProject.window],
        [`seismogram_transform`][pysmo.tools.project.PysmoProject.seismogram_transform],
        or
        [`fetch_seismogram`][pysmo.tools.project.PysmoProject.fetch_seismogram]
        is *reassigned*.

        Call this manually after any in-place mutation of
        [`entries`][pysmo.tools.project.PysmoProject.entries] (e.g. `append`,
        `remove`, or index assignment), which isn't observable by
        `on_setattr` and therefore doesn't clear the cache automatically.
        The same applies to mutating a configuration field *inside* `window`,
        `seismogram_transform`, or `fetch_seismogram` rather than reassigning
        the whole callable; a `frozen=True` attrs callable (as
        [`PhaseWindow`][pysmo.tools.project.PhaseWindow] is) rules that out.
        """
        with self._lock:
            self._cache.clear()
            self._cache_generation += 1

    def _fetch(
        self, entry: ProjectEntry[TStation, TEvent], *, _stacklevel: int = 3
    ) -> TSeismogram:
        """Fetch, transform, and cache the seismogram for one entry.

        Internal primitive; see
        [`seismogram`][pysmo.tools.project.PysmoProject.seismogram] for the
        public, station/event-based accessor built on top of this.

        Args:
            entry: The station/event/window selection to fetch.

        Returns:
            The transformed result for `entry`, from cache if an entry with
            the same [`identity`][pysmo.tools.project.ProjectEntry.identity]
            has been fetched before.

        Raises:
            ValueError: If `window` cannot resolve a window for `entry`; if
                the underlying fetch raises (e.g. no waveform data for the
                resolved window); or if the checksum no longer matches and
                `on_checksum_mismatch="raise"`.
        """
        key: _CacheKey = entry.identity
        with self._lock:
            cached = self._cache.get(key)
            generation = self._cache_generation
        if cached is None:
            # Resolve the window only on a miss: an explicit window on the
            # entry wins (it is entry data, not resolution policy), else the
            # `window` resolver derives one. Resolution and the fetch both
            # run outside the lock (see the class docstring's thread-safety
            # note): a concurrent `self.window` reassignment is a tolerated
            # torn read, `generation` (read above) guards the write-back, and
            # two threads racing the same key both fetch before one wins.
            if entry.starttime is not None and entry.endtime is not None:
                window = WindowResult(
                    starttime=entry.starttime, endtime=entry.endtime, reference=None
                )
            else:
                window = self.window(entry)
            seismogram = self.fetch_seismogram(
                entry.station, window.starttime, window.endtime
            )
            # Checksum the raw trace pre-transform: the transform's output can
            # be mutated by downstream consumers (`ICCS` edits `t0`/`t1`/`flip`
            # during a run), which would otherwise read back as false drift.
            checksum = seismogram_checksum(seismogram)
            context = FetchContext(
                entry=entry,
                starttime=window.starttime,
                endtime=window.endtime,
                reference=window.reference,
            )
            fresh = (checksum, self.seismogram_transform(seismogram, context))
            with self._lock:
                if self._cache_generation == generation:
                    cached = self._cache.setdefault(key, fresh)
                else:
                    # A parameter changed (clearing the cache) while this
                    # fetch was in flight: return this result once without
                    # caching it under a now-stale key.
                    cached = fresh

        checksum, result = cached
        with self._lock:
            recorded = entry.checksum
            if recorded is None:
                entry.checksum = checksum
        if (
            recorded is not None
            and recorded != checksum
            and (self.on_checksum_mismatch != "ignore")
        ):
            message = (
                f"Fetched data for {entry.station.network}.{entry.station.name} "
                + "no longer matches the checksum recorded when this entry was "
                + "first fetched: the source data changed, or fetch_seismogram "
                + "now yields the same samples in a different dtype."
            )
            if self.on_checksum_mismatch == "raise":
                raise ValueError(message)
            # `_stacklevel` is threaded in from the public entry point
            # (`seismogram`/`fetch_all` pass the default; `seismograms_for`
            # passes one level deeper) so the warning always points at the
            # user's own call site, not an intermediate method.
            warnings.warn(message, stacklevel=_stacklevel)
        return result

    @property
    def stations(self) -> list[TStation]:
        """Distinct stations across all entries, in first-seen order.

        A plain `@property`, not `@cached_property`: recomputed on each
        access, same as
        [`ICCS.cc_seismograms`][pysmo.tools.iccs.ICCS.cc_seismograms]'s
        precedent for a no-arg derived list view in this codebase. Compares
        with `==` (attrs-generated equality, not identity or hashing;
        `Station` is not hashable).
        """
        seen: list[TStation] = []
        for entry in self.entries:
            if entry.station not in seen:
                seen.append(entry.station)
        return seen

    @property
    def events(self) -> list[TEvent]:
        """Distinct events across all entries, excluding event-less entries.

        In first-seen order; compares with `==`, same caveat as `stations`.
        """
        seen: list[TEvent] = []
        for entry in self.entries:
            if entry.event is not None and entry.event not in seen:
                seen.append(entry.event)
        return seen

    @property
    def resolution_context_digest(self) -> str:
        """Digest over the project parameters that determine fetched content.

        Covers `window` and `seismogram_transform`. Reassigning
        `fetch_seismogram` (e.g. to an on-disk fetch cache) leaves the
        digest unchanged.

        Examples:
            >>> from pysmo.tools.project import PysmoProject
            >>> project = PysmoProject()
            >>> project.resolution_context_digest.startswith("rc2:")
            True
            >>> len(project.resolution_context_digest)
            68
        """
        return resolution_context_digest(self)

    def events_for(self, station: TStation) -> list[TEvent | None]:
        """Events available for one station, in first-seen order.

        `None` appears in the result if `station` has an event-less entry; an
        event-less selection is a first-class member of this list, not a
        special case to check for separately.
        """
        seen: list[TEvent | None] = []
        for entry in self.entries:
            if entry.station == station and entry.event not in seen:
                seen.append(entry.event)
        return seen

    def stations_for(self, event: TEvent | None) -> list[TStation]:
        """Stations available for one event, in first-seen order.

        Pass `None` for stations with an event-less entry.
        """
        seen: list[TStation] = []
        for entry in self.entries:
            if entry.event == event and entry.station not in seen:
                seen.append(entry.station)
        return seen

    def seismogram(
        self,
        station: TStation,
        event: TEvent | None = None,
        *,
        _stacklevel: int = 3,
    ) -> TSeismogram:
        """Fetch (or return from cache) the result for one station/event combination.

        Args:
            station: Station to fetch.
            event: Event to fetch for, or `None` for an event-less entry.

        Returns:
            The transformed result for the matching entry.

        Raises:
            KeyError: If no entry matches this station/event combination.
            ValueError: If more than one entry matches: an authoring
                mistake (e.g. the same station/event added twice with
                different explicit windows), surfaced rather than silently
                resolved by picking one.
        """
        matches = [e for e in self.entries if e.station == station and e.event == event]
        if not matches:
            raise KeyError("No entry for this station/event combination.")
        if len(matches) > 1:
            raise ValueError(
                "More than one entry matches this station/event combination."
            )
        return self._fetch(matches[0], _stacklevel=_stacklevel)

    def get(self, identity: str, *, _stacklevel: int = 3) -> TSeismogram:
        """Fetch (or return from cache) the result for an entry by its identity.

        Args:
            identity: An [`entry.identity`][pysmo.tools.project.ProjectEntry.identity]
                string. The entry need not have been fetched before.

        Returns:
            The transformed result for the matching entry.

        Raises:
            UnknownEntryIdentity: If no entry matches this identity.
            ValueError: If more than one entry matches: an authoring mistake,
                surfaced rather than silently resolved by picking one.

        Examples:
            >>> import pandas as pd
            >>> from pysmo import MiniEvent, MiniSeismogram, MiniStation, Station
            >>> from pysmo.tools.project import ProjectEntry, PysmoProject
            >>> def fake_fetch(station: Station, t0: pd.Timestamp, t1: pd.Timestamp):
            ...     return MiniSeismogram(
            ...         begin_time=t0, delta=pd.Timedelta(seconds=1), data=[1.0, 2.0]
            ...     )
            >>> station = MiniStation(
            ...     name="ANMO",
            ...     network="IU",
            ...     location="00",
            ...     channel="BHZ",
            ...     latitude=34.9459,
            ...     longitude=-106.4571,
            ... )
            >>> entry = ProjectEntry(
            ...     station=station,
            ...     starttime=pd.Timestamp("2020-01-01T00:00:00Z"),
            ...     endtime=pd.Timestamp("2020-01-01T00:10:00Z"),
            ... )
            >>> project = PysmoProject(entries=[entry], fetch_seismogram=fake_fetch)
            >>> seis = project.get(entry.identity)
            >>> len(seis.data)
            2
        """
        matches = [e for e in self.entries if e.identity == identity]
        if not matches:
            raise UnknownEntryIdentity(identity)
        if len(matches) > 1:
            raise ValueError(f"More than one entry resolves to identity {identity!r}.")
        return self._fetch(matches[0], _stacklevel=_stacklevel)

    def seismograms_for(self, event: TEvent) -> list[TSeismogram]:
        """All seismograms for one event, e.g. ready for `ICCS(seismograms=...)`.

        Built from
        [`stations_for`][pysmo.tools.project.PysmoProject.stations_for] and
        [`seismogram`][pysmo.tools.project.PysmoProject.seismogram], not an
        independent filter over `entries`.

        Typed to require an `Event`, unlike `stations_for`/`events_for`
        (which both treat `None` as first-class), deliberately: this
        method exists for the event-based bulk-fetch use case (`ICCS`),
        which has no equivalent "all event-less entries" workflow to
        support. `[seismogram(s, None) for s in stations_for(None)]`
        already covers that case directly if it's ever needed.
        """
        return [
            self.seismogram(station, event, _stacklevel=4)
            for station in self.stations_for(event)
        ]

    def fetch_all(self) -> list[TSeismogram]:
        """Fetch every entry in the project.

        With the default, always-fresh `fetch_seismogram`, this just warms
        `_cache` for the session. With a cache-backed `fetch_seismogram`
        (e.g. [`FetchCache`][pysmo.tools.cache.FetchCache]), this is what
        actually populates the on-disk cache: a single, explicit "get
        everything this project needs onto disk" call, rather
        than relying on incidental use of `seismogram`/`seismograms_for` to
        cover every entry eventually.

        Returns:
            One transformed result per entry, in `entries` order.
        """
        return [self._fetch(entry) for entry in self.entries]