### Install SimPy from Source Source: https://simpy.readthedocs.io/en/latest/_sources/simpy_intro/installation.rst.txt Install SimPy manually by downloading the source archive and running the setup script. ```bash $ python setup.py install ``` -------------------------------- ### SimPy Movie Renege Simulation Setup and Run Source: https://simpy.readthedocs.io/en/latest/examples/movie_renege.html Sets up the SimPy environment, initializes the movie theater with resources and events, and starts the customer arrival process. It then runs the simulation until the specified simulation time. ```python # Setup and start the simulation print('Movie renege') random.seed(RANDOM_SEED) env = simpy.Environment() # Create movie theater movies = ['Python Unchained', 'Kill Process', 'Pulp Implementation'] theater = Theater( counter=simpy.Resource(env, capacity=1), movies=movies, available=dict.fromkeys(movies, TICKETS), sold_out={movie: env.event() for movie in movies}, when_sold_out=dict.fromkeys(movies), num_renegers=dict.fromkeys(movies, 0), ) # Start process and run env.process(customer_arrivals(env, theater)) env.run(until=SIM_TIME) ``` -------------------------------- ### Upload to Test PyPI and Install Source: https://simpy.readthedocs.io/en/latest/_sources/about/release_process.rst.txt Upload the distribution files to the test PyPI server using 'twine' and then install the package from the test server to verify the upload and installation process. ```bash $ twine upload -r testpypi dist/simpy*a.b.c* $ pip install -i https://test.pypi.org/simple/ simpy ``` -------------------------------- ### SimPy Best Practice Example Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/simpy_basics.rst.txt A more concise version of the basic example, demonstrating the preferred way to create Timeout events and processes using environment shortcuts. ```python import simpy def example(env): value = yield env.timeout(1, value=42) print('now=%d, value=%d' % (env.now, value)) env = simpy.Environment() p = env.process(example(env)) env.run() ``` -------------------------------- ### SimPy Process Interaction Example Source: https://simpy.readthedocs.io/en/latest/_sources/simpy_intro/process_interaction.rst.txt This snippet demonstrates starting a SimPy environment, creating a Car and a driver process, and running the simulation until a specific time. It shows how processes can interact and influence each other's execution. ```python >>> env = simpy.Environment() >>> car = Car(env) >>> env.process(driver(env, car)) >>> env.run(until=15) Start parking and charging at 0 Was interrupted. Hope, the battery is full enough ... Start driving at 3 Start parking and charging at 5 Start driving at 10 Start parking and charging at 12 ``` -------------------------------- ### Upload to Test PyPI and Test Installation Source: https://simpy.readthedocs.io/en/latest/about/release_process.html Upload the distribution files to the test PyPI server and then install the package from there to test the release process. ```bash $ twine upload -r testpypi dist/simpy*a.b.c* $ pip install -i https://test.pypi.org/simple/ simpy ``` -------------------------------- ### Test Installation from Wheel Package Source: https://simpy.readthedocs.io/en/latest/about/release_process.html Test the installation of the SimPy package from the wheel package (.whl) in a clean virtual environment. ```bash $ rm -rf /tmp/simpy-wheel # ensure clean state if ran repeatedly $ virtualenv /tmp/simpy-wheel $ /tmp/simpy-wheel/bin/pip install dist/simpy-a.b.c-py2.py3-none-any.whl ``` -------------------------------- ### Simulate Two Clocks with Different Intervals Source: https://simpy.readthedocs.io/en/latest/_sources/index.rst.txt This example demonstrates how to simulate two processes (clocks) that run at different time intervals using SimPy. It shows the basic setup of an environment, defining a process, and running the simulation until a specified time. ```python import simpy def clock(env, name, tick): while True: print(name, env.now) yield env.timeout(tick) env = simpy.Environment() env.process(clock(env, 'fast', 0.5)) env.process(clock(env, 'slow', 1)) env.run(until=2) ``` -------------------------------- ### SimPy Simulation Setup and Run Source: https://simpy.readthedocs.io/en/latest/_sources/simpy_intro/process_interaction.rst.txt Initializes a SimPy environment, creates a Car instance, and runs the simulation until time 15. ```python import simpy env = simpy.Environment() car = Car(env) env.run(until=15) ``` -------------------------------- ### Test Installation from Source Distribution Source: https://simpy.readthedocs.io/en/latest/about/release_process.html Test the installation of the SimPy package from the source distribution (.tar.gz) in a clean virtual environment. ```bash $ rm -rf /tmp/simpy-sdist # ensure clean state if ran repeatedly $ virtualenv /tmp/simpy-sdist $ /tmp/simpy-sdist/bin/pip install dist/simpy-a.b.c.tar.gz ``` -------------------------------- ### SimPy Simulation Output Source: https://simpy.readthedocs.io/en/latest/simpy_intro/shared_resources.html Example output from the SimPy simulation, showing the arrival, charging start, and departure times for each car. Demonstrates resource contention as later cars wait for charging spots. ```text Car 0 arriving at 0 Car 0 starting to charge at 0 Car 1 arriving at 2 Car 1 starting to charge at 2 Car 2 arriving at 4 Car 0 leaving the bcs at 5 Car 2 starting to charge at 5 Car 3 arriving at 6 Car 1 leaving the bcs at 7 Car 3 starting to charge at 7 Car 2 leaving the bcs at 10 Car 3 leaving the bcs at 12 ``` -------------------------------- ### Test Source Distribution Installation Source: https://simpy.readthedocs.io/en/latest/_sources/about/release_process.rst.txt Test the installation of the source distribution (.tar.gz) in a clean virtual environment. This ensures the package can be built and installed from source. ```bash $ rm -rf /tmp/simpy-sdist # ensure clean state if ran repeatedly $ virtualenv /tmp/simpy-sdist $ /tmp/simpy-sdist/bin/pip install dist/simpy-a.b.c.tar.gz ``` -------------------------------- ### Delayed Process Start with start_delayed Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/events.rst.txt Illustrates starting a sub-process with a delay using `simpy.util.start_delayed`. The parent process yields the helper process returned by `start_delayed` and then yields the sub-process itself to get its return value. ```python from simpy.util import start_delayed def sub(env): yield env.timeout(1) return 23 def parent(env): sub_proc = yield start_delayed(env, sub(env), delay=3) ret = yield sub_proc return ret env.run(env.process(parent(env))) ``` -------------------------------- ### SimPy 3 Process Example Source: https://simpy.readthedocs.io/en/latest/about/defense_of_design.html This example demonstrates the SimPy 3 API for defining and running a process. It shows how to use an Environment and yield environment timeouts. ```python from simpy import Environment, simulate def pem(env, repeat): for i in range(repeat): yield env.timeout(i) env = Environment() env.process(pem(env, 7)) simulate(env, until=10) ``` -------------------------------- ### Pin SimPy 3.x in setup.py Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/porting_from_simpy3.rst.txt Example of specifying the SimPy version constraint in the install_requires list of a setup.py file. ```python setup( ..., install_requires=['simpy<4'], ..., ) ``` -------------------------------- ### Carwash Simulation Setup and Execution Source: https://simpy.readthedocs.io/en/latest/examples/carwash.html Sets up and runs the carwash simulation. It initializes the environment, creates the carwash and initial cars, and then continuously generates new cars. ```python """Carwash example. Covers: - Waiting for other processes - Resources: Resource Scenario: A carwash has a limited number of washing machines and defines a washing processes that takes some (random) time. Car processes arrive at the carwash at a random time. If one washing machine is available, they start the washing process and wait for it to finish. If not, they wait until they can use one. """ import itertools import random import simpy # fmt: off RANDOM_SEED = 42 NUM_MACHINES = 2 # Number of machines in the carwash WASHTIME = 5 # Minutes it takes to clean a car T_INTER = 7 # Create a car every ~7 minutes SIM_TIME = 20 # Simulation time in minutes # fmt: on class Carwash: """A carwash has a limited number of machines (``NUM_MACHINES``) to clean cars in parallel. Cars have to request one of the machines. When they got one, they can start the washing processes and wait for it to finish (which takes ``washtime`` minutes). """ def __init__(self, env, num_machines, washtime): self.env = env self.machine = simpy.Resource(env, num_machines) self.washtime = washtime def wash(self, car): """The washing processes. It takes a ``car`` processes and tries to clean it.""" yield self.env.timeout(self.washtime) pct_dirt = random.randint(50, 99) print(f"Carwash removed {pct_dirt}% of {car}'s dirt.") def car(env, name, cw): """The car process (each car has a ``name``) arrives at the carwash (``cw``) and requests a cleaning machine. It then starts the washing process, waits for it to finish and leaves to never come back ... """ print(f'{name} arrives at the carwash at {env.now:.2f}.') with cw.machine.request() as request: yield request print(f'{name} enters the carwash at {env.now:.2f}.') yield env.process(cw.wash(name)) print(f'{name} leaves the carwash at {env.now:.2f}.') def setup(env, num_machines, washtime, t_inter): """Create a carwash, a number of initial cars and keep creating cars approx. every ``t_inter`` minutes.""" # Create the carwash carwash = Carwash(env, num_machines, washtime) car_count = itertools.count() # Create 4 initial cars for _ in range(4): env.process(car(env, f'Car {next(car_count)}', carwash)) # Create more cars while the simulation is running while True: yield env.timeout(random.randint(t_inter - 2, t_inter + 2)) env.process(car(env, f'Car {next(car_count)}', carwash)) # Setup and start the simulation print('Carwash') print('Check out http://youtu.be/fXXmeP9TvBg while simulating ... ;-)') random.seed(RANDOM_SEED) # This helps to reproduce the results # Create an environment and start the setup process env = simpy.Environment() env.process(setup(env, NUM_MACHINES, WASHTIME, T_INTER)) # Execute! env.run(until=SIM_TIME) ``` -------------------------------- ### Test Wheel Package Installation Source: https://simpy.readthedocs.io/en/latest/_sources/about/release_process.rst.txt Test the installation of the wheel package (.whl) in a clean virtual environment. This verifies the pre-built distribution installs correctly. ```bash $ rm -rf /tmp/simpy-wheel # ensure clean state if ran repeatedly $ virtualenv /tmp/simpy-wheel $ /tmp/simpy-wheel/bin/pip install dist/simpy-a.b.c-py2.py3-none-any.whl ``` -------------------------------- ### Basic Event Creation Example Source: https://simpy.readthedocs.io/en/latest/_sources/about/defense_of_design.rst.txt Illustrates a simple method within an Environment class that returns an Event instance. This is a foundational example for event generation. ```python def event(self): return Event() ``` -------------------------------- ### Install SimPy from Production PyPI Source: https://simpy.readthedocs.io/en/latest/_sources/about/release_process.rst.txt Install or upgrade SimPy from the production PyPI using pip. This is the final installation test after a successful release. ```bash $ pip install -U simpy ``` -------------------------------- ### SimPy Basic Example Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/simpy_basics.rst.txt Illustrates the core SimPy concepts: creating an environment, defining a process function that yields a Timeout event, and running the simulation. ```python import simpy def example(env): event = simpy.events.Timeout(env, delay=1, value=42) value = yield event print('now=%d, value=%d' % (env.now, value)) env = simpy.Environment() example_gen = example(env) p = simpy.events.Process(env, example_gen) env.run() ``` -------------------------------- ### Install SimPy using pip Source: https://simpy.readthedocs.io/en/latest/_sources/simpy_intro/installation.rst.txt Use this command to install the latest version of SimPy from the Python Package Index. ```bash $ pip install simpy ``` -------------------------------- ### Basic Resource Usage Example Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/resources.rst.txt Demonstrates how a process requests, uses, and releases a resource. The resource capacity is set to 1, acting like a semaphore. ```python import simpy def resource_user(env, resource): request = resource.request() # Generate a request event yield request # Wait for access yield env.timeout(1) # Do something resource.release(request) # Release the resource env = simpy.Environment() res = simpy.Resource(env, capacity=1) user = env.process(resource_user(env, res)) env.run() ``` -------------------------------- ### Process Constructor Example Source: https://simpy.readthedocs.io/en/latest/_sources/about/history.rst.txt Illustrates the updated Process constructor signature introduced in version 0.5 Beta. Backward compatibility is maintained when using keyword parameters. ```python def __init__(self,name="a_process") ``` -------------------------------- ### Resource Constructor Example Source: https://simpy.readthedocs.io/en/latest/_sources/about/history.rst.txt Shows the modified Resource constructor signature from version 0.5 Beta. Backward compatibility is preserved when using keyword parameters. ```python def __init__(self,capacity=1,name="a_resource",unitName="units") ``` -------------------------------- ### PreemptiveResource Priority Example Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/resources.rst.txt Demonstrates how PreemptiveResource handles priorities, showing that lower-priority preemptive requests cannot jump ahead of higher-priority non-preemptive requests. ```python >>> def user(name, env, res, prio, preempt): ... with res.request(priority=prio, preempt=preempt) as req: ... try: ... print(f'{name} requesting at {env.now}') ... assert isinstance(env.now, int), type(env.now) ... yield req ... assert isinstance(env.now, int), type(env.now) ... print(f'{name} got resource at {env.now}') ... yield env.timeout(3) ... except simpy.Interrupt: ... print(f'{name} got preempted at {env.now}') >>> >>> env = simpy.Environment() >>> res = simpy.PreemptiveResource(env, capacity=1) >>> A = env.process(user('A', env, res, prio=0, preempt=True)) >>> env.run(until=1) # Give A a head start A requesting at 0 A got resource at 0 >>> B = env.process(user('B', env, res, prio=-2, preempt=False)) >>> C = env.process(user('C', env, res, prio=-1, preempt=True)) >>> env.run() B requesting at 1 C requesting at 1 B got resource at 3 C got resource at 6 ``` -------------------------------- ### Delayed Process Start with simpy.util.start_delayed Source: https://simpy.readthedocs.io/en/latest/topical_guides/events.html Shows how to start a process after a specified delay using `start_delayed`. Note the additional `yield` required for the helper process returned by `start_delayed`. ```python >>> from simpy.util import start_delayed >>> >>> def sub(env): ... yield env.timeout(1) ... return 23 ... >>> def parent(env): ... sub_proc = yield start_delayed(env, sub(env), delay=3) ... ret = yield sub_proc ... return ret ... >>> env.run(env.process(parent(env))) 23 ``` -------------------------------- ### SimPy 1 Process Example Source: https://simpy.readthedocs.io/en/latest/_sources/about/defense_of_design.rst.txt Illustrates how to implement a simple simulation process by sub-classing 'Process' and using module-level functions for simulation control in SimPy 1. ```python from SimPy.Simulation import Process, hold, initialize, activate, simulate class MyProcess(Process): def pem(self, repeat): for i in range(repeat): yield hold, self, 1 initialize() proc = MyProcess() activate(proc, proc.pem(3)) simulate(until=10) ``` -------------------------------- ### Process Communication Example Source: https://simpy.readthedocs.io/en/latest/_sources/examples/process_communication.rst.txt This Python script demonstrates process communication using SimPy's resources.Store for asynchronous interactions between simulation processes. It includes examples of one-to-one, many-to-one, and one-to-many communication patterns, along with a BroadcastPipe for broadcasting messages. ```python import simpy import random # A process that generates items and puts them into a store def producer(env, store, name, delay): for i in range(5): yield env.timeout(delay) item = f"item {i} from {name}" store.put(item) print(f"{env.now:.2f}: {name} produced {item}") # A process that consumes items from a store def consumer(env, store, name, delay): while True: yield env.timeout(delay) # Simulate some work before consuming item = yield store.get() print(f"{env.now:.2f}: {name} consumed {item}") # A process that broadcasts messages to multiple consumers class BroadCastPipe: def __init__(self, env): self.env = env self.consumers = [] def add_consumer(self, consumer_store): self.consumers.append(consumer_store) def put(self, item): for consumer_store in self.consumers: consumer_store.put(item) # A process that uses the BroadcastPipe def broadcaster(env, broadcast_pipe, name, delay): for i in range(3): yield env.timeout(delay) message = f"message {i} from {name}" broadcast_pipe.put(message) print(f"{env.now:.2f}: {name} broadcasted {message}") # --- Simulation Setup --- env = simpy.Environment() # One-to-one communication store_1_to_1 = simpy.Store(env) env.process(producer(env, store_1_to_1, 'Producer A', 1.0)) env.process(consumer(env, store_1_to_1, 'Consumer A', 1.5)) # Many-to-one communication store_many_to_1 = simpy.Store(env) env.process(producer(env, store_many_to_1, 'Producer B', 0.8)) env.process(producer(env, store_many_to_1, 'Producer C', 1.2)) env.process(consumer(env, store_many_to_1, 'Consumer B', 2.0)) # One-to-many communication using BroadcastPipe broadcast_pipe = BroadCastPipe(env) consumer_store_1 = simpy.Store(env) broadcast_pipe.add_consumer(consumer_store_1) env.process(consumer(env, consumer_store_1, 'Consumer C', 1.0)) consumer_store_2 = simpy.Store(env) broadcast_pipe.add_consumer(consumer_store_2) env.process(consumer(env, consumer_store_2, 'Consumer D', 1.2)) env.process(broadcaster(env, broadcast_pipe, 'Broadcaster X', 2.5)) # Run the simulation env.run(until=10) ``` -------------------------------- ### Install Latest SimPy from Production PyPI Source: https://simpy.readthedocs.io/en/latest/about/release_process.html Install or upgrade to the latest version of SimPy from the production PyPI service after a successful release. ```bash $ pip install -U simpy ``` -------------------------------- ### Install SimPy 3.x using pip Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/porting_from_simpy3.rst.txt Command to install the latest SimPy 3.x version, excluding SimPy 4 and above. ```shell pip install 'simpy<4' ``` -------------------------------- ### SimPy Process Communication Example Source: https://simpy.readthedocs.io/en/latest/examples/process_communication.html This example demonstrates one-to-one and one-to-many process communication using SimPy's Store and a custom BroadcastPipe. It covers asynchronous message generation and consumption, highlighting potential message lateness. ```python import random import simpy import simpy.core RANDOM_SEED = 42 SIM_TIME = 100 class BroadcastPipe: """A Broadcast pipe that allows one process to send messages to many. This construct is useful when message consumers are running at different rates than message generators and provides an event buffering to the consuming processes. The parameters are used to create a new :class:`~simpy.resources.store.Store` instance each time :meth:`get_output_conn()` is called. """ def __init__(self, env, capacity=simpy.core.Infinity): self.env = env self.capacity = capacity self.pipes = [] def put(self, value): """Broadcast a *value* to all receivers.""" if not self.pipes: raise RuntimeError('There are no output pipes.') events = [store.put(value) for store in self.pipes] return self.env.all_of(events) # Condition event for all "events" def get_output_conn(self): """Get a new output connection for this broadcast pipe. The return value is a :class:`~simpy.resources.store.Store`. """ pipe = simpy.Store(self.env, capacity=self.capacity) self.pipes.append(pipe) return pipe def message_generator(name, env, out_pipe): """A process which randomly generates messages.""" while True: # wait for next transmission yield env.timeout(random.randint(6, 10)) # messages are time stamped to later check if the consumer was # late getting them. Note, using event.triggered to do this may # result in failure due to FIFO nature of simulation yields. # (i.e. if at the same env.now, message_generator puts a message # in the pipe first and then message_consumer gets from pipe, # the event.triggered will be True in the other order it will be # False msg = (env.now, f'{name} says hello at {env.now}') out_pipe.put(msg) def message_consumer(name, env, in_pipe): """A process which consumes messages.""" while True: # Get event for message pipe msg = yield in_pipe.get() if msg[0] < env.now: # if message was already put into pipe, then # message_consumer was late getting to it. Depending on what # is being modeled this, may, or may not have some # significance print( f'LATE Getting Message: at time {env.now}: ' f'{name} received message: {msg[1]}' ) else: # message_consumer is synchronized with message_generator print(f'at time {env.now}: {name} received message: {msg[1]}.') # Process does some other work, which may result in missing messages yield env.timeout(random.randint(4, 8)) # Setup and start the simulation print('Process communication') random.seed(RANDOM_SEED) env = simpy.Environment() # For one-to-one or many-to-one type pipes, use Store pipe = simpy.Store(env) env.process(message_generator('Generator A', env, pipe)) env.process(message_consumer('Consumer A', env, pipe)) print('\nOne-to-one pipe communication\n') env.run(until=SIM_TIME) # For one-to many use BroadcastPipe # (Note: could also be used for one-to-one,many-to-one or many-to-many) env = simpy.Environment() bc_pipe = BroadcastPipe(env) env.process(message_generator('Generator A', env, bc_pipe)) env.process(message_consumer('Consumer A', env, bc_pipe.get_output_conn())) env.process(message_consumer('Consumer B', env, bc_pipe.get_output_conn())) print('\nOne-to-many pipe communication\n') env.run(until=SIM_TIME) ``` -------------------------------- ### Run SimPy Tests from Source Source: https://simpy.readthedocs.io/en/latest/simpy_intro/installation.html After installing from source, run this command to execute SimPy's tests using pytest. Ensure pytest is installed separately. ```bash $ py.test --pyargs simpy ``` -------------------------------- ### SimPy 3 Import Example Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/porting_from_simpy2.rst.txt Demonstrates the simplified import structure in SimPy 3, typically only requiring the 'simpy' package. ```python import simpy ``` -------------------------------- ### SimPy 2 Import Example Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/porting_from_simpy2.rst.txt Shows the typical imports required for SimPy 2, including Simulation, Process, and keywords like hold. ```python from Simpy.Simulation import Simulation, Process, hold ``` -------------------------------- ### Run a SimPy Simulation Source: https://simpy.readthedocs.io/en/latest/_sources/simpy_intro/basic_concepts.rst.txt Initializes a SimPy environment, creates a car process, and runs the simulation until a specified time. This demonstrates how to start and execute a basic SimPy process. ```python import simpy env = simpy.Environment() env.process(car(env)) env.run(until=15) ``` -------------------------------- ### SimPy Environment and Resource Initialization Source: https://simpy.readthedocs.io/en/latest/simpy_intro/shared_resources.html Initializes a SimPy environment and creates a Resource with a capacity of 2. This setup is required before creating processes that utilize the resource. ```python import simpy env = simpy.Environment() bcs = simpy.Resource(env, capacity=2) ``` -------------------------------- ### SimPy 1 Simulation Instance Example Source: https://simpy.readthedocs.io/en/latest/_sources/about/defense_of_design.rst.txt Demonstrates using a Simulation instance explicitly in SimPy 1, which was a precursor to the object-oriented API in SimPy 2. ```python sim = Simulation() proc = MyProcess(sim=sim) sim.activate(proc, proc.pem(3)) sim.simulate(until=10) ``` -------------------------------- ### SimPy 3 Environment and Generator Example Source: https://simpy.readthedocs.io/en/latest/_sources/about/defense_of_design.rst.txt Illustrates the simplified SimPy 3 API, using an Environment and a generator function (PEM) without requiring class inheritance. ```python from simpy import Environment, simulate def pem(env, repeat): for i in range(repeat): yield env.timeout(i) env = Environment() env.process(pem(env, 7)) simulate(env, until=10) ``` -------------------------------- ### SimPy Simulation with Tracing and Monitoring Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/monitoring.rst.txt This example demonstrates setting up a SimPy environment, tracing events with a custom monitor callback (using functools.partial), running a process, and printing the collected event data. ```python def test_process(env): yield env.timeout(1) data = [] # Bind *data* as first argument to monitor() # see https://docs.python.org/3/library/functools.html#functools.partial monitor = partial(monitor, data) env = simpy.Environment() trace(env, monitor) p = env.process(test_process(env)) env.run(until=p) for d in data: print(d) ``` -------------------------------- ### Build Documentation with Sphinx Source: https://simpy.readthedocs.io/en/latest/_sources/about/release_process.rst.txt Build the project documentation using Sphinx by running tox with the sphinx environment. Ensure there are no errors or undefined references. ```console $ tox -e sphinx ``` -------------------------------- ### Basic Real-time Simulation Example Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/real-time-simulations.rst.txt Compares the duration of one simulation time unit in a standard SimPy environment versus a real-time environment with a factor of 0.1. Demonstrates how to set up and run a real-time simulation. ```python import time import simpy def example(env): start = time.perf_counter() yield env.timeout(1) end = time.perf_counter() print('Duration of one simulation time unit: %.2fs' % (end - start)) env = simpy.Environment() proc = env.process(example(env)) env.run(until=proc) import simpy.rt env = simpy.rt.RealtimeEnvironment(factor=0.1) proc = env.process(example(env)) env.run(until=proc) ``` -------------------------------- ### Using Get Event as Context Manager Source: https://simpy.readthedocs.io/en/latest/api_reference/simpy.resources.html Illustrates using the `Get` event as a context manager. The `with` statement ensures that the get request is automatically cancelled if an exception is raised. ```python with res.get() as request: item = yield request ``` -------------------------------- ### Get Event Source: https://simpy.readthedocs.io/en/latest/api_reference/simpy.resources.html Represents a request to get an item from a resource. It can be used as a context manager with the `with` statement for automatic cancellation. ```APIDOC ## class simpy.resources.base.Get(resource: ResourceType) Generic event for requesting to get something from the _resource_. This event (and all of its subclasses) can act as context manager and can be used with the `with` statement to automatically cancel the request if an exception (like an `simpy.exceptions.Interrupt` for example) occurs: ```python with res.get() as request: item = yield request ``` ### Methods - **cancel()**: Cancel this get request. This method has to be called if the get request must be aborted, for example if a process needs to handle an exception like an `Interrupt`. If the get request was created in a `with` statement, this method is called automatically. ``` -------------------------------- ### Run SimPy Tests Source: https://simpy.readthedocs.io/en/latest/_sources/simpy_intro/installation.rst.txt Execute SimPy's test suite using pytest to verify the installation. Requires pytest to be installed. ```bash $ py.test --pyargs simpy ``` -------------------------------- ### Build Documentation with Sphinx Source: https://simpy.readthedocs.io/en/latest/about/release_process.html Build the project documentation using Sphinx. Verify that there are no errors or undefined references. ```bash $ tox -e sphinx ``` -------------------------------- ### Basic PreemptiveResource Usage Source: https://simpy.readthedocs.io/en/latest/topical_guides/resources.html Demonstrates how a PreemptiveResource handles requests with different priorities and how preemption occurs. Use this to understand the default preemptive behavior. ```python def resource_user(name, env, resource, wait, prio): yield env.timeout(wait) with resource.request(priority=prio) as req: print(f'{name} requesting at {env.now} with priority={prio}') yield req print(f'{name} got resource at {env.now}') try: yield env.timeout(3) except simpy.Interrupt as interrupt: by = interrupt.cause.by usage = env.now - interrupt.cause.usage_since print(f'{name} got preempted by {by} at {env.now}' f' after {usage}') env = simpy.Environment() res = simpy.PreemptiveResource(env, capacity=1) p1 = env.process(resource_user(1, env, res, wait=0, prio=0)) p2 = env.process(resource_user(2, env, res, wait=1, prio=0)) p3 = env.process(resource_user(3, env, res, wait=2, prio=-1)) env.run() ``` -------------------------------- ### Build Source Distribution and Wheel Package Source: https://simpy.readthedocs.io/en/latest/about/release_process.html Build both a source distribution (.tar.gz) and a wheel package (.whl) for the new release using the 'build' tool. ```bash $ python -m build $ ls dist/ simpy-a.b.c-py2.py3-none-any.whl simpy-a.b.c.tar.gz ``` -------------------------------- ### FilterStore Source: https://simpy.readthedocs.io/en/latest/api_reference/simpy.resources.html A resource with capacity slots for storing arbitrary objects, supporting filtered get requests. Items are retrieved in FIFO order unless a filter is applied. Get requests can be customized with a filter function. ```APIDOC ## Class: FilterStore ### Description Resource with capacity slots for storing arbitrary objects supporting filtered get requests. Like the `Store`, the capacity is unlimited by default and objects are put and retrieved from the store in a first-in, first-out order. Get requests can be customized with a filter function to only trigger for items for which said filter function returns `True`. ### Parameters * `_env` (Environment): The Environment instance the container is bound to. * `_capacity` (float | int, optional): Maximum capacity of the resource. Defaults to infinity. ### Notes In contrast to `Store`, get requests of a `FilterStore` won’t necessarily be triggered in the same order they were issued. ### Example The store is empty. Process 1 tries to get an item of type `a`, Process 2 an item of type `b`. Another process puts one item of type `b` into the store. Though Process 2 made his request after Process 1, it will receive that new item because Process 1 doesn’t want it. ``` -------------------------------- ### Build Source and Wheel Packages Source: https://simpy.readthedocs.io/en/latest/_sources/about/release_process.rst.txt Build both a source distribution (.tar.gz) and a wheel package (.whl) for the new release using the 'build' module. Inspect the contents of the 'dist/' directory. ```bash $ python -m build $ ls dist/ simpy-a.b.c-py2.py3-none-any.whl simpy-a.b.c.tar.gz ``` -------------------------------- ### Gas Station Simulation with SimPy Container Source: https://simpy.readthedocs.io/en/latest/topical_guides/resources.html This example models a gas station using SimPy's Resource for fuel dispensers and Container for the gas tank. It demonstrates how to monitor the tank level and trigger a tanker refill when it drops below a threshold. Cars arrive, request a dispenser, refuel, and leave. ```python class GasStation: def __init__(self, env): self.fuel_dispensers = simpy.Resource(env, capacity=2) self.gas_tank = simpy.Container(env, init=100, capacity=1000) self.mon_proc = env.process(self.monitor_tank(env)) def monitor_tank(self, env): while True: if self.gas_tank.level < 100: print(f'Calling tanker at {env.now}') env.process(tanker(env, self)) yield env.timeout(15) def tanker(env, gas_station): yield env.timeout(10) # Need 10 Minutes to arrive print(f'Tanker arriving at {env.now}') amount = gas_station.gas_tank.capacity - gas_station.gas_tank.level yield gas_station.gas_tank.put(amount) def car(name, env, gas_station): print(f'Car {name} arriving at {env.now}') with gas_station.fuel_dispensers.request() as req: yield req print(f'Car {name} starts refueling at {env.now}') yield gas_station.gas_tank.get(40) yield env.timeout(5) print(f'Car {name} done refueling at {env.now}') def car_generator(env, gas_station): for i in range(4): env.process(car(i, env, gas_station)) yield env.timeout(5) env = simpy.Environment() gas_station = GasStation(env) car_gen = env.process(car_generator(env, gas_station)) env.run(35) ``` -------------------------------- ### StoreGet Source: https://simpy.readthedocs.io/en/latest/api_reference/simpy.resources.html Represents a request to get an item from a Store. The request is triggered once an item is available in the store. ```APIDOC ## Class: StoreGet ### Description Request to get an _item_ from the _store_. The request is triggered once there is an item available in the store. ### Parameters * `_resource` (ResourceType): The Store resource to get the item from. ``` -------------------------------- ### Basic SimPy Process Example Source: https://simpy.readthedocs.io/en/latest/topical_guides/simpy_basics.html Illustrates the creation of a Timeout event, yielding it within a process function, and processing the event's value. Requires explicit creation of Environment and Process objects. ```python import simpy def example(env): event = simpy.events.Timeout(env, delay=1, value=42) value = yield event print('now=%d, value=%d' % (env.now, value)) env = simpy.Environment() example_gen = example(env) p = simpy.events.Process(env, example_gen) env.run() ``` -------------------------------- ### simpy.util.start_delayed Source: https://simpy.readthedocs.io/en/latest/api_reference/simpy.util.html Returns a helper process that starts another process for a given generator after a specified delay. ```APIDOC ## start_delayed(env, generator, delay) ### Description Returns a helper process that starts another process for _generator_ after a certain _delay_. ### Signature simpy.util.start_delayed(_env : Environment_, _generator : Generator[Event, Any, Any]_, _delay : int | float_) → Process ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters * **env** (Environment) - The SimPy environment. * **generator** (Generator[Event, Any, Any]) - The generator for the process to start. * **delay** (int | float) - The delay in simulation time units before starting the process. ### Returns * **Process** - A helper process that starts the target generator after the specified delay. ### Raises * **ValueError**: If `delay <= 0`. ### Example ```python from simpy import Environment from simpy.util import start_delayed def my_process(env, x): print(f'{env.now}, {x}') yield env.timeout(1) env = Environment() proc = start_delayed(env, my_process(env, 3), 5) env.run() # Output: 5, 3 ``` ``` -------------------------------- ### SimPy 4 Environment Subclass Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/porting_from_simpy3.rst.txt Example of defining a custom environment by inheriting from simpy.Environment in SimPy 4. ```python class MyEnvironment(simpy.Environment): ... ``` -------------------------------- ### SimPy 3 Environment Subclass Source: https://simpy.readthedocs.io/en/latest/_sources/topical_guides/porting_from_simpy3.rst.txt Example of defining a custom environment by inheriting from simpy.BaseEnvironment in SimPy 3. ```python class MyEnvironment(simpy.BaseEnvironment): ... ``` -------------------------------- ### SimPy Resource Initialization and Process Creation Source: https://simpy.readthedocs.io/en/latest/_sources/simpy_intro/shared_resources.rst.txt Initializes a SimPy environment and a Resource with a capacity of 2. Then, it creates and starts multiple car processes, passing the resource reference and other parameters. ```python import simpy env = simpy.Environment() bcs = simpy.Resource(env, capacity=2) for i in range(4): env.process(car(env, 'Car %d' % i, bcs, i*2, 5)) ```