Public environments and experiment records

Learning objectives

One line of termination handling can change the training target. Start by checking environment boundaries and fair policy evaluation.

Execution conditions

Run the baseline Queue example with NumPy. Compare policies on seed=100…104, which was not used for training.

Public environment activities use gymnasium and stable-baselines3. gymnasium[classic-control] is required only for visualization.

If the required package is missing, run %pip install control==0.10.2 gymnasium==1.3.0 stable-baselines3==2.9.2 in a Colab or Jupyter code cell.

Applications

Throughput and cost of equipment and communication resources

An average of 2 items arrive at the synthetic queue each time step. Increasing service capacity reduces queue length but increases energy cost.

Observation and state: Queue length $q=0$…20. Observation $q/20$.

Action: choose processing capacity 0, 2, or 4.

Reward and constraints: −(next queue length + 0.2×capacity² + 10×dropped-item count). Time limit at 100 steps.

Comparison baselines: Constant capacity of 2, and a threshold policy based on queue length.

Metrics to check: Evaluation return, mean queue length, mean capacity², and dropped items.

Choose equipment inspection times

Frame the research problem as balancing the cost of inspecting every time against the cost of missed failures.

Observation and state: Recent sensor readings, trends, and time since the last inspection. Distinguish the actual fault state from observations.

Actions: Continue operating, inspect, or replace.

Reward and constraints: Operating benefit − inspection, replacement, and failure costs.

Comparison baselines: Periodic inspection and threshold alarms.

Metrics to check: availability, failure count, and total cost. Check for failure label leakage.

Identify the boundaries of one episode

Starting conditions

Run the reset → step → termination flow using the Queue example. Use seed=7, action 1 (capacity 2), and 100 steps.

Procedure

1. Record (q,a,r,q_next,terminated,truncated) for the first 5 transitions.
2. Set the time limit to 5 steps and inspect the last transition.
3. Distinguish terminal-state termination from a data collection time limit, and write the mask used for the bootstrap target.

Results to retain

• A transition table and an explanation of the last step's mask.

"""Synthetic service queue. An action chooses capacity, including its energy cost."""
import numpy as np

class Queue:
    def reset(self, seed=0):
        self.rng = np.random.default_rng(seed)
        self.q, self.t = 0, 0
        return self.q

    def step(self, action):
        capacity = (0, 2, 4)[int(action)]
        arriving = int(self.rng.poisson(2))
        raw = max(0, self.q - capacity) + arriving
        dropped = max(0, raw - 20)
        self.q = min(20, raw)
        self.t += 1
        reward = -(self.q + .2 * capacity**2 + 10 * dropped)
        return self.q, reward, False, self.t >= 100, {'queue': self.q, 'capacity': capacity, 'dropped': dropped}

def evaluate(policy, seeds=range(100, 105)):
    rows = []
    for seed in seeds:
        env = Queue(); q = env.reset(seed)
        total = waiting = energy = dropped = 0
        while True:
            q, r, terminated, truncated, info = env.step(policy(q))
            total += r; waiting += q; energy += info['capacity']**2; dropped += info['dropped']
            if terminated or truncated: break
        rows.append([total, waiting / env.t, energy / env.t, dropped])
    return np.array(rows)

policies = {'constant': lambda q: 1, 'threshold': lambda q: 0 if q == 0 else (1 if q < 5 else 2)}
for name, policy in policies.items():
    print(name, evaluate(policy).mean(axis=0))

env=Queue();q=env.reset(7)
for k in range(5):
    nq,r,te,tr,info=env.step(1);print(k,q,1,r,nq,te,tr);q=nq
# TODO: Change the time limit to 5 and inspect the boundary.

Record results and self-check

step:

q:

action:

reward:

next q:

terminated:

truncated:

Points to check

• Did you fix the seed in reset?
• Did you avoid always setting future value to 0 solely because truncated=True?

Run Gymnasium and record transitions

Starting conditions

Use Gymnasium CartPole-v1. Compare a random policy and a simple angle/angular velocity rule at seed=100,101,102.

Procedure

1. Print observation_space, action_space, and the episode limit, and check the official documentation for the meanings of the 4 observation components and the two actions.
2. Run both policies with the same seed and record returns, lengths, and termination types.
3. Save transition records as NPZ and load them again. Locate the first transition and episode boundaries, and check the correspondence between observation and action counts.

Results to retain

• An evaluation table with 3 rows per policy and 1 transition example.
• 1 trace file containing environment ID, policy, seed, and library versions.

"""Small public-environment benchmark; evaluation traces are ordinary numeric arrays."""
import json
from importlib.metadata import version
from pathlib import Path
import numpy as np
import gymnasium as gym

def collect(env_id='CartPole-v1', policy=None, seeds=range(100,103)):
    env=gym.make(env_id); transitions=[]; summary=[]
    for seed in seeds:
        obs,info=env.reset(seed=seed);env.action_space.seed(seed);total=0.;length=0
        while True:
            action=env.action_space.sample() if policy is None else policy(obs)
            nxt,reward,terminated,truncated,info=env.step(action)
            transitions.append((obs.copy(),np.asarray(action).copy(),reward,nxt.copy(),terminated,truncated,seed))
            obs=nxt;total+=reward;length+=1
            if terminated or truncated:break
        summary.append([seed,total,length,int(terminated),int(truncated)])
    env.close()
    columns=list(zip(*transitions))
    data={k:np.asarray(v) for k,v in zip(['observations','actions','rewards','next_observations','terminated','truncated','episode_seed'],columns)}
    return np.asarray(summary),data

def cartpole_rule(obs):return int(obs[2]+.25*obs[3]>0)
def zero_torque(obs):return np.zeros(1,dtype=np.float32)

def save_trace(path, env_id, data, policy_name, seeds):
    # Contains numeric arrays and JSON text only; never requires pickle.
    metadata=dict(env_id=env_id,policy=policy_name,seeds=list(seeds),gymnasium=version('gymnasium'))
    np.savez_compressed(path,**data,metadata_json=np.array(json.dumps(metadata)))
    with np.load(path,allow_pickle=False) as saved:
        assert len(saved['observations'])==len(saved['rewards'])
        print(json.loads(str(saved['metadata_json'])))

def public_model(algorithm='DQN',env_id='CartPole-v1',seed=7,steps=2048,**changes):
    from stable_baselines3 import DQN,PPO,SAC
    options=dict(policy_kwargs=dict(net_arch=[32,32]),seed=seed,verbose=0,device='cpu')
    if algorithm=='DQN':options.update(buffer_size=5000,learning_starts=100,target_update_interval=100,batch_size=32)
    if algorithm=='PPO':options.update(n_steps=128,batch_size=64,n_epochs=3)
    if algorithm=='SAC':options.update(buffer_size=5000,learning_starts=100,batch_size=64)
    options.update(changes)
    env=gym.make(env_id)
    model={'DQN':DQN,'PPO':PPO,'SAC':SAC}[algorithm]('MlpPolicy',env,**options)
    model.learn(total_timesteps=steps)
    env.close()
    return model

print('Gymnasium',version('gymnasium'))
env=gym.make('CartPole-v1')
print('Observation:',env.observation_space,'Action:',env.action_space,'Limit:',env.spec.max_episode_steps)
env.close()
# For visual inspection, create a separate env with render_mode='rgb_array'.
# Rendering requires gymnasium[classic-control]; do not render every training step.

for name,policy in [('random',None),('rule',cartpole_rule)]:
    rows,data=collect(policy=policy);print(name,rows)
    save_trace(name+'_cartpole.npz','CartPole-v1',data,name,range(100,103))
# TODO: Locate the first transition and episode boundaries in the table.

Record results and self-check

Policy:

seed:

Return:

Episode length:

terminated:

truncated:

Points to check

• Did you distinguish the initial state seed from the action space seed?
• Did you avoid calling strong Rule results a learned policy?

Apply this to your research problem

Starting conditions

Based on the preceding experiments, choose Queue, Motor, or one of your research problems and design an experiment for real application.

Procedure

1. Define state, observation, action, reward, termination, and time limit in one table. Record each variable's units and the information you can actually obtain.
2. Write 3 expected transitions numerically and check whether rewards for good and bad actions match your intent.
3. Choose one rule-based comparison baseline, one evaluation condition, and one failure case of greatest concern. Also list the required measurements or simulator.

Results to retain

• MDP design table, 3 transition examples, and comparative experiment plan.
• If real research data are unavailable, design using synthetic conditions from an application case.

You may continue with the research problem and design table selected earlier. Revisit existing entries and transition examples using the concepts learned in this activity, and update changed entries or reasons for retaining them.

Record results and self-check

Item:

Concrete definition and units:

Observability and verification method:

Points to check

• Did you exclude future information and evaluation labels from observations?
• Did you define domain-specific metrics rather than judging the research contribution solely by training return?

Official examples and data

• Gymnasium · Basic usage
• CartPole · environment definition