SimPy机场安检仿真报错：'Airport'对象无'check'属性求助

问题背景 #

实现简化机场安检系统的SimPy仿真，需求如下：

• 乘客以λ₁=5人/分钟的Poisson分布（平均到达间隔μ₁=0.2分钟）到达ID/登机牌检查队列
• ID检查队列有多台服务器，每台服务时间为平均μ₂=0.75分钟的指数分布
• 乘客完成ID检查后，被分配到最短的人身检查队列，扫描时间服从0.5-1分钟的均匀分布

用户运行代码时出现AttributeError: 'Airport' object has no attribute 'check'错误，以下是用户提供的原始代码：

import random
import simpy

## define system constraints
# checkers = number of boarding pass checkers
# scanners = number of scanners
num_checkers = 10
num_scanners = 25
r_seed = 500

# define rates
arrival_rate = 5 # rate is passengers per minute
check_rate = .75 # rate is minutes per passenger
min_scan = .5 # scanner min time - uniform distribution
max_scan = 1.0 # scanner max time - uniform distribution

# define simulation timing
run_time = 100 # minutes per simulation
reps = 100 # number of simulation replications


# initalize global variables - will store answers in these variables later
checkWait = 0
scanWait = 0
sysTime = 0
timeWait = 0
timeChecker = 0
timeCheckerComplete = 0
timeScan = 0
timeScanComplete = 0
passenger_count = 0




## SimPy simulation framework


# create the simulation object - airport includes checkers and scanners
class Airport(object):


    # define the overall resources in our system - checkers and scanners
    def __init__(self, env):

        # set up the enviroment
        self.env = env

        # set up checkers as resource = # checkers
        self.checker = simpy.Resource(env, num_checkers)


        # set up scanners as a resource - each scanner has it's own queue!
        self.scanners = []

        # for each scanner - assign a resource - multiple scanners with 1 resource
        for i in range(0,25):
            resource = simpy.Resource(env, capacity = 1)
            self.scanners.append(resource)



    # define how long a passenger gets to the checkout
    def check(self, passenger):
        yield self.env.timeout(random.expovariate(1/check_rate))


    # define how long a passenger takes to scan themselves in
    def scan(self, passenger):
        yield self.env.timeout(random.uniform(max_scan , min_scan))





# define attributes of a passenger - how passenger moves through the system (checkers, scanners)
def passenger(env, name, s): # enviroment, name, system reference

    # initalize global variables - will store answers in these variables later
    global checkWait
    global scanWait
    global sysTime
    global timeWait
    global timeChecker
    global timeCheckerComplete
    global timeScan
    global timeScanComplete
    global passenger_count



    # time passenger arrives
    timeArrive = env.now
    print('%s arrives at time %.2f' % (name, timeArrive))



    with s.checker.request() as request:
        yield request
        print('check queue length = %d' % len(s.checker.queue))

        # time passenger arrives at checker
        timeChecker = env.now
        print('%s gets to checker at time %.2f' % (name, timeChecker))

        yield env.process(s.check(name))

        # time passenger completes checker
        timeCheckerComplete = env.now
        print('%s complete checker at time %.2f' % (name, timeCheckerComplete))



    min_q = 0

    with s.scanners[min_q].request() as request:
        yield request
        print('scanner queue length = %d' % len(s.scanners[min_q].queue))

        for i in range(1, num_scanners):
            if (len(s.scanners[i].queue) < len(s.scanners[min_q].queue)):
                min_q = i

        # time passenger arrives at scanner
        timeScan = env.now
        print('%s gets to scanner at time %.2f' % (name, timeScan))

        yield env.process(s.scan(name))

        timeScanComplete = env.now
        print('%s complete scanner at time %.2f' % (name, timeScanComplete))




    # time at the end of entire system process
    timeExit = env.now
    print('%s gets to complete system time %.2f' % (name, timeExit))




    sysTime = sysTime + (timeExit - timeArrive)
    checkWait = checkWait + (timeChecker - timeArrive)
    scanWait = scanWait + (timeScanComplete - timeCheckerComplete)
    timeWait = (checkWait + scanWait)




def setup(env):

    airport = Airport(env)

    i = 0

    while True:
        yield env.timeout(random.expovariate(1.0 / check_rate))
        i += 1
        env.process(passenger(env, 'Passenger %d' % i, airport))


# define variables to store simulation results into
avg_wait_time = []
avg_check_time = []
avg_scan_time = []
avg_sys_time = []


for i in range(0, reps):

    env = simpy.Environment()
    env.process(setup(env))
    env.run(until = run_time)

    avg_wait_time.append(timeWait / 125)
    avg_check_time.append(checkWait / 125)
    avg_scan_time.append(scanWait / 125)
    avg_sys_time.append(sysTime / 125)

    passenger_count = 0
    sysTime = 0
    checkWait = 0
    scanWait = 0
    timeWait = 0


sim_wait_avg = sum(avg_wait_time) / reps
sim_check_avg = sum(avg_check_time) / reps
sim_scan_avg = sum(avg_scan_time) / reps
sim_sys_time = sum(avg_sys_time) / reps


print('Average cummulative wait time: ' + str(sim_wait_avg))
print('Average cummulative check time: ' + str(sim_check_avg))
print('Average cummulative scan time: ' + str(sim_scan_avg))
print('Average cummulative system time: ' + str(sim_sys_time))

错误信息 #

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/var/folders/4r/z339ff2521g9hm6t9_krqctw0000gn/T/ipykernel_16484/1087764229.py in passenger(env, name, s)
     27 
---> 28         yield env.process(s.check(name))
     29 

AttributeError: 'Airport' object has no attribute 'check'

The above exception was the direct cause of the following exception:

AttributeError                            Traceback (most recent call last)
/var/folders/4r/z339ff2521g9hm6t9_krqctw0000gn/T/ipykernel_16484/2180799163.py in <module>
     22     env = simpy.Environment()
     23     env.process(setup(env))
---> 24     env.run(until = run_time)
     25 
     26     avg_wait_time.append(timeWait / 125)

~/opt/anaconda3/lib/python3.9/site-packages/simpy/core.py in run(self, until)
    252         try:
    253             while True:
---> 254                 self.step()
    255         except StopSimulation as exc:
    256             return exc.args[0]  # == until.value

~/opt/anaconda3/lib/python3.9/site-packages/simpy/core.py in step(self)
    204             exc = type(event._value)(*event._value.args)
    205             exc.__cause__ = event._value
---> 206             raise exc
    207 
    208     def run(

AttributeError: 'Airport' object has no attribute 'check'

问题分析与修复 #

核心问题 #

错误根源是代码缩进错误：Airport类中的check和scan方法没有正确缩进，导致这两个方法不属于Airport类，而是全局函数，因此调用s.check(name)时无法找到该属性。

此外代码还存在其他逻辑错误：

1. 乘客到达间隔计算错误：误用check_rate计算到达间隔，应使用arrival_rate对应需求中的0.2分钟平均间隔
2. 最短扫描队列选择逻辑错误：先请求第一个扫描器资源再找最短队列，导致乘客错误占用非目标扫描器
3. 均匀分布参数顺序错误：random.uniform(max_scan, min_scan)会生成超出预期范围的数值
4. 统计平均值时使用固定分母125，未用实际乘客数量计算

修正后的完整代码 #

import random
import simpy

## define system constraints
num_checkers = 10
num_scanners = 25
r_seed = 500
random.seed(r_seed)  # 固定随机种子保证仿真可复现

# define rates
arrival_rate = 5  # 乘客到达率：人/分钟，平均间隔1/5=0.2分钟
check_rate = 0.75  # 每个ID检查的平均服务时间：分钟/人
min_scan = 0.5  # 扫描最短时间
max_scan = 1.0  # 扫描最长时间

# define simulation timing
run_time = 100  # 单次仿真时长（分钟）
reps = 100  # 仿真重复次数


## SimPy simulation framework
class Airport(object):
    def __init__(self, env):
        self.env = env
        # ID检查资源：多台服务器共享队列
        self.checkers = simpy.Resource(env, num_checkers)
        # 扫描资源：每个扫描器独立队列
        self.scanners = [simpy.Resource(env, capacity=1) for _ in range(num_scanners)]

    # ID检查服务过程
    def check(self, passenger):
        yield self.env.timeout(random.expovariate(1/check_rate))

    # 扫描服务过程
    def scan(self, passenger):
        yield self.env.timeout(random.uniform(min_scan, max_scan))


def passenger(env, name, airport):
    # 记录到达时间
    time_arrive = env.now

    # ID检查环节
    with airport.checkers.request() as req:
        yield req
        time_start_check = env.now
        # 等待时间 = 开始服务时间 - 到达时间
        global check_wait_total
        check_wait_total += (time_start_check - time_arrive)

        yield env.process(airport.check(name))
        time_end_check = env.now
        # 检查环节总耗时（等待+服务）
        global check_total_time
        check_total_time += (time_end_check - time_arrive)

    # 选择最短队列的扫描器
    min_queue_len = len(airport.scanners[0].queue)
    selected_scanner_idx = 0
    for idx, scanner in enumerate(airport.scanners[1:], 1):
        current_len = len(scanner.queue)
        if current_len < min_queue_len:
            min_queue_len = current_len
            selected_scanner_idx = idx

    # 扫描环节
    with airport.scanners[selected_scanner_idx].request() as req:
        yield req
        time_start_scan = env.now
        # 扫描等待时间 = 开始扫描时间 - 检查完成时间
        global scan_wait_total
        scan_wait_total += (time_start_scan - time_end_check)

        yield env.process(airport.scan(name))
        time_end_scan = env.now
        # 扫描环节总耗时（等待+服务）
        global scan_total_time
        scan_total_time += (time_end_scan - time_end_check)

    # 系统总耗时
    time_exit = env.now
    global sys_total_time, passenger_count
    sys_total_time += (time_exit - time_arrive)
    passenger_count += 1


def setup(env):
    airport = Airport(env)
    i = 0
    while True:
        # 乘客到达间隔服从指数分布，参数为arrival_rate（人/分钟）
        yield env.timeout(random.expovariate(arrival_rate))
        i += 1
        env.process(passenger(env, f'Passenger {i}', airport))


# 存储仿真结果
avg_wait_times = []
avg_check_times = []
avg_scan_times = []
avg_sys_times = []

for _ in range(reps):
    # 重置全局统计变量
    check_wait_total = 0
    scan_wait_total = 0
    check_total_time = 0
    scan_total_time = 0
    sys_total_time = 0
    passenger_count = 0

    # 运行仿真
    env = simpy.Environment()
    env.process(setup(env))
    env.run(until=run_time)

    # 计算单次仿真的平均值
    if passenger_count > 0:
        avg_wait = (check_wait_total + scan_wait_total) / passenger_count
        avg_check = check_total_time / passenger_count
        avg_scan = scan_total_time / passenger_count
        avg_sys = sys_total_time / passenger_count
    else:
        avg_wait = avg_check = avg_scan = avg_sys = 0

    avg_wait_times.append(avg_wait)
    avg_check_times.append(avg_check)
    avg_scan_times.append(avg_scan)
    avg_sys_times.append(avg_sys)


# 计算多次仿真的平均结果
sim_avg_wait = sum(avg_wait_times) / reps
sim_avg_check = sum(avg_check_times) / reps
sim_avg_scan = sum(avg_scan_times) / reps
sim_avg_sys = sum(avg_sys_times) / reps

print(f'Average total wait time: {sim_avg_wait:.4f} minutes')
print(f'Average check process time (wait + service): {sim_avg_check:.4f} minutes')
print(f'Average scan process time (wait + service): {sim_avg_scan:.4f} minutes')
print(f'Average system time (total time in system): {sim_avg_sys:.4f} minutes')

主要修改点说明 #

1. 修复类方法缩进：将check和scan方法正确缩进至Airport类内部，确保类实例可调用
2. 修正乘客到达逻辑：替换到达间隔的计算参数，匹配需求中的Poisson到达率
3. 修复最短队列选择：先遍历所有扫描器找到最短队列，再请求对应资源，避免错误占用
4. 修正均匀分布参数：调整参数顺序，保证生成0.5-1分钟的有效扫描时间
5. 优化统计逻辑：用实际乘客数量计算平均值，添加随机种子保证可复现，拆分统计指标提升可读性

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