Logging in SimPy: Debugging and Tracing Your Simulations

import logging
import simpy

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def customer(env, name, server):
    logger.info(f"{name} arrives at {env.now}")

    with server.request() as req:
        yield req
        logger.debug(f"{name} starts service at {env.now}")
        yield env.timeout(5)

    logger.info(f"{name} leaves at {env.now}")

logger.debug("Detailed diagnostic info")     # For debugging
logger.info("General operational events")    # Normal operation
logger.warning("Something unexpected")       # Potential issues
logger.error("Something went wrong")         # Errors
logger.critical("System failure")            # Critical failures

class SimulationLogger:
    def __init__(self, name, level=logging.INFO):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(level)

        # Console handler
        ch = logging.StreamHandler()
        ch.setLevel(level)

        # Format with simulation context
        formatter = logging.Formatter(
            '%(asctime)s | %(levelname)-8s | %(message)s',
            datefmt='%H:%M:%S'
        )
        ch.setFormatter(formatter)
        self.logger.addHandler(ch)

    def event(self, env, entity, action, details=""):
        """Log a simulation event."""
        self.logger.info(f"t={env.now:8.2f} | {entity:15} | {action:15} | {details}")

    def debug(self, message):
        self.logger.debug(message)

# Usage
log = SimulationLogger('simulation', level=logging.INFO)

def customer(env, name, server):
    log.event(env, name, "ARRIVE")

    with server.request() as req:
        yield req
        log.event(env, name, "START_SERVICE")
        yield env.timeout(5)
        log.event(env, name, "END_SERVICE")

    log.event(env, name, "DEPART")

import logging

def setup_logging(log_file='simulation.log', level=logging.INFO):
    """Configure logging to both console and file."""
    logger = logging.getLogger('simulation')
    logger.setLevel(level)

    # File handler
    fh = logging.FileHandler(log_file, mode='w')
    fh.setLevel(logging.DEBUG)  # Capture everything in file

    # Console handler
    ch = logging.StreamHandler()
    ch.setLevel(level)

    # Formatters
    detailed_fmt = logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
    )
    simple_fmt = logging.Formatter('%(levelname)s: %(message)s')

    fh.setFormatter(detailed_fmt)
    ch.setFormatter(simple_fmt)

    logger.addHandler(fh)
    logger.addHandler(ch)

    return logger

logger = setup_logging()

class EventTracer:
    def __init__(self, env):
        self.env = env
        self.events = []

    def trace(self, entity, event_type, **kwargs):
        """Record an event with full context."""
        record = {
            'time': self.env.now,
            'entity': entity,
            'event': event_type,
            **kwargs
        }
        self.events.append(record)

    def to_dataframe(self):
        import pandas as pd
        return pd.DataFrame(self.events)

    def save(self, filename):
        df = self.to_dataframe()
        df.to_csv(filename, index=False)

# Usage
tracer = EventTracer(env)

def customer(env, name, server, tracer):
    tracer.trace(name, 'ARRIVE', queue_length=len(server.queue))

    with server.request() as req:
        yield req
        tracer.trace(name, 'START_SERVICE', wait_time=env.now - arrival)
        yield env.timeout(5)
        tracer.trace(name, 'END_SERVICE')

# After simulation
tracer.save('event_log.csv')

def customer(env, name, server, logger):
    arrival = env.now

    with server.request() as req:
        yield req
        wait = env.now - arrival

        # Only log long waits
        if wait > 10:
            logger.warning(f"{name} waited {wait:.1f} (threshold exceeded)")

        yield env.timeout(5)

def resource_monitor(env, resource, logger, interval=10):
    """Periodically log resource state."""
    while True:
        logger.info(
            f"Resource state: "
            f"in_use={resource.count}/{resource.capacity}, "
            f"queue={len(resource.queue)}"
        )
        yield env.timeout(interval)

env.process(resource_monitor(env, server, logger))

import os

DEBUG = os.environ.get('SIM_DEBUG', 'false').lower() == 'true'

logger = logging.getLogger('simulation')
logger.setLevel(logging.DEBUG if DEBUG else logging.INFO)

# Run with: SIM_DEBUG=true python simulation.py

import json
import logging

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'message': record.getMessage(),
        }
        if hasattr(record, 'sim_time'):
            log_record['sim_time'] = record.sim_time
        if hasattr(record, 'entity'):
            log_record['entity'] = record.entity
        return json.dumps(log_record)

# Setup
handler = logging.FileHandler('simulation.jsonl')
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# Usage
logger.info("Customer arrived", extra={'sim_time': env.now, 'entity': 'C1'})

import logging
import sys

def setup_simulation_logging(
    name='simulation',
    console_level=logging.INFO,
    file_level=logging.DEBUG,
    log_file=None
):
    """Configure comprehensive logging for simulation."""
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)
    logger.handlers = []  # Clear existing handlers

    # Console handler
    console = logging.StreamHandler(sys.stdout)
    console.setLevel(console_level)
    console_fmt = logging.Formatter('%(levelname)s: %(message)s')
    console.setFormatter(console_fmt)
    logger.addHandler(console)

    # File handler (if specified)
    if log_file:
        file_handler = logging.FileHandler(log_file, mode='w')
        file_handler.setLevel(file_level)
        file_fmt = logging.Formatter(
            '%(asctime)s | %(levelname)-8s | %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )
        file_handler.setFormatter(file_fmt)
        logger.addHandler(file_handler)

    return logger

# Usage
logger = setup_simulation_logging(
    console_level=logging.WARNING,  # Only warnings to console
    file_level=logging.DEBUG,       # Everything to file
    log_file='simulation_debug.log'
)