zymera.obs

obs

ObsBuilder + the named obs-channel registry.

Observations are composed from named channels — small pure functions fn(world, ctx) -> (N, H, W) | (H, W) looked up in :data:CHANNEL_FNS. A new observation idea is a new channel registered from an experiment file (:func:register_channel), never an edit to the simulator.

Builders (both frozen, hashable, closure-captured static objects):

• :class:VectorObs — v0 World._sensing: (N, 3) float32 of own (row, col) + global visited fraction.
• :class:GridObs — stacked per-agent planes (N, C, H, W) float32, plus an optional centralized (Cg, H, W) critic view (CTDE).

Doctrine: per-agent belief channels read the POST-GOSSIP belief at world.channel.shared (duck-typed attribute — any channel pytree with a .shared field works). Ground truth (world.covered, world.wall) is reserved for the central critic view; leaking it into agent_obs is how a "decentralized" policy quietly stops being one.

Parity: GridObs(("known", "own_pos", "known_walls", "neighbors", "local_frontier")) reproduces v0 CommCoverageEnv._obs bit-for-bit; the default central tuple reproduces v0 global_state.

ObsBuilder

Bases: Protocol

Composable observation builder (duck-typed; see module docstring).

requires declares the StepCtx fields agent_obs/central_obs read — it joins the env's union at __init__ so unrequested context machinery never compiles.

obs_channels property

obs_channels

Per-agent channel count C (or feature dim D for vector obs).

central_channels property

central_channels

Centralized channel count Cg, or None when there is no critic view.

agent_obs

agent_obs(world, ctx)

(N, C, H, W) — or (N, D) for vector builders.

Source code in zymera/obs.py

def agent_obs(self, world, ctx) -> chex.Array:
    """(N, C, H, W) — or (N, D) for vector builders."""
    ...

central_obs

central_obs(world, ctx)

(Cg, H, W) centralized critic view, or None.

Source code in zymera/obs.py

def central_obs(self, world, ctx) -> Optional[chex.Array]:
    """(Cg, H, W) centralized critic view, or None."""
    ...

VectorObs dataclass

VectorObs()

v0 World._sensing verbatim: (N, 3) float32 per-agent obs of own (row, col) + visited fraction (mean of world.visited over ALL cells, walls included — v0 parity).

GridObs dataclass

GridObs(channels, sense_r=1, central=_DEFAULT_CENTRAL)

Stacked named channels: per-agent (N, C, H, W) float32 planes, plus an optional centralized (Cg, H, W) critic view.

• channels — per-agent plane names, stacked in order. (H, W) team planes broadcast to every agent.
• sense_r — Chebyshev radius bound into the "local_frontier" channel (the one channel parameterized by the builder).
• central — team-plane names for central_obs; None means no critic view. Each must produce an (H, W) plane.

Channel names are validated at construction (Python time) — fail before the trace, not inside it.

register_channel

register_channel(name, fn)

Register a custom obs channel fn(world, ctx) -> (N, H, W) | (H, W).

Experiment files add their channel ideas here; proven ones graduate into the table above with a test.

Source code in zymera/obs.py

def register_channel(name: str, fn: Callable) -> None:
    """Register a custom obs channel ``fn(world, ctx) -> (N, H, W) | (H, W)``.

    Experiment files add their channel ideas here; proven ones graduate
    into the table above with a test.
    """
    if name in CHANNEL_FNS:
        raise ValueError(f"obs channel already registered: {name!r}")
    CHANNEL_FNS[name] = fn