Core

The base primitives every other subpackage builds on.

SimEnvironment

Owns the simulation clock, the priority event queue, and the list of registered entities. Stepped via env.run() or env.run(until=t).

import simweave as sw

env = sw.SimEnvironment(dt=0.1, end=10.0)
env.register(my_entity)
env.run()

Entity

The base class for anything that lives in a SimEnvironment. Override on_register(env) for setup, tick(env) for per-step work, and on_unregister(env) for teardown. Entities also carry a history list that subclasses populate however they like.

Clock and EventQueue

Lower-level building blocks. Most users never touch these directly — the SimEnvironment wraps them. They become useful when you want to schedule a one-shot event in the future:

env.schedule(at=env.clock.now + 5.0, event=ScheduledEvent(callback=fn))

API

Core runtime: clock, scheduler, entity, environment, logging.

Clock dataclass

Clock(start: float = 0.0, dt: float = 1.0, end: float | None = None, t: float = 0.0)

Fixed-step simulation clock.

Parameters:

Name	Type	Description	Default
start	float	Initial simulation time.	0.0
dt	float	Fixed timestep. Units are whatever the rest of the model uses (typically seconds, but can be hours, days, or dimensionless ticks).	1.0
end	float | None	Optional end time. If provided, :meth:is_finished returns True once t >= end.	None

advance

advance(dt: float | None = None) -> None

Advance the clock by dt (defaults to self.dt).

Source code in src/simweave/core/clock.py

def advance(self, dt: float | None = None) -> None:
    """Advance the clock by ``dt`` (defaults to ``self.dt``)."""
    step = self.dt if dt is None else dt
    if step < 0:
        raise ValueError("Cannot advance clock by a negative interval.")
    self.t += step

jump_to

jump_to(t: float) -> None

Jump the clock forward to absolute time t (no backward jumps).

Source code in src/simweave/core/clock.py

def jump_to(self, t: float) -> None:
    """Jump the clock forward to absolute time ``t`` (no backward jumps)."""
    if t < self.t:
        raise ValueError("Cannot rewind the clock.")
    self.t = t

EventQueue

EventQueue()

Min-heap priority queue of scheduled callbacks.

Source code in src/simweave/core/scheduler.py

def __init__(self) -> None:
    self._heap: list[ScheduledEvent] = []
    self._seq: int = 0

schedule

schedule(time: float, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> ScheduledEvent

Schedule callback(*args, **kwargs) to fire at simulation time.

Source code in src/simweave/core/scheduler.py

def schedule(
    self, time: float, callback: Callable[..., Any], *args: Any, **kwargs: Any
) -> ScheduledEvent:
    """Schedule ``callback(*args, **kwargs)`` to fire at simulation ``time``."""
    evt = ScheduledEvent(
        time=float(time), seq=self._seq, callback=callback, args=args, kwargs=kwargs
    )
    heapq.heappush(self._heap, evt)
    self._seq += 1
    return evt

peek_time

peek_time() -> float | None

Return the time of the next non-cancelled event, or None.

Source code in src/simweave/core/scheduler.py

def peek_time(self) -> float | None:
    """Return the time of the next non-cancelled event, or ``None``."""
    self._drop_cancelled()
    return self._heap[0].time if self._heap else None

pop_due

pop_due(now: float)

Yield every non-cancelled event whose time is <= now.

Source code in src/simweave/core/scheduler.py

def pop_due(self, now: float):
    """Yield every non-cancelled event whose time is ``<= now``."""
    self._drop_cancelled()
    while self._heap and self._heap[0].time <= now:
        evt = heapq.heappop(self._heap)
        if evt.cancelled:
            continue
        yield evt
        self._drop_cancelled()

cancel

cancel(event: ScheduledEvent) -> None

Flag an event as cancelled without paying heap-removal cost.

Source code in src/simweave/core/scheduler.py

def cancel(self, event: ScheduledEvent) -> None:
    """Flag an event as cancelled without paying heap-removal cost."""
    event.cancelled = True

ScheduledEvent dataclass

ScheduledEvent(time: float, seq: int, callback: Callable[..., Any], args: tuple = (), kwargs: dict = dict(), cancelled: bool = False)

Heap-ordered event record.

seq is a tiebreaker so that two events at the same time fire in insertion order, giving deterministic behaviour for simultaneous events.

Entity

Entity(name: str | None = None)

Base class for simulation entities.

Source code in src/simweave/core/entity.py

def __init__(self, name: str | None = None) -> None:
    self.id: int = next(Entity._id_counter)
    self.name: str = (
        name if name is not None else f"{type(self).__name__}_{self.id}"
    )
    self.created_at: float | None = None
    self.age: float = 0.0
    # Queue residency bookkeeping.
    self.current_wait_time: float = 0.0
    self.total_wait_time: float = 0.0
    # Populated by a Service when pulling this entity for processing.
    self.remaining_service_time: float = 0.0

on_register

on_register(env: 'SimEnvironment') -> None

Hook called when this entity is registered with an environment.

Source code in src/simweave/core/entity.py

def on_register(self, env: "SimEnvironment") -> None:
    """Hook called when this entity is registered with an environment."""
    if self.created_at is None:
        self.created_at = env.clock.t

tick

tick(dt: float, env: 'SimEnvironment') -> None

Advance this entity by dt simulation seconds.

Subclasses should call super().tick(dt, env) first (to keep the age counter correct) before doing their own logic.

Source code in src/simweave/core/entity.py

def tick(self, dt: float, env: "SimEnvironment") -> None:
    """Advance this entity by ``dt`` simulation seconds.

    Subclasses should call ``super().tick(dt, env)`` first (to keep the
    age counter correct) before doing their own logic.
    """
    self.age += dt

has_work

has_work(env: 'SimEnvironment') -> bool

Report whether this entity has pending work.

The default is False -- plain items waiting in a queue are not themselves "doing work" in the skip-idle-gaps sense; the queue/service holding them is.

Source code in src/simweave/core/entity.py

def has_work(self, env: "SimEnvironment") -> bool:
    """Report whether this entity has pending work.

    The default is ``False`` -- plain items waiting in a queue are not
    themselves "doing work" in the skip-idle-gaps sense; the queue/service
    holding them is.
    """
    return False

reset_id_counter classmethod

reset_id_counter() -> None

Reset the shared ID counter. Useful between replicates in tests.

Source code in src/simweave/core/entity.py

@classmethod
def reset_id_counter(cls) -> None:
    """Reset the shared ID counter. Useful between replicates in tests."""
    cls._id_counter = count(0)

SimEnvironment

SimEnvironment(start: float = 0.0, dt: float = 1.0, end: float | None = None, graph: Any = None)

Container for a clock, an event queue, and registered processes.

Source code in src/simweave/core/environment.py

def __init__(
    self,
    start: float = 0.0,
    dt: float = 1.0,
    end: float | None = None,
    graph: Any = None,
) -> None:
    self.clock = Clock(start=start, dt=dt, end=end)
    self.events = EventQueue()
    self.graph = graph
    self._processes: list[Process] = []

register

register(process: Process) -> Process

Register a process so it is ticked each simulation step.

Source code in src/simweave/core/environment.py

def register(self, process: Process) -> Process:
    """Register a process so it is ticked each simulation step."""
    if not hasattr(process, "tick"):
        raise TypeError("Registered process must implement tick(dt, env).")
    self._processes.append(process)
    on_register = getattr(process, "on_register", None)
    if callable(on_register):
        on_register(self)
    return process

step

step() -> None

Process events due at t, tick every process, then advance.

Source code in src/simweave/core/environment.py

def step(self) -> None:
    """Process events due at ``t``, tick every process, then advance."""
    for evt in self.events.pop_due(self.clock.t):
        evt.callback(*evt.args, **evt.kwargs)
    for p in self._processes:
        p.tick(self.clock.dt, self)
    self.clock.advance()

run

run(until: float | None = None, skip_idle_gaps: bool = False) -> None

Run the simulation until until or clock.end.

Source code in src/simweave/core/environment.py

def run(self, until: float | None = None, skip_idle_gaps: bool = False) -> None:
    """Run the simulation until ``until`` or ``clock.end``."""
    if until is None:
        until = self.clock.end
    if until is None:
        raise ValueError("Pass `until` or set clock.end before calling run().")
    if until <= self.clock.t:
        return

    while self.clock.t < until:
        if skip_idle_gaps and not self._any_has_work():
            next_evt = self.events.peek_time()
            if next_evt is not None and next_evt > self.clock.t:
                self.clock.jump_to(min(next_evt, until))
                continue
            # Nothing has work and no future events -> nothing will change.
            if next_evt is None:
                log.debug(
                    "No work and no pending events; halting early at t=%s.",
                    self.clock.t,
                )
                break
        self.step()

Process

Bases: Protocol

A process that participates in the tick loop.

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'

Source code in src/simweave/core/time_axis.py

def __init__(
    self,
    start: str | datetime,
    tick_unit: str = "days",
    tick_size: float = 1.0,
    date_format: str | None = None,
) -> None:
    if isinstance(start, str):
        # Accept bare dates ("2027-01-01") and full ISO strings.
        start = datetime.fromisoformat(start)
    self.start: datetime = start

    tick_unit = tick_unit.lower()
    if tick_unit not in _SECONDS_PER_UNIT:
        raise ValueError(
            f"tick_unit {tick_unit!r} not recognised. "
            f"Choose one of: {sorted(_SECONDS_PER_UNIT)}"
        )
    self.tick_unit: str = tick_unit
    self.tick_size: float = float(tick_size)
    self._secs_per_tick: float = _SECONDS_PER_UNIT[tick_unit] * self.tick_size
    self.date_format: str = date_format or _DEFAULT_FMT[tick_unit]

to_datetime

to_datetime(t: float) -> datetime

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

Source code in src/simweave/core/time_axis.py

def to_datetime(self, t: float) -> datetime:
    """Convert a scalar simulation time to a :class:`~datetime.datetime`."""
    return self.start + timedelta(seconds=float(t) * self._secs_per_tick)

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.

Source code in src/simweave/core/time_axis.py

def to_datetimes(self, times: Sequence[float] | np.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.
    """
    return [self.to_datetime(float(t)) for t in times]

label

label(t: float) -> str

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

Source code in src/simweave/core/time_axis.py

def label(self, t: float) -> str:
    """Format a simulation time as a human-readable date string."""
    return self.to_datetime(t).strftime(self.date_format)

labels

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

Vectorised :meth:label.

Source code in src/simweave/core/time_axis.py

def labels(self, times: Sequence[float] | np.ndarray) -> list[str]:
    """Vectorised :meth:`label`."""
    return [self.label(float(t)) for t in times]

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()

Source code in src/simweave/core/time_axis.py

def apply_to_figure(
    self,
    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
    ----------
    fig:
        A ``plotly.graph_objects.Figure``.
    axis:
        ``"x"`` (default) or ``"y"`` -- which axis to reformat.
    title:
        Optional replacement axis title.  If ``None``, the existing
        title is kept (or set to ``"date"`` if it was empty).

    Returns
    -------
    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()
    """
    attr = axis  # "x" or "y"
    for trace in fig.data:
        data = getattr(trace, attr, None)
        if data is None:
            continue
        arr = np.asarray(data)
        if arr.dtype.kind in ("f", "i", "u"):
            setattr(trace, attr, self.to_datetimes(arr))

    # Update the layout axis title.
    layout_axis = f"{axis}axis"
    existing = getattr(fig.layout, layout_axis, None)
    current_title = ""
    if existing is not None:
        t_obj = getattr(existing, "title", None)
        if t_obj is not None:
            current_title = getattr(t_obj, "text", "") or ""
    new_title = title if title is not None else (current_title or "date")
    fig.update_layout(**{layout_axis: {"title": new_title}})

    return fig

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)

Source code in src/simweave/core/time_axis.py

def tick_for_date(self, 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)
    """
    if isinstance(dt, str):
        dt = datetime.fromisoformat(dt)
    delta_secs = (dt - self.start).total_seconds()
    return delta_secs / self._secs_per_tick

configure

configure(level: int | str = INFO, fmt: str = '%(asctime)s [%(levelname)s] %(name)s: %(message)s', force: bool = False) -> None

Configure the simweave logger hierarchy. Safe to call many times.

Source code in src/simweave/core/logging.py

def configure(
    level: int | str = logging.INFO,
    fmt: str = "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    force: bool = False,
) -> None:
    """Configure the simweave logger hierarchy. Safe to call many times."""
    global _configured
    if _configured and not force:
        return
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(fmt))
    root = logging.getLogger(_ROOT)
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(level)
    root.propagate = False
    _configured = True

get_logger

get_logger(name: str | None = None) -> Logger

Return a logger under the simweave namespace.

Source code in src/simweave/core/logging.py

def get_logger(name: str | None = None) -> logging.Logger:
    """Return a logger under the ``simweave`` namespace."""
    if name is None or name == _ROOT:
        return logging.getLogger(_ROOT)
    if name.startswith(_ROOT + "."):
        return logging.getLogger(name)
    return logging.getLogger(f"{_ROOT}.{name}")