How I built a browser teleop simulator with FastAPI and WebSockets

I built Teleop Console Demo as a portfolio simulation: a browser drives a differential-drive robot around a small 2D arena. The console shows pose, velocity, geometric lidar, battery, network controls, and a session archive. There is no physical robot connected.

I wanted the command path to be visible alongside the moving dot. Holding Shift enables motion, releasing it stops motion, and emergency stop stays latched until Reset. Those are illustrative software interlocks. This repo contains no certified safety system, camera stream, authentication, or production deployment evidence.

Keep the browser and simulation separate

The server is one Python process. FastAPI serves the static files in web/, REST controls under /api/, and a WebSocket at /ws. There is no frontend build. The browser draws the arena on canvas and sends keyboard or gamepad input.

I split the backend into small modules: sim.py owns the world and robot, safety.py owns interlocks, netem.py delays packets, and recorder.py handles JSONL. server.py connects them. Messages carry a topic, timestamp, and data object, with names such as /cmd_vel, /odom, and /scan.

The first WebSocket connection becomes the operator. Other tabs observe telemetry, and their motion commands are rejected. When the operator disconnects, the oldest remaining connection is promoted. This is a local ownership rule, not user authentication.

Follow a command through the server

Validate input before queuing motion

teleop/server.py checks that the deadman field is a boolean and velocities are numbers. Nonfinite velocities fail validation through clamp_velocity. A valid operator command updates the deadman state before entering the delay queue.

The release path is deliberately short in teleop/server.py:

                        console.deadman.update(cid, data['deadman'])
                        if not data['deadman']:
                            console.stop()
                            continue
                        if console.replaying:
                            await diagnostic('Live commands blocked during replay')
                            continue
                        await client['net'].submit({'message': message, 'epoch': console.epoch, 'received': time.monotonic()})

A release calls console.stop() and skips network emulation. That means the synthetic loss slider cannot swallow the release. The browser also sends a release on blur, page hide, and hidden-tab transitions in web/app.js.

Console.stop() zeros velocity and increments an epoch counter. Commands carry the epoch from their arrival, so queued motion from before a stop cannot restart the robot afterward. E-stop, reset, replay transitions, and operator loss use that mechanism too.

Add delay without blocking reception

Each connection has its own NetworkEmulator, sharing the console's network settings. Its submission method in teleop/netem.py schedules delivery using monotonic time:

    async def submit(self, message):
        delivery = self.apply(message)
        if delivery:
            now = time.monotonic()
            await self.queue.put((now+delivery.delay_ms/1000, next(self.sequence), now, delivery))

Before submission, apply() samples synthetic loss and computes latency plus uniform jitter, with delay floored at zero. The priority queue orders packets by delivery time, with a sequence counter for ties. Its worker polls so a newer packet with less delay can overtake an older one.

At delivery, the server rejects an outdated epoch or a command more than 300 ms old. It then applies the per-client 30 Hz rate limit, checks interlocks, clamps velocity to 1 m/s and 2 rad/s, and feeds the watchdog. The browser sends at 20 Hz. These are configured limits, not measured throughput results.

The envelope timestamp is not trusted for freshness. The server stores its own monotonic arrival time. Setting latency above the freshness window therefore prevents motion on purpose.

Check safety during physics too

A command is not the only place to enforce a stop. Before each physics step, the console checks e-stop, replay, operator presence, and deadman state. It also runs the watchdog from teleop/safety.py:

class Watchdog:
    def __init__(self):
        self.last_command = -math.inf
    def feed(self, now=None):
        self.last_command = time.monotonic() if now is None else now
    def is_stale(self, now=None, timeout=0.3):
        return (time.monotonic() if now is None else now) - self.last_command > timeout
    def enforce(self, robot, now=None):
        if self.is_stale(now):
            robot.stop()

Only an accepted motion command feeds this timer. Pings do not keep motion alive. If commands stop arriving, the next physics safety check after the timeout stops the robot.

Reset clears the latch, respawns a fresh robot, restores battery, clears deadman state, and replaces the watchdog. It is a visible simulation reset, not a hardware recovery procedure.

Step the robot on its own clock

teleop/sim.py integrates straight motion directly and turning motion along an arc. After integration it checks the robot's circular footprint against world boundaries, rectangles, and circles. On collision it restores the previous pose and zeros velocity. Battery drain depends on absolute linear and angular speed.

The scheduler accumulates elapsed time but always gives the model the same step size in teleop/sim.py:

class PhysicsLoop:
    def __init__(self, robot, before_step=lambda: None, hz=50):
        self.robot, self.before_step, self.dt = robot, before_step, 1/hz
    async def run(self):
        previous, accumulator = time.monotonic(), 0.0
        while True:
            now = time.monotonic()
            accumulator += now-previous
            previous = now
            while accumulator >= self.dt:
                self.before_step()
                self.robot.step(self.dt)
                accumulator -= self.dt
            await asyncio.sleep(max(0, self.dt-accumulator))

The default is 50 Hz physics. Telemetry runs separately with a 0.05 second sleep, targeting 20 Hz. Those settings do not establish real-time scheduling guarantees. Telemetry sends have a timeout so one stalled client cannot hold a send indefinitely.

The lidar is geometry, not sensor data: 180 rays intersect walls and obstacles, capped at a maximum range. Collision detection happens after each step, so it is not continuous collision detection.

Record messages, replay the view

Record creates an append-only JSONL file under recordings/. Each message is serialized and flushed. The server records operator commands and outbound telemetry, excluding pings and connection events.

Replayer.messages() preserves the spacing between message timestamps. The server chooses which recorded messages to broadcast in teleop/server.py:

    async def replay(self, path):
        try:
            async for message in Replayer(path).messages():
                if message['topic'] in ('/odom', '/scan', '/battery_state'):
                    await self.broadcast(message)
        finally:
            self.replaying = False
            self.stop()

Replay forwards odometry, scan, and battery messages. It does not execute recorded commands. Starting replay stops recording and live motion, and live commands are blocked. The browser shows a REPLAY banner; Exit replay returns to live telemetry.

That distinction matters when reading a recording. A replayed pose describes a stored session, not the current position of a robot being driven by that file.

Run the local demo

The README requires Python 3.11 or newer. Its virtual environment path is in README.md:

python3 -m venv .venv
. .venv/bin/activate
pip install -r requirements.txt
python -m teleop.server

Open localhost on port 8000. PORT changes the Python server port. Alternatively, make run installs requirements and starts the server, choosing the virtual environment's Python when available. The README also provides docker compose up --build; Compose publishes port 8000 and mounts recordings/ for persistence.

Start with zero network degradation. Hold Shift and use WASD or arrow keys, then release Shift. Open another tab to check the observer role. Try e-stop and Reset before increasing latency. Record a short session, stop recording, select it, and replay it.

The RTT display uses browser ping-to-pong elapsed time. Synthetic delay is applied once on the echo path, so it is not twice the latency slider. The server diagnostic named rtt_ms reports queue delay only.

Test the boundaries

The README's test command is in README.md:

pip install -r requirements.txt
pytest -q
# or
make test

Unit tests cover straight and arc motion, rotation, obstacle collisions, lidar distances, battery depletion, fixed steps, watchdog behavior, and rate limiting. Network tests check loss, jitter bounds, and a fast packet overtaking a delayed packet.

tests/test_integration_ws.py exercises live movement, observer rejection, e-stop, disconnect, validation, stale packets, and release under synthetic loss. It also checks recorded odometry arriving over WebSocket during replay and live-command isolation. The factory's start_loops=False option lets a test step the model directly.

.github/workflows/ci.yml installs requirements and runs pytest on Python 3.11 for pushes and pull requests. That checks behavior in this simulated process. The repo has no benchmark results or measured hardware stopping distances to quote.

Limits and what I would do next

I would add browser automation for focus changes and keyboard controls, since backend tests do not exercise those browser events. I would also measure behavior with slow clients before making performance claims.

A physical robot would need a different scope: authentication, independently engineered hardware fail-safes, and deployment-specific validation. The README sketches a ROS 2 bridge and camera stream as possible extensions. Neither is implemented here.

Read the implementation

The repository contains the console, simulation, tests, and run commands. I would start with teleop/server.py, trace one command through the queue, then read the tests for release and stale packets. That is where the demo's stopping behavior becomes concrete. If you want the same ideas applied to a 3D manipulator, my humanoid arm teleop tutorial covers a simulated 6-DOF arm with damped least squares IK, a three.js URDF viewer, and joint trajectory replay.