Simuler des systèmes complexes avec SimPy en Python

name: simpy description: Framework de simulation d'événements discrets basé sur les processus en Python. Utilisez cette compétence lors de la création de simulations de systèmes avec processus, files d'attente, ressources et événements temporels tels que systèmes de fabrication, opérations de service, trafic réseau, logistique ou tout système où les entités interagissent avec des ressources partagées au fil du temps. license: MIT license tags: [scientific-skills, simpy, python, simulation] metadata: skill-author: K-Dense Inc. ------------|----------| | Resource | Capacité limitée (serveurs, machines) | | PriorityResource | Files d'attente basées sur la priorité | | PreemptiveResource | Haute priorité peut interrompre basse priorité | | Container | Matériaux en vrac (carburant, eau) | | Store | Stockage d'objets Python (FIFO) | | FilterStore | Récupération d'éléments sélectifs | | PriorityStore | Éléments ordonnés par priorité |

Référence rapide

import simpy

env = simpy.Environment()

# Basic resource (e.g., servers)
resource = simpy.Resource(env, capacity=2)

# Priority resource
priority_resource = simpy.PriorityResource(env, capacity=1)

# Container (e.g., fuel tank)
fuel_tank = simpy.Container(env, capacity=100, init=50)

# Store (e.g., warehouse)
warehouse = simpy.Store(env, capacity=10)

Motifs de simulation courants

Motif 1 : File d'attente client-serveur

import simpy
import random

def customer(env, name, server):
    arrival = env.now
    with server.request() as req:
        yield req
        wait = env.now - arrival
        print(f'{name} waited {wait:.2f}, served at {env.now}')
        yield env.timeout(random.uniform(2, 4))

def customer_generator(env, server):
    i = 0
    while True:
        yield env.timeout(random.uniform(1, 3))
        i += 1
        env.process(customer(env, f'Customer {i}', server))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
env.process(customer_generator(env, server))
env.run(until=20)

Motif 2 : Producteur-Consommateur

import simpy

def producer(env, store):
    item_id = 0
    while True:
        yield env.timeout(2)
        item = f'Item {item_id}'
        yield store.put(item)
        print(f'Produced {item} at {env.now}')
        item_id += 1

def consumer(env, store):
    while True:
        item = yield store.get()
        print(f'Consumed {item} at {env.now}')
        yield env.timeout(3)

env = simpy.Environment()
store = simpy.Store(env, capacity=10)
env.process(producer(env, store))
env.process(consumer(env, store))
env.run(until=20)

Motif 3 : Exécution de tâches parallèles

import simpy

def task(env, name, duration):
    print(f'{name} starting at {env.now}')
    yield env.timeout(duration)
    print(f'{name} done at {env.now}')
    return f'{name} result'

def coordinator(env):
    # Start tasks in parallel
    task1 = env.process(task(env, 'Task 1', 5))
    task2 = env.process(task(env, 'Task 2', 3))
    task3 = env.process(task(env, 'Task 3', 4))

    # Wait for all to complete
    results = yield task1 & task2 & task3
    print(f'All done at {env.now}')

env = simpy.Environment()
env.process(coordinator(env))
env.run()

Guide du workflow

Étape 1 : Définir le système

Identifiez :

Entités : Qu'est-ce qui se déplace dans le système ? (clients, pièces, paquets)
Ressources : Quelles sont les contraintes ? (serveurs, machines, bande passante)
Processus : Quelles sont les activités ? (arrivée, service, départ)
Métriques : Qu'a-t-on besoin de mesurer ? (temps d'attente, utilisation, débit)

Étape 2 : Implémenter les fonctions de processus

Créez des fonctions générateur pour chaque type de processus :

def entity_process(env, name, resources, parameters):
    # Arrival logic
    arrival_time = env.now

    # Request resources
    with resource.request() as req:
        yield req

        # Service logic
        service_time = calculate_service_time(parameters)
        yield env.timeout(service_time)

    # Departure logic
    collect_statistics(env.now - arrival_time)

Étape 3 : Configurer la surveillance

Utilisez les utilitaires de surveillance pour collecter des données. Consultez references/monitoring.md pour les techniques complètes.

from scripts.resource_monitor import ResourceMonitor

# Create and monitor resource
resource = simpy.Resource(env, capacity=2)
monitor = ResourceMonitor(env, resource, "Server")

# After simulation
monitor.report()

Étape 4 : Exécuter et analyser

# Run simulation
env.run(until=simulation_time)

# Generate reports
monitor.report()
stats.report()

# Export data for further analysis
monitor.export_csv('results.csv')

Fonctionnalités avancées

Interaction entre processus

Les processus peuvent interagir par des événements, des rendements de processus et des interruptions. Consultez references/process-interaction.md pour les motifs détaillés.

Mécanismes clés :

Signalisation d'événements : Événements partagés pour la coordination
Rendements de processus : Attendre que d'autres processus se terminent
Interruptions : Reprendre forcément les processus pour la préemption

Simulations en temps réel

Synchronisez la simulation avec l'heure d'horloge murale pour les applications hardware-in-the-loop ou interactives. Consultez references/real-time.md.

import simpy.rt

env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1:1 time mapping
# factor=0.5 means 1 sim unit = 0.5 seconds (2x faster)

Surveillance complète

Surveillez les processus, les ressources et les événements. Consultez references/monitoring.md pour les techniques incluant :

Suivi des variables d'état
Monkey-patching de ressources
Traçage d'événements
Collecte statistique

Scripts et modèles

basic_simulation_template.py

Modèle complet pour créer des simulations de files d'attente avec :

Paramètres configurables
Collecte de statistiques
Génération de clients
Utilisation de ressources
Génération de rapports

Utilisation :

from scripts.basic_simulation_template import SimulationConfig, run_simulation

config = SimulationConfig()
config.num_resources = 2
config.sim_time = 100
stats = run_simulation(config)
stats.report()

resource_monitor.py

Utilitaires de surveillance réutilisables :

ResourceMonitor - Suivre une seule ressource
MultiResourceMonitor - Surveiller plusieurs ressources
ContainerMonitor - Suivre les niveaux de conteneur
Calcul statistique automatique
Fonctionnalité d'export CSV