Skip to content

dotbot/calibration: take corner captures from the robot's button only #338

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
geonnave wants to merge 5 commits into
base: main
Choose a base branch
from
Open

dotbot/calibration: take corner captures from the robot's button only #338

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a4ca9c5
dotbot/calibration: take corner captures from the robot's button only
geonnave Oct 14, 2026
0e7ce7d
dotbot/console-web: drop the Capture button, the robot's button captures
geonnave Oct 14, 2026
81b0e08
dotbot/calibration: drop the session's reads per point
geonnave Oct 14, 2026
a63d3aa
dotbot/console-web: drop the Reads per point setting and the read bars
geonnave Oct 14, 2026
130b88e
dotbot/calibration: refuse a session start that still carries reads
geonnave Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 25 additions & 82 deletions dotbot/calibration/driver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,23 +4,17 @@
"""The controller's side of a calibration session: one at a time, with events.

The capture loop itself is `CalibrationSession`; this drives it from async
routes. It owns the swarmit client, built when a session starts and rebuilt
when the robot changes, and a start with no fleet in reach still opens; it
turns every state change into exactly one WebSocket notification, in the
order the changes happened; and it routes a capture that arrives from the
robot's own button to the outstanding point.

The blocking capture runs in a worker thread. Progress is published from
there by handing the coroutine back to the loop and waiting for it, so the
notifications keep the order the reads arrived in.
routes. It owns the swarmit client, built when a session starts, and a start
with no fleet in reach still opens; it turns every state change into exactly
one WebSocket notification, in the order the changes happened; and it routes
a capture that arrives from a robot's button to the outstanding point.
"""

from __future__ import annotations

import asyncio
from typing import Any, Callable, Sequence

from dotbot.calibration.ota import CAPTURE_READS_DEFAULT
from dotbot.calibration.points import field_corners, resolve_placement_points
from dotbot.calibration.push import PushRefused, gate_push, push_worklist
from dotbot.calibration.session import (
Expand All @@ -34,33 +28,19 @@
# Seconds between two checks for a button press given up incomplete.
EXPIRED_PRESS_POLL_INTERVAL = 1.0

# The swarmit log-event tag a raw-count capture carries. Imported lazily so
# the swarmit protocol registry stays out of PyDotBot test collection.
_CAPTURE_TAG: int | None = None


def capture_tag() -> int:
global _CAPTURE_TAG
if _CAPTURE_TAG is None:
from swarmit.testbed.protocol import LH2_CALIB_TAG

_CAPTURE_TAG = LH2_CALIB_TAG
return _CAPTURE_TAG


class SessionDriver:
"""One calibration session at a time, with its transport and its events.

`client_factory` takes the device address and returns a swarmit client,
injected so a test drives the whole loop without a fleet.
`client_factory` returns a swarmit client, injected so a test drives the
whole loop without a fleet.
"""

def __init__(
self,
client_factory: Callable[[str], Any],
client_factory: Callable[[], Any],
notify: Callable[[dict | None], Any],
site: Site | None = None,
stream_factory: Callable[[Any, str, Callable], Any] | None = None,
stream_factory: Callable[[Any, Callable], Any] | None = None,
):
self._client_factory = client_factory
self._notify = notify
Expand All @@ -70,7 +50,6 @@ def __init__(
self.logger = LOGGER.bind(context=__name__)
self._client: Any = None
self._stream: Any = None
self._device = ""
self._lock = asyncio.Lock()
self._loop: asyncio.AbstractEventLoop | None = None
self._expiry_watch: asyncio.Task | None = None
Expand Down Expand Up @@ -98,60 +77,31 @@ def preview(self, specs: Sequence[str]) -> dict:
"points": [
placement_dict(index, placement)
for index, placement in enumerate(placements)
],
"reads": CAPTURE_READS_DEFAULT,
]
}

async def start(
self,
specs: Sequence[str],
device: str = "",
area: str = "",
reads: int | None = None,
) -> dict:
"""Resolve the points and open a session with point 0 outstanding."""
async with self._lock:
self._close_stream()
extra = {} if reads is None else {"reads": reads}
session = CalibrationSession.resolve(list(specs), site=self.site, **extra)
session = CalibrationSession.resolve(list(specs), site=self.site)
session.device = device.upper()
session.area = area
self.session = session
self._loop = asyncio.get_running_loop()
# Button captures come unrequested from any robot, so the stream
# is listening from the start rather than from the first capture.
await asyncio.to_thread(self._listen_for_buttons, session.device)
# is listening from the start.
await asyncio.to_thread(self._listen_for_buttons)
if self._expiry_watch is None or self._expiry_watch.done():
self._expiry_watch = asyncio.create_task(self._watch_expired_presses())
await self._emit()
return session.as_dict()

async def capture(self, device: str = "") -> dict:
"""Take the outstanding point's reads from `device`, then advance."""
async with self._lock:
session = self._require()
if device:
session.device = device.upper()
if not session.device:
raise SessionError(
"no robot chosen: click one on the map, or name it in the "
"capture request"
)
loop = asyncio.get_running_loop()
self._loop = loop
stream = await asyncio.to_thread(self._ensure_stream, session.device)

def on_progress(_point) -> None:
asyncio.run_coroutine_threadsafe(self._emit(), loop).result()

try:
await asyncio.to_thread(session.capture, stream, on_progress)
session.error = ""
except TimeoutError as exc:
session.error = str(exc)
await self._emit()
return session.as_dict()

async def redo(self) -> dict:
"""Discard the last captured point's reads and re-open it."""
async with self._lock:
Expand Down Expand Up @@ -186,7 +136,7 @@ async def push(
async with self._lock:
session = self._require()
payload = session.push_payload()
client = await asyncio.to_thread(self._ensure_client, "")
client = await asyncio.to_thread(self._ensure_client)
try:
try:
check = await asyncio.to_thread(
Expand All @@ -205,7 +155,7 @@ async def push(
check.info_gens,
)
finally:
await asyncio.to_thread(self._listen_for_buttons, "")
await asyncio.to_thread(self._listen_for_buttons)
await self._emit()
return {
"id": session.saved_id,
Expand Down Expand Up @@ -301,30 +251,25 @@ def _require(self) -> CalibrationSession:
)
return self.session

def _ensure_client(self, device: str) -> Any:
if self._client is None or device != self._device:
self._close_stream()
if self._client is not None:
self._client.__exit__(None, None, None)
self._client = None
self._client = self._client_factory(device)
def _ensure_client(self) -> Any:
if self._client is None:
self._client = self._client_factory()
self._client.__enter__()
self._device = device
return self._client

def _ensure_stream(self, device: str) -> Any:
client = self._ensure_client(device)
def _ensure_stream(self) -> Any:
client = self._ensure_client()
if self._stream is None:
self._stream = self._stream_factory(client, device, self.on_button_capture)
self._stream = self._stream_factory(client, self.on_button_capture)
self._stream.__enter__()
return self._stream

def _listen_for_buttons(self, device: str) -> None:
def _listen_for_buttons(self) -> None:
try:
self._ensure_stream(device)
self._ensure_stream()
except Exception as exc: # a fleet out of reach is not a failure
self.logger.warning(
"Not listening for button captures until a capture reaches the fleet",
"Not listening for button captures: the fleet is out of reach",
error=str(exc),
)

Expand All @@ -338,9 +283,7 @@ def bind_loop(self, loop: asyncio.AbstractEventLoop) -> None:
self._loop = loop


def _default_stream(client: Any, device: str, on_button_capture: Callable) -> Any:
def _default_stream(client: Any, on_button_capture: Callable) -> Any:
from dotbot.calibration.ota import CaptureSession

return CaptureSession(
client, device, capture_tag(), on_button_capture=on_button_capture
)
return CaptureSession(client, on_button_capture=on_button_capture)
125 changes: 9 additions & 116 deletions dotbot/calibration/ota.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,9 @@

"""Over-the-air LH2 capture collection (swarmit transport).

A capture comes either from the calibrate app, taken on the robot's own
button, or from the secure bootloader, which samples its raw LH2 counts on
request (READY mode only). Both arrive inside a SWARMIT_EVENT_LOG. This
module decodes the records and triggers the bootloader captures; the solve
and the file live in `lighthouse2`.

DEPRECATED: the bootloader capture request, in favour of the calibrate app.
A capture is taken by the calibrate app on the robot's own button and
arrives inside SWARMIT_EVENT_LOGs. This module decodes and assembles the
records; the solve and the file live in `lighthouse2`.

A capture is n reads per point, not one: the per-point error is the
placement sigma and the lighthouse's own over sqrt(n) in quadrature, so a
Expand All @@ -21,7 +17,6 @@

from __future__ import annotations

import queue
import threading
import time
from collections.abc import Callable
Expand All @@ -30,9 +25,6 @@
from dotbot.calibration.lighthouse2 import LH_PERIODS, LH2CalibrationSample, Sample
from dotbot.calibration.points import CORNERS # noqa: F401 - the capture order

CAPTURE_TIMEOUT_DEFAULT = 5.0
CAPTURE_RETRIES_DEFAULT = 3
CAPTURE_READS_DEFAULT = 25

# Each raw sample inside the LOG payload is [lh_index:1][count1:4 LE][count2:4 LE].
_SAMPLE_SIZE = 9
Expand All @@ -55,18 +47,6 @@
STATION_PRESENCE_RATIO_MIN = 0.5


def parse_capture_payload(data: bytes, tag: int) -> list[LH2CalibrationSample]:
"""Decode a SWARMIT_EVENT_LOG payload of raw LH2 samples.

Layout (mirrors the swarmit bootloader): a 1-byte `tag`, then N
fixed-size records. Returns [] for any payload that is not a capture
(regular text log lines do not carry `tag` as their first byte).
"""
if len(data) < 1 or data[0] != tag:
return []
return _parse_records(data[1:])


def _parse_records(body: bytes) -> list[LH2CalibrationSample]:
samples: list[LH2CalibrationSample] = []
for off in range(0, len(body) - _SAMPLE_SIZE + 1, _SAMPLE_SIZE):
Expand Down Expand Up @@ -307,25 +287,15 @@ class CaptureSession:
A capture from the calibrate app's button can come from any device and
at any time; its chunks are assembled into one `ButtonCapture` per press
and handed to `on_button_capture`.

With a `device`, `capture()` also triggers that device's READY-mode
capture (DEPRECATED, in favour of the button) and waits for the reply,
re-triggering on timeout because the trigger send is best-effort. The
bootloader only answers a trigger, so one stream serves every point.
"""

def __init__(
self,
client,
device: str | None,
tag: int,
on_button_capture: Callable[[ButtonCapture], None] | None = None,
button_timeout: float = BUTTON_CAPTURE_TIMEOUT_DEFAULT,
):
self._client = client
self._device = (device or "").upper()
self._tag = tag
self._queue: queue.Queue = queue.Queue()
self._stop = threading.Event()
self._on_button_capture = on_button_capture
self._assembler = ButtonAssembler(timeout=button_timeout)
Expand All @@ -340,23 +310,12 @@ def __exit__(self, *exc) -> None:
self._stop.set()

def _reader(self) -> None:
try:
for event in self._client.watch_log_events():
if self._stop.is_set():
break
addr = str(event.get("addr", "")).upper()
data = bytes.fromhex(event.get("data_hex", ""))
chunk = parse_button_payload(data)
if chunk is not None:
self._on_button_chunk(addr, chunk)
continue
if not self._device or addr != self._device:
continue
decoded = parse_capture_payload(data, self._tag)
if decoded:
self._queue.put(decoded)
except Exception as exc: # surfaced on the next capture() get()
self._queue.put(exc)
for event in self._client.watch_log_events():
if self._stop.is_set():
break
chunk = parse_button_payload(bytes.fromhex(event.get("data_hex", "")))
if chunk is not None:
self._on_button_chunk(str(event.get("addr", "")).upper(), chunk)

def _on_button_chunk(self, addr: str, chunk: ButtonChunk) -> None:
if self._on_button_capture is None:
Expand All @@ -366,73 +325,7 @@ def _on_button_chunk(self, addr: str, chunk: ButtonChunk) -> None:
if capture is not None:
self._on_button_capture(capture)

@property
def device(self) -> str:
"""The device `capture()` triggers, or "" for a button-only stream."""
return self._device

def expired_presses(self) -> list[tuple[str, int]]:
"""Button presses given up on since the last call: (device, press)."""
with self._assembler_lock:
return self._assembler.expired()

def capture(
self,
timeout: float,
retries: int,
on_attempt: Callable[[int, int], None] | None = None,
) -> list[LH2CalibrationSample]:
"""Trigger one capture and return every station's record from the reply.

Raises TimeoutError if nothing arrives within `retries + 1` triggers,
and ValueError on a stream opened without a device.
"""
if not self._device:
raise ValueError(
"no device to trigger: this stream only takes button captures"
)
# Discard anything left over from the previous point.
while not self._queue.empty():
self._queue.get_nowait()

attempts = retries + 1
for attempt in range(attempts):
if on_attempt is not None:
on_attempt(attempt + 1, attempts)
self._client.request_lh2_capture(self._device)
deadline = time.monotonic() + timeout
remaining = deadline - time.monotonic()
while remaining > 0:
try:
item = self._queue.get(timeout=remaining)
except queue.Empty:
break
if isinstance(item, Exception):
raise item
return item

raise TimeoutError(
f"no LH2 samples from {self._device} after {retries + 1} attempt(s); "
f"is the DotBot in READY (app stopped) and in view of a lighthouse?"
)

def capture_point(
self,
point: int,
reads: int = CAPTURE_READS_DEFAULT,
timeout: float = CAPTURE_TIMEOUT_DEFAULT,
retries: int = CAPTURE_RETRIES_DEFAULT,
on_attempt: Callable[[int, int], None] | None = None,
on_read: (
Callable[[int, int, list[list[LH2CalibrationSample]]], None] | None
) = None,
) -> PointCapture:
"""Take `reads` captures at one point and group them per station."""
collected: list[list[LH2CalibrationSample]] = []
for index in range(reads):
collected.append(
self.capture(timeout=timeout, retries=retries, on_attempt=on_attempt)
)
if on_read is not None:
on_read(index + 1, reads, collected)
return samples_from_reads(collected, point)
Loading
Loading