Add direct spectrogram generation to voltage backend

- Implement `to_spectrogram` method in `RawVoltageBackend` for generating spectrograms directly from voltage data without intermediate RAW file processing.
- Enhance CLI options to include frequency range and fine-channel FFT method for spectrogram generation.
- Update `PolyphaseFilterbank` to support selected coarse channelization methods and validate input parameters.
- Modify `RawReductionSpec` to include frequency range and methods for coarse and fine channelization.
- Add tests for new functionality, ensuring selected channelization methods match full slices and validating spectrogram outputs.
- Introduce benchmarking script for comparing performance of full FFT slicing against selected-bin PFB channelization.
- Create `spectrogram.py` module to handle spectrogram specifications and results, including writing output to files.

Author: bbrzycki

benchmarks/voltage_selected_bins.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import argparse
+import os
+import time
+from collections.abc import Iterable
+
+import numpy as np
+
+import setigen as stg
+
+
+def _parse_counts(raw_counts: str) -> list[int]:
+    """Parse a comma-separated list of selected coarse-channel counts."""
+    return [int(item.strip()) for item in raw_counts.split(",") if item.strip()]
+
+
+def _time_call(repeats: int, func: object) -> float:
+    """Return the best runtime across repeated calls."""
+    timings = []
+    for _ in range(repeats):
+        start = time.perf_counter()
+        func()
+        timings.append(time.perf_counter() - start)
+    return min(timings)
+
+
+def run_benchmark(
+    *,
+    num_taps: int,
+    num_branches: int,
+    windows: int,
+    counts: Iterable[int],
+    repeats: int,
+    seed: int,
+) -> None:
+    """Benchmark full-FFT slicing against selected-bin PFB channelization."""
+    rng = np.random.default_rng(seed)
+    samples = rng.standard_normal(num_taps * num_branches * windows)
+
+    print(
+        "backend={backend} taps={taps} branches={branches} windows={windows} repeats={repeats}".format(
+            backend="cupy" if os.getenv("SETIGEN_ENABLE_GPU") == "1" else "numpy",
+            taps=num_taps,
+            branches=num_branches,
+            windows=windows,
+            repeats=repeats,
+        )
+    )
+    print("channels,full_seconds,selected_seconds,speedup")
+
+    for count in counts:
+        full_filterbank = stg.voltage.PolyphaseFilterbank(
+            num_taps=num_taps,
+            num_branches=num_branches,
+        )
+        selected_filterbank = stg.voltage.PolyphaseFilterbank(
+            num_taps=num_taps,
+            num_branches=num_branches,
+        )
+
+        full_seconds = _time_call(
+            repeats,
+            lambda: full_filterbank.channelize(
+                samples,
+                cache=False,
+                start_chan=0,
+                num_chans=count,
+                method="full",
+            ),
+        )
+        selected_seconds = _time_call(
+            repeats,
+            lambda: selected_filterbank.channelize(
+                samples,
+                cache=False,
+                start_chan=0,
+                num_chans=count,
+                method="selected",
+            ),
+        )
+        speedup = full_seconds / selected_seconds if selected_seconds else float("inf")
+        print(f"{count},{full_seconds:.8f},{selected_seconds:.8f},{speedup:.3f}")
+
+
+def main() -> None:
+    """Run the selected-bin voltage benchmark."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--num-taps", type=int, default=8)
+    parser.add_argument("--num-branches", type=int, default=1024)
+    parser.add_argument("--windows", type=int, default=16)
+    parser.add_argument("--counts", default="1,2,4,8,16,64")
+    parser.add_argument("--repeats", type=int, default=5)
+    parser.add_argument("--seed", type=int, default=0)
+    args = parser.parse_args()
+
+    run_benchmark(
+        num_taps=args.num_taps,
+        num_branches=args.num_branches,
+        windows=args.windows,
+        counts=_parse_counts(args.counts),
+        repeats=args.repeats,
+        seed=args.seed,
+    )
+
+
+if __name__ == "__main__":
+    main()

docs/source/setigen.voltage.rst

@@ -62,3 +62,18 @@ setigen.voltage.waterfall module
     :undoc-members:
     :show-inheritance:
 
+setigen.voltage.reduction module
+--------------------------------
+
+.. automodule:: setigen.voltage.reduction
+    :members:
+    :undoc-members:
+    :show-inheritance:
+
+setigen.voltage.spectrogram module
+----------------------------------
+
+.. automodule:: setigen.voltage.spectrogram
+    :members:
+    :undoc-members:
+    :show-inheritance:

docs/source/voltages.rst

@@ -139,6 +139,61 @@ acceleration, and rawspec-style polarization semantics for ``pol_mode=1``,
 directly and are not wrapped in :class:`~setigen.frame.Frame`, which remains a
 2D total-power interface.
 
+Direct voltage spectrograms
+---------------------------
+
+When the desired product is a dynamic spectrum rather than a persistent RAW
+file, :meth:`~setigen.voltage.backend.RawVoltageBackend.to_spectrogram` can run
+the same exact voltage path and skip the intermediate RAW serialization step.
+The backend still generates voltage samples, applies the digitizer, coarse PFB,
+and complex requantizer, and then performs fine channelization and integration
+directly in Python.
+
+.. code-block:: python
+
+    spec = stg.voltage.VoltageSpectrogramSpec(
+        fftlength=1024,
+        integration_factor=4,
+        pol_mode=stg.voltage.PolarizationMode.TOTAL_POWER,
+        coarse_method='auto',
+        fine_method='auto',
+    )
+
+    result = rvb.to_spectrogram(spec,
+                                num_blocks=1,
+                                length_mode='num_blocks',
+                                verbose=False)
+    frame = result.to_frame()
+
+The ``coarse_method`` and ``fine_method`` options accept ``'auto'``,
+``'full'``, and ``'selected'``. Automatic selection is used only for exact
+methods that preserve the current channel response and FFT normalization. For
+small coarse-channel or fine-channel selections, ``'auto'`` may compute only
+the requested DFT bins; otherwise it falls back to the full FFT path.
+
+For targeted analysis, the direct spectrogram and RAW reduction APIs support
+frequency-region selection on the final fine-channel axis:
+
+.. code-block:: python
+
+    roi = stg.voltage.VoltageSpectrogramSpec(
+        fftlength=1024,
+        integration_factor=4,
+        frequency_range=(6001.0e6, 6001.5e6),
+        fine_method='selected',
+    )
+
+    frame = rvb.to_spectrogram(roi,
+                               num_blocks=1,
+                               length_mode='num_blocks',
+                               verbose=False).to_frame()
+
+The frequency range is mutually exclusive with ``start_chan`` and
+``num_chans``. Existing RAW generation remains available through
+:meth:`~setigen.voltage.backend.RawVoltageBackend.record`; direct spectrogram
+generation does not write a RAW file unless users separately call
+``record``.
+
 Using GPU acceleration
 ----------------------
 

setigen/voltage/__init__.py

@@ -24,6 +24,10 @@
     reduce_raw,
     reduce_raw_to_frame,
 )
+from setigen.voltage.spectrogram import (
+    VoltageSpectrogramResult,
+    VoltageSpectrogramSpec,
+)
 
 __all__ = [
     "DataStream",
@@ -52,4 +56,6 @@
     "RawReductionSpec",
     "reduce_raw",
     "reduce_raw_to_frame",
+    "VoltageSpectrogramResult",
+    "VoltageSpectrogramSpec",
 ]

setigen/voltage/_backend/pipeline.py

@@ -227,6 +227,9 @@ def _channelize_voltage(
     antenna: int,
     pol: int,
     digitize: bool,
+    start_chan: int | None = None,
+    num_chans: int | None = None,
+    coarse_method: str = "auto",
 ) -> Any:
     """Digitize, PFB-channelize, and coarse-channel select one voltage stream.
 
@@ -236,18 +239,33 @@ def _channelize_voltage(
         antenna: Antenna index.
         pol: Polarization index.
         digitize: Whether to digitize before PFB channelization.
+        start_chan: Optional first coarse channel to return. Defaults to the
+            backend recording start channel.
+        num_chans: Optional number of coarse channels to return. Defaults to
+            the backend recording channel count.
+        coarse_method: Coarse-channel transform method for the PFB.
 
     Returns:
         Complex coarse-channel voltages for the selected channel slice.
     """
+    if start_chan is None:
+        start_chan = backend.start_chan
+    if num_chans is None:
+        num_chans = backend.num_chans
+
     if digitize:
         start = time.time()
         v = backend.digitizer[antenna][pol].quantize(v)
         backend.digitizer_stage_t += time.time() - start
 
     start = time.time()
-    v = backend.filterbank[antenna][pol].channelize(v, cache=True)
-    v = v[:, backend.start_chan : backend.start_chan + backend.num_chans]
+    v = backend.filterbank[antenna][pol].channelize(
+        v,
+        cache=True,
+        start_chan=start_chan,
+        num_chans=num_chans,
+        method=coarse_method,
+    )
     backend.filterbank_stage_t += time.time() - start
     return v