Skip to content

Platform Adapters

Platform adapters bridge hardware-specific APIs (battery sensors, robot APIs, ROS topics, etc.) to the PhyTrace UDM pipeline. They sit below the Emitters layer and implement a standard interface so that any hardware source can be used with a PeriodicEmitter without writing custom collector code each time.

Architecture

Hardware / Platform API
        │
        ▼
PlatformAdapter (phytrace.adapters.base)
        │  wraps
        ▼
AdapterCollector (phytrace.adapters.collector)
        │  conforms to TelemetryCollector
        ▼
PeriodicEmitter  →  UDMEvent  →  Transport

PlatformAdapter

PlatformAdapter

Bases: ABC

Abstract base class for platform-specific hardware adapters.

Implement this class to integrate hardware callbacks (battery sensors, robot APIs, ROS topics, etc.) with the PhyTrace SDK without writing custom collector code each time.

The adapter is intended to be wrapped by an :class:AdapterCollector which makes it compatible with :class:~phytrace.emitters.PeriodicEmitter.

Lifecycle

The optional :meth:setup and :meth:teardown hooks are called by :class:AdapterCollector on start and stop respectively, allowing adapters to open/close hardware connections.

Example

::

class BatteryAdapter(PlatformAdapter):
    def __init__(self, robot_api, source_id: str):
        self._api = robot_api
        self._source_id = source_id

    @property
    def source_id(self) -> str:
        return self._source_id

    @property
    def source_type(self) -> SourceType:
        return SourceType.AMR

    async def collect(self) -> UDMEvent:
        level = await self._api.get_battery_level()
        return (
            UDMEventBuilder(
                source_id=self.source_id,
                source_type=self.source_type,
            )
            .with_power(battery_soc_pct=level)
            .build()
        )

Attributes

source_id abstractmethod property 

source_id: str

Return the unique source identifier for this adapter.

source_type abstractmethod property 

source_type: SourceType

Return the UDM source type for this adapter.

Methods:

collect abstractmethod async 

collect() -> UDMEvent

Collect the current telemetry snapshot from the platform.

RETURNS DESCRIPTION
UDMEvent

A fully constructed :class:~phytrace.models.event.UDMEvent.
Source code in phytrace/adapters/base.py 
72
73
74
75
76
77
78
79
80
@abstractmethod
async def collect(self) -> UDMEvent:
    """
    Collect the current telemetry snapshot from the platform.

    Returns:
        A fully constructed :class:`~phytrace.models.event.UDMEvent`.
    """
    ...

setup async 

setup() -> None

Optional hook called before the first :meth:collect invocation.

Use this to open hardware connections, subscribe to topics, or perform any other initialisation that should happen once.

Source code in phytrace/adapters/base.py 
82
83
84
85
86
87
88
async def setup(self) -> None:  # noqa: B027
    """
    Optional hook called before the first :meth:`collect` invocation.

    Use this to open hardware connections, subscribe to topics, or
    perform any other initialisation that should happen once.
    """

teardown async 

teardown() -> None

Optional hook called after the last :meth:collect invocation.

Use this to cleanly close hardware connections or unsubscribe from topics when the collector is stopped.

Source code in phytrace/adapters/base.py 
90
91
92
93
94
95
96
async def teardown(self) -> None:  # noqa: B027
    """
    Optional hook called after the last :meth:`collect` invocation.

    Use this to cleanly close hardware connections or unsubscribe from
    topics when the collector is stopped.
    """

AdapterCollector

AdapterCollector 

AdapterCollector(adapter: PlatformAdapter)

Bases: TelemetryCollector

A :class:~phytrace.emitters.periodic.TelemetryCollector that delegates to a :class:PlatformAdapter.

This is the primary bridge between the Platform Adapters layer and the Event Emitters layer. Pass an AdapterCollector to a :class:~phytrace.emitters.PeriodicEmitter to start emitting UDM events from your hardware adapter without writing custom collector code.

Lifecycle methods (:meth:setup / :meth:teardown) on the wrapped adapter are exposed and should be called explicitly (or via an async context manager pattern) when the collector is started/stopped.

Example

::

adapter  = BatteryAdapter(robot_api, source_id="robot-001")
collector = AdapterCollector(adapter)

emitter = PeriodicEmitter(
    collector=collector,
    interval_ms=1000,
    on_event=lambda e: agent.emit(e),
)

await collector.setup()
await emitter.start()
# … later …
await emitter.stop()
await collector.teardown()

Initialise the collector with a platform adapter.

PARAMETER DESCRIPTION
adapter

The :class:PlatformAdapter to delegate collection to.

TYPE: PlatformAdapter

Source code in phytrace/adapters/collector.py 
48
49
50
51
52
53
54
55
def __init__(self, adapter: PlatformAdapter) -> None:
    """
    Initialise the collector with a platform adapter.

    Args:
        adapter: The :class:`PlatformAdapter` to delegate collection to.
    """
    self._adapter = adapter

Attributes

adapter property 

adapter: PlatformAdapter

Return the underlying :class:PlatformAdapter.

Methods:

setup async 

setup() -> None

Call :meth:~PlatformAdapter.setup on the wrapped adapter.

Source code in phytrace/adapters/collector.py 
76
77
78
async def setup(self) -> None:
    """Call :meth:`~PlatformAdapter.setup` on the wrapped adapter."""
    await self._adapter.setup()

teardown async 

teardown() -> None

Call :meth:~PlatformAdapter.teardown on the wrapped adapter.

Source code in phytrace/adapters/collector.py 
80
81
82
async def teardown(self) -> None:
    """Call :meth:`~PlatformAdapter.teardown` on the wrapped adapter."""
    await self._adapter.teardown()

SimplePlatformAdapter

SimplePlatformAdapter 

SimplePlatformAdapter(source_id: str, source_type: SourceType, collect_func: Callable[[], UDMEventBuilder | Awaitable[UDMEventBuilder]], setup_func: Callable[[], None | Awaitable[None]] | None = None, teardown_func: Callable[[], None | Awaitable[None]] | None = None)

Bases: PlatformAdapter

A :class:PlatformAdapter backed by a plain callback function.

Useful for quick hardware integrations and testing without needing to subclass :class:PlatformAdapter directly.

The collect_func may be either a synchronous callable returning a :class:~phytrace.core.builder.UDMEventBuilder, or an async callable returning the same. Both forms are supported.

Example – synchronous callback

::

def read_battery() -> UDMEventBuilder:
    level = hardware_api.battery_level()          # blocking HW call
    return (
        UDMEventBuilder(
            source_id="robot-001",
            source_type=SourceType.AMR,
        )
        .with_power(battery_soc_pct=level)
    )

adapter = SimplePlatformAdapter(
    source_id="robot-001",
    source_type=SourceType.AMR,
    collect_func=read_battery,
)
collector = AdapterCollector(adapter)
emitter   = PeriodicEmitter(collector=collector, interval_ms=1000)

Example – async callback

::

async def read_battery_async() -> UDMEventBuilder:
    level = await async_hardware_api.battery_level()
    return (
        UDMEventBuilder(
            source_id="robot-001",
            source_type=SourceType.AMR,
        )
        .with_power(battery_soc_pct=level)
    )

adapter = SimplePlatformAdapter(
    source_id="robot-001",
    source_type=SourceType.AMR,
    collect_func=read_battery_async,
)

Initialise the simple adapter.

PARAMETER DESCRIPTION
source_id

Unique source identifier passed through to the UDM event.

TYPE: str

source_type

UDM source type classification.

TYPE: SourceType

collect_func

Callable (sync or async) that returns a configured :class:~phytrace.core.builder.UDMEventBuilder.

TYPE: Callable[[], UDMEventBuilder | Awaitable[UDMEventBuilder]]

setup_func

Optional callable (sync or async) invoked during :meth:setup.

TYPE: Callable[[], None | Awaitable[None]] | None DEFAULT: None

teardown_func

Optional callable (sync or async) invoked during :meth:teardown.

TYPE: Callable[[], None | Awaitable[None]] | None DEFAULT: None

Source code in phytrace/adapters/simple.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def __init__(
    self,
    source_id: str,
    source_type: SourceType,
    collect_func: Callable[[], UDMEventBuilder | Awaitable[UDMEventBuilder]],
    setup_func: Callable[[], None | Awaitable[None]] | None = None,
    teardown_func: Callable[[], None | Awaitable[None]] | None = None,
) -> None:
    """
    Initialise the simple adapter.

    Args:
        source_id: Unique source identifier passed through to the UDM event.
        source_type: UDM source type classification.
        collect_func: Callable (sync or async) that returns a configured
            :class:`~phytrace.core.builder.UDMEventBuilder`.
        setup_func: Optional callable (sync or async) invoked during
            :meth:`setup`.
        teardown_func: Optional callable (sync or async) invoked during
            :meth:`teardown`.
    """
    self._source_id = source_id
    self._source_type = source_type
    self._collect_func = collect_func
    self._setup_func = setup_func
    self._teardown_func = teardown_func

Methods:

Example – Battery Hardware Callback

The following example shows how to integrate a battery hardware callback into the PhyTrace telemetry pipeline using SimplePlatformAdapter:

import asyncio
from phytrace import (
    SimplePlatformAdapter,
    AdapterCollector,
    PeriodicEmitter,
    UDMEventBuilder,
    SourceType,
    EventType,
)

# --- Simulate a hardware API ---
class HardwareBatteryAPI:
    async def get_level(self) -> float:
        return 78.5  # read from real hardware

hw = HardwareBatteryAPI()

# --- Define a simple adapter using an async callback ---
async def read_battery() -> UDMEventBuilder:
    level = await hw.get_level()
    return (
        UDMEventBuilder(source_id="robot-001", source_type=SourceType.AMR)
        .with_event_type(EventType.TELEMETRY_PERIODIC)
        .with_power(battery_soc_pct=level)
    )

adapter   = SimplePlatformAdapter(
    source_id="robot-001",
    source_type=SourceType.AMR,
    collect_func=read_battery,
)
collector = AdapterCollector(adapter)
emitter   = PeriodicEmitter(
    collector=collector,
    interval_ms=1000,
    on_event=lambda e: print(f"emitted: {e.event_id}"),
)

async def main():
    await collector.setup()
    await emitter.start()
    await asyncio.sleep(5)
    await emitter.stop()
    await collector.teardown()

asyncio.run(main())

For more control, subclass PlatformAdapter directly:

from phytrace import PlatformAdapter, UDMEventBuilder, SourceType, EventType
from phytrace.models.event import UDMEvent

class BatteryAdapter(PlatformAdapter):
    def __init__(self, api, source_id: str):
        self._api = api
        self._source_id = source_id

    @property
    def source_id(self) -> str:
        return self._source_id

    @property
    def source_type(self) -> SourceType:
        return SourceType.AMR

    async def setup(self) -> None:
        await self._api.connect()

    async def teardown(self) -> None:
        await self._api.disconnect()

    async def collect(self) -> UDMEvent:
        level = await self._api.battery_level()
        return (
            UDMEventBuilder(source_id=self.source_id, source_type=self.source_type)
            .with_event_type(EventType.TELEMETRY_PERIODIC)
            .with_power(battery_soc_pct=level)
            .build()
        )