Discrete event simulation

Queueing-network primitives.

Building blocks

• Queue(maxlen, name) — FIFO buffer of Entity instances.
• PriorityQueue — min-heap by entity property.
• Resource / ResourcePool — countable concurrency limiters.
• Service(capacity, buffer_size, next_q, default_service_time, rng) — multi-channel server that pulls from its own buffer and forwards finished entities.
• ArrivalGenerator(interarrival, factory, target, rng) — Poisson-by- default source that builds entities and pushes them into a target.
• EntityProperties — typed attribute bag attached to each entity for routing decisions.

RNG distributions

exponential, uniform, normal, deterministic — convenience factories that return zero-arg samplers bound to a numpy.random.Generator.

rng = np.random.default_rng(0)
service_time = sw.exponential(rate=1.0, rng=rng)
service_time()           # callable per draw

Example: M/M/2

See the Quickstart for the full snippet. The recorded outputs render as:

Calendar time axis

Simulation clocks are dimensionless floats. :class:~simweave.core.time_axis.SimTimeAxis maps any simulation time to a real-world calendar date so that plot axes show months and years rather than raw tick numbers.

import simweave as sw

# 1 tick = 1 day, simulation starts 1 January 2027
tax = sw.SimTimeAxis(start="2027-01-01", tick_unit="days")

tax.label(90.0)                    # "2027-04-01"
tax.tick_for_date("2027-07-01")    # 181.0  (schedule events by date)

# Pass to any time-series plot helper
fig = sw.plot_queue_length(recorder, time_axis=tax)

# Or apply after the fact
fig = sw.plot_warehouse_stock(recorder)
tax.apply_to_figure(fig)

Supported tick_unit values: "seconds", "minutes", "hours", "days", "weeks", "months" (≈ 30.44 days), "years" (≈ 365.25 days).

Use tick_size to express coarser steps — e.g. tick_unit="hours", tick_size=4 makes each simulation unit equal to 4 real hours.

SimTimeAxis

SimTimeAxis(start: str | datetime, tick_unit: str = 'days', tick_size: float = 1.0, date_format: str | None = None)

Map simulation tick time to real-world calendar dates.

Parameters:

Name	Type	Description	Default
start	str | datetime	Calendar date/time that corresponds to simulation t = 0. Strings are parsed via :meth:datetime.fromisoformat; a bare date string such as "2027-01-01" works on Python 3.7+.	required
tick_unit	str	Duration of one simulation time unit. Must be one of "seconds", "minutes", "hours", "days", "weeks", "months", or "years".	'days'
tick_size	float	Scale factor: how many tick_unit durations one simulation unit represents. Defaults to 1.0. For example, setting tick_unit="hours" and tick_size=4 makes each sim unit equal to 4 hours.	1.0
date_format	str | None	:func:strftime format string for :meth:label and :meth:labels. Defaults to a sensible format for the chosen tick_unit.	None

Examples:

>>> tax = SimTimeAxis("2027-06-01", tick_unit="weeks")
>>> tax.label(4.0)
'29 Jun 2027'

to_datetime

to_datetime(t: float) -> datetime

Convert a scalar simulation time to a :class:~datetime.datetime.

to_datetimes

to_datetimes(times: Sequence[float] | ndarray) -> list[datetime]

Convert an iterable of simulation times to :class:~datetime.datetime objects.

The returned list is suitable for Plotly's x parameter; Plotly renders datetime objects natively with automatic axis formatting.

label

label(t: float) -> str

Format a simulation time as a human-readable date string.

labels

labels(times: Sequence[float] | ndarray) -> list[str]

Vectorised :meth:label.

apply_to_figure

apply_to_figure(fig: Any, axis: str = 'x', title: str | None = None) -> Any

Replace numeric tick values on axis with calendar dates.

Iterates through every trace in fig.data and, wherever the nominated axis data is a numeric array, substitutes :class:~datetime.datetime objects. Plotly then renders the axis as a date axis with its own smart tick-label formatter.

Parameters:

Name	Type	Description	Default
fig	Any	A plotly.graph_objects.Figure.	required
axis	str	"x" (default) or "y" -- which axis to reformat.	'x'
title	str | None	Optional replacement axis title. If None, the existing title is kept (or set to "date" if it was empty).	None

Returns:

Type	Description
The same figure object (modified in-place), so calls can be chained::	fig = plot_fleet_availability(rec) fig = time_axis.apply_to_figure(fig, title="Calendar date") fig.show()

tick_for_date

tick_for_date(dt: str | datetime) -> float

Return the simulation tick corresponding to a given calendar date.

Useful for scheduling events at specific real-world dates::

t_start = tax.tick_for_date("2027-03-01")
env.schedule_at(t_start, my_callback)

API

Discrete-event primitives: queues, services, resources, arrival generators.

EntityProperties dataclass

EntityProperties(entity_type: str = 'default', service_time: Distribution = (lambda: exponential(1.0))(), balk_on_length: int | None = None, renege_after: float | None = None, extras: dict[str, Any] = dict())

Per-entity sim properties. Pass the same instance to many entities.

Attributes:

Name	Type	Description
entity_type	str	Free-form tag, useful for stratified summaries.
service_time	Distribution	Distribution drawn whenever a Service pulls this entity into a work channel.
balk_on_length	int | None	If the waiting queue has more than this many items on arrival, the entity refuses to join. None disables balking.
renege_after	float | None	If the entity's current_wait_time exceeds this threshold, it leaves the queue. None disables reneging.
extras	dict[str, Any]	Arbitrary additional fields the modeller wants to track.

Queue

Queue(maxlen: int = 10, name: str | None = None, next_q: 'Queue | str' = 'terminus')

Bases: Entity

Bounded FIFO queue.

Parameters:

Name	Type	Description	Default
maxlen	int	Maximum number of items held. Extra arrivals are dropped (and the drop counter is incremented).	10
name	str | None	Human-readable name.	None
next_q	'Queue | str'	Where forwarded items go -- another Queue/Service or the sentinel string "terminus" meaning "remove from the system".	'terminus'

Source code in src/simweave/discrete/queues.py

def __init__(
    self,
    maxlen: int = 10,
    name: str | None = None,
    next_q: "Queue | str" = "terminus",
) -> None:
    super().__init__(name=name)
    if maxlen <= 0:
        raise ValueError("Queue maxlen must be positive.")
    self._deque: deque[Entity] = deque(maxlen=maxlen)
    self.maxlen = maxlen
    self._next_q: "Queue | str" = "terminus"
    self.next_q = next_q  # validate via setter
    self.dropped_count: int = 0
    self.reneged_count: int = 0
    self.balked_count: int = 0
    # Metrics accumulated over the lifetime of the queue.
    self.cumulative_length_time: float = 0.0  # integral of len(q) * dt
    self.cumulative_wait_time: float = 0.0  # sum of total_wait_time on departure
    self.arrivals: int = 0
    self.departures: int = 0

enqueue

enqueue(item: Entity) -> bool

Attempt to append an item. Returns True on success.

Honours item.sim_properties.balk_on_length if present.

Source code in src/simweave/discrete/queues.py

def enqueue(self, item: Entity) -> bool:
    """Attempt to append an item. Returns ``True`` on success.

    Honours ``item.sim_properties.balk_on_length`` if present.
    """
    sim_props = getattr(item, "sim_properties", None)
    balk = (
        getattr(sim_props, "balk_on_length", None)
        if sim_props is not None
        else None
    )
    if balk is not None and len(self._deque) >= balk:
        self.balked_count += 1
        log.debug(
            "%s: %s balked (len %d >= %d).",
            self.name,
            item.name,
            len(self._deque),
            balk,
        )
        return False

    if self.is_full:
        self.dropped_count += 1
        log.debug("%s: dropped %s (full).", self.name, item.name)
        return False

    item.current_wait_time = 0.0
    self._deque.append(item)
    self.arrivals += 1
    return True

forward

forward(override_target: 'Queue | str | None' = None) -> bool

Dequeue head and enqueue into override_target (or self.next_q).

Returns True if forwarded, False if downstream is blocked (in which case the head item is not removed).

Source code in src/simweave/discrete/queues.py

def forward(self, override_target: "Queue | str | None" = None) -> bool:
    """Dequeue head and enqueue into ``override_target`` (or ``self.next_q``).

    Returns ``True`` if forwarded, ``False`` if downstream is blocked (in
    which case the head item is *not* removed).
    """
    target = override_target if override_target is not None else self._next_q
    if not self._deque:
        return False
    if target == "terminus":
        self.dequeue()
        return True
    assert isinstance(target, Queue)
    if target.is_full:
        return False
    item = self._deque.popleft()
    self.departures += 1
    self.cumulative_wait_time += item.total_wait_time
    accepted = target.enqueue(item)
    if not accepted:
        # Shouldn't normally happen because we checked is_full, but handle
        # balking on the downstream queue: silently drop.
        self.dropped_count += 1
    return True

average_length

average_length(elapsed: float) -> float

Mean queue length over the elapsed simulation time (L in Little's law).

Source code in src/simweave/discrete/queues.py

def average_length(self, elapsed: float) -> float:
    """Mean queue length over the elapsed simulation time (L in Little's law)."""
    if elapsed <= 0:
        return 0.0
    return self.cumulative_length_time / elapsed

average_wait

average_wait() -> float

Mean residence time of completed items (W in Little's law).

Source code in src/simweave/discrete/queues.py

def average_wait(self) -> float:
    """Mean residence time of completed items (W in Little's law)."""
    if self.departures == 0:
        return 0.0
    return self.cumulative_wait_time / self.departures

PriorityQueue

PriorityQueue(maxlen: int = 10, name: str | None = None, next_q: 'Queue | str' = 'terminus')

Bases: Queue

Min-heap priority queue. Lower priority values dequeue first.

Items are wrapped in (priority, seq, item) tuples internally so the payload doesn't need to be comparable.

Source code in src/simweave/discrete/queues.py

def __init__(
    self,
    maxlen: int = 10,
    name: str | None = None,
    next_q: "Queue | str" = "terminus",
) -> None:
    super().__init__(maxlen=maxlen, name=name, next_q=next_q)
    self._heap: list[_PItem] = []
    self._seq_counter = 0

Resource

Resource(name: str | None = None, home_pool: 'ResourcePool | str' = 'terminus')

Bases: Entity

A reusable resource that can be checked out of a pool.

Source code in src/simweave/discrete/resources.py

def __init__(
    self, name: str | None = None, home_pool: "ResourcePool | str" = "terminus"
) -> None:
    super().__init__(name=name)
    self.home_pool: "ResourcePool | str" = home_pool
    self.times_acquired: int = 0
    self.busy_time: float = 0.0
    self._is_busy: bool = False

release

release(to: 'ResourcePool | None' = None) -> None

Return the resource to to or its home pool.

Source code in src/simweave/discrete/resources.py

def release(self, to: "ResourcePool | None" = None) -> None:
    """Return the resource to ``to`` or its home pool."""
    target = to if to is not None else self.home_pool
    self._is_busy = False
    if isinstance(target, ResourcePool):
        target.deposit(self)
    elif target == "terminus":
        return
    else:
        raise TypeError(
            "Resource.release target must be a ResourcePool or 'terminus'."
        )

ResourcePool

ResourcePool(maxlen: int = 10, name: str | None = None)

Bases: Queue

Pool of interchangeable resources that a Service can check out.

Source code in src/simweave/discrete/resources.py

def __init__(self, maxlen: int = 10, name: str | None = None) -> None:
    super().__init__(maxlen=maxlen, name=name, next_q="terminus")

deposit

deposit(resource: Resource) -> None

Return a resource to the pool. Raises if the pool is at capacity.

Source code in src/simweave/discrete/resources.py

def deposit(self, resource: Resource) -> None:
    """Return a resource to the pool. Raises if the pool is at capacity."""
    if not isinstance(resource, Resource):
        raise TypeError("ResourcePool only accepts Resource instances.")
    resource.home_pool = self
    if self.is_full:
        raise RuntimeError(f"{self.name}: pool full when returning {resource.name}")
    self._deque.append(resource)
    self.arrivals += 1

try_acquire

try_acquire() -> Resource | None

Non-blocking acquire. Returns a Resource or None if empty.

Source code in src/simweave/discrete/resources.py

def try_acquire(self) -> Resource | None:
    """Non-blocking acquire. Returns a Resource or ``None`` if empty."""
    if not self._deque:
        return None
    resource = self._deque.popleft()
    assert isinstance(resource, Resource)
    resource.home_pool = self
    resource._is_busy = True
    resource.times_acquired += 1
    self.departures += 1
    return resource

release

release(resource: Resource) -> None

Release a previously acquired resource back to this pool.

Source code in src/simweave/discrete/resources.py

def release(self, resource: Resource) -> None:
    """Release a previously acquired resource back to this pool."""
    resource.release(to=self)

Service

Service(capacity: int = 1, buffer_size: int = 10, next_q: 'Queue | str' = 'terminus', resources: ResourcePool | None = None, default_service_time: float = 1.0, rng: Generator | None = None, name: str | None = None)

Bases: Queue

Multi-channel server with optional resource pool.

Parameters:

Name	Type	Description	Default
capacity	int	Number of parallel work channels (servers).	1
buffer_size	int	Max queue length before service. Further arrivals are dropped.	10
next_q	'Queue | str'	Where to forward completed items.	'terminus'
resources	ResourcePool | None	Optional :class:ResourcePool from which to acquire one resource per item served. If None, the service is unconstrained.	None
default_service_time	float	Fallback when an arriving entity has no sim_properties.	1.0
rng	Generator | None	Optional numpy Generator used when drawing service times.	None

Source code in src/simweave/discrete/services.py

def __init__(
    self,
    capacity: int = 1,
    buffer_size: int = 10,
    next_q: "Queue | str" = "terminus",
    resources: ResourcePool | None = None,
    default_service_time: float = 1.0,
    rng: np.random.Generator | None = None,
    name: str | None = None,
) -> None:
    super().__init__(maxlen=buffer_size, name=name, next_q=next_q)
    if capacity < 1:
        raise ValueError("Service capacity must be >= 1.")
    self.capacity = capacity
    self.resources = resources
    self.default_service_time = float(default_service_time)
    self.rng = rng if rng is not None else np.random.default_rng()
    self.channels = [_WorkChannel(self, i) for i in range(capacity)]
    self.completed_count: int = 0
    self.completed_total_time: float = (
        0.0  # sum of (wait + service) across completed
    )

utilisation

utilisation(elapsed: float) -> float

Average utilisation across all channels over elapsed time.

Source code in src/simweave/discrete/services.py

def utilisation(self, elapsed: float) -> float:
    """Average utilisation across all channels over ``elapsed`` time."""
    if elapsed <= 0 or self.capacity == 0:
        return 0.0
    busy = sum(ch.busy_time for ch in self.channels)
    return busy / (self.capacity * elapsed)

average_residence

average_residence() -> float

Mean total time (waiting + service) for completed items.

Source code in src/simweave/discrete/services.py

def average_residence(self) -> float:
    """Mean total time (waiting + service) for completed items."""
    if self.completed_count == 0:
        return 0.0
    return self.completed_total_time / self.completed_count

ArrivalGenerator

ArrivalGenerator(interarrival: Callable[[Generator], float], factory: Callable[[SimEnvironment], Entity], target: Queue, rng: Generator | None = None, name: str | None = None)

Bases: Entity

Generates new entities according to an inter-arrival distribution.

On each tick it adds to an internal clock and, whenever the clock passes the next scheduled arrival, invokes factory(env) to mint a new entity and pushes it into target. Multiple arrivals within a single tick are handled correctly.

Source code in src/simweave/discrete/services.py

def __init__(
    self,
    interarrival: Callable[[np.random.Generator], float],
    factory: Callable[[SimEnvironment], Entity],
    target: Queue,
    rng: np.random.Generator | None = None,
    name: str | None = None,
) -> None:
    super().__init__(name=name)
    self.interarrival = interarrival
    self.factory = factory
    self.target = target
    self.rng = rng if rng is not None else np.random.default_rng()
    self._countdown: float = float(self.interarrival(self.rng))
    self.generated: int = 0
    self.rejected: int = 0

exponential

exponential(mean: float) -> Distribution

Exponential inter-arrival / service time.

Source code in src/simweave/discrete/properties.py

def exponential(mean: float) -> Distribution:
    """Exponential inter-arrival / service time."""
    if mean <= 0:
        raise ValueError("Exponential mean must be positive.")

    def draw(rng: RNG) -> float:
        return float(rng.exponential(mean))

    return draw

normal

normal(mean: float, std: float, *, clip_nonnegative: bool = True) -> Distribution

Normal distribution; by default clips at 0 since negative service times are nonsense.

Source code in src/simweave/discrete/properties.py

def normal(mean: float, std: float, *, clip_nonnegative: bool = True) -> Distribution:
    """Normal distribution; by default clips at 0 since negative service times are nonsense."""
    if std <= 0:
        raise ValueError("normal std must be positive.")

    def draw(rng: RNG) -> float:
        x = float(rng.normal(mean, std))
        return max(0.0, x) if clip_nonnegative else x

    return draw

set_default_seed

set_default_seed(seed: int) -> None

Replace the module-level generator with a seeded one.

Source code in src/simweave/discrete/properties.py

def set_default_seed(seed: int) -> None:
    """Replace the module-level generator with a seeded one."""
    global _DEFAULT_RNG
    _DEFAULT_RNG = np.random.default_rng(seed)