如何在SimPy中使用AnyOf？解决Bed对象无env属性报错

错误原因 #

你错误地将Bed对象传入了AnyOf，但AnyOf要求传入的是SimPy的事件对象（比如Store.get()返回的事件）。原代码中edbed = yield self.clean_beds[pt.care_area].get()已经完成了获取床位的操作，拿到的是Bed实例而非事件，因此AnyOf尝试访问Bed对象的env属性时触发错误。

你的需求是优先获取目标诊室（ED1/ED2）床位，若目标床位全占满则使用Extra床位。正确逻辑应是同时发起两个床位的获取请求，等待任意一个完成后，取消未完成的请求，并将未使用的床位放回对应Store。

修复后的代码 #

重点修改ed_process中的床位获取逻辑：

import simpy
import random
from dataclasses import dataclass
from simpy.events import AnyOf

@dataclass
class BedAtt:
    name: str
    type: str
    capacity: int

class Pre_Define:
    allbeds = [
        BedAtt('ED1', 'ED', 5),
        BedAtt('ED2', 'ED', 14),
        BedAtt('Extra', 'ED', 10),
        BedAtt('IU', 'IU', 15), 
        ]  

    pt_inter_arrival = 6
    m_triage = 17
    m_ED_physician = 23
    m_IUbed_occupied = 30
    m_bed_cleaning = 50
    m_additionaltest = 150
    n_nurses = 5
    n_ed_physician = 5
    n_bed_cleanler = 5
    n_transporter = 5
    
    prob_ED_Bed1 = 30
    prob_ED_Bed2 = 70
    
    prob_Additionaltest = 60
    prob_Dis = 75

    warmup_period = 0
    sim_duration = 720

class Bed():
    def __init__(self, id, bed_type: str, care_area: str, name: str):
        self.id = id
        self.type = bed_type
        self.care_area = care_area
        self.name = name

class Patient:
    def __init__(self, p_id: int) -> None:
        self.id = p_id
        self.type = None
        self.care_area = None
        self.name = None

        self.admission_decision = ""
        self.addtional_test_boolean = None
        # For copying from assigned Bed information
        self.edbed_info = None
        self.iubed_info = None
        self.bed_id = None
        self.bed_care_area = None
        self.bed_name = None
        self.bed_status = None
        
    def addtional_test(self):
        additionaltest_prob = random.uniform(0, 1)*100
        if additionaltest_prob <= Pre_Define.prob_Additionaltest:
            self.addtional_test_boolean = True
            
    def admin_decision(self):
        admin_decision_prob = random.uniform(0, 1)
        if admin_decision_prob <= Pre_Define.prob_Dis/100:
            self.admission_decision = "IU"
        else:
            self.admission_decision = "Admit"

    def ed_care_area(self):
        edcarearea_decision_prob = random.uniform(0, 1)
        if edcarearea_decision_prob <= Pre_Define.prob_ED_Bed1/100:
            self.care_area = "ED1"
        else:
            self.care_area = "ED2"

class EDModel:
    def __init__(self, run_number):
        self.run_number = run_number
        self.env = simpy.Environment()
        self.pt_counter = 0
        self.tg = simpy.Resource(self.env, capacity = Pre_Define.n_nurses)
        self.physician = simpy.Resource(self.env, capacity = Pre_Define.n_ed_physician)
        self.bed_cleaner = simpy.Resource(self.env, capacity = Pre_Define.n_bed_cleanler)
        self.bed_transporter = simpy.Resource(self.env, capacity = Pre_Define.n_transporter)
        self.clean_beds = {
            care_area.name: simpy.Store(self.env, capacity=care_area.capacity)
            for care_area in Pre_Define.allbeds
        }

        self.clean_request_Q = simpy.Store(
            self.env,
            capacity = sum(care_area.capacity for care_area in Pre_Define.allbeds)
        )

        for care_area in Pre_Define.allbeds:
            self.clean_beds[care_area.name].items = [
                Bed(i, care_area.type, care_area=care_area.name, name=f'{care_area.name}_bed_{i}')
                for i in range(1, care_area.capacity + 1)  
            ]

    def generate_pt_arrivals(self):
        while True:
            pt = Patient(self.pt_counter)
            yield self.env.timeout(Pre_Define.pt_inter_arrival)
            self.env.process(self.ed_process(pt))
            self.pt_counter += 1

    def bed_cleaners(self, cleaner_id):
        while True:
            bed = yield self.clean_request_Q.get()
            with self.bed_cleaner.request() as bedreq:
                yield bedreq
                yield self.env.timeout(Pre_Define.m_bed_cleaning)
                print(f'{self.env.now:.2f} bed_cleaners complete {bed.name} cleaning')
                yield self.clean_beds[bed.care_area].put(bed)
                print(f'{self.env.now:.2f} bed_cleaners put {bed.name} to {bed.care_area}')

    def ed_process(self, pt: Patient):
        pt.ed_care_area()
        print(f'{self.env.now:.2f} patient {pt.id} {pt.care_area} arrival')
        with self.tg.request() as req:
            yield req
            yield self.env.timeout(Pre_Define.m_triage)

        # 创建两个床位获取事件（不立即yield，同时发起请求）
        ed_bed_event = self.clean_beds[pt.care_area].get()
        extra_bed_event = self.clean_beds['Extra'].get()

        # 等待任意一个事件完成
        selected_events = yield AnyOf(self.env, [ed_bed_event, extra_bed_event])
        
        # 处理选中的床位和未使用的请求/床位
        edbed = None
        if ed_bed_event in selected_events:
            edbed = selected_events[ed_bed_event]
            print(f'{self.env.now:.2f} patient {pt.id} got target bed: {edbed.name}')
            # 处理Extra床位请求：未触发则取消，已触发则放回床位
            if not extra_bed_event.triggered:
                extra_bed_event.cancel()
            else:
                extrabed = selected_events[extra_bed_event]
                yield self.clean_beds['Extra'].put(extrabed)
                print(f'{self.env.now:.2f} patient {pt.id} return unused extra bed: {extrabed.name}')
        else:
            edbed = selected_events[extra_bed_event]
            print(f'{self.env.now:.2f} patient {pt.id} used extra bed: {edbed.name}')
            # 取消未完成的目标床位请求
            if not ed_bed_event.triggered:
                ed_bed_event.cancel()

        print(f'{self.env.now:.2f} ED patient {pt.id} {pt.care_area} final assigned bed: {edbed.name}')

        with self.physician.request() as req:
            yield req
            yield self.env.timeout(Pre_Define.m_ED_physician)
        pt.addtional_test()
        if pt.addtional_test_boolean:
            yield self.env.timeout(Pre_Define.m_additionaltest)
        pt.admin_decision()
        if pt.admission_decision == 'IU':
            iubed = yield self.clean_beds[pt.admission_decision].get()
            pt.iubed_info = iubed
            print(f'{self.env.now:.2f} IU patient {pt.id} get a bed: {iubed.name}')

            with self.bed_transporter.request() as transreq:
                yield transreq

            print(f'{self.env.now:.2f} IU patient {pt.id} return ED bed: {edbed.name}')

            yield self.clean_request_Q.put(edbed)

            yield self.env.timeout(Pre_Define.m_IUbed_occupied)
            print(f'{self.env.now:.2f} patient {pt.id} IU and return IU bed: {iubed.name}')
            yield self.clean_request_Q.put(iubed)
        else:
            print(f'{self.env.now:.2f} patient {pt.id} discharge and return ED bed: {edbed.name}')
            yield self.clean_request_Q.put(edbed)

    def run(self):
        self.env.process(self.generate_pt_arrivals())
        for i in range(Pre_Define.n_bed_cleanler):
            self.env.process(self.bed_cleaners(i+1))
        self.env.run(until = Pre_Define.warmup_period + Pre_Define.sim_duration)


EDModel(1).run()

关键修改说明 #

1. 事件创建而非直接获取：先创建ed_bed_event和extra_bed_event两个事件对象，不立即yield，确保两个床位请求同时发起。
2. AnyOf等待事件完成：通过AnyOf等待任意一个请求完成，返回的selected_events包含完成的事件与对应的Bed对象映射。
3. 处理未使用的资源：
   • 若目标床位先获取成功，检查Extra床位请求状态：未触发则取消请求，已触发则将床位放回Store。
   • 若Extra床位先获取成功，取消未完成的目标床位请求，将Extra床位作为最终分配床位。

此修改既实现了优先使用目标诊室床位、 fallback到Extra床位的业务逻辑，又彻底解决了原代码中的AttributeError问题。

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