SimPy - biblioteka dla DES

Biblioteka simpy dostarcza gotowe klasy obiektów do budowy symulacji:

  • procesy (Process) - reprezentują procesy zachodzące w czasie (czynności wykonywane przez agentów)
  • zdarzenia (Event) - reprezentują szczególny 'punktowy' w czasie rodzaj procesu
  • zasoby (Resource) - reprezentują zasoby o ograniczonej pojemności, współdzielone przez procesy
  • środowisko (Environment), które zarządzą upływem czasu i następstwem zdarzeń oraz koordynuje procesy i dostęp do zasobów

instalacja pakietu simpy

Pakiet simpy nie jest czescia dysrubucji Anaconda. Aby go zainstalowac z poziomu wiersza poleceń systemu operacyjnego wykonujemy komende:

pip3 install simpy

albo

pip3 install --user simpy

jesli nie mamy uprawnień aministratora.

Przygotowanie przebiegu symulacji zaczynamy od utworzenia środowiska, w którym bedzię się ona odbywać:

In [ ]:
import simpy
In [ ]:
# utworz srodowisko symulacji
env = simpy.Environment() 

srodowisko symulacji w konkretnym przebiegu jest obiektem klasy Environment

In [ ]:
env
In [ ]:
type(env)

Funkcja car definiuje prostego agenta, który ma dwa stany - jazdę (driving) albo parkowanie (parking) i przełącza się między nimi w deterministyczny sposób.

In [ ]:
def car(env):
    trip_duration = 3
    parking_duration = 5
    while True:
        print('Start parking at %d' % env.now)
        yield env.timeout(parking_duration)
        print('Start driving at %d' % env.now)
        yield env.timeout(trip_duration)
In [ ]:
car(env)

dodajemy agenta do srodowiska symulacji

In [ ]:
env.process(car(env))

uruchamiamy sumulacje:

In [ ]:
env.run(until=25)

zegar środowiska symulacji przesunął sie o 25 jednostek, możemy kontunuować jej wykonanie:

In [ ]:
env.run(40)

wprowadzmy element losowosci do zachowania agentów:

In [ ]:
import random as rnd
LAMBDA = 1

def car(env):
    while True:
        print('Start parking at %f' % env.now)
        yield env.timeout(rnd.expovariate(LAMBDA))
        print('Start driving at %f' % env.now)
        yield env.timeout(rnd.expovariate(LAMBDA))
In [ ]:
env = simpy.Environment() # utworz srodowisko symulacji
In [ ]:
env.process(car(env)) # dodaj agenta do srodowiska
In [ ]:
env.run(10)

Urozmaicimy symulację przez heterogenizację populacji samochodów. Wprowadzimy też nowy typ agenta - stację paliwową.

In [ ]:
def car(env, name, bcs, driving_time, charge_duration):
    # Simulate driving to the BCS
    yield env.timeout(driving_time)

    # Request one of its charging spots
    print('%s arriving at %d' % (name, env.now))
    with bcs.request() as req:
        yield req

        # Charge the battery
        print('%s starting to charge at %s' % (name, env.now))
        yield env.timeout(charge_duration)
        print('%s leaving the bcs at %s' % (name, env.now))

Przygotowanie przebiegu symulacji

In [ ]:
env = simpy.Environment() # utworz srodowisko symulacji
bcs = simpy.Resource(env, capacity=2) # stacja paliwowa z dwoma dystrybutorami

tworzymy populacje agentów, różniących sie nazwą i czasem jazdy i dodajemy do srodowiska

In [ ]:
N = 15
for i in range(N):
    c = car(env, 'Car %d' % i, bcs, i*2, 5)
    env.process(c)

Uruchamiamy przebieg symulacji. Zauważmy, że nie podajemy ograniczenia czasowego. Dlaczego?

In [ ]:
env.run()

Wydzielamy proces tankowania/ładowania do osobnej funkcji. Agenci typu car beda oczekiwac na zakonczenie procesu ładowania a nie wystąpienie zdarzenia typu timeout. W ten sposób logika symulacji staje się jaśniejsza.

In [ ]:
def charge(env, name, duration):
    # Charge the battery
    print('%s starting to charge at %s' % (name, env.now))
    yield env.timeout(duration)
    print('%s finished charging at %s' % (name, env.now))
In [ ]:
def car(env, name, bcs, driving_time, charge_duration):
    # Simulate driving to the BCS
    yield env.timeout(driving_time)

    # Request one of its charging spots
    print('%s arriving at %d' % (name, env.now))
    with bcs.request() as req:
        yield req

        # We yield the process that process() returns
        # to wait for it to finish
        yield env.process(charge(env, name, charge_duration))
        print('%s leaving the bcs at %s' % (name, env.now))
In [ ]:
env = simpy.Environment()
bcs = simpy.Resource(env, capacity=2) # stacja paliwowa z dwoma dystrybutorami
In [ ]:
N = 8
for i in range(N):
    c = car(env, 'Car %d' % i, bcs, i*2, 5)
    env.process(c)
env.run()

Złożoność naszej symulacji rośnie dlatego opakujemy kod agenta typu car w klasę. Wprowadzimy też nowe elementy

  • kierowcę - agenta typu driver
  • możliwość przerwania trwającego procesu tankowania/ładowania przed zaplanowanym czasem jego zakończenia przez kierowcę
In [ ]:
def driver(env, car):
    yield env.timeout(3)
    car.action.interrupt()
    yield env.timeout(6)
    car.action.interrupt()
In [ ]:
class Car(object):
    def __init__(self, env):
        self.env = env
        self.action = env.process(self.run())

    def run(self):
        trip_duration = 2
        charge_duration = 5
        while True:
            print('Start parking and charging at %d' % self.env.now)
            # We may get interrupted while charging the battery
            try:
                yield self.env.process(self.charge(charge_duration))
            except simpy.Interrupt:
                # When we received an interrupt, we stop charing and
                # switch to the "driving" state
                print('Was interrupted! Hope, the battery is full enough...')

            print('Start driving at %d' % self.env.now)
            yield self.env.timeout(trip_duration)

    def charge(self, duration):
        yield self.env.timeout(duration)
In [ ]:
env = simpy.Environment()
car = Car(env)
env.process(driver(env, car))
In [ ]:
env.run(until=50)