katgpucbf.fgpu.delay module

A collection of classes and methods for delay-tracking.

It should be noted that the classes in this module use a slightly different model than the public katcp interface. The reference channel for phase change is channel 0, rather than the centre channel. The difference is dealt with by the request handler for the ?delays katcp request.

class katgpucbf.fgpu.delay.AbstractDelayModel

Bases: ABC

Abstract base class for delay models.

All units are samples rather than SI units.

abstractmethod range(start: int, stop: int, step: int) → tuple[ndarray, ndarray, ndarray]

Find input timestamps corresponding to a range of output samples.

For each output sample with timestamp in range(start, stop, step), it determines a corresponding input sample.

Parameters:

• start – First timestamp (inclusive).

• stop – Last timestamp (exclusive)

• step – Interval between timestamps (must be positive).

Returns:

• orig_time – Undelayed integer timestamps corresponding to range(start, stop, step)

• residual – Fractional sample delay not accounted for by time - orig_time.

• phase – Fringe-stopping phase to be added.

abstractmethod skip(target: int, start: int, step: int) → int

Find the next output time for which the input time is at least target.

The output time must also be at least start and a multiple of step.

class katgpucbf.fgpu.delay.AlignedDelayModel(base: DM, align: int)

Bases: AbstractDelayModel, Generic

Wrap another delay model and enforce an alignment on original timestamp.

Note that this can cause residual delays to be larger than 1.

range(start: int, stop: int, step: int) → tuple[ndarray, ndarray, ndarray]

Find input timestamps corresponding to a range of output samples.

For each output sample with timestamp in range(start, stop, step), it determines a corresponding input sample.

Parameters:

• start – First timestamp (inclusive).

• stop – Last timestamp (exclusive)

• step – Interval between timestamps (must be positive).

Returns:

• orig_time – Undelayed integer timestamps corresponding to range(start, stop, step)

• residual – Fractional sample delay not accounted for by time - orig_time.

• phase – Fringe-stopping phase to be added.

skip(target: int, start: int, step: int) → int

Find the next output time for which the input time is at least target.

The output time must also be at least start and a multiple of step.

class katgpucbf.fgpu.delay.LinearDelayModel(start: int, delay: float, delay_rate: float, phase: float, phase_rate: float)

Bases: AbstractDelayModel

Delay model that adjusts delay linearly over time.

Parameters:

• start – Output sample at which the model should start being used.

• delay – Delay to apply at start. [seconds]

• delay_rate – Rate of change of delay. [seconds/second]

• phase – Fringe-stopping phase to apply with the fine delay. [radians]

• phase_rate – Rate of change of the fringe-stopping phase. [radians/second]

Raises:

ValueError – If rate is greater than or equal to 1 or start is negative

range(start: int, stop: int, step: int) → tuple[ndarray, ndarray, ndarray]

Find input timestamps corresponding to a range of output samples.

For each output sample with timestamp in range(start, stop, step), it determines a corresponding input sample.

Parameters:

• start – First timestamp (inclusive).

• stop – Last timestamp (exclusive)

• step – Interval between timestamps (must be positive).

Returns:

• orig_time – Undelayed integer timestamps corresponding to range(start, stop, step)

• residual – Fractional sample delay not accounted for by time - orig_time.

• phase – Fringe-stopping phase to be added.

skip(target: int, start: int, step: int) → int

Find the next output time for which the input time is at least target.

The output time must also be at least start and a multiple of step.

class katgpucbf.fgpu.delay.MultiDelayModel(callback_func: Callable[[Sequence[LinearDelayModel]], None] | None = None)

Bases: AbstractDelayModel

Piece-wise linear delay model.

The model evolves over time by calling add(). It must only be queried with monotonically increasing start values, because as soon as a query is made beyond the end of the first piece it is discarded. Additionally, after calling skip(), the return value should be treated as a lower bound for future start values.

In the initial state it has a model with zero delay.

It accepts an optional callback function that takes in the LinearDelayModels attached to this MultiDelayModel. This callback is called whenever the first linear piece changes. It is also called immediately by the constructor.

add(model: LinearDelayModel) → None

Extend the model with a new linear model.

The new model is applicable from its start time forever. If the new model has an earlier start time than some previous model, the previous model will be discarded.

range(start: int, stop: int, step: int) → tuple[ndarray, ndarray, ndarray]

Find input timestamps corresponding to a range of output samples.

For each output sample with timestamp in range(start, stop, step), it determines a corresponding input sample.

Parameters:

• start – First timestamp (inclusive).

• stop – Last timestamp (exclusive)

• step – Interval between timestamps (must be positive).

Returns:

• orig_time – Undelayed integer timestamps corresponding to range(start, stop, step)

• residual – Fractional sample delay not accounted for by time - orig_time.

• phase – Fringe-stopping phase to be added.

skip(target: int, start: int, step: int) → int

Find the next output time for which the input time is at least target.

The output time must also be at least start and a multiple of step.

exception katgpucbf.fgpu.delay.NonMonotonicQueryWarning

Bases: UserWarning

Delay model was queried non-monotonically.

katgpucbf.fgpu.delay.wrap_angle(angle)

Restrict an angle to [-pi, pi].

This works on both Python scalars and numpy arrays.