API reference

Public API of the imuninn package, generated from the docstrings. Start with imuninn.connect() / imuninn.Device; everything else is for specific tasks.

imuninn

iMuninn PC-side client API. The phone is the server; PCs connect to it.

Re-exported for convenience: discover, connect, Device, Phone, LeaseError (documented in their modules below).

• PROTOCOL_VERSION = 5
• SERVICE_TYPE = '_imuninn._tcp.local.'
• DEFAULT_CONTROL_PORT = 7420

imuninn.session

Synchronous facade for scripts: runs the asyncio client in a background thread.

Device

class Device

A connected phone, driven synchronously.

Wraps PhoneClient and runs its asyncio loop on a background thread, so scripts and notebooks can call plain methods. Create one with connect and use it as a context manager:

with imuninn.connect() as dev:
    dev.start()
    for pose in dev.iter("ar_pose"):
        ...

Data arrives through on callbacks or iter queues. Config changes go through set, set_config or apply_profile and are validated by the phone; the result says what was applied, clamped or rejected.

def __init__(host: str, port: int, name: str | None = None, data_transport: str = 'tcp')

Create an unconnected device; call open (or use with).

Args: host: Phone address (IPv4, IPv6 or hostname). port: Control WebSocket port (default 7420). name: Client name shown on the phone; defaults to this machine’s hostname. data_transport: "tcp" (video / depth on TCP, pose / IMU / health on UDP) or "websocket" (everything on one WebSocket; the browser path).

Device.open

def open(timeout: float = 10.0) -> 'Device'

Connect to the phone (handshake, clock sync, subscribe to all streams).

Returns self; raises if the phone does not answer within timeout seconds.

Device.close

def close() -> None

Disconnect and stop the background thread. Safe to call twice.

Device.info (property -> pb.DeviceInfo | None)

Device identity from the Welcome message (model, OS, LiDAR, protocol version).

Device.capabilities (property -> pb.Capabilities | None)

Self-describing parameter specs for every stream plus the current config.

Device.config (property -> pb.DeviceConfig | None)

Current effective configuration; updated on every ConfigChanged broadcast.

Device.stats (property -> dict[str, ‘StreamStats’])

Per-stream counters (received, dropped by seq gaps, rate in Hz).

Device.clock_offset_ns (property -> int | None)

pc_time_ns - device_time_ns from the last time sync, or None.

Device.connected (property -> bool)

True while the control channel is up.

Device.set

def set(key: str, value: Any) -> pb.ConfigResult

Set one parameter by dotted key and return the phone’s verdict.

Keys: "session.mode" ("ar" / "camera"), "session.<param>" (e.g. "session.black_screen"), "<stream>.enabled" or "<stream>.<param>" (e.g. "imu.rate_hz", "rgb.exposure_mode"). Out-of-range numbers are clamped, unknown keys rejected; see ConfigResult.fields. Raises LeaseError when another client holds the config lease.

Device.set_config

def set_config(patch: pb.DeviceConfig) -> pb.ConfigResult

Apply a partial DeviceConfig; only the fields present in patch change.

Device.get_config

def get_config() -> pb.DeviceConfig

Fetch the current configuration from the phone.

Device.subscribe

def subscribe(stream_ids: list[str], unsubscribe: bool = False) -> None

Choose which streams this client receives (all are subscribed after open).

Device.start

def start() -> None

Start the capture session on the phone (camera / ARKit / IMU).

Device.stop

def stop() -> None

Stop the capture session on the phone.

Device.start_recording

def start_recording() -> None

Start a phone-side recording (.imuninn file in the app’s Documents).

Device.stop_recording

def stop_recording() -> None

Stop the phone-side recording.

Device.time_sync

def time_sync() -> int | None

Re-estimate the clock offset (NTP-style, lowest RTT of 8 rounds); returns it in ns.

Device.health (property -> pb.DeviceHealth | None)

Latest DeviceHealth broadcast (thermal state, battery, phone-side rates), every 5 s.

Device.lease_holder (property -> str)

Name of the client holding the config lease, or "" when free.

Device.acquire_lease

def acquire_lease() -> bool

Ask for exclusive config control; True if granted. Released on disconnect.

Device.release_lease

def release_lease() -> None

Give the config lease back.

Device.apply_profile

def apply_profile(name_or_path: str) -> pb.ConfigResult

Apply a YAML profile (built-in name like “calibration”, or a file path).

Device.save_profile

def save_profile(path: str, name: str | None = None) -> None

Write the phone’s current config as a YAML profile.

Device.to_pc_time_ns

def to_pc_time_ns(device_time_ns: int) -> int

Device timestamp -> PC wall-clock nanoseconds (needs a completed time sync).

Device.to_pc_time

def to_pc_time(device_time_ns: int) -> float

Device timestamp -> PC wall-clock seconds.

Device.on

def on(stream: str, cb: Callable) -> None

Callback for a stream id (“imu”, “ar_pose”, “rgb”), “*” for all (cb(stream, sample)), or “event” for connected/config/disconnected.

Device.iter

def iter(stream: str | None = None, maxsize: int = 1024, timeout: float | None = None) -> Iterator

Iterate over incoming samples.

Args: stream: A stream id ("imu", "ar_pose", "rgb", "depth", "health") to yield its samples, or None for (stream, sample) tuples of every stream. maxsize: Queue depth; when the consumer falls behind, newer samples are dropped rather than blocking the client. timeout: Stop iterating after this many seconds without a sample.

connect

def connect(target: str = '', name: str | None = None, timeout: float = 10.0, data_transport: str = 'tcp') -> Device

Connect to a phone by Bonjour name, host, host:port, or ‘’ for the first one found.

data_transport="websocket" takes every stream over one WebSocket (protocol v5, the browser path): one port, no UDP, slightly more latency for pose / IMU under congestion. Returns an opened Device; use as a context manager or call close().

imuninn.discovery

Find phones advertising _imuninn._tcp on the local network (Bonjour / mDNS).

Phone

class Phone

A phone found on the LAN: Bonjour name, preferred address and control port.

Phone.address (property -> str)

host:port of the control channel.

discover

def discover(timeout: float = 2.0, first: bool = False) -> list[Phone]

Browse for timeout seconds (or until the first phone when first=True).

resolve

def resolve(target: str, timeout: float = 3.0) -> tuple[str, int]

‘host’, ‘host:port’, a Bonjour service name, or ‘’ (first phone found) -> (host, port).

imuninn.client

Asyncio client for one phone (the phone is the server, §6.7).

PhoneClient

class PhoneClient

Asyncio connection to one phone (advanced API; Device wraps it).

Create, await connect(), register listeners, await close(). Control messages travel on a WebSocket; data on TCP + UDP or on a second WebSocket.

def __init__(host: str, port: int = DEFAULT_CONTROL_PORT, name: str | None = None, data_transport: str = 'tcp')

data_transport: “tcp” (video / depth on TCP, pose / IMU / health on UDP) or “websocket” (protocol v5: everything on a second WebSocket to the control port, the path browsers use; single port, no UDP).

PhoneClient.name (property -> str)

Device name once connected, else the host.

PhoneClient.add_data_listener

def add_data_listener(cb: DataListener) -> None

Call cb(DataMessage) for every sample, on the event loop thread.

PhoneClient.add_event_listener

def add_event_listener(cb: Callable[[str], None]) -> None

Events: “connected”, “config”, “disconnected”.

PhoneClient.connect

async def connect(subscribe_all: bool = True, time_sync: bool = True) -> 'PhoneClient'

Handshake (Hello / Welcome), open the data channel, sync clocks, subscribe.

PhoneClient.close

async def close() -> None

Cancel the receive tasks and close every socket.

PhoneClient.send

async def send(msg: pb.ControlMessage) -> None

Send a control message without waiting for a reply.

PhoneClient.request

async def request(msg: pb.ControlMessage, timeout: float = 5.0) -> pb.ControlMessage

Send a control message and await the reply with the same request_id.

PhoneClient.fetch_capabilities

async def fetch_capabilities() -> pb.Capabilities

Ask the phone for its parameter specs and current config.

PhoneClient.fetch_config

async def fetch_config() -> pb.DeviceConfig

Ask the phone for its current configuration.

PhoneClient.set_config

async def set_config(patch: pb.DeviceConfig) -> pb.ConfigResult

Send a partial config; raises LeaseError on 409, RuntimeError on other errors.

PhoneClient.set

async def set(key: str, value: bool | int | float | str) -> pb.ConfigResult

Set one parameter by dotted key.

“session.mode” -> DeviceConfig.mode (“ar” | “camera”) “session.” -> DeviceConfig.session_params[param] (e.g. “session.black_screen”) “.enabled” -> StreamConfig.enabled “.” -> StreamConfig.params[param]

PhoneClient.acquire_lease

async def acquire_lease() -> bool

Request the config lease; True when granted.

PhoneClient.release_lease

async def release_lease() -> None

Release the config lease.

PhoneClient.apply_profile

async def apply_profile(name_or_path: str) -> pb.ConfigResult

Apply a YAML profile (built-in name or file path) as one partial config.

PhoneClient.subscribe

async def subscribe(stream_ids: list[str], unsubscribe: bool = False) -> None

Subscribe to (or unsubscribe from) the given stream ids.

PhoneClient.start

async def start() -> None

Start the capture session on the phone.

PhoneClient.stop

async def stop() -> None

Stop the capture session on the phone.

PhoneClient.start_recording

async def start_recording() -> None

Start a phone-side recording.

PhoneClient.stop_recording

async def stop_recording() -> None

Stop the phone-side recording.

PhoneClient.time_sync

async def time_sync(rounds: int = 8) -> int | None

NTP-style offset estimate; keeps the lowest-RTT sample.

PhoneClient.to_pc_time_ns

def to_pc_time_ns(device_time_ns: int) -> int

Device timestamp -> PC wall-clock ns; raises until time_sync has run.

StreamStats

class StreamStats

Receive-side counters for one stream (kept by PhoneClient).

StreamStats.observe

def observe(header: pb.Header) -> None

Account for one received sample: counts, seq-gap drops, 1 s rate window.

LeaseError

class LeaseError

Raised when another client holds the configuration lease (phone replied 409).

imuninn.depth

Decode DepthFrame payloads: uint16 millimetres (+ uint8 confidence), LZ4 via Apple’s Compression framing (“bv41” blocks).

decode

def decode(frame: pb.DepthFrame, upright: bool = True) -> tuple[np.ndarray, np.ndarray | None]

-> (depth_mm uint16 (H, W), confidence uint8 (H, W) or None). 0 mm means invalid. ARKit confidence: 0 = low, 1 = medium, 2 = high.

to_metres

def to_metres(depth_mm: np.ndarray) -> np.ndarray

uint16 mm -> float32 m with NaN for invalid (0).

imuninn.video

H.264 decoding for VideoFrame payloads (Annex-B NAL units) via PyAV.

H264Decoder

class H264Decoder

Feed VideoFrame messages, get RGB numpy arrays (H, W, 3) in sensor orientation.

H264Decoder.decode

def decode(frame: pb.VideoFrame, upright: bool = True) -> Iterator['np.ndarray']

Decode one VideoFrame; yields zero or more RGB arrays.

Frames before the first keyframe are skipped. With upright=True the image is rotated by frame.rotation_deg so it matches how the phone was held.

H264Decoder.reset

def reset() -> None

Drop decoder state and wait for the next keyframe (after a gap or format change).

imuninn.mcap_io

MCAP recording / conversion / replay (§7.3): the phone’s Protobuf messages are written as-is, one topic per stream, so the dataset format is the protocol.

McapRecorder

class McapRecorder

Write phone messages to an MCAP file, one topic per stream, Protobuf-encoded as received.

Use as a context manager; feed write from a data listener. log_time is PC wall clock when clock_offset_ns is known, publish_time is device time.

McapRecorder.write

def write(msg: pb.DataMessage, log_time_ns: int | None = None) -> None

Append one DataMessage to its topic.

McapRecorder.write_config

def write_config(config: pb.DeviceConfig) -> None

Record the configuration in effect on the /config topic.

McapRecorder.close

def close() -> None

Finish the MCAP index and close the file.

replay

def replay(path: str | Path, realtime: bool = False, topics: list[str] | None = None) -> Iterator[tuple[str, object]]

Yield (stream_id, sample) from an MCAP file. realtime=True sleeps to reproduce timing.

convert

def convert(src: str | Path, dst: str | Path) -> int

Phone recording -> MCAP. log_time = device time (no PC clock available).

summary

def summary(path: str | Path) -> dict[str, int]

Message count per topic, from the file’s summary section.

imuninn.profiles

Configuration presets (§6.8 Profile) as YAML.

builtin_profiles

def builtin_profiles() -> list[str]

Names of the YAML presets shipped with the package (calibration, dataset, …).

load

def load(name_or_path: str) -> dict

Read a profile (built-in name or YAML file) as a dict.

to_patch

def to_patch(profile: dict) -> pb.DeviceConfig

Profile dict -> partial DeviceConfig for set_config.

from_config

def from_config(config: pb.DeviceConfig, name: str = 'snapshot') -> dict

Current DeviceConfig -> profile dict (for saving what is in effect now).

dump

def dump(profile: dict, path: str | Path) -> None

Write a profile dict as YAML.

resolve_path

def resolve_path(name_or_path: str) -> Path

A file path as given, or the built-in profile of that name.