Create a new environment
An environment is the task/world only — it owns the hidden target, the per-actor
measurement, and the team reward. Communication (delay, partial observability) is handled by
network.py, so you do not re-implement it.
The contract
from coadapt.env import Env, EnvSpec
from coadapt.spaces import Discrete # or MultiDiscrete / Box
import chex, jax, jax.numpy as jnp
@chex.dataclass
class MyWorld: # the hidden task state (a JAX pytree)
target: jnp.ndarray
t: jnp.ndarray
class MyEnv(Env):
def __init__(self, N=8, horizon=60, nu=0.0):
self.spec = EnvSpec(num_agents=N, meas_dim=3,
action_space=Discrete(5), horizon=horizon)
def reset(self, key):
# -> (measurement [N, meas_dim], world)
...
def step(self, key, world, actions):
# -> (measurement, world, reward_scalar, done, info)
...
Rules that keep it compatible
- reset/step are functional (take a key + state, return new state) so they
scan/vmap. - The measurement is what an actor senses of ITSELF/the world — not its neighbours. Neighbour information reaches an agent only through the network's delayed channel.
- reward is one shared team scalar in [0, 1] (cooperative).
- info should carry
per_actor_errorso the standard plots work.
Register it
# src/coadapt/envs/__init__.py
from .myenv import MyEnv
ENVS = { ..., "myenv": MyEnv }
It is now selectable as --env myenv everywhere.