MassRobotics AMR Interop · Vecna APT / ATG / AFL / CPJ
A Vecna fleet on the interop wire, no robots required
A spec-faithful emulator of the MassRobotics AMR
Interoperability Standard — the open interop layer Vecna Robotics
co-developed and certifies its fleet against. Both roles ship: emulated
robots that deliver an identityReport and stream
statusReports from a live movement model, and a fleet-side
receiver that validates every message against the official schema and keeps
a per-robot registry. All over a hand-rolled, stdlib RFC 6455
WebSocket — zero runtime network dependencies.
Judged by the standard's own referee
Self-validation is circular, so this emulator is run against the reference receiver from the official MassRobotics-AMR/AMR_Interop_Standard repository — the Node/Ajv implementation, built unmodified in Docker at the same commit the vendored schema is pinned to.
Official Node/Ajv reference receiver
All four Vecna vehicle models connect at once; the harness witnesses
the receiver's own verdicts on its /ui WebSocket, where it
broadcasts {message, isValid, errors} for everything it
validates. Every identityReport and statusReport we send — idle,
navigating with path and destinations, erroring, charging — is judged
valid, and a deliberately-invalid probe is judged invalid, proving the
witness channel detects bad traffic rather than rubber-stamping.
Vendored byte-for-byte, zero repairs
AMR_Interop_Standard.json at commit
f9357a4 (2021-10-08), retrieved 2026-08-16 — vendored with
no normalizations; upstream and vendored hashes are identical:
sha256 816c8f78e1d280d10983a1bcea69a3c8e6572e5bb6f777a0863e6a5b2c3fc288
Every outgoing message is validated against it and raises on violation; every incoming message is validated and recorded — problems are kept per robot, never silently dropped.
Five minutes to a reporting fleet
$ uv run massrobotics-emulator receiver --port 3000 --verbose $ uv run massrobotics-emulator robot --receiver ws://127.0.0.1:3000 --model APT --count 2 --patrol # --model APT|ATG|AFL|CPJ picks a Vecna vehicle preset; # --count N runs a fleet on one connection each; # --patrol drives a loop instead of idling; # --interval sets the statusReport period (seconds).
import asyncio
from massrobotics_emulator import InteropReceiver, MassRoboticsAMR, vecna_config
async def main():
receiver = InteropReceiver()
await receiver.start()
robot = MassRoboticsAMR(vecna_config("APT"))
await robot.start(receiver.uri)
(record,) = await receiver.wait_for_robots(1)
robot.navigate_to(25.0, 10.0)
print(await receiver.next_status(
record.uuid, lambda s: s["operationalState"] == "navigating"))
await robot.stop()
await receiver.stop()
asyncio.run(main())
The wire, replayed
This session was recorded from the emulator, exactly as sent:
one APT delivers its identityReport, is told
navigate_to(25, 10), and streams statusReports —
position, quaternion heading, velocity, battery, predicted path and
destinations — until it arrives. Every message shown was judged
schema-valid by the receiver when recorded.
Full messages, verbatim (note the schema's own field-naming
split — current location says planarDatum, predicted locations
say planarDatumUUID):
What the schema actually says
Vendoring the official artifact surfaced upstream quirks —
documented in the package's schemas/registry.json and honored
on the wire, never silently repaired.
| Quirk | Detail |
|---|---|
| The examples fail the schema | Both example messages in the standard's own examples/
directory use uppercase UUIDs, but every uuid pattern in
the schema requires lowercase hex (^[0-9a-f]{8}-…). This
emulator always emits lowercase UUIDs. |
cargoMaxWeight is a string |
Typed string despite its description saying "Max
weight of cargo in kg". Emitted as a string here, schema-faithfully. |
| Two names for the map reference | The current location object calls it planarDatum;
predicted locations (path, destinations)
call it planarDatumUUID. Both spellings are intentional
per the schema and used as-is. |
| Location is stricter than the prose | statusReport.location requires angle
(a quaternion) and planarDatum — stricter than the PDF
suggests. The reference receiver enforces the schema, so the emulator
always sends both. |
uniqueItems on predictions |
path and destinations declare
uniqueItems, so consecutive identical predicted locations
would be invalid; the movement model guarantees distinct entries. |
What ships in the package
| Surface | Coverage |
|---|---|
| Robot role | MassRoboticsAMR: WebSocket client delivering one
identityReport and streaming statusReports while a movement
model drives position, heading, velocity, battery, and predicted path/destinations |
| Fleet role | InteropReceiver: WebSocket server validating every
incoming message and keeping a live per-robot registry — identity, status history,
connection state, recorded validation problems |
| Vehicle presets | Vecna's certified fleet via vecna_config():
APT pallet truck, ATG tugger, AFL counterbalanced forklift, CPJ cobot pallet jack;
any identity via AMRConfig |
| Validation | Outbound messages that violate the vendored schema raise; inbound violations are recorded per robot, never dropped |
| Transport | Hand-rolled stdlib RFC 6455 client + server (ws.py),
strict about masking, fragmentation, control frames, and close codes |
| Tests | Schema-quirk regressions, RFC 6455 conformance with adversarial raw-byte peers, and full sender↔receiver loops on simulated time |
Want this for your fleet manager?
MiR REST, VDA 5050, Omron ARCL, MassRobotics Interop — same pattern each time: official spec as the oracle, provenance with hashes, adversarial tests, live validation against independent implementations. If your integration needs an emulator for another manufacturer, that pattern is repeatable.