Streaming¶

The client.ws surface opens a long-lived WebSocket subscription to one or more server channels and yields typed messages as they arrive.

Streaming is the live-push counterpart to the paginated-pull iterators on Events (events.iter_pws, events.iter_zone_events). Use streaming when you need updates as they happen; use events when you need to back-fill history.

Quick reference¶

Method	What it does
`client.ws.list_channels()`	Return the server's channel inventory.
`client.ws.subscribe(channels, *, project_uid=None, reconnect=True, queue_max=1024, include_heartbeats=False)`	Open a session subscribed to `channels`; returns a `WsSession`.
`session.__iter__()`	Blocking iterator yielding `WsChannelEnvelope` instances.
`session.add_channels(channels)`	Add channels to an open session.
`session.remove_channels(channels)`	Remove channels from an open session.
`session.close()`	Best-effort unsubscribe and close. Also fires on `__exit__`.
`session.messages_dropped`	Counter of messages dropped due to backpressure.

Channels¶

Channel	Typed class	Carries
`pos`	`PositionMessage`	Live position fixes from trackable objects.
`notify`	`NotifyMessage`	Object-lifecycle events (created / updated / deleted) on tracked entities.
`alarm`	`AlarmMessage`	Alarm set / clear events.
`zone_event`	`ZoneEventMessage`	Trackable enter / leave events on a zone.
`sensor`	`SensorMessage`	Sensor-reading broadcasts (full ZMQ payload on `.raw`).
`ota`	`OtaMessage`	OTA-firmware status broadcasts.
`param_p`	`ParamPMessage`	Parameter-publication broadcasts from devices.
`user_msg`	`UserMsgMessage`	Badge-side replies to `client.user_messages.send`.

Any channel the server emits that isn't in this table decodes to the fallback WsChannelMessage — iteration continues, and the channel name and full wire dict are preserved on .channel and .raw.

Subscribing¶

from rtls_sdk import PositionMessage, NotifyMessage

with client.ws.subscribe(["pos", "notify"]) as session:
    for msg in session:
        match msg:
            case PositionMessage(x=x, y=y, mac=mac):
                print(f"{mac} → ({x}, {y})")
            case NotifyMessage(object=obj, action=action, plist=p):
                print(f"{action} {obj}: {p.get('uid')}")

subscribe blocks until the server acks (raising WsAuthError on rejected credentials). Iteration is blocking and runs until session.close() or a terminal condition. The context manager closes the socket on exit.

Mutating an open session¶

with client.ws.subscribe(["pos"]) as session:
    session.add_channels(["alarm"])
    session.remove_channels(["pos"])
    for msg in session:
        ...

Both methods update the session's local channel set; no-op if a channel is already present (for add) or absent (for remove).

Specialized operations¶

Reconnect¶

By default subscribe(..., reconnect=True) retries with backoff 1, 2, 4, 8, 15, 30, 30, 30 seconds; exhaustion raises WsError. Pass reconnect=False to end the iterator cleanly on the first disconnect. Auth-flavoured close codes (1008, 4001, 4003) skip the backoff and raise WsAuthError immediately.

Heartbeats¶

The server periodically emits a notify-channel frame with action="HB". These are filtered at the reader by default. To receive them — for example, as a liveness probe — pass include_heartbeats=True:

with client.ws.subscribe(["notify"], include_heartbeats=True) as session:
    for msg in session:
        if isinstance(msg, NotifyMessage) and msg.action == "HB":
            print("server is alive")

Backpressure¶

The session uses a bounded queue (queue_max=1024). On overflow the oldest message is dropped, session.messages_dropped is incremented, and a WARNING logs every 100 drops. Offload non-trivial per-message work to a worker thread.

REST → WS round-trip¶

Any REST create / update / delete on a project-scoped entity fires a notify event on that project's channel. The pattern: subscribe first, let the ack settle, then trigger the side-effect, then drain the queue for the matching event.

import queue
import threading
import time

from rtls_sdk import NotifyMessage

session = client.ws.subscribe(["notify"])
sink: queue.Queue = queue.Queue()


def drain() -> None:
    for msg in session:
        sink.put(msg)


threading.Thread(target=drain, daemon=True).start()
time.sleep(1.0)  # let the subscribe ack settle

tag = client.tags.create(name="Forklift 9")

deadline = time.monotonic() + 30
while time.monotonic() < deadline:
    try:
        msg = sink.get(timeout=0.5)
    except queue.Empty:
        continue
    if isinstance(msg, NotifyMessage) and msg.plist.get("uid") == tag.uid:
        print(f"{msg.action} {msg.object}: {tag.uid}")
        break

session.close()

The same shape works for client.user_messages.send → user_msg channel (subscribe to ["user_msg"], match on UserMsgMessage and the round-tripped hex payload).

Model¶

Every typed channel class inherits from WsChannelEnvelope:

Field	Type	Notes
`channel`	`str`	The channel that produced this message.
`raw`	`dict[str, Any]`	The full wire dict — use for forward-compat access to fields the typed class doesn't surface.

Per-channel tables show the fields callers most often read. Full pydantic shape: reference/models.md.

PositionMessage (pos)

Field	Type	Notes
`mac`	`str`	Bare 12-hex tag MAC.
`x`, `y`, `z`	`int`	Position in server-native units.
`ts`, `ms`	`int`	Epoch seconds and millisecond fraction.
`dev_id`	`str`	Server device id.
`project_uid`	`str`	Project scope.

NotifyMessage (notify)

Field	Type	Notes
`object`	`str`	Entity kind, e.g. `"TRACKABLE_OBJECT"`, `"GROUP"`, `"ZONE"`.
`action`	`str`	`"created"`, `"updated"`, `"deleted"`, or `"HB"`.
`plist`	`dict[str, Any]`	Entity payload; `plist["uid"]` identifies the affected resource.
`project_uid`	`str \\| None`	Project scope (absent on `HB` frames).
`ts`, `ms`	`int`	Timestamps.

AlarmMessage (alarm)

Field	Type	Notes
`uid`	`str`	Alarm uid.
`alarm_name`	`str`	Display name.
`state`	`str`	`"set"` or `"clear"`.
`obj_uid`	`str`	The triggering entity.
`x`, `y`, `z`	`int \\| None`	Location at firing time.

ZoneEventMessage (zone_event)

Field	Type	Notes
`action`	`str`	Enter / leave.
`obj_uid`	`str`	The trackable that crossed the boundary.
`zone_uid`	`str`	The zone crossed.
`zone_type_name`	`str`	E.g. `"keep_in"`, `"keep_out"`.
`sublocation_uid`	`str`	Where it happened.
`x`, `y`, `z`	`int`	Position at the event.

SensorMessage (sensor) — only project_uid and company_uid are typed; the full ZMQ payload is on .raw.

OtaMessage (ota) — only mac and project_uid are typed; full payload on .raw.

ParamPMessage (param_p)

Field	Type	Notes
`mac`	`str`	Device MAC.
`plist`	`dict[str, Any]`	Published parameters; contents are firmware-defined.
`ts`, `ms`	`int`	Timestamps.

UserMsgMessage (user_msg)

Field	Type	Notes
`mac`	`str`	Badge MAC.
`dev_id`	`str`	Server device id.
`hex`	`str`	Hex payload, up to 512 characters (256 bytes).
`project_uid`	`str`	Project scope.
`ts`, `ms`	`int`	Timestamps.

Fallback. WsChannelMessage carries .channel and .raw only. The typed classes also inherit from WsChannelEnvelope, so use an exact-type check rather than isinstance when distinguishing the fallback: type(msg) is WsChannelMessage.

User messages — REST send path whose badge replies arrive on the user_msg channel.
Tags, Groups, Zones — any create / update / delete fires a notify event scoped to the active project.
System — client.system.ws_host() is the internal REST hop the SDK uses to resolve the WS endpoint; callers should never need to invoke it directly.