基于NetworkX构建单线铁路列车调度优化模型的技术咨询

Great question—single-track rail scheduling is a classic constrained optimization problem, and pairing NetworkX for topology modeling with pySim for discrete-event simulation is a solid approach. Let’s break this down into actionable steps tailored to your setup (stations: Desmet → Missoula → Bonner → McQuarrie → Clinton; 5 eastbound, 3 westbound trains).

1. Model the Track Topology with NetworkX #

First, we need to represent the single-track line as a graph where physical track segments are the critical constrained resources (since a single segment can only hold one train at a time, regardless of direction).

Key Choices: #

• Use a directed graph to model east/west travel, but link opposite-direction segments to the same shared resource (so we can enforce single-occupancy constraints).
• Add travel time attributes to each segment (replace the example values with your real-world data).

import networkx as nx

# Define stations (west to east)
stations = ["Desmet", "Missoula", "Bonner", "McQuarrie", "Clinton"]

# Track segments with travel times (minutes) – include both directions
segment_details = [
    ("Desmet", "Missoula", 20),
    ("Missoula", "Bonner", 15),
    ("Bonner", "McQuarrie", 18),
    ("McQuarrie", "Clinton", 22),
    ("Missoula", "Desmet", 20),
    ("Bonner", "Missoula", 15),
    ("McQuarrie", "Bonner", 18),
    ("Clinton", "McQuarrie", 22),
]

# Build directed graph
track_graph = nx.DiGraph()
track_graph.add_nodes_from(stations)

for u, v, travel_time in segment_details:
    # Assign a shared resource ID to opposite segments (e.g., "Desmet-Missoula" for both directions)
    resource_id = tuple(sorted((u, v)))
    track_graph.add_edge(
        u, v,
        travel_time=travel_time,
        resource_id=resource_id,
        capacity=1  # Enforce single occupancy per resource
    )

2. Define Train Attributes #

Next, formalize your trains with operational constraints. Include details like earliest departure time, priority (if any), and origin/destination (fixed for east/west in your case).

# Eastbound trains (Desmet → Clinton)
eastbound = [
    {"id": "E1", "direction": "east", "origin": "Desmet", "dest": "Clinton", "earliest_depart": 0, "priority": 1},
    {"id": "E2", "direction": "east", "origin": "Desmet", "dest": "Clinton", "earliest_depart": 10, "priority": 2},
    {"id": "E3", "direction": "east", "origin": "Desmet", "dest": "Clinton", "earliest_depart": 25, "priority": 1},
    {"id": "E4", "direction": "east", "origin": "Desmet", "dest": "Clinton", "earliest_depart": 40, "priority": 3},
    {"id": "E5", "direction": "east", "origin": "Desmet", "dest": "Clinton", "earliest_depart": 55, "priority": 2},
]

# Westbound trains (Clinton → Desmet)
westbound = [
    {"id": "W1", "direction": "west", "origin": "Clinton", "dest": "Desmet", "earliest_depart": 30, "priority": 1},
    {"id": "W2", "direction": "west", "origin": "Clinton", "dest": "Desmet", "earliest_depart": 60, "priority": 2},
    {"id": "W3", "direction": "west", "origin": "Clinton", "dest": "Desmet", "earliest_depart": 90, "priority": 1},
]

all_trains = eastbound + westbound

3. Build the Scheduling Model with pySim #

pySim excels at discrete-event simulation, which is perfect for modeling train movements and enforcing constraints. We’ll create event handlers for train departures, segment entries/exits, and station stops, with core logic to avoid segment conflicts.

Core Constraints to Enforce: #

1. No two trains can occupy the same track segment at the same time.
2. Trains must follow the fixed station sequence (no skipping stops).
3. Trains can wait at stations to let opposing trains pass (the key to single-track scheduling).

from pysim import Simulation, Event

class TrainDeparture(Event):
    def __init__(self, train, station, time):
        self.train = train
        self.station = station
        super().__init__(time)

    def process(self, sim):
        # Determine next station based on direction
        station_idx = stations.index(self.station)
        if self.train["direction"] == "east":
            next_station = stations[station_idx + 1] if station_idx < len(stations)-1 else None
        else:
            next_station = stations[station_idx - 1] if station_idx > 0 else None

        if not next_station:
            # Train reached destination – record arrival
            sim.results[f"{self.train['id']}_arrival"] = self.time
            return

        # Get segment details
        edge = track_graph[self.station][next_station]
        travel_time = edge["travel_time"]
        resource_id = edge["resource_id"]

        # Check if segment is available; wait if not
        if resource_id not in sim.segment_usage:
            # Segment is free – schedule entry
            entry_time = self.time
            sim.segment_usage[resource_id] = entry_time + travel_time
        else:
            # Wait for segment to become free
            entry_time = max(self.time, sim.segment_usage[resource_id])
            sim.segment_usage[resource_id] = entry_time + travel_time

        # Schedule arrival at next station
        sim.schedule(TrainArrival(self.train, next_station, entry_time + travel_time))

class TrainArrival(Event):
    def __init__(self, train, station, time):
        self.train = train
        self.station = station
        super().__init__(time)

    def process(self, sim):
        # If not at destination, schedule next departure (add optional station dwell time here)
        if self.station != self.train["dest"]:
            # Add 2-minute dwell time for passenger boarding/alighting (adjust as needed)
            departure_time = self.time + 2
            sim.schedule(TrainDeparture(self.train, self.station, departure_time))
        else:
            sim.results[f"{self.train['id']}_arrival"] = self.time

# Initialize simulation
sim = Simulation()
sim.segment_usage = {}  # Tracks when each segment becomes free
sim.results = {}  # Stores arrival times for each train

# Schedule initial departures
for train in all_trains:
    sim.schedule(TrainDeparture(train, train["origin"], train["earliest_depart"]))

# Run simulation for 24 hours (1440 minutes)
sim.run(until=1440)

# Print results
print("Train Arrival Times (minutes since start):")
for train_id, arr_time in sim.results.items():
    print(f"{train_id}: {arr_time}")

4. Optimize for Maximum Throughput #

Your goal is to maximize the number of trains passing through in 24 hours. Here’s how to refine the model:

• Adjust Departure Windows: Use a heuristic (like genetic algorithms or greedy scheduling) to shift train departure times to minimize conflicts. For example, cluster eastbound trains in batches separated by enough time to let westbound trains pass.
• Prioritize High-Impact Trains: If some trains have higher priority (e.g., freight vs. passenger), modify the event logic to let priority trains jump the queue at stations.
• Analyze Bottlenecks: Use NetworkX to identify segments with the highest utilization (e.g., nx.degree_centrality(track_graph) can highlight busy segments) and adjust scheduling to reduce congestion there.
• Iterate and Validate: Run the simulation multiple times with different departure time combinations, then select the schedule that fits the most trains within 24 hours without conflicts.

5. Key Tips for Success #

• Use Realistic Travel/Dwell Times: Inaccurate times will lead to invalid schedules. Make sure to include acceleration/deceleration time for segments if needed.
• Test Edge Cases: What if a train is delayed? Add random delay events to the simulation to test robustness.
• Combine with Optimization Libraries: For more advanced scheduling, pair pySim with libraries like scipy.optimize or DEAP (genetic algorithms) to automatically search for optimal departure times.

内容的提问来源于stack exchange，提问作者Jasher