scene_service.map_binding

Map identity binding — how scene learns WHICH map it is scoped to.

The mapping service broadcasts its current identity {map_id, mode, generation} latched on the robonix/service/map/lifecycle topic (see capabilities/service/map/lifecycle.v1.toml). At startup scene probes that broadcast once and, when present, binds to it — the broadcast is authoritative, so the deploy no longer needs to hand-copy the same map_id into two config blocks. Static fallbacks (manifest config / SCENE_MAP_ID env / “default”) remain for deployments where mapping is not up yet when scene starts (the normal full-boot order) or does not broadcast at all.

generation is mapping’s map-frame epoch: it bumps whenever the map origin may have changed (mapping-mode session start, reset_map) and stays put across localization round-trips. P2 records it and warns on runtime change; acting on it (flush + re-anchor) is the P3 lifecycle linkage.

Split from service.py so the pure precedence rule and the one-shot ROS read are unit-testable without atlas.

Functions

choose_map_binding(broadcast, config_map_id, ...)

Precedence: mapping's lifecycle broadcast (authoritative source of map identity) > manifest config.map_id > SCENE_MAP_ID env > "default".

read_latched_lifecycle(topic, timeout_s)

One-shot read of the latched MapLifecycle sample on topic.

sanitize_map_id(raw)

Classes

MapBinding(map_id, source[, mode, generation])

The resolved binding plus where it came from (for the startup log and for the runtime mismatch watcher).

class scene_service.map_binding.MapBinding(map_id: str, source: str, mode: str = '', generation: int | None = None)[source]

Bases: object

The resolved binding plus where it came from (for the startup log and for the runtime mismatch watcher).

generation: int | None = None
map_id: str
mode: str = ''
source: str
scene_service.map_binding.choose_map_binding(broadcast: dict | None, config_map_id: object, env_map_id: str | None) MapBinding[source]

Precedence: mapping’s lifecycle broadcast (authoritative source of map identity) > manifest config.map_id > SCENE_MAP_ID env > “default”. A broadcast with an empty map_id means mapping runs ephemeral (no named map) — treated as “no broadcast”, scene falls back to static binding.

scene_service.map_binding.read_latched_lifecycle(topic: str, timeout_s: float) dict | None[source]

One-shot read of the latched MapLifecycle sample on topic.

Runs on a PRIVATE rclpy context + executor so it neither initialises nor disturbs the default context the ingest hub owns (the hub calls rclpy.init() later and would raise on a double init). Returns {map_id, mode, generation} or None on timeout. Degrades to None with one warning when rclpy or the generated map interface package is missing (ros2_idl overlay not built) — binding then falls back to config/env, scene itself keeps working.

scene_service.map_binding.sanitize_map_id(raw: str | None) str[source]