Using 5G Urllc Timing Beacons For Deterministic Multi-Robot Handovers With Edge Inference

Why I went down this rabbit hole

I was building a tiny “multi-robot relay” prototype for edge inference: one robot captures sensor data, an edge server runs the model, and another robot needs to act on the result with very tight timing. The annoying part wasn’t the model—it was handover timing.

On real 5G/6G deployments, handovers and scheduling can introduce jitter (random small timing shifts). For most apps that’s fine, but for “capture → transmit → infer → respond” loops, jitter can become the difference between stable behavior and oscillation.

So I tried a very specific trick: a deterministic timing beacon sent using an Ultra-Reliable Low-Latency Communications (URLLC)-like periodic pattern, then using that beacon to align handovers and inference windows across robots. The beacon itself isn’t magic; it’s just a timestamp reference that keeps every participant “speaking the same time language.”

Below is what I built and the code I used to simulate the mechanism.

---

The niche idea: “Edge Inference Windows” anchored to Timing Beacons

Timing beacon: a periodic message (think: “tick-tick-tick”) that carries a reference timestamp t0.

Inference window: a fixed time slot where the system is allowed to accept a specific sensor sample and produce an output.

My goal: when a robot switches networks (handover), it may experience delays, but the beacon alignment lets the edge server:

• map incoming sensor samples to the correct inference window, and
• decide whether to accept/reject a sample based on timing tolerance.

That turns handover jitter into a manageable “late/early sample” problem rather than a chaotic one.

---

What I implemented (in code): a beacon sender + edge receiver + robotic handover simulator

I built everything as a local simulation so I could reason about timing behavior. The key components:

1. Beacon server

   • broadcasts beacon messages with a timestamp
   • sends them at a fixed period (e.g. every 50 ms)

2. Robot simulator

   • generates “sensor frames” with timestamps
   • simulates handovers by injecting random network delay and sometimes dropping frames
   • reports the frame to the edge along with an observed receive time

3. Edge inference scheduler

   • computes which inference window a frame belongs to based on the last known beacon
   • accepts frames that land in-bounds
   • runs a dummy inference (so the timing pipeline is the focus)
   • returns an “action time” that the next robot could use

---

Step 1: Beacon server (periodic timing reference)

import asyncio
import json
import time
from dataclasses import dataclass

@dataclass
class Beacon:
    seq: int
    t0: float  # sender timestamp (seconds since epoch on sender side)
    period_s: float

async def beacon_server(host="127.0.0.1", port=9000, period_s=0.05):
    """
    Periodically emits a JSON beacon.
    In a real deployment, this would correspond to a URLLC-friendly periodic signal.
    """
    clients = set()
    start = time.time()
    seq = 0

    async def handle_client(reader, writer):
        clients.add(writer)
        try:
            while True:
                await asyncio.sleep(3600)
        finally:
            clients.remove(writer)
            writer.close()
            await writer.wait_closed()

    server = await asyncio.start_server(handle_client, host, port)

    async with server:
        while True:
            await asyncio.sleep(period_s)
            seq += 1
            beacon = Beacon(
                seq=seq,
                t0=time.time(),      # in a real system, this is a network-time synchronized timestamp
                period_s=period_s
            )
            msg = (json.dumps(beacon.__dict__) + "\n").encode("utf-8")
            for w in list(clients):
                try:
                    w.write(msg)
                    await w.drain()
                except Exception:
                    clients.discard(w)

async def main():
    await beacon_server()

if __name__ == "__main__":
    asyncio.run(main())

What matters here

• period_s is the “tick rate.” I used 50ms for clarity.
• Every beacon carries seq and t0. The edge uses these to anchor inference windows.
• In production, you’d align to a standardized time source (network time synchronization). In the sim, I use time.time().

---

Step 2: Edge scheduler + inference acceptance logic

This is where the deterministic behavior happens.

Definitions I used

• Inference window width: window_s (e.g. 20ms)
• Each frame has a frame_time_s (when it was captured) and an arrival_time_s (when the edge receives it).
• The scheduler estimates the time index (which window number) relative to the last beacon.

import asyncio
import json
import time
from dataclasses import dataclass

@dataclass
class Frame:
    robot_id: int
    seq: int
    frame_time_s: float
    payload: dict

@dataclass
class FrameResult:
    robot_id: int
    seq: int
    window_index: int
    accepted: bool
    reason: str
    action_time_s: float | None

class EdgeScheduler:
    def __init__(self, window_s=0.02, slack_s=0.008):
        """
        window_s: size of each inference window
        slack_s: how late a frame can arrive and still be accepted
        """
        self.window_s = window_s
        self.slack_s = slack_s
        self.last_beacon_t0 = None
        self.last_beacon_seq = None
        self.beacon_period_s = None

    def update_beacon(self, beacon: dict):
        self.last_beacon_t0 = beacon["t0"]
        self.last_beacon_seq = beacon["seq"]
        self.beacon_period_s = beacon["period_s"]

    def window_index_for_frame(self, frame_time_s: float) -> int:
        if self.last_beacon_t0 is None:
            # No timing anchor yet—force a "reject" scenario upstream.
            return -1
        # Compute how many full beacon periods have elapsed, then map to windows.
        elapsed_s = frame_time_s - self.last_beacon_t0
        # Each beacon period is period_s, but we split into fixed inference windows.
        # This makes windows deterministic even if beacons are the coarse tick.
        windows_per_period = int(round(self.beacon_period_s / self.window_s))
        if windows_per_period <= 0:
            windows_per_period = 1
        beacon_periods_elapsed = int(elapsed_s // self.beacon_period_s)
        within_period_s = elapsed_s - beacon_periods_elapsed * self.beacon_period_s
        within_period_index = int(within_period_s // self.window_s)
        return beacon_periods_elapsed * windows_per_period + within_period_index

    async def run_inference(self, frame: Frame, arrival_time_s: float) -> FrameResult:
        if self.last_beacon_t0 is None:
            return FrameResult(
                robot_id=frame.robot_id,
                seq=frame.seq,
                window_index=-1,
                accepted=False,
                reason="no_beacon_anchor",
                action_time_s=None,
            )

        widx = self.window_index_for_frame(frame.frame_time_s)

        # Determine when the frame was "supposed" to be in the window
        # by computing the start of that window in terms of the beacon anchor.
        # This is the core acceptance test.
        windows_per_period = int(round(self.beacon_period_s / self.window_s))
        if windows_per_period <= 0:
            windows_per_period = 1

        beacon_periods = widx // windows_per_period
        within_period_index = widx % windows_per_period
        scheduled_window_start = (
            self.last_beacon_t0
            + beacon_periods * self.beacon_period_s
            + within_period_index * self.window_s
        )

        # Accept if arrival is not later than window_start + slack
        latest_ok = scheduled_window_start + self.window_s + self.slack_s

        if arrival_time_s <= latest_ok:
            # Dummy inference: in reality you’d run an edge model here.
            # I include a small processing delay to simulate edge compute.
            await asyncio.sleep(0.002)

            action_time = arrival_time_s + 0.001  # when the action becomes available
            return FrameResult(
                robot_id=frame.robot_id,
                seq=frame.seq,
                window_index=widx,
                accepted=True,
                reason="accepted_within_slack",
                action_time_s=action_time,
            )
        else:
            return FrameResult(
                robot_id=frame.robot_id,
                seq=frame.seq,
                window_index=widx,
                accepted=False,
                reason="late_after_window",
                action_time_s=None,
            )

async def beacon_listener_and_scheduler(beacon_host="127.0.0.1", beacon_port=9000):
    """
    Connects to beacon server and keeps timing anchor updated.
    """
    reader, writer = await asyncio.open_connection(beacon_host, beacon_port)
    scheduler = EdgeScheduler(window_s=0.02, slack_s=0.008)

    # keep reading beacons forever
    buf = b""
    while True:
        data = await reader.readline()
        if not data:
            break
        beacon = json.loads(data.decode("utf-8"))
        scheduler.update_beacon(beacon)

    writer.close()
    await writer.wait_closed()

# The scheduler above is embedded in the combined demo below.

Why this works (and what I observed)

• The beacon provides a stable reference (last_beacon_t0).
• The inference window index is derived deterministically from the frame’s capture time.
• If handover jitter delays arrival too long, the frame becomes “late,” and the edge rejects it rather than feeding stale data into the control loop.

That makes system behavior more predictable under network transitions.

---

Step 3: Robot handover simulator (injecting jitter like a real handover)

I simulated:

• “good” network delay when not handover-ing
• a brief handover period where delay spikes
• occasional drops

import asyncio
import random
import time
from dataclasses import dataclass

@dataclass
class Frame:
    robot_id: int
    seq: int
    frame_time_s: float
    payload: dict

async def robot_simulator(robot_id, frame_rate_hz=40, run_s=2.5):
    """
    Generates frames at a fixed capture rate.
    Simulates network handover by increasing delay mid-run.
    """
    interval_s = 1.0 / frame_rate_hz
    seq = 0
    t_end = time.time() + run_s

    # Define a handover region
    handover_start = time.time() + run_s * 0.45
    handover_end = handover_start + run_s * 0.18

    while time.time() < t_end:
        frame_time_s = time.time()
        seq += 1
        payload = {"signal": random.random()}

        # Base network delay (seconds)
        if handover_start <= time.time() <= handover_end:
            # handover: bigger jitter + occasional drop
            if random.random() < 0.08:
                # drop
                await asyncio.sleep(interval_s)
                continue
            delay_s = random.uniform(0.010, 0.040)
        else:
            delay_s = random.uniform(0.002, 0.010)

        await asyncio.sleep(delay_s)  # simulate network transit
        arrival_time_s = time.time()

        yield Frame(robot_id=robot_id, seq=seq, frame_time_s=frame_time_s, payload=payload), arrival_time_s

        # wait until next capture time
        await asyncio.sleep(max(0.0, interval_s - delay_s))

---

Step 4: Putting it together (single script demo)

This combines:

• beacon server running in the background
• edge scheduler reading beacons and deciding acceptance
• two robots with slightly different handover timings

import asyncio
import json
import time
from dataclasses import dataclass

# --- Beacon server code (same as earlier, shortened to fit) ---
@dataclass
class Beacon:
    seq: int
    t0: float
    period_s: float

async def beacon_server(host="127.0.0.1", port=9000, period_s=0.05):
    clients = set()
    seq = 0

    async def handle_client(reader, writer):
        clients.add(writer)
        try:
            while True:
                await asyncio.sleep(3600)
        finally:
            clients.remove(writer)
            writer.close()
            await writer.wait_closed()

    server = await asyncio.start_server(handle_client, host, port)
    async with server:
        while True:
            await asyncio.sleep(period_s)
            seq += 1
            beacon = Beacon(seq=seq, t0=time.time(), period_s=period_s)
            msg = (json.dumps(beacon.__dict__) + "\n").encode("utf-8")
            for w in list(clients):
                try:
                    w.write(msg)
                    await w.drain()
                except Exception:
                    clients.discard(w)

# --- Edge scheduler code ---
@dataclass
class Frame:
    robot_id: int
    seq: int
    frame_time_s: float
    payload: dict

@dataclass
class FrameResult:
    robot_id: int
    seq: int
    window_index: int
    accepted: bool
    reason: str
    action_time_s: float | None

class EdgeScheduler:
    def __init__(self, window_s=0.02, slack_s=0.008):
        self.window_s = window_s
        self.slack_s = slack_s
        self.last_beacon_t0 = None
        self.beacon_period_s = None

    def update_beacon(self, beacon: dict):
        self.last_beacon_t0 = beacon["t0"]
        self.beacon_period_s = beacon["period_s"]

    def window_index_for_frame(self, frame_time_s: float) -> int:
        if self.last_beacon_t0 is None:
            return -1
        elapsed_s = frame_time_s - self.last_beacon_t0
        windows_per_period = int(round(self.beacon_period_s / self.window_s))
        windows_per_period = max(1, windows_per_period)
        beacon_periods_elapsed = int(elapsed_s // self.beacon_period_s)
        within_period_s = elapsed_s - beacon_periods_elapsed * self.beacon_period_s
        within_period_index = int(within_period_s // self.window_s)
        return beacon_periods_elapsed * windows_per_period + within_period_index

    async def handle_frame(self, frame: Frame, arrival_time_s: float) -> FrameResult:
        if self.last_beacon_t0 is None:
            return FrameResult(frame.robot_id, frame.seq, -1, False, "no_beacon_anchor", None)

        widx = self.window_index_for_frame(frame.frame_time_s)
        windows_per_period = int(round(self.beacon_period_s / self.window_s))
        windows_per_period = max(1, windows_per_period)

        beacon_periods = widx // windows_per_period
        within_period_index = widx % windows_per_period

        scheduled_window_start = (
            self.last_beacon_t0
            + beacon_periods * self.beacon_period_s
            + within_period_index * self.window_s
        )

        latest_ok = scheduled_window_start + self.window_s + self.slack_s

        if arrival_time_s <= latest_ok:
            await asyncio.sleep(0.002)  # simulate edge inference time
            action_time = arrival_time_s + 0.001
            return FrameResult(frame.robot_id, frame.seq, widx, True, "accepted_within_slack", action_time)

        return FrameResult(frame.robot_id, frame.seq, widx, False, "late_after_window", None)

# --- Robot simulator ---
async def robot_simulator(robot_id, frame_rate_hz=40, run_s=2.5, handover_offset_s=0.0):
    interval_s = 1.0 / frame_rate_hz
    seq = 0
    t_end = time.time() + run_s

    handover_start = time.time() + run_s * 0.45 + handover_offset_s
    handover_end = handover_start + run_s * 0.18

    while time.time() < t_end:
        frame_time_s = time.time()
        seq += 1
        payload = {"signal": robot_id + random.random()}

        if handover_start <= time.time() <= handover_end:
            if random.random() < 0.08:
                await asyncio.sleep(interval_s)
                continue
            delay_s = random.uniform(0.010, 0.040)
        else:
            delay_s = random.uniform(0.002, 0.010)

        await asyncio.sleep(delay_s)
        arrival_time_s = time.time()

        yield Frame(robot_id, seq, frame_time_s, payload), arrival_time_s
        await asyncio.sleep(max(0.0, interval_s - delay_s))

# --- Main integration ---
async def main():
    import random
    beacon_host, beacon_port = "127.0.0.1", 9000

    # Start beacon server
    beacon_task = asyncio.create_task(beacon_server(host=beacon_host, port=beacon_port, period_s=0.05))
    await asyncio.sleep(0.1)  # give server time to start

    # Connect edge to beacon stream
    reader, _ = await asyncio.open_connection(beacon_host, beacon_port)
    scheduler = EdgeScheduler(window_s=0.02, slack_s=0.008)

    stop_at = time.time() + 2.7
    results = []

    async def beacon_reader():
        buf = b""
        while time.time() < stop_at:
            line = await reader.readline()
            if not line:
                break
            beacon = json.loads(line.decode("utf-8"))
            scheduler.update_beacon(beacon)

    async def consume_robots():
        async def feed_robot(robot_id, offset):
            async for frame, arrival in robot_simulator(robot_id, run_s=2.5, handover_offset_s=offset):
                res = await scheduler.handle_frame(frame, arrival)
                results.append(res)

        t1 = asyncio.create_task(feed_robot(1, handover_offset_s=0.00))
        t2 = asyncio.create_task(feed_robot(2, handover_offset_s=0.06))
        await asyncio.gather(t1, t2)

    await asyncio.gather(beacon_reader(), consume_robots())
    beacon_task.cancel()

    # Print a small summary
    total = len(results)
    accepted = sum(1 for