FrbSifterClient

class pirate_frb.rpc.FrbSifterClient(server_address: str = 'localhost:50051', timeout=1, raise_exception_on_timeout=True)

The FrbSifterClient manages grouper-sifter communication (from the grouper side).

The grouper sends two types of messages to the sifter: a ConfigMessage (sent once), and FrbEventsMessages (sent once per FPGA window, typically one per time chunk).

If FrbSifterClient is used as a context manager, then close() is automatically called.

Attributes (read-only):

  • server_address (str) – the sifter’s ip:port address.

  • timeout (float) – per-send RPC deadline, in seconds. If > 0 (default 1), send_events() is fire-and-forget and an event is dropped if it isn’t delivered within this many seconds. 0 means synchronous: send_events() blocks until the sifter replies.

  • raise_exception_on_timeout (bool, default True) – what happens when an async send (timeout > 0) expires: if True the timeout is re-raised from the next send_events() (or printed by close() if there is no next send); if False the event is dropped and a message is printed to stdout (so a sifter that is offline or not responding is visible).

Usage:

with FrbSifterClient("localhost:7100") as sifter:
    sifter.send_configuration(pirate_yaml, xengine_yaml,
                               dedispersion_plan_yaml, grouper_yaml,
                               search_ip_addr)
    sifter.send_events(beam_set_id, events, coarsegrain_snr)

where ‘events’ is a FrbSifterEvents.

close(drain_timeout=2.0)

Wait (briefly) for in-flight async sends, then close the gRPC channel.

send_events() dispatches its RPCs asynchronously; close() waits up to ‘drain_timeout’ seconds (total) for the pending ones to finish – so a clean shutdown doesn’t cancel the last events – then closes the channel (which cancels anything still in flight). A leftover deferred timeout (raise_exception_on_timeout=True) that no subsequent send_events() will report is printed here rather than raised, so close() can’t mask an exception that is unwinding through __exit__.

Automatically called if the FrbSifterClient is used as a context manager.

send_configuration(pirate_yaml, xengine_yaml, dedispersion_plan_yaml, grouper_yaml, search_ip_addr)

Send the initial ConfigMessage (CheckConfiguration RPC).

Sent once, before any events. Unlike send_events(), this send is always synchronous: it blocks until the sifter replies and raises on failure (the client’s timeout governs only send_events()). Each yaml argument may be either a str (sent unmodified) or any yaml-serializable object (yaml.dump()’d to a string before sending).

Parameters:

  • pirate_yaml (str or object) – Pirate dedispersion-configuration yaml.

  • xengine_yaml (str or object) – X-engine metadata yaml.

  • dedispersion_plan_yaml (str or object) – Dedispersion-plan yaml.

  • grouper_yaml (str or object) – Grouper-specific configuration yaml.

  • search_ip_addr (str) – The producer FrbServer’s FrbSearch RPC endpoint (an ip:port string, typically grouper.search_ip_addr), so the sifter can call back to pirate. Sent as-is (not yaml-serialized).

Return type:

None

Raises:

RuntimeError – On failure: a yaml-serialization error, a gRPC transport error, or a not-ok reply from the sifter.

send_events(beam_set_id, events, coarsegrain_snr, from_simulator=False)

Send an FrbEventsMessage (FrbEvents RPC).

Sent once per FPGA window, after send_configuration().

The request is built and validated synchronously (so the caller’s arrays are snapshotted before returning). Delivery then depends on the client’s timeout. If timeout > 0 (the default) this is fire-and-forget: the RPC is dispatched on a grpc background thread with a timeout-second deadline and this method returns immediately; an event not delivered within timeout seconds expires (handled per raise_exception_on_timeout – see the class docstring). If timeout == 0 the send is synchronous: it blocks until the sifter replies and raises on any transport error or not-ok reply. (In the async case those are instead handled by the background callback: a not-ok reply is logged, and a failed/expired send follows raise_exception_on_timeout.) close() waits briefly for in-flight sends to finish.

Parameters:

  • beam_set_id (int) – X-engine beam-set id.

  • events (FrbSifterEvents) – The per-event triggers (one FrbEvent per array element) AND the FPGA window [chunk_fpga_start, chunk_fpga_end] this message covers – the window is taken from this object. For a coarse-grain-only message, pass an FrbSifterEvents with empty per-event arrays but a valid window.

  • coarsegrain_snr (array_like) – Per-beam coarse-grained max SNR over the same FPGA window, as a 1-d numpy or cupy array (or any 1-d iterable). A cupy array is copied to the host in a single shot via .get() (cupy raises on implicit np.asarray()), then cast to float32.

  • from_simulator (bool, optional) – False (default) for events produced by the real search pipeline; True for the parallel “ideal” events the fake X-engine emits when simulating pulses upstream (so the sifter can tell the two apart).

Return type:

None

Raises:

RuntimeError – Synchronously, for malformed input (a bad ‘events’ type, or a coarsegrain_snr that can’t be converted to a 1-d float32 array); with timeout=0, also on a transport error or not-ok reply. With raise_exception_on_timeout=True, a previous async send that timed out is re-raised here (deferred from its background callback thread).