BaseComponent

BaseComponent — abstract base class for UMAA application components.

A component composes multiple BaseService instances into a cohesive UMAA application (e.g. an autopilot, a mission executor).

Lifecycle (managed automatically by DDSContext):

1. __init__(ctx, name) — create DDS services, register with context.

2. on_start() — one-shot startup (publish initial messages, etc.).

3. _run() — async coroutine for periodic / event-driven work.

4. close() — publish final messages, release resources.

class rtiumaapy.base_component.BaseComponent(ctx: DDSContext, component_name: str)

Bases: ABC

Abstract base class for UMAA application components.

Subclasses must implement _run() and close(). Subclasses may override on_start() for one-shot startup work.

The component registers itself with the DDSContext so that DDSContext.run_until_shutdown() calls on_start() once, then starts _run(), and calls close() during shutdown (while DDS writers are still open).

__init__(ctx: DDSContext, component_name: str) → None

Initialise the component and register it with the context.

Parameters:

• ctx – The DDSContext that owns shared DDS infrastructure.

• component_name – A unique name for this component instance (e.g. "Autopilot", "MissionExecutor").

property ctx: DDSContext

The DDSContext this component belongs to.

property component_name: str

The registered name of this component.

async on_start() → None

One-shot startup hook — called before _run().

Override to publish initial announcements (specs, capabilities, startup log entries, etc.). The default implementation is a no-op.

abstractmethod async close() → None

Cleanup — called by DDSContext.shutdown() while writers are still open.

Override to publish final status messages, log shutdown events, etc.

Usage

A component groups multiple services into a cohesive UMAA application. The lifecycle is managed by DDSContext:

1. __init__() — create services, register with context

2. on_start() — one-shot startup (publish announcements)

3. _run() — async coroutine for periodic work

4. close() — publish final messages, release resources

from rtiumaapy import DDSContext, BaseComponent

class MyComponent(BaseComponent):
    def __init__(self, ctx: DDSContext, source_id):
        super().__init__(ctx, "MyComponent")
        # Create services here...

    async def on_start(self):
        print("Component started — publishing specs")

    async def _run(self):
        while True:
            await asyncio.sleep(1.0)

    async def close(self):
        print("Shutting down")