Skip to content

Gateway Analysis Compute

Poll-loop FFT, filter, and reference-stat computations

Pure analysis computations for the session poll loop.

Everything here operates on already-acquired WaveformData (no instrument I/O) and returns stream-frame dicts or None when uncomputable. The poll loop turns None into a one-shot cleared frame (the _shown pattern).

filtered_waveform 

filtered_waveform(config: Dict[str, Any], acquired: Dict[str, Any])

Apply the configured Butterworth filter to its source channel's waveform.

Source code in scpi_control/server/compute.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def filtered_waveform(config: Dict[str, Any], acquired: Dict[str, Any]):
    """Apply the configured Butterworth filter to its source channel's waveform."""
    data = acquired.get("C{0}".format(config["source"]))
    if data is None or len(data.voltage) < 2:
        return None
    kind = config["kind"]
    low = config["cutoff_low"]
    high = config["cutoff_high"]
    order = config["order"]
    if kind == "lowpass" and high is not None:
        return _analyzer.apply_lowpass_filter(data, high, order)
    if kind == "highpass" and low is not None:
        return _analyzer.apply_highpass_filter(data, low, order)
    if kind == "bandpass" and low is not None and high is not None:
        return _analyzer.apply_bandpass_filter(data, low, high, order)
    return None

reference_stats 

reference_stats(reference: Dict[str, Any], acquired: Dict[str, Any]) -> Dict[str, Any]

Correlation + max deviation of the live source-channel trace vs the reference.

Mirrors ReferenceWaveform.calculate_correlation/_difference semantics (interpolate onto the live grid when lengths differ) without instantiating the store. Degrades every failure to nulls — never raises.

Source code in scpi_control/server/compute.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def reference_stats(reference: Dict[str, Any], acquired: Dict[str, Any]) -> Dict[str, Any]:
    """Correlation + max deviation of the live source-channel trace vs the reference.

    Mirrors ReferenceWaveform.calculate_correlation/_difference semantics
    (interpolate onto the live grid when lengths differ) without instantiating
    the store. Degrades every failure to nulls — never raises.
    """
    stats: Dict[str, Any] = {"type": "reference_stats", "correlation": None, "max_deviation": None}
    channel = reference.get("channel")
    data = acquired.get("C{0}".format(channel)) if channel else None
    if data is None or len(data.voltage) < 2:
        return stats
    try:
        ref_voltage = np.asarray(reference["data"]["voltage"], dtype=float)
        if len(data.voltage) != len(ref_voltage):
            ref_time = np.asarray(reference["data"]["time"], dtype=float)
            ref_voltage = np.interp(data.time, ref_time, ref_voltage)
        stats["max_deviation"] = float(np.max(np.abs(data.voltage - ref_voltage)))
        with np.errstate(divide="ignore", invalid="ignore"):
            correlation = float(np.corrcoef(data.voltage, ref_voltage)[0, 1])
        stats["correlation"] = correlation if np.isfinite(correlation) else None
    except Exception:
        return {"type": "reference_stats", "correlation": None, "max_deviation": None}
    return stats