multi_topic_support

Shared UMAA support runtime: GUID helpers, path/navigation, combined sample/builder, editable single-object facade, element views, and top-level reader/writer adapters.

This module is the glue that lets users interact with any nested UMAA multi-topic structure (generalization/specialization, large sets, large lists) as if it were one normal Python object — both when READING (assembled views) and WRITING (editable combined builders).

Key ideas

GUID helpers: robust hashing/equality for UMAA’s 16-digit NumericGUID.
Paths: every node in the multi-topic tree (root, specialization nodes, set/list elements) has a path; special tokens address collection elements.
CombinedSample (reader) & CombinedBuilder (writer): carry overlays and per-node collection bags keyed by these paths.
OverlayView / ElementView (reader) and BuilderEditView / CombinedEditHandle (writer): “single-object” interfaces that let users read/edit nested content naturally (e.g., v.objective.speed or cmd.objective.collections["waypoints"]).
Adapters: UmaaReaderAdapter / UmaaFilteredReaderAdapter / UmaaWriterAdapter expose DDS-like ergonomics (listeners, read/take) while the UMAA graph handles the heavy lifting behind the scenes.

Classes:

`BuilderEditView`(builder[, path])	Write-through overlay view for CombinedBuilder, path-aware.
`CombinedBuilder`(base[, overlays_by_path, ...])	Editable, path-aware combined sample for publishing nested UMAA graphs.
`CombinedEditHandle`(builder)	One-object facade around a CombinedBuilder for ergonomic editing.
`CombinedSample`(base[, collections, ...])	Assembled, read-only sample composed from multiple UMAA topics.
`ElementHandle`(builder, elem_path, base_elem)	Tiny writer facade rooted at a collection element node path.
`ElementView`(combined, elem, path)	Read-only proxy for a collection element that knows its absolute path.
`ForwardingReaderListener`(adapter)	Internal listener installed on the root RTI DataReader.
`ForwardingWriterListener`(adapter)	Internal listener installed on each RTI DataWriter in the UMAA writer tree.
`ListCollection`()	Mutable list-like collection for building UMAA Large Lists at runtime.
`ListEditor`(builder, node_path, list_name, ...)	Convenience API to append/edit elements of a list at a given node path.
`OverlayView`(base, collections[, ...])	Read-only overlay view supporting nested overlays and per-node collections.
`SetCollection`()	Mutable set-like collection for building UMAA Large Sets at runtime.
`SetEditor`(builder, node_path, set_name, ...)	Convenience API to add/edit elements of a set at a given node path.
`UmaaFilteredReaderAdapter`(root_node, ...)	Like `UmaaReaderAdapter`, but for a root bound to a ContentFilteredTopic.
`UmaaReaderAdapter`(root_node, root_reader)	Adapter that makes a UMAA reader graph feel like an RTI DataReader.
`UmaaWriterAdapter`(root_node, top_level, ...)	Adapter that makes a UMAA writer graph feel like an RTI DataWriter.

Functions:

`get_at_path`(obj, path)	Navigate attributes using a path of names.
`set_at_path`(root, path, value)	Set an attribute at a nested path.

class umaapy.util.multi_topic_support.BuilderEditView(builder, path=())[source]

Bases: object

Write-through overlay view for CombinedBuilder, path-aware.

Getting nested attributes returns scoped BuilderEditView`s when the value looks like a user object (has `__dict__/__slots__).
Setting attributes writes into the overlay object when present; otherwise into the base at the current path.
collections at any path returns the per-node bag for lists/sets.

class umaapy.util.multi_topic_support.CombinedBuilder(base, overlays_by_path=<factory>, collections_by_path=<factory>)[source]

Bases: object

Editable, path-aware combined sample for publishing nested UMAA graphs.

Parameters:

base (Any) – The base/root object to publish.
collections_by_path (Dict[Tuple[Any, ...], Dict[str, Any]], optional) – Per-node collections bags keyed by absolute path.
overlays_by_path (Dict[Tuple[Any, ...], Any], optional) – Per-node specialization overlays keyed by absolute path.

Attributes:

`base`
`collections_by_path`
`overlays_by_path`

Methods:

`collections_at`([path])	Get (and create if absent) the per-node collections bag at a given path.
`ensure_collection_at`(name, kind[, path])	Ensure a per-node collection exists at a given path.
`overlay_at`([path])	Get the specialization overlay at a given path, if any.
`spawn_child`(base_obj[, path])	Spawn a child builder scoped to path, rebasing nested overlays/collections.
`use_specialization_at`(spec_obj[, path])	Set a specialization overlay at a given path.

base: Any

collections_at(path=())[source]

Get (and create if absent) the per-node collections bag at a given path.

Parameters:: path (Sequence[Any]) – Absolute node path.
Returns:: The collections bag dictionary.
Return type:: Dict[str, Any]

collections_by_path: Dict[Tuple[Any, ...], Dict[str, Any]]

ensure_collection_at(name, kind, path=())[source]

Ensure a per-node collection exists at a given path.

Parameters:

path (Sequence[Any]) – Absolute node path (attributes and/or element tokens).
name (str) – Collection logical name (e.g., ‘waypoints’).
kind ({'set', 'list'}) – Collection kind.

Returns:

A SetCollection or ListCollection instance.

Return type:

Any

overlay_at(path=())[source]

Get the specialization overlay at a given path, if any.

Return type:: Optional[Any]

overlays_by_path: Dict[Tuple[Any, ...], Any]

spawn_child(base_obj, path=())[source]

Spawn a child builder scoped to path, rebasing nested overlays/collections.

This is used by writer decorators to publish elements and nested content under a collection element or a specialization node.

Parameters:

path (Sequence[Any]) – Absolute node path for the child.
base_obj (Any) – The child’s base object.

Returns:

A child builder that carries only the relevant nested bags/overlays.

Return type:

CombinedBuilder

use_specialization_at(spec_obj, path=())[source]

Set a specialization overlay at a given path.

Parameters:

path (Sequence[Any]) – Absolute path where the generalization field lives (e.g., (‘objective’,)).
spec_obj (Any) – Specialization object instance.

Return type:

None

class umaapy.util.multi_topic_support.CombinedEditHandle(builder)[source]

Bases: object

One-object facade around a CombinedBuilder for ergonomic editing.

Access attributes directly (write-through), and use nested .collections to add list/set elements at any node.

Attributes:

builder

Return the underlying CombinedBuilder.

property builder: CombinedBuilder: Return the underlying CombinedBuilder.

class umaapy.util.multi_topic_support.CombinedSample(base, collections=<factory>, overlays_by_path=<factory>)[source]

Bases: object

Assembled, read-only sample composed from multiple UMAA topics.

Parameters:

base (Any) – The base/root sample (e.g., the metadata or generalization-containing message).
collections (Dict[str, Any], optional) – Per-node collections bag at the current node. Nested nodes use overlays_by_path.
overlays_by_path (Dict[Tuple[Any, ...], Any], optional) – Nested overlays keyed by their absolute attribute/element path.

Methods:

`add_overlay_at`(overlay_obj[, path])	Register a nested overlay at an absolute path.
`clone_with_collections`(updates)	Return a new CombinedSample with updated local collections bag.

Attributes:

`base`
`collections`
`overlays_by_path`
`view`	Return a read-only overlay view for user access.

add_overlay_at(overlay_obj, path=())[source]

Register a nested overlay at an absolute path.

Parameters:

path (Sequence[Any]) – Absolute path (attributes and/or element tokens).
overlay_obj (Any) – Overlay object to merge at that path.

Returns:

A new CombinedSample with the overlay registered.

Return type:

CombinedSample

base: Any

clone_with_collections(updates)[source]

Return a new CombinedSample with updated local collections bag.

Return type:: CombinedSample

collections: Dict[str, Any]

overlays_by_path: Dict[Tuple[Any, ...], Any]

property view: OverlayView: Return a read-only overlay view for user access.

class umaapy.util.multi_topic_support.ElementHandle(builder, elem_path, base_elem)[source]

Bases: object

Tiny writer facade rooted at a collection element node path.

Provides convenience for choosing specializations and adding nested collections.

Attributes:

collections

Return the per-node collections bag under this element.

Methods:

`ensure_collection`(name, kind)	Ensure a collection under this element node.
`use_specialization`(spec_type, *[, at])	Attach a specialization under the element node.

property collections: Dict[str, Any]: Return the per-node collections bag under this element.

ensure_collection(name, kind)[source]

Ensure a collection under this element node.

Return type:: Any

use_specialization(spec_type, *, at=None)[source]

Attach a specialization under the element node.

Parameters:

spec_type (type) – The specialization class to instantiate.
at (Sequence[str], optional) – Path under this element where the generalization lives (defaults to the only generalization if exactly one exists).

Returns:

The specialization object instance for further editing.

Return type:

Any

class umaapy.util.multi_topic_support.ElementView(combined, elem, path)[source]

Bases: object

Read-only proxy for a collection element that knows its absolute path.

This proxy allows nested overlay access beneath a collection element, e.g.:

for task in combined.view.missionPlan.collections["taskPlan"]:
    print(task.objective.speed)  # specialization under that element

class umaapy.util.multi_topic_support.ForwardingReaderListener(adapter)[source]

Bases: NoOpDataReaderListener

Internal listener installed on the root RTI DataReader.

On DATA_AVAILABLE: triggers UMAA assembly via root_node.poll_once(), then forwards on_data_available if enabled by the user’s mask.
All other reader events are forwarded per the user’s mask as-is.

Methods:

`on_data_available`(reader)
`on_liveliness_changed`(reader, status)
`on_reliable_reader_activity_changed`(reader, ...)
`on_reliable_reader_cache_changed`(reader, status)
`on_requested_deadline_missed`(reader, status)
`on_requested_incompatible_qos`(reader, status)
`on_sample_lost`(reader, status)
`on_sample_rejected`(reader, status)
`on_subscription_matched`(reader, status)

on_data_available(reader)[source]

on_liveliness_changed(reader, status)[source]

on_reliable_reader_activity_changed(reader, status)[source]

on_reliable_reader_cache_changed(reader, status)[source]

on_requested_deadline_missed(reader, status)[source]

on_requested_incompatible_qos(reader, status)[source]

on_sample_lost(reader, status)[source]

on_sample_rejected(reader, status)[source]

on_subscription_matched(reader, status)[source]

class umaapy.util.multi_topic_support.ForwardingWriterListener(adapter)[source]

Bases: NoOpDataWriterListener

Internal listener installed on each RTI DataWriter in the UMAA writer tree.

Forwards events to UmaaWriterAdapter, which filters per the user’s mask.

Methods:

`on_instance_replaced`(writer, handle)
`on_liveliness_lost`(writer, status)
`on_offered_deadline_missed`(writer, status)
`on_offered_incompatible_qos`(writer, status)
`on_publication_matched`(writer, status)
`on_reliable_reader_activity_changed`(writer, ...)
`on_reliable_writer_cache_changed`(writer, status)

on_instance_replaced(writer, handle)[source]

on_liveliness_lost(writer, status)[source]

on_offered_deadline_missed(writer, status)[source]

on_offered_incompatible_qos(writer, status)[source]

on_publication_matched(writer, status)[source]

on_reliable_reader_activity_changed(writer, status)[source]

on_reliable_writer_cache_changed(writer, status)[source]

class umaapy.util.multi_topic_support.ListCollection[source]

Bases: object

Mutable list-like collection for building UMAA Large Lists at runtime.

Elements must have an elementID attribute.
Append/insert supported; element linking is handled by the writer decorator.

Methods:

`append`(elem)	Append an element.
`insert`(index, elem)	Insert an element at a given index.
`pop`([index])	Pop and return an element.
`to_runtime`()	Return list snapshot of elements.

append(elem)[source]

Append an element.

Return type:: None

insert(index, elem)[source]

Insert an element at a given index.

Return type:: None

pop(index=-1)[source]

Pop and return an element.

Return type:: Any

to_runtime()[source]

Return list snapshot of elements.

Return type:: List[Any]

class umaapy.util.multi_topic_support.ListEditor(builder, node_path, list_name, element_type)[source]

Bases: object

Convenience API to append/edit elements of a list at a given node path.

Methods:

`append`(elem)	Append an existing element instance and return its handle.
`append_new`(*[, element_id])	Append a new element and return its handle.

append(elem)[source]

Append an existing element instance and return its handle.

Return type:: ElementHandle

append_new(*, element_id=None)[source]

Append a new element and return its handle.

Return type:: ElementHandle

class umaapy.util.multi_topic_support.OverlayView(base, collections, overlays_by_path=(), path=())[source]

Bases: object

Read-only overlay view supporting nested overlays and per-node collections.

Attribute resolution order:

If a nested overlay is registered for the next attribute hop, return a new OverlayView scoped to that subpath.
If the top-level overlay has the attribute, return it.
Otherwise, return the attribute from the base object.
If the attribute equals a collection name, return the collection.

class umaapy.util.multi_topic_support.SetCollection[source]

Bases: object

Mutable set-like collection for building UMAA Large Sets at runtime.

Elements must have an elementID attribute.
Order is not guaranteed.

Methods:

`add`(elem)	Add an element; replaces existing entry with same ID.
`discard`(element_id)	Remove an element by ID if present.
`to_runtime`()	Return a stable list snapshot of elements.

add(elem)[source]

Add an element; replaces existing entry with same ID.

Return type:: None

discard(element_id)[source]

Remove an element by ID if present.

Return type:: None

to_runtime()[source]

Return a stable list snapshot of elements.

Return type:: List[Any]

class umaapy.util.multi_topic_support.SetEditor(builder, node_path, set_name, element_type)[source]

Bases: object

Convenience API to add/edit elements of a set at a given node path.

Methods:

`add`(elem)	Add an existing element instance and return its handle.
`add_new`(*[, element_id])	Create a new element with (optional) elementID and return its handle.

add(elem)[source]

Add an existing element instance and return its handle.

Return type:: ElementHandle

add_new(*, element_id=None)[source]

Create a new element with (optional) elementID and return its handle.

Return type:: ElementHandle

class umaapy.util.multi_topic_support.UmaaFilteredReaderAdapter(root_node, root_reader, cft)[source]

Bases: UmaaReaderAdapter

Like UmaaReaderAdapter, but for a root bound to a ContentFilteredTopic.

Attributes:

content_filtered_topic

Access the root ContentFilteredTopic.

Methods:

topic_name()

Return the filtered topic name.

property content_filtered_topic: rti.connextdds.ContentFilteredTopic: Access the root ContentFilteredTopic.

topic_name()[source]

Return the filtered topic name.

Return type:: str

class umaapy.util.multi_topic_support.UmaaReaderAdapter(root_node, root_reader)[source]

Bases: object

Adapter that makes a UMAA reader graph feel like an RTI DataReader.

Supports: - set_listener(listener, status_mask): full reader listener surface. - read()/take() -> (samples, infos) where infos[i].valid can be False (disposes). - read_data()/take_data() -> valid samples only (no infos). - __getattr__: delegates unknown attributes to the underlying RTI reader.

Attributes:

raw_reader

Access the underlying RTI DataReader.

Methods:

`read`()	Return a snapshot of buffered records without clearing.
`read_data`()	Return valid `CombinedSample`s only (no infos), without clearing.
`set_listener`(listener[, status_mask])	Register a user DataReaderListener-like object with a status mask.
`take`()	Return and clear buffered records.
`take_data`()	Return and clear valid `CombinedSample`s only (no infos).

property raw_reader: rti.connextdds.DataReader: Access the underlying RTI DataReader.

read()[source]

Return a snapshot of buffered records without clearing.

Returns:: (samples, infos) – samples[i] is a CombinedSample or None if infos[i].valid == False.
Return type:: (list, list)

read_data()[source]: Return valid `CombinedSample`s only (no infos), without clearing.

set_listener(listener, status_mask=rti.connextdds.StatusMask.NONE)[source]

Register a user DataReaderListener-like object with a status mask.

The listener is not installed on the underlying root reader; instead, we install a single internal listener and forward events to user callbacks after UMAA assembly when appropriate (e.g., on_data_available).

Return type:: None

take()[source]

Return and clear buffered records.

Returns:: (samples, infos)
Return type:: (list, list)

take_data()[source]: Return and clear valid `CombinedSample`s only (no infos).

class umaapy.util.multi_topic_support.UmaaWriterAdapter(root_node, top_level, root_writer)[source]

Bases: object

Adapter that makes a UMAA writer graph feel like an RTI DataWriter.

Supports: - .new() returning a CombinedBuilder, - .new_combined(…) returning a CombinedEditHandle (single-object facade), - .write(…) accepting either, splitting and linking metadata automatically, - set_listener(listener, status_mask) forwarding DataWriter events after internal handling, - __getattr__ delegation to the root writer.

Methods:

`editor_for_list`(handle_or_builder, path, ...)	Create a `ListEditor` rooted at a node path.
`editor_for_set`(handle_or_builder, path, ...)	Create a `SetEditor` rooted at a node path.
`new`()	Create a new CombinedBuilder using the top-level writer's base factory.
`new_combined`(*[, spec_at, spec_type, ...])	Create a one-object editable combined handle with an optional nested specialization and pre-created collections at that path.
`set_listener`(listener[, status_mask])	Register a user DataWriterListener-like object with a status mask.
`topic_name`()	Return the topic name of the root writer.
`write`(builder_or_handle)	Publish a combined sample.

Attributes:

raw_writer

Access the underlying RTI DataWriter.

editor_for_list(handle_or_builder, path, list_name, element_type)[source]

Create a ListEditor rooted at a node path.

Usage Examples

from umaapy import get_configurator, reset_dds_participant
from umaapy.umaa_types import (
    UMAA_MM_ObjectiveExecutorControl_ObjectiveExecutorCommandType as ObjectiveExecutorCommandType,
    UMAA_MM_BaseType_RouteObjectiveType as RouteObjectiveType,
)

reset_dds_participant()
w = get_configurator().get_umaa_writer(ObjectiveExecutorCommandType)
r = get_configurator().get_umaa_reader(ObjectiveExecutorCommandType)

cmd = w.new_combined(spec_at=("objective",), spec_type=RouteObjectiveType)
cmd.objective.name = "My objective"
w.write(cmd)

for cs in r.take_data():
    print(cs.view.objective.name)