Visualization

coordination_oru.viz.web_viewer

Browser-based live viewer for a running :class:TrajectoryEnvelopeCoordinatorSimulation.

A small starlette + uvicorn server serves the prebuilt Vite/React frontend (shipped inside the wheel at coordination_oru/viz/static/) and streams coordinator state over a /ws websocket. Like :class:PygletViewer it is a pure polling observer — it reads public coordinator state and never calls back into the core.

Wire protocol (all messages carry seq, monotonic int, and ts, unix ms):

• {"kind": "static", "title", "world": {"size", "center"}, "robots": [{"id", "envelopeID", "path": [[x, y], ...], "envelope": [ring, ...]}], "footprints": [{"id", "ring"}], "interactive": bool} — per-robot path polyline and swept-envelope polygon rings for every driving robot, plus each known robot's footprint outline centered at the origin. Sent on client connect and whenever the set of robots or driving envelopes changes (missions start/finish). Paths are the heavy payload, so they are only re-sent on change. When the viewer has an occupancy map, the message also carries "map": {"dataUri", "resolution", "origin": [x, y], "width", "height"} (a base64 PNG data URI plus the world-frame placement); interactive tells the frontend to enable the goal-posting UX.
• {"kind": "state", "robots": [{"id", "driving", "pose": [x, y, theta], "pathIndex", "pathLength", "velocity", "criticalPoint"}], "criticalSections": [{"robot1", "start1", "end1", "robot2", "start2", "end2"}], "dependencies": [{"waiting", "driving", "waitingPoint"}], "counts": {"driving", "parked", "criticalSections", "orders"}} — sent every poll tick (poll_hz). The frontend places the static footprint outline at pose (translate+rotate, cheap enough to CSS-animate); critical sections reference path indices into the static paths, the frontend slices the highlight segments from those; dependencies are the current yielder → leader precedence orders.

Inbound messages: {"kind": "postGoal", "robot": int, "goal": [x, y, theta]} — a goal pose for a robot, dispatched to the on_goal callback when one is configured. Malformed or unknown inbound messages are silently ignored.

The server runs inside the simulation's asyncio event loop (the coordinator is asyncio-native, so no thread bridge is needed): create the viewer, then await viewer.serve() alongside the sim driver.

__all__ = ['WebViewer', 'build_static_message', 'build_state_message'] module-attribute

STATIC_MISSING_MESSAGE = 'coordination_oru/viz/static/ is missing an index.html (this looks like a source checkout without a frontend build). Build it with:\n npm --prefix frontend install && npm --prefix frontend run build' module-attribute

TrajectoryEnvelopeTrackerDummy

Bases: AbstractTrajectoryEnvelopeTracker

onTrajectoryEnvelopeUpdate()

startTracking()

setCriticalPoint(criticalPoint)

getRobotReport()

finishParking()

isParkingFinished()

run() async

onPositionUpdate()

getCurrentTimeInMillis()

AbstractTrajectoryEnvelopeCoordinator

Bases: ABC

CONTROL_PERIOD = CONTROL_PERIOD instance-attribute

TEMPORAL_RESOLUTION = TEMPORAL_RESOLUTION instance-attribute

DEFAULT_ROBOT_TRACKING_PERIOD = DEFAULT_ROBOT_TRACKING_PERIOD instance-attribute

overlay = False instance-attribute

quiet = False instance-attribute

totalMsgsSent = 0 instance-attribute

totalMsgsReTx = 0 instance-attribute

criticalSectionCounter = 0 instance-attribute

solver = None instance-attribute

missionsPool = [] instance-attribute

envelopesToTrack = [] instance-attribute

currentParkingEnvelopes = [] instance-attribute

allCriticalSections = set() instance-attribute

CSToDepsOrder = {} instance-attribute

depsToCS = {} instance-attribute

escapingCSToWaitingRobotIDandCP = {} instance-attribute

stoppingPoints = {} instance-attribute

stoppingTimes = {} instance-attribute

stoppingPointTimers = {} instance-attribute

trackers = {} instance-attribute

currentDependencies = {} instance-attribute

communicatedCPs = {} instance-attribute

externalCPCounters = {} instance-attribute

comparators = [] instance-attribute

forwardModels = {} instance-attribute

footprints = {} instance-attribute

maxFootprintDimensions = {} instance-attribute

robotTrackingPeriodInMillis = {} instance-attribute

robotMaxVelocity = {} instance-attribute

robotMaxAcceleration = {} instance-attribute

muted = set() instance-attribute

yieldIfParking = True instance-attribute

checkEscapePoses = True instance-attribute

trackingCallbacks = {} instance-attribute

inferenceCallback = None instance-attribute

motionPlanners = {} instance-attribute

packetLossProbability = network_configuration.PROBABILITY_OF_PACKET_LOSS instance-attribute

MAX_TX_DELAY = network_configuration.getMaximumTxDelay() instance-attribute

maxFaultsProbability = network_configuration.PROBABILITY_OF_PACKET_LOSS instance-attribute

numberOfReplicas = 1 instance-attribute

isDriving = {} instance-attribute

getMaxFootprintDimension(robotID)

getFootprint(robotID)

getFootprintPolygon(robotID)

setFootprint(robotID, *coordinates)

computeMaxFootprintDimension(coords)

setForwardModel(robotID, fm)

getForwardModel(robotID)

setRobotTrackingPeriodInMillis(robotID, trackingPeriodInMillis)

getRobotTrackingPeriodInMillis(robotID)

setRobotMaxVelocity(robotID, maxVelocity)

setRobotMaxAcceleration(robotID, maxAcceleration)

getRobotMaxVelocity(robotID)

getRobotMaxAcceleration(robotID)

setNetworkParameters(packetLossProbability, max_tx_delay, maxFaultsProbability)

setInferenceCallback(cb)

getControlPeriod()

getTemporalResolution()

setYieldIfParking(value)

setCheckEscapePoses(value)

toggleMute(robotID)

mute(robotID)

unMute(robotID)

getMuted()

getCurrentTimeInMillis() abstractmethod

setupSolver(max_envelopes=64)

startInference() async

stopInference() async

isStartedInference()

onNewMissionDispatched(robotID)

onCriticalSectionUpdate()

getDrivingEnvelopes()

isParked(robotID)

isDrivingRobot(robotID)

getIdleRobots()

getAllRobotIDs()

getRobotReport(robotID)

getCurrentDependencies()

getCurrentSuperEnvelope(robotID)

getCurrentTrajectoryEnvelope(robotID)

addTrackingCallback(robotID, cb)

setVisualization(viz)

addComparator(c)

setMotionPlanner(robotID, mp)

getMotionPlanner(robotID)

inParkingPose(robotID)

setCriticalPoint(robotID, criticalPoint, retransmitt)

placeRobot(robotID, currentPose=None, parking=None, location=None)

isFree(robotID)

atStoppingPoint(robotID)

spawnWaitingThread(robotID, index, duration)

getObstaclesInCriticalPoints(robotIDs)

getObstaclesFromWaitingRobots(robotID)

makeObstacles(robotID, *obstaclePoses)

doReplanning(mp, fromPose, toPose, obstaclesToConsider=())

updateDependencies() abstractmethod

canExitCriticalSection(drivingCurrentIndex, waitingCurrentIndex, drivingTE, waitingTE, lastIndexOfCSDriving)

getCriticalPoint(yieldingRobotID, cs, leadingRobotCurrentPathIndex)

isAhead(cs, rr1, rr2)

computeCriticalSections()

filterCriticalSections()

getCriticalSections(te1, minStart1, te2, minStart2, checkEscapePoses, maxDimensionOfSmallestRobot) staticmethod

cleanUpRobotCS(robotID, lastWaitingPoint)

startTrackingAddedMissions()

addMissions(*missions)

getNewTracker(te, cb) abstractmethod

OccupancyMap

A y-up occupancy grid with world<->grid transforms and inflation.

image = image instance-attribute

resolution = resolution instance-attribute

origin = origin instance-attribute

occupied = occupied instance-attribute

height property

width property

bounds property

World-frame (xmin, ymin, xmax, ymax) of the map.

from_yaml(yaml_path, *, unknown_is_occupied=True) classmethod

Load a ROS map_server YAML descriptor and its image.

Cells with occupancy probability above occupied_thresh are occupied; the unknown band between free_thresh and occupied_thresh counts as occupied unless unknown_is_occupied=False. Only trinary mode and unrotated maps (origin yaw 0) are supported.

world_to_grid(x, y)

The (row, col) cell containing world point (x, y).

grid_to_world(row, col)

The world (x, y) of the center of cell (row, col).

in_bounds(row, col)

inflated(radius)

The occupancy grid dilated by radius metres (a disk structuring element). Cached per pixel radius; callers must .copy() before writing.

to_png_bytes()

The map as a grayscale 8-bit PNG (image orientation, top row first), encoded with the stdlib only.

WebViewer

Read-only browser viewer; runs in the simulation's asyncio loop.

Usage::

viewer = WebViewer(tec, world_size=14.0)
server_task = asyncio.create_task(viewer.serve())
...  # drive the sim in the same loop
await server_task  # serves until Ctrl+C or viewer.request_stop()

coordinator = coordinator instance-attribute

host = host instance-attribute

port = port instance-attribute

poll_hz = poll_hz instance-attribute

world_size = world_size instance-attribute

world_center = world_center instance-attribute

title = title instance-attribute

map = map instance-attribute

on_goal = on_goal instance-attribute

app = self._build_app() instance-attribute

serve() async

Serve until Ctrl+C (SIGINT/SIGTERM) or :meth:request_stop.

Raises :class:RuntimeError when the frontend build is missing — i.e. a source checkout where npm run build has not been run.

request_stop()

Ask the running server to shut down gracefully.

_now_ms()

_static_dir()

The frontend build directory, if one has been built into the installed package (see the npm build step in .github/workflows/deploy.yml).

_rings(geometry)

Exterior rings of a (Multi)Polygon as [[x, y], ...] lists, coordinates rounded to 3 decimals (mm resolution) to slim the JSON.

driving_envelope_ids(coordinator)

robotID → envelope ID for every robot currently driving a mission.

static_content_key(coordinator)

Changes exactly when the static message content would: a robot appears, or the set of driving envelopes changes.

build_static_message(coordinator, *, title='coordination_oru', world_size=20.0, world_center=(0.0, 0.0), occupancy_map=None, map_data_uri=None, interactive=False)

The per-mission payload: paths and swept envelopes of driving robots.

build_state_message(coordinator)

The per-tick payload: placed footprints, reports, critical sections.

coordination_oru.viz.pyglet_viewer

Pyglet-based visualisation for a running :class:TrajectoryEnvelopeCoordinatorSimulation.

The viewer takes a snapshot of coordinator state on every draw frame (60 Hz by default) and re-creates a small set of pyglet shapes from it. State is read without locking — the brief race against the asyncio sim loop is harmless for visualisation purposes (worst case: one frame's worth of jitter).

Usage pattern (run sim in a daemon thread, viewer in the main thread):

.. code-block:: python

tec = TrajectoryEnvelopeCoordinatorSimulation(...)

async def run_sim():
    await tec.startInference()
    tec.addMissions(...)
    ...

loop = asyncio.new_event_loop()
threading.Thread(
    target=lambda: (asyncio.set_event_loop(loop), loop.run_until_complete(run_sim())),
    daemon=True,
).start()

viewer = PygletViewer(tec, world_size=15.0)
viewer.run()

Layers drawn (back to front):

1. Path polylines (faint, per-robot colour).
2. Swept-envelope outlines (very faint fill).
3. Critical-section index ranges highlighted in red on each path.
4. Current footprints filled in robot colour.
5. Status text (top-left).

ROBOT_COLORS = ((255, 99, 71), (51, 168, 255), (255, 195, 51), (130, 255, 51), (200, 51, 255), (51, 255, 200)) module-attribute

CS_HIGHLIGHT = (240, 90, 90, 200) module-attribute

PATH_COLOR = (180, 180, 180, 120) module-attribute

SWEPT_OPACITY = 28 module-attribute

TEXT_COLOR = (220, 220, 220, 255) module-attribute

LABEL_COLOR = (10, 10, 10, 255) module-attribute

BACKGROUND = (24, 24, 28, 255) module-attribute

TrajectoryEnvelopeTrackerDummy

Bases: AbstractTrajectoryEnvelopeTracker

onTrajectoryEnvelopeUpdate()

startTracking()

setCriticalPoint(criticalPoint)

getRobotReport()

finishParking()

isParkingFinished()

run() async

onPositionUpdate()

getCurrentTimeInMillis()

TrajectoryEnvelopeCoordinatorSimulation

Bases: TrajectoryEnvelopeCoordinator

DEFAULT_ROBOT_TRACKING_PERIOD = DEFAULT_ROBOT_TRACKING_PERIOD instance-attribute

DEFAULT_MAX_VELOCITY = MAX_VELOCITY instance-attribute

DEFAULT_MAX_ACCELERATION = MAX_ACCELERATION instance-attribute

useInternalCPs = True instance-attribute

checkCollisions = False instance-attribute

collisionsList = [] instance-attribute

totalMsgsLost = 0 instance-attribute

totalPacketsLost = 0 instance-attribute

DEFAULT_FOOTPRINT = Polygon(DEFAULT_FOOTPRINT) instance-attribute

MAX_DEFAULT_FOOTPRINT_DIMENSION = self.computeMaxFootprintDimension(DEFAULT_FOOTPRINT) instance-attribute

setCheckCollisions(enable)

incrementLostMsgsCounter()

incrementLostPacketsCounter()

setUseInternalCriticalPoints(value)

getRobotMaxVelocity(robotID)

getRobotMaxAcceleration(robotID)

getMaxFootprintDimension(robotID)

getDefaultFootprint()

getFootprint(robotID)

setDefaultFootprint(*coordinates)

getNewTracker(te, cb)

getCurrentTimeInMillis()

addMissions(*missions)

onCriticalSectionUpdate()

PygletViewer

Read-only viewer for a :class:TrajectoryEnvelopeCoordinatorSimulation.

coordinator = coordinator instance-attribute

world_size = world_size instance-attribute

world_center = world_center instance-attribute

draw_swept_envelope = draw_swept_envelope instance-attribute

window = pyglet.window.Window(width=width, height=height, caption=title) instance-attribute

batch = pyglet.graphics.Batch() instance-attribute

stop_when_idle()

Auto-close the window once every robot is parked (idle).

run()

place_footprint(footprint, pose)

Rotate footprint (centered at origin) by theta then translate to pose.