katgpucbf.dsim.server module

katcp server.

class katgpucbf.dsim.server.DEngine(sender: Sender, descriptor_sender: DescriptorSender, heap_sets: list[HeapSet], adc_sample_rate: float, sample_bits: int, dither_seed: int, *args, **kwargs)[source]

Bases: Engine

katcp server.

Parameters:

  • sender – Sender which is streaming data out. It is halted when the server is stopped.

  • heap_sets – Heap sets which can be transmitted (must have length 3). The first element must be currently set on sender.

  • adc_sample_rate – Sampling rate in Hz

  • sample_bits – Number of bits per output sample

  • dither_seed – Dither seed (used only to populate a sensor).

  • *args – Passed to base class

  • **kwargs – Passed to base class

VERSION: str = 'katgpucbf-dsim-1.0'
async on_stop() → None[source]

Cancel tasks registered with add_cancel_task() on shutdown.

async request_signals(ctx, signals_str: str, period: int | None = None) → int[source]

Update the signals that are generated.

Parameters:

  • signals_str – Textural description of the signals. See the docstring for parse_signals for the language description. The description must produce one signal per polarisation.

  • period – Period for the generated signal. It must divide into the value indicated by the max-period sensor. If not specified, the value of max-period is used.

Returns:

First timestamp which will use the new signals

Return type:

timestamp

async request_time(ctx) → float[source]

Return the current UNIX timestamp.

async set_signals(signals: Sequence[Signal], signals_str: str, period: int | None = None) → int[source]

Change the signals request_signals().

This is the implementation of request_signals(). See that method for description of the parameters and return value (signals is the parsed version of signals_str).