Port Terminal Example

A container port terminal simulation with resource allocation and RL integration.

Overview

This example models a container port terminal inspired by the Winter Simulation Conference (WSC) 2025 challenge. It demonstrates:

  • Multiple resource types (berths, quay cranes, AGVs, yard blocks)

  • Complex entity lifecycles (vessels, containers)

  • Decision points for optimization

  • RL integration patterns

  • Multi-resource coordination

Terminal Operations

Vessel Arrival → Berthing → Discharge Containers → Load Containers → Departure
                    │              │                     │
                    └── Berth ─────┴── QC + AGV + Yard ──┘

Key resources:

  • Berths: Locations where vessels dock

  • Quay Cranes (QC): Load/unload containers from vessels

  • AGVs: Transport containers between quay and yard

  • Yard Blocks: Storage areas for containers

Code Walkthrough

Domain Entities

@dataclass
class Vessel(TimedEntity):
    """A container vessel."""
    vessel_id: str
    containers_to_discharge: int
    containers_to_load: int
    berth: Optional[int] = None

@dataclass
class Container(TimedEntity):
    """A container being handled."""
    container_id: str
    vessel: Vessel
    operation: str  # "discharge" or "load"
    yard_block: Optional[int] = None

Resource Setup

class PortTerminal(Simulation):
    def __init__(
        self,
        num_berths: int = 4,
        num_quay_cranes: int = 8,
        num_agvs: int = 20,
        num_yard_blocks: int = 6,
    ):
        super().__init__()

        # Berths - distinguishable resources
        self.berths = ResourcePool(
            sim=self,
            resources=[f"Berth-{i}" for i in range(num_berths)],
        )

        # Quay cranes - can be assigned to berths
        self.quay_cranes = ResourcePool(
            sim=self,
            resources=[f"QC-{i}" for i in range(num_quay_cranes)],
        )

        # AGVs - shared transport fleet
        self.agvs = ResourcePool(
            sim=self,
            resources=[f"AGV-{i}" for i in range(num_agvs)],
        )

        # Yard blocks - container storage
        self.yard_blocks = [
            YardBlock(f"YB-{i}", capacity=500)
            for i in range(num_yard_blocks)
        ]

Vessel Lifecycle

def on_init(self):
    """Schedule vessel arrivals."""
    self.schedule(self.vessel_arrival, delay=0)

def vessel_arrival(self):
    """Handle vessel arrival."""
    vessel = self.create_vessel()
    vessel.record_entry(self.now)

    # Request berth allocation (decision point)
    berth = self.decision_maker.allocate_berth(vessel, self.berths)

    if berth:
        self.berth_vessel(vessel, berth)
    else:
        self.vessel_queue.enqueue(vessel)

    # Schedule next arrival
    self.schedule(
        self.vessel_arrival,
        delay=self.rng.exponential(self.inter_arrival_time)
    )

def berth_vessel(self, vessel: Vessel, berth: str):
    """Berth a vessel and start operations."""
    vessel.berth = berth
    self.berths.acquire(vessel, resource_id=berth)

    # Allocate quay cranes (decision point)
    num_cranes = self.decision_maker.allocate_cranes(vessel)
    self.assigned_cranes[vessel] = []

    for _ in range(num_cranes):
        crane = self.quay_cranes.acquire(vessel)
        self.assigned_cranes[vessel].append(crane)

    # Start discharge operations
    self.start_discharge(vessel)

Container Handling

def start_discharge(self, vessel: Vessel):
    """Start discharging containers from vessel."""
    for i in range(vessel.containers_to_discharge):
        container = Container(
            container_id=f"{vessel.vessel_id}-D{i}",
            vessel=vessel,
            operation="discharge",
        )
        self.schedule(
            self.discharge_container,
            delay=i * 0.1,  # Staggered start
            args=(container,)
        )

def discharge_container(self, container: Container):
    """Discharge a single container."""
    # Use quay crane
    crane = self.get_available_crane(container.vessel)
    crane_time = self.rng.uniform(2, 4)  # minutes

    self.schedule(
        self.container_lifted,
        delay=crane_time,
        args=(container,)
    )

def container_lifted(self, container: Container):
    """Container lifted from vessel, needs AGV transport."""
    # Request AGV
    agv = self.agvs.acquire(container)

    # Select yard block (decision point)
    yard_block = self.decision_maker.select_yard_block(container)
    container.yard_block = yard_block

    # Transport to yard
    transport_time = self.get_transport_time(container.vessel.berth, yard_block)
    self.schedule(
        self.container_stored,
        delay=transport_time,
        args=(container, agv)
    )

def container_stored(self, container: Container, agv):
    """Container stored in yard."""
    self.agvs.release(agv)
    self.yard_blocks[container.yard_block].store(container)
    container.record_exit(self.now)

    self.containers_discharged += 1
    self.check_vessel_complete(container.vessel)

Decision Maker Interface

class PortDecisionMaker:
    """Decision maker for port operations (pluggable for RL)."""

    def allocate_berth(self, vessel: Vessel, berths: ResourcePool) -> Optional[str]:
        """Decide which berth to assign to vessel."""
        available = berths.get_available()
        if not available:
            return None
        # Default: first available
        return available[0]

    def allocate_cranes(self, vessel: Vessel) -> int:
        """Decide how many cranes to assign."""
        # Default: 2 cranes per vessel
        return min(2, self.available_cranes())

    def select_yard_block(self, container: Container) -> int:
        """Decide which yard block for container."""
        # Default: least utilized block
        return min(range(len(self.yard_blocks)),
                   key=lambda i: self.yard_blocks[i].utilization)

RL Integration

Replace the decision maker with an RL agent:

class RLPortDecisionMaker(PortDecisionMaker, RLInterface):
    """RL-based decision maker for port operations."""

    def get_state(self) -> np.ndarray:
        """Return current port state."""
        return np.array([
            self.berths.utilization,
            self.quay_cranes.utilization,
            self.agvs.utilization,
            len(self.vessel_queue),
            *[yb.utilization for yb in self.yard_blocks],
        ])

    def get_action_space(self) -> ActionSpace:
        """Define action space."""
        return ActionSpace(
            space_type="multi_discrete",
            nvec=[
                self.num_berths + 1,  # berth selection (0 = wait)
                4,                     # num cranes (1-4)
                self.num_yard_blocks,  # yard block
            ]
        )

    def apply_action(self, action: np.ndarray):
        """Apply RL agent's decision."""
        berth_action, crane_action, yard_action = action
        self.pending_decisions = {
            "berth": berth_action,
            "cranes": crane_action + 1,
            "yard": yard_action,
        }

Running the Example

from simcraft.examples.port_terminal import PortTerminal

# Create simulation
sim = PortTerminal(
    num_berths=4,
    num_quay_cranes=8,
    num_agvs=20,
    num_yard_blocks=6,
)

# Run for one week
sim.run(until=7 * 24 * 60)  # minutes

# Print results
sim.report()

Sample Output

Port Terminal Simulation Results
============================================================

Vessel Statistics:
  Vessels arrived: 42
  Vessels served: 40
  Vessels in queue: 2
  Average wait time: 45.3 minutes
  Average turnaround: 8.2 hours

Container Statistics:
  Containers discharged: 12,450
  Containers loaded: 11,890
  Average handling time: 4.2 minutes

Resource Utilization:
  Berths: 82.5%
  Quay Cranes: 71.3%
  AGVs: 68.9%
  Yard Blocks: 45.2% - 67.8%

Key Patterns

Multiple Resource Types

The simulation coordinates berths, cranes, AGVs, and yard blocks - each with different characteristics.

Decision Points

Key decisions (berth allocation, crane assignment, yard block selection) are isolated for easy RL integration.

Complex Entity Lifecycle

Vessels go through multiple stages: arrival → queue → berthing → discharge → load → departure.

Resource Pools

Distinguishable resources (berths, cranes) are managed via ResourcePool with custom selection policies.

Source Code

The complete source code is in simcraft/examples/port_terminal.py.