def hilbert_instantaneous_phase(signal, time, t_eval: float):
    """Instantaneous phase and envelope of a band-limited signal at one time.

    Parameters
    ----------
    signal : array_like
        A band-passed, roughly monochromatic trace [V].
    time : array_like
        Its time base [s].
    t_eval : float
        Time at which to report the phase [s].

    Returns
    -------
    tuple
        The phase in degrees and the envelope in volts at the nearest sample, that
        sample's index, and the full phase and envelope series [-].

    Convention
    ----------
    Phase comes from the analytic signal and is wrapped to the principal branch,
    so it runs within a full turn and jumps rather than accumulating. The sample
    nearest the requested time is used; no interpolation.

    Applicability
    -------------
    Machine-independent.

    Limitations
    -----------
    **Meaningful only for a band-passed input.** The analytic signal of a
    broadband trace has a phase that does not correspond to any single mode, so
    filter first with :func:`sxr_band_signals`. The Hilbert transform is also
    poorly behaved at the record edges, so a time near either end is unreliable.

    Provenance
    ----------
    .. [1] The analytic-signal phase estimate used by the validated VEST SXR
       viewer's mode analysis.
    """
    time = np.asarray(time, dtype=float)
    analytic = hilbert(np.asarray(signal, dtype=float))
    index = int(np.argmin(np.abs(time - float(t_eval))))
    phase_series = np.degrees(np.angle(analytic))
    envelope_series = np.abs(analytic)
    return (
        float(phase_series[index]),
        float(envelope_series[index]),
        index,
        phase_series,
        envelope_series,
    )

def load_te_ratio_calibration(path=None):
    """Load a ratio-to-temperature table and build its interpolator.

    Parameters
    ----------
    path : str or path-like, optional
        Calibration table with ``te`` and ``ratio`` columns. Defaults to the
        packaged VEST beryllium/aluminium table [-].

    Returns
    -------
    tuple
        The interpolator mapping ratio to temperature, the temperature column in
        electron-volts, and the ratio column, dimensionless [-].

    Raises
    ------
    ValueError
        The table lacks either column, or holds fewer than two valid points.

    Convention
    ----------
    **The interpolator extrapolates rather than clipping**, so a ratio outside the
    table returns an extrapolated temperature, not the nearest tabulated one. A
    caller that needs out-of-range samples rejected must test the ratio itself.
    Column names are matched case-insensitively after stripping.

    Defaults
    --------
    The packaged VEST beryllium/aluminium table is a diagnostic calibration,
    shipped with the package and described in the data directory's own notes.

    Applicability
    -------------
    Machine-independent.  The table is an argument; only the packaged default is
    the VEST filter pair, and a different filter combination supplies its own.

    Limitations
    -----------
    Non-finite rows are dropped silently. The two-filter method assumes a thermal
    spectrum, so the temperature it returns is meaningless for a non-thermal one.

    Provenance
    ----------
    .. [1] The packaged VEST beryllium/aluminium ratio calibration,
       ``vaft/data/legacy/sxr_te_ratio_be_al.csv``.
    """
    if path is None:
        from vaft.data.resources import data_path

        path = data_path(TE_RATIO_TABLE)
    table = pd.read_csv(path)
    columns = {c.strip().lower(): c for c in table.columns}
    if "te" not in columns or "ratio" not in columns:
        raise ValueError(
            "the calibration table needs 'te' and 'ratio' columns; "
            f"found {list(table.columns)}"
        )
    te = table[columns["te"]].to_numpy(dtype=float)
    ratio = table[columns["ratio"]].to_numpy(dtype=float)
    good = np.isfinite(te) & np.isfinite(ratio)
    if good.sum() < 2:
        raise ValueError("the calibration table holds fewer than two valid points.")
    interpolator = interp1d(
        ratio[good], te[good], fill_value="extrapolate", bounds_error=False
    )
    return interpolator, te[good], ratio[good]

def sxr_baseline_correction(time, data, baseline_start: float) -> np.ndarray:
    """Zero each channel against the quiet tail after the plasma.

    Parameters
    ----------
    time : array_like
        Sample times, shared by every channel [s].
    data : array_like
        Signals as ``(n_channels, n_samples)``, or one trace [V].
    baseline_start : float
        Time from which the record is taken to be quiet [s].

    Returns
    -------
    np.ndarray
        The signals with each channel's own zero level removed [V].

    Raises
    ------
    ValueError
        The time base and the sample axis disagree in length.

    Convention
    ----------
    The zero level is per channel, not shared, because the detectors have
    different dark levels. The mean of the tail is used, not a fitted trend, so a
    drift through the shot survives.

    Defaults
    --------
    When no sample lies at or beyond *baseline_start* the last 1000 samples are
    used instead. That fallback is a validated-workflow default carried from the
    VEST SXR viewer.

    Applicability
    -------------
    Machine-independent.

    Limitations
    -----------
    Assumes the tail is genuinely quiet. A record that ends while the plasma is
    still radiating has its baseline biased by that signal, and the fallback makes
    that silent rather than an error.

    Provenance
    ----------
    .. [1] The validated VEST SXR viewer, whose baseline convention and
       last-1000-sample fallback this reproduces.
    """
    time = np.asarray(time, dtype=float)
    values = _as_channel_matrix(data)
    if time.size != values.shape[1]:
        raise ValueError(
            f"time and data must share the sample axis; got {time.size} and {values.shape[1]}"
        )
    mask = time >= float(baseline_start)
    if np.any(mask):
        baseline = values[:, mask].mean(axis=1, keepdims=True)
    else:
        baseline = values[:, -1000:].mean(axis=1, keepdims=True)
    return values - baseline

def sxr_cwt_spectrogram(signal, fs: float, f0: float, f1: float, n_freq: int):
    """Continuous-wavelet magnitude scalogram of one trace.

    Parameters
    ----------
    signal : array_like
        The trace [V].
    fs : float
        Sample rate [Hz].
    f0 : float
        Lowest frequency of the scalogram [Hz].
    f1 : float
        Highest frequency [Hz].
    n_freq : int
        Number of frequencies between them [-].

    Returns
    -------
    tuple of np.ndarray
        The frequency axis, ascending, and the magnitude as
        ``(n_freq, n_samples)`` [-].

    Raises
    ------
    ImportError
        The optional wavelet package is not installed.

    Convention
    ----------
    Frequencies are returned ascending, which reverses the underlying library's
    own descending order, and the magnitude rows are reversed to match. A wavelet
    transform trades frequency resolution for time resolution as frequency rises,
    unlike the fixed-window short-time transform.

    Defaults
    --------
    Single-threaded execution is a numerical convenience, keeping the call
    deterministic and avoiding a thread pool inside a per-channel loop.

    Applicability
    -------------
    Machine-independent.

    Limitations
    -----------
    Needs an optional package that is not a VAFT dependency. The short-time
    transform in :func:`vaft.process.fluctuation.compute_spectrogram` covers the
    default time-frequency need without it. The input is cast to single precision
    by that library.

    Provenance
    ----------
    .. [1] The optional ``fcwt`` continuous-wavelet implementation; the
       alternative path is :mod:`vaft.process.fluctuation`.
    """
    try:
        import fcwt as fcwt_lib
    except ImportError as error:
        raise ImportError(
            "the optional 'fcwt' package is not installed; run 'pip install fcwt' "
            "or use vaft.process.fluctuation.compute_spectrogram (STFT) instead."
        ) from error
    frequencies, out = fcwt_lib.cwt(
        np.asarray(signal, dtype=np.float32),
        int(fs),
        float(f0),
        float(f1),
        int(n_freq),
        nthreads=1,
    )
    return np.asarray(frequencies)[::-1], np.abs(out[::-1, :])

def sxr_te_pairs_from_ods(ods: Any, array: str) -> tuple[tuple[int, int], ...]:
    """Pair the two filters' channels of one array by the chord they view.

    Parameters
    ----------
    ods : ODS
        Carries ``soft_x_rays.channel`` with the identifiers the VEST mapper
        writes [-].
    array : str
        Which array to pair within [-].

    Returns
    -------
    tuple of tuple of int
        One ``(beryllium_index, aluminium_index)`` pair per shared chord, ordered
        by chord number [-].

    Raises
    ------
    ValueError
        No chord in that array carries both filters.

    Convention
    ----------
    Channel identifiers are ``{daq}:{array}:{filter}:{chord}``, written by the
    VEST digitizer mapper. Pairing on the chord number rather than on position in
    the channel list is what absorbs a per-block wiring reversal: the two filter
    blocks need not be ordered the same way.

    Returns index pairs, not data, so it changes no measurement.

    Applicability
    -------------
    VEST-specific.  Depends on the identifier grammar and the filter names written
    by :func:`vaft.machine_mapping.soft_x_rays.soft_x_rays_from_digitizer_csv`;
    another machine's mapper would need its own pairing rule.

    Limitations
    -----------
    A channel whose identifier does not split into four parts, or whose chord is
    not an integer, is skipped silently, so a mis-mapped channel is absent from
    the result rather than reported.

    Provenance
    ----------
    .. [1] :func:`vaft.machine_mapping.soft_x_rays.soft_x_rays_from_digitizer_csv`,
       which writes the identifiers this parses.
    """
    be: dict[int, int] = {}
    al: dict[int, int] = {}
    channels = ods["soft_x_rays.channel"]
    for index in range(len(channels)):
        identifier = str(channels[index]["identifier"])
        parts = identifier.split(":")
        if len(parts) != 4 or parts[1] != str(array):
            continue
        try:
            chord = int(parts[3])
        except ValueError:
            continue
        if parts[2] == "Be":
            be[chord] = index
        elif parts[2] == "Al":
            al[chord] = index
    shared = sorted(set(be) & set(al))
    if not shared:
        raise ValueError(
            f"array {array!r} has no chords with both Be and Al channels; "
            "is this a two-filter array mapped by soft_x_rays_from_digitizer_csv?"
        )
    return tuple((be[chord], al[chord]) for chord in shared)