joint-sim 0.1.1

MAPF+JSSP Joint Simulation Environment for Reinforcement Learning

Joint-Sim: MAPF+JSSP Reinforcement Learning Environment

A Gymnasium-compatible simulation environment for the joint optimization of Multi-Agent Path Finding (MAPF) and Job Shop Scheduling Problem (JSSP).

Features

• Gymnasium-compatible interface: Standard reset(), step(), observation_space, action_space
• Joint optimization: Simultaneously handles AGV routing (MAPF) and job scheduling (JSSP)
• Cost-based assignment: Hungarian algorithm for optimal task-AGV matching
• Collision-free routing: Reservation table for conflict prevention
• Flexible configuration: Customizable machines, AGVs, jobs, and grid layouts
• SVG visualization: Real-time factory state rendering

Installation

pip install joint-sim

For additional features:

pip install "joint-sim[visualization]"  # Matplotlib for charts
pip install "joint-sim[optimal-solver]"  # OR-Tools for optimal JSSP
pip install "joint-sim[all]"             # All optional dependencies

Quick Start

Basic Usage

from joint_sim import JointSimGymEnv, FactoryConfig

# Create environment
config = FactoryConfig(
    n_machines=4,
    n_agvs=2,
    n_jobs=5,
    grid_size=(10, 10),
)

env = JointSimGymEnv(config, assigner_type='cost')

# Standard Gymnasium interface
obs, info = env.reset(seed=42)

for _ in range(1000):
    # Random action (use your RL policy here)
    action = env.action_space.sample()

    obs, reward, terminated, truncated, info = env.step(action)

    if terminated or truncated:
        break

print(f"Completed jobs: {info['completed_jobs']}/{info['total_jobs']}")

With Custom Policy

import numpy as np
from joint_sim import JointSimGymEnv, FactoryConfig

env = JointSimGymEnv(
    config=FactoryConfig(n_machines=4, n_agvs=2, n_jobs=5),
    assigner_type='cost'
)

obs, info = env.reset(seed=42)

def heuristic_policy(env, obs):
    """Assign nearest task to each AGV"""
    requests = env.get_transport_requests()
    n_agvs = env.config.n_agvs

    if not requests:
        return {'agv_assignments': np.full(n_agvs, -1, dtype=np.int32)}

    assignments = np.full(n_agvs, -1, dtype=np.int32)

    for i in range(min(n_agvs, len(requests))):
        assignments[i] = requests[i].job_id

    return {'agv_assignments': assignments}

while True:
    action = heuristic_policy(env, obs)
    obs, reward, terminated, truncated, info = env.step(action)

    if terminated or truncated:
        break

print(f"Total reward: {reward}, Completed: {info['completed_jobs']}")

Environment Specification

Observation Space

observation_space = Dict({
    'grid': Box(0, 3, shape=(height, width), dtype=int32),      # Factory grid
    'agv_positions': Box(0, max_size, shape=(n_agvs, 2)),       # AGV (x, y) positions
    'agv_status': Box(0, 3, shape=(n_agvs,), dtype=int32),      # 0=idle, 1=moving, 2=loading, 3=unloading
    'agv_carrying': Box(-1, n_jobs, shape=(n_agvs,), dtype=int32),  # Job ID being carried (-1=none)
    'machine_status': Box(0, 1, shape=(n_machines,), dtype=int32),  # 0=idle, 1=working
    'machine_queue_length': Box(0, n_jobs, shape=(n_machines,)),   # Queue length per machine
    'time': Box(0, max_time, shape=(), dtype=int32),            # Current timestep
})

Action Space

action_space = Dict({
    'agv_assignments': Box(-1, n_jobs-1, shape=(n_agvs,), dtype=int32),
})
# -1: No assignment (AGV stays idle)
# 0 to n_jobs-1: Assign job to AGV

Reward Function

• Job completion bonus: +100 per completed job
• Terminal bonus: +100 when all jobs complete

API Reference

JointSimGymEnv

env = JointSimGymEnv(
    config: FactoryConfig = None,       # Factory configuration
    assigner_type: str = 'cost',        # 'cost' or 'greedy'
    assigner_config: dict = None,       # Assigner parameters
    max_episode_steps: int = 10000,     # Maximum steps per episode
    reward_scale: float = 1.0           # Reward scaling factor
)

FactoryConfig

config = FactoryConfig(
    n_machines: int = 6,                # Number of machines
    n_agvs: int = 3,                    # Number of AGVs
    n_jobs: int = 10,                   # Number of jobs
    grid_size: Tuple[int, int] = (20, 20),  # Grid dimensions
    n_ops_per_job: Tuple[int, int] = (3, 6),  # Operations per job range
    op_duration_range: Tuple[int, int] = (3, 10),  # Operation duration range
    seed: int = 42,                     # Random seed
    max_time: int = 5000,               # Maximum simulation time
)

Key Methods

Method	Description
reset(seed=None)	Reset environment, returns (obs, info)
step(action)	Execute action, returns (obs, reward, terminated, truncated, info)
get_transport_requests()	Get list of pending transport requests
get_state()	Get current FactoryState
get_metrics()	Get simulation metrics
render(filepath=None)	Render SVG visualization

License

MIT License - see LICENSE file for details.

Citation

If you use this environment in your research, please cite:

@software{joint_sim,
  title = {Joint-Sim: MAPF+JSSP Reinforcement Learning Environment},
  author = {Skyrim Forestsea},
  year = {2026},
  url = {https://github.com/skyrimforest/joint-sim}
}

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.