import logging
import os
from pprint import pformat
import numpy as np
from rich.logging import RichHandler
from rich.console import Console
import simpy
from sim_tools.distributions import ExponentialLogging
Learning objectives:
- Learn how to set up alternative logs for your model.
- Add parameter list to your log.
- Understand how to save the log to a file.
Relevant reproducibility guidelines:
- NHS Levels of RAP (🥈): Logs are automatically recorded by the pipeline to ensure outputs are as expected.
Pre-reading:
This page continues on from: Entity processing.
Entity generation → Entity processing → Logging
Required packages:
These should be available from environment setup in the “Test yourself” section of Environments.
library(simmer)In the basic model we built on the previous pages, we printed out the arrival and consultation times as the simulation ran.
In the basic simmer model we built on the previous pages, we ran the simulation with verbose = TRUE and looked at the logs output by simmer.
This allowed us to check that the model is working as expected. It is important to check this when building a simulation, so you can have confidence that events are coded correctly - e.g., that arrivals happen at the right time, that patients get seen in the right order by the right people.
Instead of print statements, this page shows you how to make more detailed visual logs, with functionality including saving them to a file.
The default simmer logs are useful but can be quite confusing to use. Instead, we can make custom logs that are easier to read and interpret.
Logging should be optional. It’s best to use them while developing and testing the model with short runs and no replications. They will be excessive and slow down the simulation if used for long runs or multiple replications.
Setting up alternative logs for the model
Parameter class
No changes required.
Patient class
No changes required.
Logging class
We create a new class SimLogger for logging, which is explained line-by-line below.
class SimLogger:
"""
Provides log of events as the simulation runs.
Attributes
----------
verbose : bool
Whether to print messages as simulation runs.
"""
def __init__(self, verbose):
"""
Initialise logging class.
Parameters
----------
verbose : bool
Whether to print messages as simulation runs.
"""
self.verbose = verbose
if self.verbose:
self.logger = logging.getLogger(__name__)
self._configure_logging()
def _configure_logging(self):
"""
Configure the logger.
"""
# Ensure any existing handlers are removed to avoid duplication
for handler in self.logger.handlers[:]:
self.logger.removeHandler(handler)
# Configure RichHandler without INFO/ERROR labels, times or paths
# to log message. Set up console with jupyter-specific behaviour
# disabled to prevent large gaps between each log message on ipynb.
console = Console()
console.is_jupyter = False
rich_handler = RichHandler(
console=console, show_time=False,
show_level=False, show_path=False
)
# Add handler to the logger
self.logger.addHandler(rich_handler)
# Set logging level and format. If don't set level info, it would
# only show log messages which are warning, error or critical.
self.logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(message)s")
rich_handler.setFormatter(formatter)
def log(self, msg, sim_time=None):
"""
Log a message if logging is enabled.
Parameters
----------
msg : str
Message to log.
sim_time : float or None, optional
Current simulation time. If provided, prints before message.
"""
if self.verbose:
if sim_time is not None:
self.logger.info("%0.3f: %s", sim_time, msg)
else:
self.logger.info(msg)Explaining SimLogger
class SimLogger:
"""
Provides log of events as the simulation runs.
Attributes
----------
verbose : bool
Whether to print messages as simulation runs.
"""
def __init__(self, verbose):
"""
Initialise logging class.
Parameters
----------
verbose : bool
Whether to print messages as simulation runs.
"""
self.verbose = verbose
if self.verbose:
self.logger = logging.getLogger(__name__)
self._configure_logging()If verbose is True, the class configures a logger using Python’s built-in logging module, then calls our _configure_logging method.
If verbose is False, the logger is not set up and later calls to the log() method will not produce an output.
def _configure_logging(self):
"""
Configure the logger.
"""
# Ensure any existing handlers are removed to avoid duplication
for handler in self.logger.handlers[:]:
self.logger.removeHandler(handler)The _configure_logging method begins _, a Python convention which indicates that it is a method for internal use within the class (i.e., users wouldn’t call the method directly).
This method starts by removing any existing logging handlers. This is because Python loggers are persistent and global, so if you create multiple SimLogger instances or re-run code in an interactive environment like Jupyter, each new handler would be add to the logger, causing duplicate logging messages.
# Configure RichHandler without INFO/ERROR labels, times or paths
# to log message. Set up console with jupyter-specific behaviour
# disabled to prevent large gaps between each log message on ipynb.
console = Console()
console.is_jupyter = False
rich_handler = RichHandler(
console=console, show_time=False,
show_level=False, show_path=False
)A RichHandler is created for styled, readable messages in the console. The handler’s settings are adjusted so that log messages are clear and compact: labels for log level, timestamps, and file paths are hidden to keep each line focused on the actual message.
Setting console.is_jupyter = False ensures that when working in a Jupyter notebook, the log lines are formatted without large gaps between them.
# Add handler to the logger
self.logger.addHandler(rich_handler)The prepared handler is then added to the logger, which means all log output from this logger will be processed and shown using the RichHandler’s formatting rules.
# Set logging level and format. If don't set level info, it would
# only show log messages which are warning, error or critical.
self.logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(message)s")
rich_handler.setFormatter(formatter)The logger’s level is set to INFO, ensuring that our simulation logging messages (which use logger.info) are shown. If you do not set the level, the default is WARNING, which means only warnings and severe messages woudl appear.
def log(self, msg, sim_time=None):
"""
Log a message if logging is enabled.
Parameters
----------
msg : str
Message to log.
sim_time : float or None, optional
Current simulation time. If provided, prints before message.
"""
if self.verbose:
if sim_time is not None:
self.logger.info("%0.3f: %s", sim_time, msg)
else:
self.logger.info(msg)The log() method is used to record messages as the simulation runs. If verbose is True, this method will feed the provided message to the logger.
Model class
In Model, a new attribute logger stores as instance of SimLogger.
Where previously we had statements “if verbose print…”, these are replace with self.param.log().
class Model:
"""
Simulation model.
Attributes
----------
param : Parameters
Simulation parameters.
run_number : int
Run number for random seed generation.
env : simpy.Environment
The SimPy environment for the simulation.
doctor : simpy.Resource
SimPy resource representing doctors.
patients : list
List of Patient objects.
arrival_dist : Exponential
Distribution used to generate random patient inter-arrival times.
consult_dist : Exponential
Distribution used to generate length of a doctor's consultation.
logger : SimLogger
The logging instance used for logging messages.
"""
def __init__(self, param, run_number):
"""
Create a new Model instance.
Parameters
----------
param : Parameters
Simulation parameters.
run_number : int
Run number for random seed generation.
"""
self.param = param
self.run_number = run_number
# Create SimPy environment
self.env = simpy.Environment()
# Create resource
self.doctor = simpy.Resource(
self.env, capacity=self.param.number_of_doctors
)
# Create a random seed sequence based on the run number
ss = np.random.SeedSequence(self.run_number)
seeds = ss.spawn(2)
# Set up attributes to store results
self.patients = []
# Initialise distributions
self.arrival_dist = Exponential(mean=self.param.interarrival_time,
random_seed=seeds[0])
self.consult_dist = Exponential(mean=self.param.consultation_time,
random_seed=seeds[1])
# Initialise logger
self.logger = SimLogger(verbose=self.param.verbose)
def generate_arrivals(self):
"""
Process that generates patient arrivals.
"""
while True:
# Sample and pass time to next arrival
sampled_iat = self.arrival_dist.sample()
yield self.env.timeout(sampled_iat)
# Create a new patient
patient = Patient(patient_id=len(self.patients)+1,
arrival_time=self.env.now)
self.patients.append(patient)
# Log arrival time
self.logger.log(msg=f"Patient {patient.patient_id} arrives.",
sim_time=self.env.now)
# Start process of consultation
self.env.process(self.consultation(patient))
def consultation(self, patient):
"""
Process that simulates a consultation.
Parameters
----------
patient :
Instance of the Patient() class representing a single patient.
"""
# Patient requests access to a doctor (resource)
with self.doctor.request() as req:
yield req
# Log consultation start time
self.logger.log(
msg=f"Patient {patient.patient_id} starts consultation.",
sim_time=self.env.now
)
# Sample consultation duration and pass time spent with doctor
time_with_doctor = self.consult_dist.sample()
yield self.env.timeout(time_with_doctor)
def run(self):
"""
Run the simulation.
"""
# Schedule arrival generator
self.env.process(self.generate_arrivals())
# Run the simulation
self.env.run(until=self.param.run_length)Run the model
param = Parameters()
model = Model(param=param, run_number=0)
model.run()16.468: Patient 1 arrives.
16.468: Patient 1 starts consultation.
20.283: Patient 2 arrives.
20.283: Patient 2 starts consultation.
26.545: Patient 3 arrives.
26.545: Patient 3 starts consultation.
27.675: Patient 4 arrives.
27.675: Patient 4 starts consultation.
28.779: Patient 5 arrives.
28.779: Patient 5 starts consultation.
37.778: Patient 6 arrives.
37.778: Patient 6 starts consultation.
38.108: Patient 7 arrives.
38.108: Patient 7 starts consultation.
42.611: Patient 8 arrives.
44.088: Patient 9 arrives.
47.598: Patient 8 starts consultation. If we run the model with verbose=False in parameters, no logs are printed:
param = Parameters(verbose=False)
model = Model(param=param, run_number=0)
model.run()Parameter function
No changes required.
Model function
The required changes are explained line-by-line below:
#' Run simulation.
#'
#' @param param List. Model parameters.
#' @param run_number Numeric. Run number for random seed generation.
#'
#' @importFrom simmer add_generator add_resource get_attribute log_ release run
#' @importFrom simmer seize set_attribute simmer timeout trajectory
#'
#' @export
model <- function(param, run_number) {
# Set random seed based on run number
set.seed(run_number)
# Create simmer environment
env <- simmer("simulation", verbose = FALSE,
log_level = if (param[["verbose"]]) 1L else 0L)
# Define the patient trajectory
patient <- trajectory("consultation") |>
log_("🚶 Arrives.", level = 1L) |>
seize("doctor", 1L) |>
# Record sampled length of consultation
set_attribute("doctor_serve_length", function() {
rexp(n = 1L, rate = 1L / param[["consultation_time"]])
}) |>
log_(function() {
paste0("🩺 Doctor consultation begins (length: ",
round(get_attribute(env, "doctor_serve_length"), 5L), ")")
}, level = 1L) |>
timeout(function() get_attribute(env, "doctor_serve_length")) |>
release("doctor", 1L) |>
log_("🚪 Leaves.", level = 1L)
env <- env |>
# Add doctor resource
add_resource("doctor", param[["number_of_doctors"]]) |>
# Add patient generator
add_generator("patient", patient, function() {
rexp(n = 1L, rate = 1L / param[["interarrival_time"]])
}) |>
# Run the simulation
simmer::run(until = param[["run_length"]])
}Explaining changes to model()
env <- simmer("simulation", verbose = FALSE,
log_level = if (param[["verbose"]]) 1L else 0L)Simmer’s default logging is disabled by setting verbose = FALSE in simmer::simmer().
The display of our custom log messages is then managed using the log_level argument in simmer::simmer().
log_("🚶 Arrives.", level = 1L) |>Custom log messages are created using simmer’s log_() function, which allows us to insert our own messages into the simulation trajectory.
Each message can be assigned a level. We’ve set level = 1, meaning it will only be displayed if the simulation’s log_level is greater than or equal to 1.
# Record sampled length of consultation
set_attribute("doctor_serve_length", function() {
rexp(n = 1L, rate = 1L / param[["consultation_time"]])
}) |>
log_(function() {
paste0("🩺 Doctor consultation begins (length: ",
round(get_attribute(env, "doctor_serve_length"), 5L), ")")
}, level = 1L) |>
timeout(function() get_attribute(env, "doctor_serve_length")) |>We need to sample the length of the consultation within the trajectory and use this value in two places: within log_() for reporting and in timeout() to control the duration.
To achieve this, we use simmer’s get_attribute() to generate and store the value as an attribute. We then access this value later using get_attribute().
log_("🚪 Leaves.", level = 1L)Finally, we create a log message to record when the patient leaves.
Run the model
param <- create_params()
model(param = param, run_number = 1L)3.77591: patient0: 🚶 Arrives.
3.77591: patient0: 🩺 Doctor consultation begins (length: 11.81643)
4.50444: patient1: 🚶 Arrives.
4.50444: patient1: 🩺 Doctor consultation begins (length: 1.39795)
5.9024: patient1: 🚪 Leaves.
6.68479: patient2: 🚶 Arrives.
6.68479: patient2: 🩺 Doctor consultation begins (length: 28.94969)
12.8326: patient3: 🚶 Arrives.
12.8326: patient3: 🩺 Doctor consultation begins (length: 5.39683)
15.5923: patient0: 🚪 Leaves.
17.6154: patient4: 🚶 Arrives.
17.6154: patient4: 🩺 Doctor consultation begins (length: 1.47046)
18.2294: patient3: 🚪 Leaves.
19.0859: patient4: 🚪 Leaves.
24.5691: patient5: 🚶 Arrives.
24.5691: patient5: 🩺 Doctor consultation begins (length: 7.6203)
30.7571: patient6: 🚶 Arrives.
30.7571: patient6: 🩺 Doctor consultation begins (length: 44.23934)
32.1894: patient5: 🚪 Leaves.
35.6345: patient2: 🚪 Leaves.
36.0298: patient7: 🚶 Arrives.
36.0298: patient7: 🩺 Doctor consultation begins (length: 10.35244)
45.41: patient8: 🚶 Arrives.
45.41: patient8: 🩺 Doctor consultation begins (length: 6.54747)
46.3823: patient7: 🚪 Leaves.
47.0947: patient9: 🚶 Arrives.
47.0947: patient9: 🩺 Doctor consultation begins (length: 5.8848)If we run the model with verbose = FALSE in parameters, no logs are printed:
param <- create_params(verbose = FALSE)
model(param = param, run_number = 1L)Adding parameters to the logs
As mentioned on the Parameters from script page, one of the benefits of storing parameters in a single object is that we can easily print a copy of all parameters used for a scenario by fetching the object.
Parameter class
No changes required.
Patient class
No changes required.
Logging class
We add a new method sanitise_object() to SimLogger which tidies up the provided parameter object so it is easily readable and doesn’t contain memory addresses.
In Python, when you print objects directly, the log output often shows their memory address (e.g., <simpy.core.Environment object at 0x752feb454d50>), which is a reference to where the object is stored in the computer’s memory and not relevant for us to include in our log.
The sanitise_object() method is applied in log() for when the provided msg is a dictionary.
class SimLogger:
"""
Provides log of events as the simulation runs.
Attributes
----------
verbose : bool
Whether to print messages as simulation runs.
"""
def __init__(self, verbose):
"""
Initialise logging class.
Parameters
----------
verbose : bool
Whether to print messages as simulation runs.
"""
self.verbose = verbose
if self.verbose:
self.logger = logging.getLogger(__name__)
self._configure_logging()
def _configure_logging(self):
"""
Configure the logger.
"""
# Ensure any existing handlers are removed to avoid duplication
for handler in self.logger.handlers[:]:
self.logger.removeHandler(handler)
# Configure RichHandler without INFO/ERROR labels, times or paths
# to log message. Set up console with jupyter-specific behaviour
# disabled to prevent large gaps between each log message on ipynb.
console = Console()
console.is_jupyter = False
rich_handler = RichHandler(
console=console, show_time=False,
show_level=False, show_path=False
)
# Add handler to the logger
self.logger.addHandler(rich_handler)
# Set logging level and format. If don't set level info, it would
# only show log messages which are warning, error or critical.
self.logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(message)s")
rich_handler.setFormatter(formatter)
def sanitise_object(self, obj):
"""
Sanitise object references to avoid memory addresses in logs.
Parameters
----------
obj : object
Object to sanitise.
Returns
-------
str
Sanitised version of the object. For basic types (int, float,
bool, str, list, dict, tuple, set), returns the string
representation. For other objects, returns the class name in
angle brackets.
"""
if isinstance(obj, object) and not isinstance(
obj, (int, float, bool, str, list, dict, tuple, set)
):
# Return the class name instead of the memory address
return f"<{obj.__class__.__module__}.{obj.__class__.__name__}>"
return obj
def log(self, msg, sim_time=None):
"""
Log a message if logging is enabled.
Parameters
----------
msg : str
Message to log.
sim_time : float or None, optional
Current simulation time. If provided, prints before message.
"""
# Format dictionaries so easier to read
if isinstance(msg, dict):
msg = {key: self.sanitise_object(value)
for key, value in msg.items()}
msg = pformat(msg, indent=4)
if self.verbose:
if sim_time is not None:
self.logger.info("%0.3f: %s", sim_time, msg)
else:
self.logger.info(msg)Model class
Within Model.__init__(), we now provide the attributes of Model and our parameter object to the log.
class Model:
"""
Simulation model.
Attributes
----------
param : Parameters
Simulation parameters.
run_number : int
Run number for random seed generation.
env : simpy.Environment
The SimPy environment for the simulation.
doctor : simpy.Resource
SimPy resource representing doctors.
patients : list
List of Patient objects.
arrival_dist : Exponential
Distribution used to generate random patient inter-arrival times.
consult_dist : Exponential
Distribution used to generate length of a doctor's consultation.
logger : SimLogger
The logging instance used for logging messages.
"""
def __init__(self, param, run_number):
"""
Create a new Model instance.
Parameters
----------
param : Parameters
Simulation parameters.
run_number : int
Run number for random seed generation.
"""
self.param = param
self.run_number = run_number
# Create SimPy environment
self.env = simpy.Environment()
# Create resource
self.doctor = simpy.Resource(
self.env, capacity=self.param.number_of_doctors
)
# Create a random seed sequence based on the run number
ss = np.random.SeedSequence(self.run_number)
seeds = ss.spawn(2)
# Set up attributes to store results
self.patients = []
# Initialise distributions
self.arrival_dist = Exponential(mean=self.param.interarrival_time,
random_seed=seeds[0])
self.consult_dist = Exponential(mean=self.param.consultation_time,
random_seed=seeds[1])
# Initialise logger
self.logger = SimLogger(verbose=self.param.verbose)
self.logger.log(sim_time=self.env.now, msg="Initialise model:")
self.logger.log(vars(self))
self.logger.log(sim_time=self.env.now, msg="Parameters:")
self.logger.log(vars(self.param))
def generate_arrivals(self):
"""
Process that generates patient arrivals.
"""
while True:
# Sample and pass time to next arrival
sampled_iat = self.arrival_dist.sample()
yield self.env.timeout(sampled_iat)
# Create a new patient
patient = Patient(patient_id=len(self.patients)+1,
arrival_time=self.env.now)
self.patients.append(patient)
# Log arrival time
self.logger.log(msg=f"Patient {patient.patient_id} arrives.",
sim_time=self.env.now)
# Start process of consultation
self.env.process(self.consultation(patient))
def consultation(self, patient):
"""
Process that simulates a consultation.
Parameters
----------
patient :
Instance of the Patient() class representing a single patient.
"""
# Patient requests access to a doctor (resource)
with self.doctor.request() as req:
yield req
# Log consultation start time
self.logger.log(
msg=f"Patient {patient.patient_id} starts consultation.",
sim_time=self.env.now
)
# Sample consultation duration and pass time spent with doctor
time_with_doctor = self.consult_dist.sample()
yield self.env.timeout(time_with_doctor)
def run(self):
"""
Run the simulation.
"""
# Schedule arrival generator
self.env.process(self.generate_arrivals())
# Run the simulation
self.env.run(until=self.param.run_length)Run the model
param = Parameters()
model = Model(param=param, run_number=0)0.000: Initialise model:
{ 'arrival_dist': '<sim_tools.distributions.Exponential>',
'consult_dist': '<sim_tools.distributions.Exponential>',
'doctor': '<simpy.resources.resource.Resource>',
'env': '<simpy.core.Environment>',
'logger': '<__main__.SimLogger>',
'param': '<__main__.Parameters>',
'patients': [],
'run_number': 0}
0.000: Parameters:
{ 'consultation_time': 10,
'interarrival_time': 5,
'number_of_doctors': 3,
'run_length': 50,
'verbose': True} model.run()16.468: Patient 1 arrives.
16.468: Patient 1 starts consultation.
20.283: Patient 2 arrives.
20.283: Patient 2 starts consultation.
26.545: Patient 3 arrives.
26.545: Patient 3 starts consultation.
27.675: Patient 4 arrives.
27.675: Patient 4 starts consultation.
28.779: Patient 5 arrives.
28.779: Patient 5 starts consultation.
37.778: Patient 6 arrives.
37.778: Patient 6 starts consultation.
38.108: Patient 7 arrives.
38.108: Patient 7 starts consultation.
42.611: Patient 8 arrives.
44.088: Patient 9 arrives.
47.598: Patient 8 starts consultation. Parameter function
No changes required.
Model function
The log_() function is designed to be used within simmer trajectories. To record the model’s parameter set as part of the simmer log, we create a simple trajectory param_traj containing a single log_() action. Here, the parameters are converted into string format and logged.
This is add to the simulation environment using add_generator() with at(0) ensuring that it just triggers once at the very start of the simulation.
#' Run simulation.
#'
#' @param param List. Model parameters.
#' @param run_number Numeric. Run number for random seed generation.
#'
#' @importFrom simmer add_generator add_resource get_attribute log_ release run
#' @importFrom simmer seize set_attribute simmer timeout trajectory
#'
#' @export
model <- function(param, run_number) {
# Set random seed based on run number
set.seed(run_number)
# Create simmer environment
env <- simmer("simulation", verbose = FALSE,
log_level = if (param[["verbose"]]) 1L else 0L)
# Add blank trajectory used to add parameters to the log
param_string <- paste(names(param), param, sep = "=", collapse = ";\n ")
param_traj <- trajectory() |>
log_(param_string, level = 1L)
# Define the patient trajectory
patient <- trajectory("consultation") |>
log_("🚶 Arrives.", level = 1L) |>
seize("doctor", 1L) |>
# Record sampled length of consultation
set_attribute("doctor_serve_length", function() {
rexp(n = 1L, rate = 1L / param[["consultation_time"]])
}) |>
log_(function() {
paste0("🩺 Doctor consultation begins (length: ",
round(get_attribute(env, "doctor_serve_length"), 5L), ")")
}, level = 1L) |>
timeout(function() get_attribute(env, "doctor_serve_length")) |>
release("doctor", 1L) |>
log_("🚪 Leaves.", level = 1L)
env <- env |>
# Add parameter log
add_generator("Parameters", param_traj, at(0L)) |>
# Add doctor resource
add_resource("doctor", param[["number_of_doctors"]]) |>
# Add patient generator
add_generator("patient", patient, function() {
rexp(n = 1L, rate = 1L / param[["interarrival_time"]])
}) |>
# Run the simulation
simmer::run(until = param[["run_length"]])
}Run the model
param <- create_params()
model(param = param, run_number = 1L)0: Parameters0: interarrival_time=5;
consultation_time=10;
number_of_doctors=3;
run_length=50;
verbose=TRUE
3.77591: patient0: 🚶 Arrives.
3.77591: patient0: 🩺 Doctor consultation begins (length: 11.81643)
4.50444: patient1: 🚶 Arrives.
4.50444: patient1: 🩺 Doctor consultation begins (length: 1.39795)
5.9024: patient1: 🚪 Leaves.
6.68479: patient2: 🚶 Arrives.
6.68479: patient2: 🩺 Doctor consultation begins (length: 28.94969)
12.8326: patient3: 🚶 Arrives.
12.8326: patient3: 🩺 Doctor consultation begins (length: 5.39683)
15.5923: patient0: 🚪 Leaves.
17.6154: patient4: 🚶 Arrives.
17.6154: patient4: 🩺 Doctor consultation begins (length: 1.47046)
18.2294: patient3: 🚪 Leaves.
19.0859: patient4: 🚪 Leaves.
24.5691: patient5: 🚶 Arrives.
24.5691: patient5: 🩺 Doctor consultation begins (length: 7.6203)
30.7571: patient6: 🚶 Arrives.
30.7571: patient6: 🩺 Doctor consultation begins (length: 44.23934)
32.1894: patient5: 🚪 Leaves.
35.6345: patient2: 🚪 Leaves.
36.0298: patient7: 🚶 Arrives.
36.0298: patient7: 🩺 Doctor consultation begins (length: 10.35244)
45.41: patient8: 🚶 Arrives.
45.41: patient8: 🩺 Doctor consultation begins (length: 6.54747)
46.3823: patient7: 🚪 Leaves.
47.0947: patient9: 🚶 Arrives.
47.0947: patient9: 🩺 Doctor consultation begins (length: 5.8848)Saving logs to a file
Saving simulation logs to a file is very handy for reproducibility because it creates a permanent, detailed record of exactly what happened during the simulation run. This can be useful for debugging, peer review and similar.
Parameter class
Our argument verbose has been replace by three new arguments: log_to_console, log_to_file and file_path…
class Parameters:
"""
Parameter class.
Attributes
----------
interarrival_time : float
Mean time between arrivals (minutes).
consultation_time : float
Mean length of doctor's consultation (minutes).
number_of_doctors : int
Number of doctors.
run_length : int
Total duration of simulation (minutes).
log_to_console : bool
Whether to print activity log to console.
log_to_file : bool
Whether to save activity log to file.
file_path : str
Path to save log to file.
"""
def __init__(
self, interarrival_time=5, consultation_time=10,
number_of_doctors=3, run_length=50,
log_to_console=False, log_to_file=False, file_path=None
):
"""
Initialise Parameters instance.
Parameters
----------
interarrival_time : float
Time between arrivals (minutes).
consultation_time : float
Length of consultation (minutes).
number_of_doctors : int
Number of doctors.
run_length : int
Total duration of simulation (minutes).
log_to_console : bool
Whether to print activity log to console.
log_to_file : bool
Whether to save activity log to file.
file_path : str
Path to save log to file.
"""
self.interarrival_time = interarrival_time
self.consultation_time = consultation_time
self.number_of_doctors = number_of_doctors
self.run_length = run_length
self.log_to_console = log_to_console
self.log_to_file = log_to_file
self.file_path = file_pathPatient class
No changes required.
Logging class
The new parameter arguments log_to_console, log_to_file and file_path are passed to the SimLogger class. Where previously the class checked if verbose was true, it now checks whether log to console or file are true.
The class has a new _validate_log_path() which checks that the provided path is suitable (i.e., it exists and the filename ends with .log).
Within _configure_logging(), it either adds:
- The
RichHandler(as before) iflog_to_consoleis True. - A
FileHandleriflog_to_fileis True.
These are both add to logger and their logging level and format are set.
class SimLogger:
"""
Provides log of events as the simulation runs.
Attributes
----------
log_to_console : bool
Whether to print activity log to console.
log_to_file : bool
Whether to save activity log to file.
file_path : str
Path to save log to file.
"""
def __init__(self, log_to_console, log_to_file, file_path):
"""
Initialise logging class.
Parameters
----------
log_to_console : bool
Whether to print activity log to console.
log_to_file : bool
Whether to save activity log to file.
file_path : str
Path to save log to file.
"""
self.log_to_console = log_to_console
self.log_to_file = log_to_file
self.file_path = file_path
# If saving to file, check path is valid
if self.log_to_file:
self._validate_log_path()
if self.log_to_console or self.log_to_file:
self.logger = logging.getLogger(__name__)
self._configure_logging()
def _validate_log_path(self):
"""
Validate the log file path.
Raises
------
ValueError
If log path is invalid.
"""
directory = os.path.dirname(self.file_path)
if directory and not os.path.exists(directory):
raise ValueError(
f"The directory '{directory}' for the log file does not exist."
)
if not self.file_path.endswith(".log"):
raise ValueError(
f"The log file path '{self.file_path}' must end with '.log'."
)
def _configure_logging(self):
"""
Configure the logger.
"""
# Ensure any existing handlers are removed to avoid duplication
for handler in self.logger.handlers[:]:
self.logger.removeHandler(handler)
# Configure RichHandler without INFO/ERROR labels, times or paths
# to log message. Set up console with jupyter-specific behaviour
# disabled to prevent large gaps between each log message on ipynb.
console = Console()
console.is_jupyter = False
rich_handler = RichHandler(
console=console, show_time=False,
show_level=False, show_path=False
)
# Add handlers to the logger
handlers = []
if self.log_to_console:
handlers.append(rich_handler)
if self.log_to_file:
handlers.append(logging.FileHandler(self.file_path, mode="w"))
for handler in handlers:
self.logger.addHandler(handler)
# Set logging level and format. If don't set level info, it would
# only show log messages which are warning, error or critical.
self.logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(message)s")
for handler in handlers:
handler.setFormatter(formatter)
def sanitise_object(self, obj):
"""
Sanitise object references to avoid memory addresses in logs.
Parameters
----------
obj : object
Object to sanitise.
Returns
-------
str
Sanitised version of the object. For basic types (int, float,
bool, str, list, dict, tuple, set), returns the string
representation. For other objects, returns the class name in
angle brackets.
"""
if isinstance(obj, object) and not isinstance(
obj, (int, float, bool, str, list, dict, tuple, set)
):
# Return the class name instead of the memory address
return f"<{obj.__class__.__module__}.{obj.__class__.__name__}>"
return obj
def log(self, msg, sim_time=None):
"""
Log a message if logging is enabled.
Parameters
----------
msg : str
Message to log.
sim_time : float or None, optional
Current simulation time. If provided, prints before message.
"""
# Format dictionaries so easier to read
if isinstance(msg, dict):
msg = {key: self.sanitise_object(value)
for key, value in msg.items()}
msg = pformat(msg, indent=4)
if self.log_to_console or self.log_to_file:
if sim_time is not None:
self.logger.info("%0.3f: %s", sim_time, msg)
else:
self.logger.info(msg)Model class
The new Parameter attributes (log_to_console, log_to_file, file_path) are passed to the logger.
class Model:
"""
Simulation model.
Attributes
----------
param : Parameters
Simulation parameters.
run_number : int
Run number for random seed generation.
env : simpy.Environment
The SimPy environment for the simulation.
doctor : simpy.Resource
SimPy resource representing doctors.
patients : list
List of Patient objects.
arrival_dist : Exponential
Distribution used to generate random patient inter-arrival times.
consult_dist : Exponential
Distribution used to generate length of a doctor's consultation.
logger : SimLogger
The logging instance used for logging messages.
"""
def __init__(self, param, run_number):
"""
Create a new Model instance.
Parameters
----------
param : Parameters
Simulation parameters.
run_number : int
Run number for random seed generation.
"""
self.param = param
self.run_number = run_number
# Create SimPy environment
self.env = simpy.Environment()
# Create resource
self.doctor = simpy.Resource(
self.env, capacity=self.param.number_of_doctors
)
# Create a random seed sequence based on the run number
ss = np.random.SeedSequence(self.run_number)
seeds = ss.spawn(2)
# Set up attributes to store results
self.patients = []
# Initialise distributions
self.arrival_dist = Exponential(mean=self.param.interarrival_time,
random_seed=seeds[0])
self.consult_dist = Exponential(mean=self.param.consultation_time,
random_seed=seeds[1])
# Initialise logger
self.logger = SimLogger(log_to_console=self.param.log_to_console,
log_to_file=self.param.log_to_file,
file_path=self.param.file_path)
self.logger.log(sim_time=self.env.now, msg="Initialise model:")
self.logger.log(vars(self))
self.logger.log(sim_time=self.env.now, msg="Parameters:")
self.logger.log(vars(self.param))
def generate_arrivals(self):
"""
Process that generates patient arrivals.
"""
while True:
# Sample and pass time to next arrival
sampled_iat = self.arrival_dist.sample()
yield self.env.timeout(sampled_iat)
# Create a new patient
patient = Patient(patient_id=len(self.patients)+1,
arrival_time=self.env.now)
self.patients.append(patient)
# Log arrival time
self.logger.log(msg=f"Patient {patient.patient_id} arrives.",
sim_time=self.env.now)
# Start process of consultation
self.env.process(self.consultation(patient))
def consultation(self, patient):
"""
Process that simulates a consultation.
Parameters
----------
patient :
Instance of the Patient() class representing a single patient.
"""
# Patient requests access to a doctor (resource)
with self.doctor.request() as req:
yield req
# Log consultation start time
self.logger.log(
msg=f"Patient {patient.patient_id} starts consultation.",
sim_time=self.env.now
)
# Sample consultation duration and pass time spent with doctor
time_with_doctor = self.consult_dist.sample()
yield self.env.timeout(time_with_doctor)
def run(self):
"""
Run the simulation.
"""
# Schedule arrival generator
self.env.process(self.generate_arrivals())
# Run the simulation
self.env.run(until=self.param.run_length)Run the model
With log_to_console and log_to_file both set to False, no logs are generated.
param = Parameters()
model = Model(param=param, run_number=0)
model.run()If we set log_to_console=True, the log is printed as before.
param = Parameters(log_to_console=True)
model = Model(param=param, run_number=0)0.000: Initialise model:
{ 'arrival_dist': '<sim_tools.distributions.Exponential>',
'consult_dist': '<sim_tools.distributions.Exponential>',
'doctor': '<simpy.resources.resource.Resource>',
'env': '<simpy.core.Environment>',
'logger': '<__main__.SimLogger>',
'param': '<__main__.Parameters>',
'patients': [],
'run_number': 0}
0.000: Parameters:
{ 'consultation_time': 10,
'file_path': '<builtins.NoneType>',
'interarrival_time': 5,
'log_to_console': True,
'log_to_file': False,
'number_of_doctors': 3,
'run_length': 50} model.run()16.468: Patient 1 arrives.
16.468: Patient 1 starts consultation.
20.283: Patient 2 arrives.
20.283: Patient 2 starts consultation.
26.545: Patient 3 arrives.
26.545: Patient 3 starts consultation.
27.675: Patient 4 arrives.
27.675: Patient 4 starts consultation.
28.779: Patient 5 arrives.
28.779: Patient 5 starts consultation.
37.778: Patient 6 arrives.
37.778: Patient 6 starts consultation.
38.108: Patient 7 arrives.
38.108: Patient 7 starts consultation.
42.611: Patient 8 arrives.
44.088: Patient 9 arrives.
47.598: Patient 8 starts consultation. If we set log_to_file=True and provide a file_path, a file is generated containing the log output.
log_file = os.path.join("logs_resources", "python_log_example.log")
param = Parameters(log_to_file=True, file_path=log_file)
model = Model(param=param, run_number=0)
model.run()If we import the log file, we’ll see it contains the same output.
# Read all lines from the file
with open(log_file, "r", encoding="utf-8") as file:
log_contents = file.readlines()
# Print each line
for line in log_contents:
print(line, end="")0.000: Initialise model:
{ 'arrival_dist': '<sim_tools.distributions.Exponential>',
'consult_dist': '<sim_tools.distributions.Exponential>',
'doctor': '<simpy.resources.resource.Resource>',
'env': '<simpy.core.Environment>',
'logger': '<__main__.SimLogger>',
'param': '<__main__.Parameters>',
'patients': [],
'run_number': 0}
0.000: Parameters:
{ 'consultation_time': 10,
'file_path': 'logs_resources/python_log_example.log',
'interarrival_time': 5,
'log_to_console': False,
'log_to_file': True,
'number_of_doctors': 3,
'run_length': 50}
16.468: Patient 1 arrives.
16.468: Patient 1 starts consultation.
20.283: Patient 2 arrives.
20.283: Patient 2 starts consultation.
26.545: Patient 3 arrives.
26.545: Patient 3 starts consultation.
27.675: Patient 4 arrives.
27.675: Patient 4 starts consultation.
28.779: Patient 5 arrives.
28.779: Patient 5 starts consultation.
37.778: Patient 6 arrives.
37.778: Patient 6 starts consultation.
38.108: Patient 7 arrives.
38.108: Patient 7 starts consultation.
42.611: Patient 8 arrives.
44.088: Patient 9 arrives.
47.598: Patient 8 starts consultation.Parameter function
Our argument verbose has been replace by three new arguments: log_to_console, log_to_file and file_path…
#' Generate parameter list.
#'
#' @param interarrival_time Numeric. Time between arrivals (minutes).
#' @param consultation_time Numeric. Mean length of doctor's
#' consultation (minutes).
#' @param number_of_doctors Numeric. Number of doctors.
#' @param run_length Numeric. Total duration of simulation (minutes).
#' @param log_to_console Whether to print activity log to console.
#' @param log_to_file Whether to save activity log to file.
#' @param file_path Path to save log to file.
#'
#' @return A named list of parameters.
#' @export
create_params <- function(
interarrival_time = 5L,
consultation_time = 10L,
number_of_doctors = 3L,
run_length = 50L,
log_to_console = FALSE,
log_to_file = FALSE,
file_path = NULL
) {
list(
interarrival_time = interarrival_time,
consultation_time = consultation_time,
number_of_doctors = number_of_doctors,
run_length = run_length,
log_to_console = log_to_console,
log_to_file = log_to_file,
file_path = file_path
)
}Model function
The model() function now determines whether verbose is True based on log_to_console and log_to_file (checking if either are True).
So that we are able to save the log messages, we use capture.output to store the logging outputs when our simulation is run into an object called sim_log. This can then be printed to the console or saved to a file.
#' Run simulation.
#'
#' @param param List. Model parameters.
#' @param run_number Numeric. Run number for random seed generation.
#'
#' @importFrom simmer add_generator add_resource get_attribute log_ release run
#' @importFrom simmer seize set_attribute simmer timeout trajectory
#'
#' @export
model <- function(param, run_number) {
# Set random seed based on run number
set.seed(run_number)
# Determine whether to get verbose activity logs
verbose <- any(c(param[["log_to_console"]], param[["log_to_file"]]))
# Create simmer environment
env <- simmer("simulation", verbose = FALSE,
log_level = if (verbose) 1L else 0L)
# Add blank trajectory used to add parameters to the log
param_string <- paste(names(param), param, sep = "=", collapse = ";\n ")
param_traj <- trajectory() |>
log_(param_string, level = 1L)
# Define the patient trajectory
patient <- trajectory("consultation") |>
log_("🚶 Arrives.", level = 1L) |>
seize("doctor", 1L) |>
# Record sampled length of consultation
set_attribute("doctor_serve_length", function() {
rexp(n = 1L, rate = 1L / param[["consultation_time"]])
}) |>
log_(function() {
paste0("🩺 Doctor consultation begins (length: ",
round(get_attribute(env, "doctor_serve_length"), 5L), ")")
}, level = 1L) |>
timeout(function() get_attribute(env, "doctor_serve_length")) |>
release("doctor", 1L) |>
log_("🚪 Leaves.", level = 1L)
sim_log <- capture.output(
env <- env |> # nolint
# Add parameter log
add_generator("Parameters", param_traj, at(0L)) |>
# Add doctor resource
add_resource("doctor", param[["number_of_doctors"]]) |>
# Add patient generator
add_generator("patient", patient, function() {
rexp(n = 1L, rate = 1L / param[["interarrival_time"]])
}) |>
# Run the simulation
simmer::run(until = param[["run_length"]])
)
# Print log to console
if (isTRUE(param[["log_to_console"]])) {
print(sim_log)
}
# Save log to file
if (isTRUE(param[["log_to_file"]])) {
writeLines(sim_log, param[["file_path"]])
}
}Run the model
With log_to_console and log_to_file both set to FALSE, no logs are generated.
param <- create_params()
model(param = param, run_number = 1L)If we set log_to_console = TRUE, the log is printed as before.
param <- create_params(log_to_console = TRUE)
model(param = param, run_number = 1L) [1] "0: Parameters0: interarrival_time=5;"
[2] " consultation_time=10;"
[3] " number_of_doctors=3;"
[4] " run_length=50;"
[5] " log_to_console=TRUE;"
[6] " log_to_file=FALSE;"
[7] " file_path=NULL"
[8] "3.77591: patient0: 🚶 Arrives."
[9] "3.77591: patient0: 🩺 Doctor consultation begins (length: 11.81643)"
[10] "4.50444: patient1: 🚶 Arrives."
[11] "4.50444: patient1: 🩺 Doctor consultation begins (length: 1.39795)"
[12] "5.9024: patient1: 🚪 Leaves."
[13] "6.68479: patient2: 🚶 Arrives."
[14] "6.68479: patient2: 🩺 Doctor consultation begins (length: 28.94969)"
[15] "12.8326: patient3: 🚶 Arrives."
[16] "12.8326: patient3: 🩺 Doctor consultation begins (length: 5.39683)"
[17] "15.5923: patient0: 🚪 Leaves."
[18] "17.6154: patient4: 🚶 Arrives."
[19] "17.6154: patient4: 🩺 Doctor consultation begins (length: 1.47046)"
[20] "18.2294: patient3: 🚪 Leaves."
[21] "19.0859: patient4: 🚪 Leaves."
[22] "24.5691: patient5: 🚶 Arrives."
[23] "24.5691: patient5: 🩺 Doctor consultation begins (length: 7.6203)"
[24] "30.7571: patient6: 🚶 Arrives."
[25] "30.7571: patient6: 🩺 Doctor consultation begins (length: 44.23934)"
[26] "32.1894: patient5: 🚪 Leaves."
[27] "35.6345: patient2: 🚪 Leaves."
[28] "36.0298: patient7: 🚶 Arrives."
[29] "36.0298: patient7: 🩺 Doctor consultation begins (length: 10.35244)"
[30] "45.41: patient8: 🚶 Arrives."
[31] "45.41: patient8: 🩺 Doctor consultation begins (length: 6.54747)"
[32] "46.3823: patient7: 🚪 Leaves."
[33] "47.0947: patient9: 🚶 Arrives."
[34] "47.0947: patient9: 🩺 Doctor consultation begins (length: 5.8848)" If we set log_to_file = TRUE and provide a file_path, a file is generated containing the log output.
log_file <- file.path("logs_resources", "r_log_example.log")
param <- create_params(log_to_file = TRUE, file_path = log_file)
model(param = param, run_number = 1L)If we import the log file, we’ll see it contains the same output:
log_contents <- readLines(log_file)
print(log_contents, sep = "\n") [1] "0: Parameters0: interarrival_time=5;"
[2] " consultation_time=10;"
[3] " number_of_doctors=3;"
[4] " run_length=50;"
[5] " log_to_console=FALSE;"
[6] " log_to_file=TRUE;"
[7] " file_path=logs_resources/r_log_example.log"
[8] "3.77591: patient0: 🚶 Arrives."
[9] "3.77591: patient0: 🩺 Doctor consultation begins (length: 11.81643)"
[10] "4.50444: patient1: 🚶 Arrives."
[11] "4.50444: patient1: 🩺 Doctor consultation begins (length: 1.39795)"
[12] "5.9024: patient1: 🚪 Leaves."
[13] "6.68479: patient2: 🚶 Arrives."
[14] "6.68479: patient2: 🩺 Doctor consultation begins (length: 28.94969)"
[15] "12.8326: patient3: 🚶 Arrives."
[16] "12.8326: patient3: 🩺 Doctor consultation begins (length: 5.39683)"
[17] "15.5923: patient0: 🚪 Leaves."
[18] "17.6154: patient4: 🚶 Arrives."
[19] "17.6154: patient4: 🩺 Doctor consultation begins (length: 1.47046)"
[20] "18.2294: patient3: 🚪 Leaves."
[21] "19.0859: patient4: 🚪 Leaves."
[22] "24.5691: patient5: 🚶 Arrives."
[23] "24.5691: patient5: 🩺 Doctor consultation begins (length: 7.6203)"
[24] "30.7571: patient6: 🚶 Arrives."
[25] "30.7571: patient6: 🩺 Doctor consultation begins (length: 44.23934)"
[26] "32.1894: patient5: 🚪 Leaves."
[27] "35.6345: patient2: 🚪 Leaves."
[28] "36.0298: patient7: 🚶 Arrives."
[29] "36.0298: patient7: 🩺 Doctor consultation begins (length: 10.35244)"
[30] "45.41: patient8: 🚶 Arrives."
[31] "45.41: patient8: 🩺 Doctor consultation begins (length: 6.54747)"
[32] "46.3823: patient7: 🚪 Leaves."
[33] "47.0947: patient9: 🚶 Arrives."
[34] "47.0947: patient9: 🩺 Doctor consultation begins (length: 5.8848)" Explore the example models
Nurse visit simulation
Click to visit pydesrap_mms repository
| Key files | notebooks/logs.ipynbsimulation/simlogger.pysimulation/param.pysimulation/model.py |
| What to look for? | Using SimLogger with a similar set-up to above. |
Click to visit rdesrap_mms repository
| Key files | rmarkdown/logs.md |
| What to look for? | Creates custom logs (not with parameters or saved to file - kept simple). |
| Why it matters? | In this example, we’ve kept the model simpler and stuck with the default simmer log message. Instead, custom logs are demonstrated in an rmarkdown file. |
Stroke pathway simulation
Click to visit pydesrap_stroke repository
| Key files | notebooks/logs.ipynbsimulation/logging.pysimulation/parameters.pysimulation/model.py |
| What to look for? | Using SimLogger with a similar set-up to above. |
Click to visit rdesrap_stroke repository
| Key files | rmarkdown/logs.mdR/model.RR/parameters.RR/create_asu_trajectory.RR/create_rehab_trajectory.R |
| What to look for? | Custom logs are implemented like above, but with parameters appended to sim_log rather than being a trajectory and part of log_(). |
| Why it matters? | Custom logs are part of the model for this example. They’re set up a little differently - indeed, there are many ways you could do this! |
Test yourself
Try adding logs to your model. Experiment with different log messages and parameter options, then run your code and observe how the logs reflect your changes at each step.
If you extend your model (for example, with different patient types, destinations, or additional resources), continue making use of logs. Logging is especially valuable during model development - it helps you check that each part of your logic works as intended and makes it easier to troubleshoot issues as your simulation grows in complexity.