Environment Wrappers

NetworkedEnv 

class netrl.NetworkedEnv(env, config, channel_config=None, node_id='agent_0')[source]

Bases: Wrapper

Gymnasium wrapper simulating networked observation transmission.

Parameters:

env (gymnasium.Env) – The base environment to wrap. Must have a Box observation space.
config (NetworkConfig) – Channel and buffer configuration. For the Gilbert-Elliott backend this also carries the Markov-chain and loss parameters. Validated on construction.
channel_config (NS3WifiConfig | NS3WiFiChannelFastConfig | NS3MmWaveConfig | NS3LenaConfig | None, optional) –
Selects and configures the channel backend:

None (default)
Use the Gilbert-Elliott channel. All GE parameters are taken from config (p_gb, p_bg, loss_good, loss_bad, delay_steps).

NS3WifiConfig(...)
Use the ns-3 802.11a WiFi channel (subprocess-based). The binary src/ns3_wifi_sim must be built first (bash src/build_ns3_sim.sh).

NS3WiFiChannelFastConfig(...)
Use the ns-3 802.11a WiFi channel (pybind11 fast binding, 15-20x faster). The pybind11 extension must be built first (bash build_pybind11.sh).

NS3MmWaveConfig(...)
Use the ns-3 5G mmWave EPC channel. The binary src/ns3_mmwave_sim must be built first (bash src/build_ns3_mmwave_sim.sh).

NS3LenaConfig(...)
Use the ns-3 5G-LENA NR EPC channel. The binary src/ns3_lena_sim must be built first (bash src/build_ns3_lena_sim.sh).
node_id (str) – Identifier for this agent’s transmission node. Default “agent_0”.

__init__(env, config, channel_config=None, node_id='agent_0')[source]

Wraps an environment to allow a modular transformation of the step() and reset() methods.

Args:: env: The environment to wrap

Parameters:

env (Env)
config (NetworkConfig)
channel_config (NS3WifiConfig | NS3MmWaveConfig | NS3LenaConfig | None)
node_id (str)

Return type:

None

reset(*, seed=None, options=None)[source]

Reset the wrapped environment and clear all channel / buffer state.

Returns the initial Dict observation (all zeros, all recv_mask=False) because no observation has been transmitted or received yet.

Parameters:

seed (int | None)
options (dict | None)

Return type:

Tuple[Dict[str, ndarray], dict]

step(action, packet_size=None)[source]

Step the underlying environment and run the channel simulation.

Parameters:

action (Any Action compatible with the wrapped env.)
packet_size (int | None Payload bytes to use for the packet) – transmitted this step. None means use the channel’s own default (NS3WifiConfig. packet_size_bytes for the ns-3 backend, ignored for GE / Perfect channels).
step (Sequence per)
-----------------
raw_obs (1. env.step(action) ->)
reward
term
trunc
info
channel (2. central.receive_from(...) -> transmit through)
None) (3. central.flush_and_update() -> flush + buffer.add(obs or)
1 (4. step_count +=)
(obs_array (5. buffer.get_padded() ->)
recv_mask)
observation (6. Return Dict)
reward
flags
info. (augmented)
with (The info dict is extended) – “channel_info” : dict from get_channel_info() (state, params…) “arrived_this_step” : bool, True if a packet arrived at this step.

Return type:

Tuple[Dict[str, ndarray], float, bool, bool, dict]

property step_count: int: Current integer step counter (0-indexed; incremented after each step).

property config: NetworkConfig: The NetworkConfig used to configure this wrapper.

property central_node: CentralNode: Direct access to the underlying CentralNode (for multi-agent use).

Observation space

The original Box(obs_shape) is replaced with:

Dict({
    "observations": Box(shape=(buffer_size, *obs_shape), dtype=obs_dtype),
    "recv_mask":    MultiBinary(buffer_size),
})

Extended info keys

Key	Description
`"channel_info"`	`dict` from `get_channel_info()`.
`"arrived_this_step"`	`bool` — packet arrived this step.

MultiViewNetworkedEnv 

class netrl.MultiViewNetworkedEnv(env, config, observer_ids, multi_view_model, channel_factory)[source]

Bases: Wrapper

Gymnasium wrapper with multiple observers and per-step transmission control.

Parameters:

env (gymnasium.Env) – The base environment. Must have a Box observation space.
config (NetworkConfig) – Shared channel and buffer configuration. buffer_size applies to every observer’s buffer. For GE channels, the Markov parameters are also read from here. Call config.validate() before passing.
observer_ids (List[str]) – Unique string identifier for each observer. The length of this list determines the number of independent transmission paths. These ids are also used as keys in the returned observation dict and in the transmit_mask / packet_sizes arguments of step().
channel_factory (Callable[[NetworkConfig], CommChannel]) –
Called once per observer (in the order of observer_ids) during construction to create each observer’s CommChannel. Choose from:

GEChannel
Independent Gilbert–Elliott channel per observer. All GE parameters come from config.

PerfectChannel
Lossless zero-delay channel (useful for debugging).

lambda nc: NS3WifiChannel(nc, NS3WifiConfig(...))
Independent ns-3 802.11a WiFi subprocess per observer.

make_multi_ue_wifi_factory(NS3WifiMultiUEConfig(...))
Shared ns-3 infrastructure BSS for all observers. All observers contend for the same wireless medium. The factory must be created before passing it here, and NS3WifiMultiUEConfig.n_ues must equal len(observer_ids).
multi_view_model (MultiViewModel)

Examples

Three observers sharing a single 802.11a WiFi channel:

from netrl import NetworkConfig, MultiViewNetworkedEnv
from netrl import NS3WifiMultiUEConfig, make_multi_ue_wifi_factory

factory = make_multi_ue_wifi_factory(
    NS3WifiMultiUEConfig(
        n_ues=3,
        distances_m=[10.0, 30.0, 60.0],
        step_duration_ms=2.0,
    )
)
env = MultiViewNetworkedEnv(
    gym.make("CartPole-v1"),
    NetworkConfig(buffer_size=10),
    observer_ids=["near", "mid", "far"],
    channel_factory=factory,
)
obs, info = env.reset()
# obs["near"]["observations"].shape == (10, 4)
# obs["near"]["recv_mask"].shape    == (10,)

# Step with all observers transmitting (default)
obs, r, term, trunc, info = env.step(action)

# Step: only "near" transmits, with 256 bytes
obs, r, term, trunc, info = env.step(
    action,
    transmit_mask={"near": True, "mid": False, "far": False},
    packet_sizes={"near": 256},
)

__init__(env, config, observer_ids, multi_view_model, channel_factory)[source]

Wraps an environment to allow a modular transformation of the step() and reset() methods.

Args:: env: The environment to wrap

Parameters:

env (Env)
config (NetworkConfig)
observer_ids (List[str])
multi_view_model (MultiViewModel)
channel_factory (Callable[[NetworkConfig], CommChannel])

Return type:

None

reset(*, seed=None, options=None)[source]

Reset the wrapped environment and all channel / buffer state.

Returns all-zeros observations with recv_mask=False for every observer because no transmission has occurred yet.

Parameters:

seed (int | None)
options (dict | None)

Return type:

Tuple[Dict[str, Dict[str, ndarray]], dict]

step(action, *, transmit_mask=None, packet_sizes=None)[source]

Step the environment and run the multi-observer channel simulation.

Parameters:

action (Any) – Action compatible with the wrapped environment.
transmit_mask (Dict[str, bool] | None, optional) –
Controls which observers transmit this step.

None (default)
All observers transmit.

Dict mapping observer_id → bool
Only observers mapped to True transmit. Observers absent from the dict default to True (i.e. the mask is an opt-out, not opt-in).
packet_sizes (Dict[str, int] | None, optional) –
Per-observer payload in bytes for this step.

None (default)
Every transmitting observer uses its channel’s own default packet size (e.g. NS3WifiMultiUEConfig.packet_size_bytes).

Dict mapping observer_id → int
Overrides the packet size for the named observer. Absent observers use the channel default.

Returns:

obs (Dict[observer_id → Dict{“observations”, “recv_mask”}]) – Observation buffers for all observers.
reward (float)
terminated (bool)
truncated (bool)
info (dict) – Extended with:

"channel_info"
Dict[observer_id → channel diagnostic dict from CommChannel.get_channel_info()].

"arrived_this_step"
Dict[observer_id → bool] — True if a packet from that observer arrived at the AP during this step.

"transmitted_this_step"
Dict[observer_id → bool] — True if that observer attempted to transmit this step (i.e. was not masked out).

Return type:

Tuple[Dict[str, Dict[str, ndarray]], float, bool, bool, dict]

Notes

flush_and_update is called for all observers every step, regardless of transmit_mask. This ensures every buffer advances by exactly one slot per step and that delayed packets from previous steps are still collected for silent observers.

property step_count: int: Current step counter (0-indexed; incremented after each step call).

property observer_ids: List[str]: Ordered list of observer identifiers.

property config: NetworkConfig: The NetworkConfig used to configure this wrapper.

property central_node: CentralNode: Direct access to the underlying CentralNode.

Observation space

Dict({
    "<observer_id>": Dict({
        "observations": Box(shape=(buffer_size, *obs_shape_i), dtype=obs_dtype_i),
        "recv_mask":    MultiBinary(buffer_size),
    }),
    ...
})

Extended info keys

Key	Description
`"channel_info"`	`Dict[observer_id → channel_info_dict]`
`"arrived_this_step"`	`Dict[observer_id → bool]`
`"transmitted_this_step"`	`Dict[observer_id → bool]`

Environment Wrappers

NetworkedEnv

MultiViewNetworkedEnv

NetworkedEnv 

MultiViewNetworkedEnv 