API

SDP Queue Connector Device

class ska_sdp_lmc_queue_connector.sdp_queue_connector.SDPQueueConnector(*args: Any, **kwargs: Any)

A dynamically configured tango device for sending and receiving data to and from data services.

attributes: str

Static attributes defined in fandango json atttributes format.

exchanges_config_path: str

Configuration Database template parent path for exchange configs.

Brace templates are substituted with the device subarray_id.

Examples: * /component/lmc-queueconnector-01/owner * /component/lmc-queueconnector-{}/owner

experimental_flow: Annotated[bool, deprecated.deprecated]

Enable to automatically configure using configuration database flow configs.

experimental_kafka: bool

Backwards compatibility flag to select Kafka specialized exchanges.

generic_read(attr: tango.Attribute)

Generic read function conforming to attribute.fread interface.

Parameters:

attr – output tango attribute

generic_write(wattr: tango.WAttribute)

Generic read function conforming to attribute.fwrite interface.

Parameters:

wattr – input tango write attribute

async group_configure(configuration_json: str, group_name_override: str | None = None)

Commands the device to load exchanges using a provided config json string.

Parameters:

• configuration_json – QueueConnectorDescriptor json. None value is considered as an empty config.

• group_name_override – When provided, overrides the group name inside the QueueConnectorDescriptor.

async reconfigure(group_name: str, config_json: str | None)

Tries commanding to stop, reconfigure and start the device depending on the current state.

Parameters:

• group_name – name of group to reconfigure.

• config_json – new group config json or None for empty config.

set_attr(name: str, value: Tuple[str, bytes] | numpy.ndarray | Dict[str, object] | int | float | bool | str | None, *, push_event=True)

Sets the internal attribute and optionally pushes a change event.

Parameters:

• name – name of attribute to set

• value – value to set

• push_event – Whether to push an event. Setting to None will push if polling is not active. Defaults to None.

async start_db_monitor()

Starts the database monitoring in a background task.

property static_attribute_names: list[str]

Get collection of static attribute names.

subarray_id: str

Subarray ID used when filtering configuration database configs.

async watch_database_and_reconfigure()

Long running awaitable that watches the configuration database and automatically tries reconfiguring and the device.

class ska_sdp_lmc_queue_connector.sdp_queue_connector.QueueConnectorDescriptor(*args: Any, **kwargs: Any)

Primary JSON serializable descriptor for configuring a queue connector device.

exchanges: dict[str | None, list[ExchangeDescriptor]] | list[ExchangeDescriptor] | None = None

Mapping or list of exchanges for a queue connector to process when running.

Exchange Interfaces

class ska_sdp_lmc_queue_connector.exchange.ExchangeDescriptor(*args: Any, **kwargs: Any)

Descriptor for instantiating an exchange.

dtype: Annotated[numpy.dtype, _DType] = 'str'

Python primitive, numpy dtype or tango dtype of the dynamic attribute

pipe: Annotated[DefaultPipeDescriptor | BufferWithTimePipeDescriptor, pydantic.Field]

A pipe operator to be applied between source and sink read and write

shape: list = []

The in-memory data shape in C-order

sink: Annotated[InMemorySinkDescriptor | TangoLocalAttributeSinkDescriptor | TangoArrayScatterAttributeSinkDescriptor | TangoObjectScatterAttributeSinkDescriptor | TangoRemoteAttributeSinkDescriptor | KafkaProducerSinkDescriptor | DataQueueProducerSinkDescriptor, pydantic.Field]

A data sink descriptor to be written to by the exchange

source: Annotated[InMemorySourceDescriptor | TangoSubscriptionSourceDescriptor | TangoDishPointingSubscriptionSourceDescriptor | TangoMccsPointingSubscriptionSourceDescriptor | TangoMccsTrackingSubscriptionSourceDescriptor | KafkaConsumerSourceDescriptor | DataQueueConsumerSourceDescriptor, pydantic.Field] | list[Annotated[InMemorySourceDescriptor | TangoSubscriptionSourceDescriptor | TangoDishPointingSubscriptionSourceDescriptor | TangoMccsPointingSubscriptionSourceDescriptor | TangoMccsTrackingSubscriptionSourceDescriptor | KafkaConsumerSourceDescriptor | DataQueueConsumerSourceDescriptor, pydantic.Field]]

One or more data source descriptors to be read by the exchange

class ska_sdp_lmc_queue_connector.exchange.Exchange(sources: Sequence[DataSource], sink: DataSink, pipe: DataPipe)

A container representing a connection between a source and sink and handles asynchronous streaming between them.

async run()

Asynchronously runs the exchange connecting source payloads to the sink.

Successful completion without exceptions guarantees that no payloads are lost between sources and sinks.

async start()

Invokes start on the sinks and sources.

async stop()

Invokes stop on the sinks and sources.

Data Source, Sink and Pipe Interfaces

class ska_sdp_lmc_queue_connector.sourcesink.DataSource

Interface for an object containing data that can be asynchronously read.

abstract async start()

Asynchronously performs all additional initialization before reading (e.g. connecting to socket endpoints)

abstract async stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

class ska_sdp_lmc_queue_connector.sourcesink.DataSink

Interface for an object that receives data that can be asynchronously written to.

abstract async awrite(value: Tuple[str, bytes] | numpy.ndarray | Dict[str, object] | int | float | bool | str)

Writes a single data entry to the sink.

abstract async start()

Asynchronously performs all additional initialization before writing (e.g. waiting for socket connections)

abstract async stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

class ska_sdp_lmc_queue_connector.sourcesink.DataPipe

Functor interface for pipe operators to perform on python data between a DataSource and DataSink.

abstract property output_dtype: numpy.dtype

Dtype of the functor stream output

abstract property output_shape: list[int]

Shape of the functor stream output

Tango

TangoSubscriptionSource

class ska_sdp_lmc_queue_connector.tango_sourcesink.TangoSubscriptionSourceDescriptor(*args: Any, **kwargs: Any)

attribute_name: str

Attribute name the subscription is to.

device_name: str

Device name containing the attribute the subscription is to.

etype: SchemaEventType = 0

The type of attribute event to listen for.

stateless: bool = True

When True will retry subscribing every 10 seconds if failed subscription, otherwise will raise an exception on start if False.

class ska_sdp_lmc_queue_connector.tango_sourcesink.TangoSubscriptionSource(desc: TangoSubscriptionSourceDescriptor, device: SDPQueueConnector, dtype: np.dtype, shape: list[int])

A DataSource populated from a subscription to tango attribute events.

start()

Asynchronously performs all additional initialization before reading (e.g. connecting to socket endpoints)

stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

TangoDishPointingSubscriptionSource

class ska_sdp_lmc_queue_connector.tango_pointing_source.TangoDishPointingSubscriptionSourceDescriptor(*args: Any, **kwargs: Any)

Descriptor to TangoDishPointingSubscriptionSource.

antenna_name: str | None = None

Antenna name to prepend to the data.

class ska_sdp_lmc_queue_connector.tango_pointing_source.TangoDishPointingSubscriptionSource(desc: TangoDishPointingSubscriptionSourceDescriptor, device: SDPQueueConnector, dtype: np.dtype, shape: list[int])

Specialized tango attribute source for converting Dish pointings to a structured ndarray.

TangoMccsPointingSubscriptionSource

class ska_sdp_lmc_queue_connector.tango_pointing_source.TangoMccsPointingSubscriptionSourceDescriptor(*args: Any, **kwargs: Any)

Descriptor to TangoMccsPointingSubscriptionSource.

station_beam_name: str | None = None

MCCS station beam name/id to prepend to the data.

class ska_sdp_lmc_queue_connector.tango_pointing_source.TangoMccsPointingSubscriptionSource(desc: TangoMccsPointingSubscriptionSourceDescriptor, device: SDPQueueConnector, dtype: np.dtype, shape: list[int])

Specialized tango attribute source for converting MCCS station beam pointings to a structured ndarray.

TangoMccsTrackingSubscriptionSource

class ska_sdp_lmc_queue_connector.tango_pointing_source.TangoMccsTrackingSubscriptionSourceDescriptor(*args: Any, **kwargs: Any)

Descriptor to TangoMccsTrackingSubscriptionSource.

station_beam_name: str | None = None

MCCS station beam name/id to prepend to the data.

class ska_sdp_lmc_queue_connector.tango_pointing_source.TangoMccsTrackingSubscriptionSource(desc: TangoMccsTrackingSubscriptionSourceDescriptor, device: SDPQueueConnector, dtype: np.dtype, shape: list[int])

Specialized tango attribute source for converting MCCS station beam tracking flags to a structured ndarray.

TangoAttributeSink

class ska_sdp_lmc_queue_connector.tango_sourcesink.TangoLocalAttributeSinkDescriptor(*args: Any, **kwargs: Any)

attribute_name: str

Attribute name to dynamically add to the SDPQueueConnector

default_value: SchemaDataType | None = None

Starting value of the dynamic attribute before a source is read

class ska_sdp_lmc_queue_connector.tango_sourcesink.TangoLocalAttributeSink(desc: TangoLocalAttributeSinkDescriptor, device: SDPQueueConnector, dtype: np.dtype, shape: list[int])

A DataSink that publishes data to a tango attribute on the Tango Device member.

awrite(value: Tuple[str, bytes] | numpy.ndarray | Dict[str, object] | int | float | bool | str)

Writes a single data entry to the sink.

start()

Asynchronously performs all additional initialization before writing (e.g. waiting for socket connections)

stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

TangoArrayScatterAttributeSink

class ska_sdp_lmc_queue_connector.tango_array_scatter_sink.TangoArrayScatterAttributeSinkDescriptor(*args: Any, **kwargs: Any)

A Tango Attribute Sink for splitting and scattering of ndarray data.

attribute_names: list[str]

Ordered attributute names to dynamically add the SDPQueueConnector

attribute_shape_names: list[str] | None = None

Optional attribute shapes names

axis: int = 0

Axis to split on. Non-zero values may reduce performance.

default_value: SchemaDataType = ''

Starting value of the dynamic attribute before a source is read

indices: list[int] | None = None

Optional ordered indexes of the split boundaries

class ska_sdp_lmc_queue_connector.tango_array_scatter_sink.TangoArrayScatterAttributeSink(desc: TangoArrayScatterAttributeSinkDescriptor, device: SDPQueueConnector, dtype: np.dtype, shape: list[int])

awrite(value: Tuple[str, bytes] | numpy.ndarray | Dict[str, object] | int | float | bool | str)

Writes a single data entry to the sink.

start()

Asynchronously performs all additional initialization before writing (e.g. waiting for socket connections)

stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

TangoObjectScatterAttributeSink

class ska_sdp_lmc_queue_connector.tango_object_scatter_sink.TangoObjectScatterAttributeSinkDescriptor(*args: Any, **kwargs: Any)

A Tango Attribute Sink for splitting and scattering object heirarchy data.

attributes: list[TangoAttributeDescriptor]

Attribute names to dynamically add the SDPQueueConnector

class ska_sdp_lmc_queue_connector.tango_object_scatter_sink.TangoAttributeDescriptor(*args: Any, **kwargs: Any)

An attribute descriptor describing a filter and path to a value in JSON.

attribute_name: str

Attribute name to dynamically add to the SDPQueueConnector

default_value: SchemaDataType | None = None

Starting value of the dynamic attribute before a source is read

dtype: DType

Python primitive, numpy dtype or tango dtype of the dynamic attribute

filter: str | None = None

JMESPath predicate expression for whether to update the attribute.

For testing on https://play.jmespath.org, the deserialized message can be converted to the equivalent JSON using the following:

>>> import json, numpy
>>> class NumpyArrayJSONEncoder(json.JSONEncoder):
...     def default(self, obj):
...         if isinstance(obj, numpy.ndarray):
...             return obj.tolist()
...         return json.JSONEncoder.default(self, obj)
...
>>> example = numpy.array([("a", 1)], dtype=[("name", object),("value", int)])
>>> json.dumps(example, cls=NumpyArrayJSONEncoder)
'[["a", 1]]'

path: str = '@'

JMESPath expression for the location of the data to extract

shape: list[int] = []

Maximum shape of the dynamic attribute

class ska_sdp_lmc_queue_connector.tango_object_scatter_sink.TangoObjectScatterAttributeSink(desc: TangoObjectScatterAttributeSinkDescriptor, device: SDPQueueConnector, dtype: np.dtype, shape: list[int])

awrite(value: Tuple[str, bytes] | numpy.ndarray | Dict[str, object] | int | float | bool | str)

Writes a single data entry to the sink.

start()

Asynchronously performs all additional initialization before writing (e.g. waiting for socket connections)

stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

Kafka

KafkaConsumerSource

class ska_sdp_lmc_queue_connector.kafka_sourcesink.KafkaConsumerSourceDescriptor(*args: Any, **kwargs: Any)

format: Literal['utf-8', 'ascii', 'python', 'json', 'msgpack_numpy', 'npy', 'carray'] = 'python'

Format of the message bytes

servers: str | list[str]

The Kafka broker(s) to query for metadata and setup the connection

topic: str

The Kafka topic to read from

class ska_sdp_lmc_queue_connector.kafka_sourcesink.KafkaConsumerSource(descriptor: KafkaConsumerSourceDescriptor, dtype: numpy.dtype, shape: list[int])

A DataSource which consumes messages from a Kafka topic

start()

Asynchronously performs all additional initialization before reading (e.g. connecting to socket endpoints)

async stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

KafkaProducerSink

class ska_sdp_lmc_queue_connector.kafka_sourcesink.KafkaProducerSinkDescriptor(*args: Any, **kwargs: Any)

class TimestampOptions(*args: Any, **kwargs: Any)

A set of kafka producer options related to overriding Kafka message timestamps from dynamic data instead of the system time when messages are produced. Timestamps in dynamic data must be one of:

• An offset to unix epoch in milliseconds.

• A numpy datetime64 on TAI scale.

key: str | None = None

Timestamp key for dictionary-like data types.

reducer: Literal['min', 'max', 'mean'] | None = None

Axes reduce operation for timestamps.

slices: tuple[int | Annotated[slice, _Slice], ...] = ()

Timestamp slice location for multidimensional data. Size must match number of dimensions.

format: Literal['utf-8', 'ascii', 'python', 'json', 'msgpack_numpy', 'npy', 'carray'] = 'python'

The format of the Kafka message.

message_max_bytes: int = 1048576

The max size of encoded messages in bytes. NOTE: The 1MiB default is recommended by Kafka, sending larger messages higher than this will linearly increase latency.

servers: str | list[str]

The Kafka broker(s) to query for metadata and setup the connection

timestamp_options: TimestampOptions | None = None

An optional group of settings related to extracting Kafka timestamps from dynamic data. None results in using the current time.

topic: str

The Kafka topic to write messages to

class ska_sdp_lmc_queue_connector.kafka_sourcesink.KafkaProducerSink(descriptor: KafkaProducerSinkDescriptor, dtype: numpy.dtype, shape: list[int])

A DataSink which produces messages for a Kafka topic

awrite(value: Tuple[str, bytes] | numpy.ndarray | Dict[str, object] | int | float | bool | str)

Writes a single data entry to the sink.

start()

Asynchronously performs all additional initialization before writing (e.g. waiting for socket connections)

stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

In-Memory

InMemorySource

class ska_sdp_lmc_queue_connector.in_memory_sourcesink.InMemorySourceDescriptor(*args: Any, **kwargs: Any)

Descriptor for instantiating an InMemorySource

data: list[Annotated[tuple[str, Annotated[bytes, _B64Bytes]] | Annotated[numpy.ndarray, _NDArray] | dict[str, Any] | int | float | bool | str, pydantic.Field]]

Data values to be read from the source to a sink

delay: float = 0

Time delay in seconds before the next data value is available to read

class ska_sdp_lmc_queue_connector.in_memory_sourcesink.InMemorySource(desc: InMemorySourceDescriptor, dtype: numpy.dtype, shape: list[int] | None = None)

An in-memory implementation of a DataSource for testing.

start()

Asynchronously performs all additional initialization before reading (e.g. connecting to socket endpoints)

stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

InMemorySink

class ska_sdp_lmc_queue_connector.in_memory_sourcesink.InMemorySinkDescriptor(*args: Any, **kwargs: Any)

Descriptor for instantiating an InMemorySink

key: str

Key for accessing the stored data queue via classmethod

class ska_sdp_lmc_queue_connector.in_memory_sourcesink.InMemorySink(desc: InMemorySinkDescriptor)

An in-memory implementation of a DataSink for testing. Instances of

sink data queues exist as class members referenced via a lookup key.

awrite(value: Tuple[str, bytes] | numpy.ndarray | Dict[str, object] | int | float | bool | str)

Writes a single data entry to the sink.

start()

Asynchronously performs all additional initialization before writing (e.g. waiting for socket connections)

stop()

Asynchronously performs all end of stream destruction after reading (e.g. closing sockets)

Pipes

class ska_sdp_lmc_queue_connector.pipe.default_pipe.DefaultPipe(dtype: numpy.dtype, shape: list[int])

Simple pipe that does not modify the data stream.

property output_dtype: numpy.dtype

Dtype of the functor stream output

property output_shape: list[int]

Shape of the functor stream output

class ska_sdp_lmc_queue_connector.pipe.buffer_pipe.BufferWithTimePipe(desc: BufferWithTimePipeDescriptor, dtype: numpy.dtype, shape: list[int])

Buffers data with time window by dynamically by adding a dynamic dimension to the ouput shape and appending data until the time window closes.

property output_dtype: numpy.dtype

Dtype of the functor stream output

property output_shape: list[int]

Shape of the functor stream output

Format

ska_sdp_lmc_queue_connector.serialization.Format

Supported serialized format of structured data.

• utf-8: string data format using python __str__ method encoded as utf-8

• ascii: string data format using python __str__ method encoded as ascii

• python: python literal from python __repr__ method encoded as utf-8

• json: json format using utf-8 format

• msgpack_numpy: binary format for JSON-like data combined with numpy support

• npy: binary format for numpy arrays containing type and shape

• carray: raw data buffer in C/row major order