Search code examples
pythonsimulationsimpy

How to use 'AnyOf' of Simpy?


I tried to solve this error, but I haven't been able to solve this error for a week.

The hospital has beds at ED rooms, ED1 and ED2, and when the available beds here are used up and there are no more beds, I'm going to implement support for Extra beds in one of the two rooms needed. At this time, I tried to implement by using "AnyOf" for the assignment of a bed to the new patient when the beds in the hospital room ED1, ED2, and Extra are all in use.

However, I got an error as below.

AttributeError: 'Bed' object has no attribute 'env'

Please help me solve this error.

My simplified code is below.

Thanks.

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)

        edbed = yield self.clean_beds[pt.care_area].get()
        #print('edbed:', edbed.name)
        
        extrabed = yield self.clean_beds['Extra'].get()
        print('Extra:', extrabed.name)

        seledted_bed_ = yield AnyOf(self.env, [edbed, extrabed]) 
        print('seledted_bed_: ', seledted_bed_)

        if edbed in seledted_bed_:
            # need to remove the extra bed
            if extrabed in seledted_bed_:
                # got two beds but extra bed back
                self.clean_beds['Extra'].put(seledted_bed_)
            else:
                # stil need to cancel req
                extrabed.cancel()
        else:
            # need to deal with the extra bed
            if extrabed in seledted_bed_:
                edbed = extrabed
        
        print(f'{self.env.now:.2f} ED patient {pt.id} {pt.care_area} get a 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)) #, pt
        self.env.run(until = Pre_Define.warmup_period + Pre_Define.sim_duration)


EDModel(1).run()

Solution

  • I did not test this, but I think I see two issues

    anyOf takes a list of events you are passing it a list of beds

    edbed = yield self.clean_beds[pt.care_area].get()
    

    returns a bed

    change to (just dropping the yield)

    edbed = self.clean_beds[pt.care_area].get()
    

    to returns a event

    The same with

    extrabed = yield self.clean_beds['Extra'].get()
    

    anyOf returns a ordered dictionary with an entry for every triggered event. You need this because more then one event can fire at the same time. The dict key is the request and value of the returned value of the fired event.

    so in

    seledted_bed_ = yield AnyOf(self.env, [edbed, extrabed])
    

    seledted_bed will be a dict that you will need to process for two use cases

    Case 1:
    the dict has only one key and value pair, where the value is the bed you are looking for. You will still need to cancel the request that is not in the dict.

    Case 2: The dict has two key/value pairs. This mean both requests fired at the same time. You need to pick one pair to get the bed you want, but you also need to return the bed you did not use back to the store it was pulled from.