Utils

Utility functions relevant to the entire ska-low-mccs package.

class LatencyInjector(min_latency: float = 0.0, max_latency: float = 2.0, probability_of_slow: float = 0.1)

Injects configurable latency into method calls on any object.

Useful for testing race conditions and timing-sensitive code by simulating variable network or hardware delays.

Usage examples:

# Example 1: Wrap specific methods
injector = LatencyInjector(
    min_latency=0.0, max_latency=1.8, probability_of_slow=0.15
)
injector.wrap_method(proxy, 'load_pointing_delays')
injector.wrap_method(proxy, 'apply_pointing_delays')

# Example 2: Wrap all public methods on a proxy
injector = LatencyInjector(
    min_latency=0.0, max_latency=2.0, probability_of_slow=0.2
)
injector.wrap_proxy(proxy)

# Example 3: Wrap only specified methods
injector = LatencyInjector(
    min_latency=0.0, max_latency=1.5, probability_of_slow=0.1
)
injector.wrap_proxy(
    proxy,
    methods=['load_pointing_delays', 'apply_pointing_delays']
)

# Example 4: Wrap all methods except specific ones
injector = LatencyInjector(
    min_latency=0.0, max_latency=1.0, probability_of_slow=0.05
)
injector.wrap_proxy(
    proxy, exclude=['get_change_event_callbacks']
)

__init__(min_latency: float = 0.0, max_latency: float = 2.0, probability_of_slow: float = 0.1) → None

Initialize the latency injector.

Parameters:

• min_latency – Minimum latency in seconds for fast calls

• max_latency – Maximum latency in seconds for slow calls

• probability_of_slow – Probability (0.0-1.0) of triggering slow latency

wrap_method(obj: object, method_name: str) → None

Wrap a method with latency injection.

Parameters:

• obj – The object containing the method to wrap

• method_name – Name of the method to wrap

wrap_proxy(proxy: object, methods: list[str] | None = None, exclude: list[str] | None = None) → None

Wrap methods on a proxy with latency injection.

Parameters:

• proxy – The proxy object to wrap

• methods – List of method names to wrap. If None, wraps all public callable methods

• exclude – List of method names to exclude from wrapping

subarraybeam_trl_from_ids(subarray_id: int, local_beam_id: int) → str

Return TRL in new XX-YY format from subarray + local ids.

Parameters:

• subarray_id – 1-based subarray identifier

• local_beam_id – 1-based local beam identifier

Returns:

TRL string ‘low-mccs/subarraybeam/XX-YY’

Raises:

ValueError – for out of range inputs

summarize_device_config(self: Any, name: str, version: str, properties: list[str], envvars: list[str] | None = None, redacted_envvars: list[str] | None = None) → str

Construct a string summary of device configuration.

Parameters:

• self – the tango device invoking this function

• name – device TRL

• version – software version of the device

• properties – list of device properties whose values should be included. It is assumed that these device properties have already been loaded into the device as object attributes.

• envvars – list of environment variables whose values should be included

• redacted_envvars – list of environment variables whose presence should be reported, but for which the value should not be included. Use this for environment variables that contain secret information such as passwords and other credentials.

Returns:

a string summary of the device configuration.