FrbGrouper

class pirate_frb.rpc.FrbGrouper

The FrbGrouper manages pirate-grouper communication (from the grouper side).

This is a complex class, and the docstrings just summarize syntax. For a lot more info, see the grouper-specific parts of the sphinx docs.

Constructor:

g = pirate_frb.rpc.FrbGrouper(ip_addr)

where ip_addr (str) is the grouper’s own ip:port listen address (e.g. ‘127.0.0.1:7000’); the producer (an FrbServer running pirate) connects to it and hands off the GpuDedisperser output ring buffer over CUDA IPC. Use the object as a context manager (see below).

Usage summary:

with pirate_frb.rpc.FrbGrouper('127.0.0.1:7000') as g:
    for ichunk in itertools.count():       # outer loop over time chunks
        for ibatch in range(g.nbatches):   # inner loop over beam batches
           with g.get_output(ichunk, ibatch) as outputs:
               pass
        events = g.create_events(...)

Attributes (all read-only):

  • is_stopped (bool) – whether the grouper is in the stopped state.

  • cuda_device_id (int) – CUDA device where the IPC-mapped outputs live.

  • dtype (ksgpu.Dtype) – data type of the out_max arrays.

  • nt_in (int) – input time samples per time chunk.

  • total_beams (int) – total beams per chunk (= beams_per_gpu).

  • beams_per_batch (int) – beams per output batch.

  • nbatches (int) – beam-batches per chunk (= total_beams / beams_per_batch); producer seq_id = ichunk*nbatches + ibatch.

  • num_batch_slots (int) – output ring-buffer depth; leading beam axis = num_batch_slots*beams_per_batch (<= total_beams).

  • initial_chunk (int) – chunk index of the producer’s first output vs FPGA seq 0 (sets GpuDedisperserOutputs.ichunk_fpga_based).

  • ntrees (int) – number of dedispersion trees.

  • ndm_out (list of int) – per-tree output DM-channel counts (length ntrees).

  • nt_out (list of int) – per-tree output time-sample counts (length ntrees).

  • dedispersion_config (DedispersionConfig) – producer’s dedispersion config (from the handshake).

  • xengine_metadata (XEngineMetadata) – X-engine metadata (from the handshake).

  • xengine_metadata_yaml_string (str) – X-engine metadata as a YAML string.

  • dedispersion_config_yaml_string (str) – dedispersion config as a YAML string.

  • dedispersion_plan_yaml_string (str) – dedispersion plan as a YAML string.

  • grouper_ip_addr (str) – the grouper’s own listen address (ip:port), set at construction.

  • search_ip_addr (str) – producer FrbServer’s FrbSearch RPC endpoint (ip:port), from the handshake.

The Python context manager also attaches (available inside the with block):

  • xengine_yaml (dict) – parsed X-engine metadata (from xengine_metadata_yaml_string).

  • dedispersion_config_yaml (dict) – parsed dedispersion config.

  • dedispersion_plan_yaml (dict) – parsed dedispersion plan.

  • dm_per_unit_delay (float) – DM of a full-band delay of one input time sample.

  • steady_state_it0 (list of int arrays). A dedispersion output array element (ichunk, beam, itree, idm, it) is “steady-state”, i.e. unaffected by initial zero-padding, if: ichunk*nt_out + it >= steady_state_it0[itree][idm].

create_events(ichunk, itrees, ibeams, idm, itime, snr)

Build a FrbSifterEvents from GPU arrays of (tree, beam, dm, time) indices.

The grouper finds peaks at index quadruples (itree, ibeam, idm, itime), and needs to translate these indices into physical quantities (DM, time, etc) before sending events to the sifter.

The create_events() method performs this translation, including subtleties like early triggers (which mean that event arrival times can be in the future!) For more info, see the grouper-specific parts of the sphinx docs, and/or the tex notes.

Currently, when we assign dm/time value to each event, we ignore out_argmax arrays, and use the center of each coarse dm/time pixel. I plan to improve this in the future.

The index arrays (itrees, ibeams, idm, itime, snr) must all be cupy arrays with the same shape – one event per element. Index values are assumed to be in range; they are not bounds-checked, to avoid a device->host sync.

Parameters:

  • ichunk (int) – Time-chunk index, using the same zero-based indexing as get_output(). (Note: when create_events() computes the arrival time, it adds the value of ‘initial_chunk’, which was sent in the pirate-grouper handshake.)

  • itrees (cupy.ndarray) – Per-event dedispersion-tree index.

  • ibeams (cupy.ndarray) – Per-event global beam index.

  • idm (cupy.ndarray) – Per-event index along the selected tree’s output DM axis.

  • itime (cupy.ndarray) – Per-event index along the selected tree’s output time axis.

  • snr (cupy.ndarray) – Per-event SNR.

Returns:

The events translated into physical units (beam id, absolute FPGA timestamp, DM, SNR), ready to pass to FrbSifterClient.send_events(). Also carries the chunk’s absolute FPGA window (chunk_fpga_start, chunk_fpga_end). The per-event quantities not measured by the toy grouper get placeholders: rfi_prob = 0, width_ms = one time sample, and the subband = the full band (subband_freq_{lo,hi}_MHz).

Return type:

FrbSifterEvents

get_output(ichunk, ibatch)

Acquire one beam-batch’s outputs, as a GpuDedisperserOutputs object.

Usage summary (see grouper-specific parts of sphinx docs for more info):

# Warning: output arrays are only valid inside the context manager!!
# Therefore, each batch must be fully processed, before seeing the next batch.
with grouper.get_output(ichunk, ibatch) as outputs:
    # Loop over dedispersion trees.
    for itree, tree_out in enumerate(outputs.out_max):
        # 'tree_out' has shape (beams_per_batch, coarse_ndm, coarse_ntime)
Parameters:

  • ichunk (int) – Zero-based time-chunk index (i.e. first call to get_output() should specify ichunk=0, and ignore the value of FrbGrouper.initial_chunk).

  • ibatch (int) – Beam-batch index within the chunk (must satisfy 0 <= ibatch < self.nbatches).

Yields:

GpuDedisperserOutputs – Per-batch slice with .out_max / .out_argmax (lists of ksgpu Arrays, convertible to cupy via DLPack).