SimPy

https://simpy.readthedocs.io/en/latest
Admin
SimPy is a process-based discrete-event simulation framework for Python. It allows modeling active...
Tokens:69,480
Snippets:665
Trust Score:8.3
Update:3 months ago
Show doc for...
Tokens:
Raw
### 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))
<Process(driver) object at 0x...>
>>> 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))
```
Tokens:
Raw