katgpucbf.fgpu.recv module

Recv module.

class katgpucbf.fgpu.recv.Chunk(*args, sink: ChunkRingPair, device: object = None, **kwargs)[source]

Bases: Chunk

Collection of heaps passed to the GPU at one time.

It extends the spead2 base class to store a timestamp (computed from the chunk ID when the chunk is received), and optionally store a vkgdr device array.

When used as a context manager, it will call recycle() on exit.

device: object

recycle() → None[source]: Return the chunk to the owning stream/group.

sink: ReferenceType

timestamp: int

class katgpucbf.fgpu.recv.Layout(sample_bits: int, heap_samples: int, chunk_samples: int, mask_timestamp: bool)[source]

Bases: BaseLayout

Parameters controlling the sizes of heaps and chunks.

property batch_heaps: int: Number of heaps per chunk, on the axes other than time.

property chunk_batches: int: Number of batches per chunk.

chunk_samples: int: Samples in time i.e. per polarisation

property chunk_timestamp_step: int: Expected increase in timestamp from one chunk to the next.

property heap_bytes: int: Number of payload bytes per heap.

property heap_sample_count: int

Number of samples per heap.

The meaning of samples is up to the concrete subclass. It is used only for statistics.

heap_samples: int

mask_timestamp: bool

property pol_chunk_bytes: int: Number of bytes for the data in one polarisation of a chunk.

sample_bits: int

property timestamp_mask: uint64: Mask to AND with incoming timestamps.

katgpucbf.fgpu.recv.iter_chunks(ringbuffer: ChunkRingbuffer, layout: Layout, sensors: SensorSet, time_converter: TimeConverter) → AsyncGenerator[Chunk, None][source]

Iterate over the chunks and update sensors.

It also populates the chunk timestamp.

Parameters:

ringbuffer – Source of chunks.
layout – Structure of the streams.
sensors – Sensor set containing at least the sensors created by make_sensors().
time_converter – Converter to turn data timestamps into sensor timestamps.