Tile

High-level asynchronous API for a single OmniTile.

class omnitiles.tile.Tile(info, *, transport=None)[source]

Bases: object

Async interface to a single OmniTile over BLE.

A Tile owns a Transport and a StreamParser. Every command method returns once the packet has been handed to the transport — the protocol does not ACK, so a successful await means “sent”, not “executed”. Subscribe to telemetry to observe effects.

Tiles are async context managers:

async with Tile(info) as tile:
    await tile.m1_extend()

Parameters:

info (TileInfo)
transport (Transport | None)

property name: str

property address: str

property connected: bool

async connect()[source]

Return type:: None

async disconnect()[source]

Return type:: None

async ping()[source]

Return type:: None

async m1_extend(speed=255)[source]

Drive M1 in the extend direction at the given PWM speed (0-255).

Parameters:: speed (int)
Return type:: None

async m1_retract(speed=255)[source]

Parameters:: speed (int)
Return type:: None

async m1_brake()[source]

Return type:: None

async m1_set_position(position)[source]

Command a closed-loop M1 target, scaled 0-255 over the stroke.

Parameters:: position (int)
Return type:: None

async m1_set_position_mm(mm)[source]

Convenience: set M1 target in millimeters along the stroke.

Parameters:: mm (float)
Return type:: None

async m2_extend(speed=255)[source]

Parameters:: speed (int)
Return type:: None

async m2_retract(speed=255)[source]

Parameters:: speed (int)
Return type:: None

async m2_brake()[source]

Return type:: None

async m2_set_position(position)[source]

Parameters:: position (int)
Return type:: None

async m2_set_position_mm(mm)[source]

Parameters:: mm (float)
Return type:: None

async base_velocity(vx, vy, omega)[source]

Command open-loop mobile-base velocity. Each component is int8.

Parameters:

vx (int)
vy (int)
omega (int)

Return type:

None

async base_brake()[source]

Return type:: None

property telemetry: Telemetry | None: Most recently received telemetry frame, or None.

on_telemetry(callback)[source]

Register callback to run on every incoming telemetry frame.

Returns a zero-argument function that removes the subscription. Callbacks fire on whatever thread/loop the transport delivers notifications on, so they must not block.

Parameters:: callback (Callable[[Telemetry], None])
Return type:: Callable[[], None]

async wait_for_telemetry(timeout=None)[source]

Await the next telemetry frame and return it.

Parameters:: timeout (float | None)
Return type:: Telemetry