zainar-halo

ZaiNar RF halo awareness — egocentric proximity for Device Connect nodes.

A robot's onboard sensors see line-of-sight only. ZaiNar's RF network sees every tagged entity in a facility, through walls and around blind corners. zainar-halo packages that awareness as Device Connect–native primitives: a halo (per-node view of surrounding entities) computed locally by the robot from a zone broadcast, with no GPU and no training run required.

Install

pip install zainar-halo              # consume location: halos, zones, robots
pip install 'zainar-halo[cloud]'     # also PUBLISH a device's own fixes

Requires device-connect-edge>=0.2.5. The cloud extra declares and pins nats-py>=2.6,<3, the client ZainarLocationPublisher uses — it is the only class here that talks to a broker directly; the rest is in-process halo math. The extra is not a gate: device-connect-edge already depends on nats-py, so a bare install usually has one. It is how you get a version the publisher was tested against.

What's in the box

Two meanings of "publish", which is worth getting straight once

ZainarLocationService broadcasts halos — in-process computation, emitted over device-connect for robots in a zone to consume. It reaches the ZaiNar cloud not at all. Reading it as the device-to-cloud path is the most expensive misread this package has caused: one integration concluded there was no way for a device to report its own position and spent a day hand-rolling one.

Telling the cloud where a device is, is ZainarLocationPublisher. The two are independent and most robots want both — the halo for local safety math, the publisher so the rest of the platform can see it.

Quick start — publishing your own position

A device is a first-class location source. Onboard perception — ArUco, lidar, VIO, depth, wheel odometry — is equal to the RF network, not a lesser input.

import asyncio, time
from zainar_halo import ZainarLocationPublisher, InvalidFix

# site_id, credentials and nats_url all come from the registration response
# of POST /sites/{site_id}/devices. Do not hardcode the URL — it is per-platform.
async def main():
    async with ZainarLocationPublisher(
        site_id="robo-warehouse-2",
        device_id="elephant-2",
        credentials="device.creds",
        nats_url="nats://nats-dev.example.com:4222",
        source="aruco",          # provenance, recorded on every fix
        entity_type="amr",       # a wheeled mobile robot is "amr"
    ) as pub:
        while True:
            x_cm, y_cm = my_perception()
            seq = await pub.publish_fix(x_cm=x_cm, y_cm=y_cm)
            # `seq` is JetStream's sequence number: proof the fix was
            # PERSISTED, not merely sent. A core NATS publish gives no such
            # proof and is simply gone if the stream is briefly unavailable.
            await asyncio.sleep(1.0)

asyncio.run(main())

publish(position) takes a Position directly, and publish_batch(positions) validates every fix before publishing any — a batch with one bad fix publishes nothing rather than applying halfway.

Why this is a class and not a documented subject

Position.measurement_time is a float in epoch seconds (what time.time() returns). The cloud wire field is timestamp_ms, an int in epoch milliseconds. Two names and two units for the one field every freshness filter reads — so a hand-rolled publish that sends measurement_time, or sends seconds under the right key, is accepted, persisted, acked, and then dated to 1970 and filtered out as ~57 years stale. The publish looks successful and the data is invisible.

This class owns that conversion, in one tested place. It is also deliberately louder than the rest of the package: an entity type that is not a wire value, a timestamp that is milliseconds-as-seconds, a non-finite coordinate all raise InvalidFix before anything is sent. Position normalizes the same input in silence because it sits under a live feed where one odd record must not stop a robot; at a publish boundary the caller is an author who can still fix it.

Quick start — location service (halo zone broadcaster)

from zainar_halo import ZainarLocationService, HaloConfig, Position
import time

service = ZainarLocationService(
    operator_configs={"amr-7": HaloConfig(radius_cm=1500, entity_types=["forklift", "human"])},
)
# register with DeviceRuntime and connect

# In your kinesis-gw subscriber callback:
service.ingest_position(Position(
    device_id="amr-7", entity_type="amr",
    x_cm=2200, y_cm=500, z_cm=0,
    measurement_time=time.time(),
))
# The @periodic tick broadcasts automatically at 1Hz

Quick start — robot subscriber

from zainar_halo import HaloConfig, RobotDriver

class MyRobot(RobotDriver):
    async def _on_halo_computed(self, halo: dict) -> None:
        # feed to costmap, planner, alerting...
        nearest = halo["nearest_entity_type"]
        dist_m  = (halo["nearest_distance_cm"] or 0) / 100
        print(f"{nearest} at {dist_m:.1f}m")

robot = MyRobot(halo_config=HaloConfig(radius_cm=1500, entity_types=["forklift", "human"]))
# register with DeviceRuntime and connect — the @on subscription is automatic

Design note — awareness not firehose

get_halo(device_id) returns a filtered, egocentric view — entities within radius_cm, capped by entity_types and max_age_s. It is not a raw position feed. This is intentional:

• An MCP agent asking "what's around amr-7?" gets a safety-relevant subset, not the full facility tracking log.
• Operators set the filter via set_halo_config(). The config is stamped into every broadcast tick so all consumers use the same safety perimeter.
• For raw zone position lists, use get_zone_entities(zone).

Dense zones and the k-cap

When a zone has many tracked entities, HaloConfig.max_members limits the halo to the nearest k. This bounds per-robot compute to O(k) regardless of zone density:

# Nearest 10 only — ignores everything beyond the 10th closest
HaloConfig(radius_cm=1500, max_members=10)

Default: None (unbounded — all entities within radius).

Performance

From the S5 scale test (200 robots, one zone, NATS on localhost):

Payload is O(N) in entity count — one JSON list, one NATS publish per zone, N local computes (no central aggregation, no fan-out amplification).

Units and conventions

• Lengths: centimeters as int (distance_cm, radius_cm, x_cm/y_cm/z_cm)

• Speeds: cm/s as int (closing_speed_cmps, vx_cmps)

• Timestamps: measurement_time is epoch seconds as a float, here and everywhere in this package. The cloud wire format uses timestamp_ms, integer milliseconds; ZainarLocationPublisher is the only thing that should be doing that conversion.

• 3D throughout — distance is 3D Euclidean; a drone 4m overhead is 400cm away

• Entity types: a closed set of spatial_vocab wire values —

  human  amr  forklift  vehicle  ambulance  drone  gurney  pallet  tool  anchor  unknown
  

  Importable as WIRE_ENTITY_TYPES. A wheeled mobile robot is amr: robot, agv and mobile_robot are not wire types. Anything outside the set becomes unknown (with a warning), and every entity_types filter then skips it — so a mistyped entity is present and unfindable rather than rejected. ZainarLocationPublisher raises on it instead.

See also

• zainar-halo-agent-tools — FastMCP server exposing halo tools to LLM agents
• CLAUDE.md at repo root — full architecture, scenario, and integration guide