API Reference
Simulation Context
SimulationContext(sim_id, factor=1.0, random_seed=None, logical_start_time=None, go_live_at=None, warmup=None, initial_time=0.0, lag_interval=None, timezone=None, strict=False, environment=None)
Builder and Orchestrator for Event-Driven Digital Twins.
The SimulationContext implements the Builder Pattern to construct the
foundational architecture of a Digital Twin. It acts as the central control
plane, aggregating discrete machine capacities, continuous physical states
(wear and tear), and statistical distributions into a unified SimParameter
object.
Architectural Guarantees:
1. Immutability During Configuration: The underlying DynamicRealtimeEnvironment
and its background asynchronous I/O threads are completely deferred until
.run() is explicitly invoked. This prevents partial state leaks or
unintended clock starts.
2. One registry: All registered configurations are flattened
and managed by the internal SimulationRegistry, allowing external
Kafka Admin commands to mutate these states safely during runtime.
3. Deterministic Stochasticity: The Sampler gives each distribution a
random sequence of its own, made from random_seed, so a seeded run repeats
exactly, and adding a distribution leaves the others' draws unchanged. A run
without random_seed draws a seed and logs it, so it can be repeated.
Attributes:
| Name | Type | Description |
|---|---|---|
sim_id |
str
|
The unique namespace for this simulation instance (e.g., 'HotRolling'). |
factor |
float
|
The real-time synchronization multiplier. A factor of 1.0 syncs strictly with wall-clock time, while 0.0 executes instantly in a fast-forward loop (ideal for historical data processing). |
go_live_at |
Optional[datetime]
|
Logical instant at which |
warmup |
Optional[float]
|
The end of the warm-up, in simulation seconds. |
random_seed |
Optional[int]
|
The run's seed. When None, |
sampler |
Optional[Sampler]
|
The centralized Random Number Generator engine.
Accessible to user-defined raw generators only after |
Initializes the context builder with a designated namespace and temporal factor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sim_id
|
str
|
The unique prefix for this group of parameters. All generated telemetry and event paths will be prefixed with this ID. |
required |
factor
|
float
|
The real-time synchronization factor. Defaults to 1.0. |
1.0
|
random_seed
|
Optional[int]
|
Optional seed for deterministic execution. |
None
|
logical_start_time
|
Optional[datetime]
|
Optional override for the environment's base clock,
forwarded to |
None
|
go_live_at
|
Optional[datetime]
|
Optional logical instant at which the run stops using |
None
|
warmup
|
Union[float, str, None]
|
The end of the warm-up, in simulation seconds or as a duration
such as |
None
|
initial_time
|
Union[float, str]
|
The simulation clock's starting value, in seconds or as a
duration such as |
0.0
|
lag_interval
|
Union[float, str, None]
|
Simulation seconds, or a duration such as |
None
|
timezone
|
Optional[str]
|
An IANA time zone name, such as |
None
|
strict
|
bool
|
When True a paced step raises |
False
|
environment
|
Optional[Type[DynamicRealtimeEnvironment]]
|
A subclass of |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/core/context.py
131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 | |
Attributes
env
property
Read-only handle to the active DynamicRealtimeEnvironment.
Grants raw generators access to environment primitives such as
publish_event, timeout, now, and start_datetime without
reaching into private builder state.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If accessed during the Builder Phase (before |
Functions
add_arrival(name, dist, *, rate=None, mean=None, std=None, low=None, value=None, high=None, mode=None, shape=None, scale=None, values=None, weights=None, edges=None, counts=None, attributes=None, key=None, spawn=None)
Registers a statistical distribution defining the inter-arrival times of entities.
Each arrival starts a new entity, which wait_for_arrival returns. Its key,
the record key of every event it causes, is the arrival name and a counter
(orders-1, orders-2) unless key gives another.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The internal identifier (e.g., 'orders'). |
required |
dist
|
str
|
The family, as for |
required |
rate
|
Optional[float]
|
Events per second, for exponential. |
None
|
mean
|
Optional[float]
|
The mean, for normal, lognormal and truncated_normal. |
None
|
std
|
Optional[float]
|
The standard deviation, for the same three (0 when not given). |
None
|
low
|
Optional[float]
|
The lowest value, for truncated_normal, uniform and triangular. |
None
|
value
|
Optional[float]
|
The time a constant distribution always gives. |
None
|
high
|
Optional[float]
|
The highest value, for uniform and triangular. |
None
|
mode
|
Optional[float]
|
The most likely value, for triangular. |
None
|
shape
|
Optional[float]
|
The shape, for gamma and weibull. |
None
|
scale
|
Optional[float]
|
The scale, for gamma and weibull. |
None
|
values
|
Optional[List[float]]
|
The values an empirical draws from, taken at load. |
None
|
weights
|
Optional[List[float]]
|
One weight per empirical value. |
None
|
edges
|
Optional[List[float]]
|
The bin edges of an empirical histogram. |
None
|
counts
|
Optional[List[float]]
|
One count per empirical histogram bin. |
None
|
attributes
|
Optional[Mapping[str, Any]]
|
The attributes each entity starts with, read in conditions
as |
None
|
key
|
Optional[Callable[[int], str]]
|
A function of the entity's counter, from 1, returning its key. |
None
|
spawn
|
Optional[str]
|
The task each arrival starts, with the arrival's entity and the
task's next instance id, as YAML's |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the family is unknown, or a parameter is missing, out of
range or not one the family takes. A |
Source code in src/desflow/core/context.py
add_arrival_replay(name, source, *, time='time', unit='s', fields='all', on_invalid='error', attributes=None, key=None, spawn=None)
Registers an arrival taken from recorded rows instead of a distribution.
Each row becomes one arrival at its recorded time, in the order given, and
its fields become the entity's attributes. wait_for_arrival(name) returns
the wait for the next row, so an arrival loop is written as for any other
arrival. Rows at equal times arrive at that time in their order; rows at or
after until never arrive; after the last row the arrival stops, with one
INFO line. The rows are read and checked when the run starts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The arrival's name. |
required |
source
|
Union[str, Path, Iterable[Mapping[str, Any]]]
|
A CSV or Parquet file (Parquet needs the |
required |
time
|
str
|
The column of each row's time. |
'time'
|
unit
|
str
|
|
's'
|
fields
|
Union[str, List[str]]
|
The columns each entity carries, or |
'all'
|
on_invalid
|
str
|
|
'error'
|
attributes
|
Optional[Mapping[str, Any]]
|
Attributes every entity starts with; a row's fields of the same name replace them. |
None
|
key
|
Optional[Callable[[int], str]]
|
A function of the entity's counter, from 1, returning its key. |
None
|
spawn
|
Optional[str]
|
The task each row starts, as for |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/core/context.py
add_breakdown(resource, time_to_failure, repair, work='interrupt')
Makes a resource break down at random and be repaired.
The failure clock runs on calendar time: after each repair, or the start,
the time to the next failure is drawn, busy or idle. While the resource is
down its capacity is 0, its current_cap keeps the planned value, and a
change to it during the repair sets the capacity the resource returns to.
Each failure and repair publishes an event keyed by the resource's name,
{"status": "down", "reason": "failure"} and {"status": "repaired", ...}.
The two distributions are registry paths,
<sim_id>.breakdown.<resource>.time_to_failure and .repair.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resource
|
str
|
The resource's name. |
required |
time_to_failure
|
Mapping[str, Any]
|
A distribution, such as
|
required |
repair
|
Mapping[str, Any]
|
A distribution, such as |
required |
work
|
str
|
|
'interrupt'
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a distribution or |
Source code in src/desflow/core/context.py
add_container(name, current_cap=None, max_cap=None, level=0.0, type=None)
Registers a container: an amount that tasks take from and add to, such as stock.
current_cap is the capacity, the largest level, as for a resource. The
level starts at level and has a live registry path of its own,
<sim_id>.containers.<name>.level, so a scenario or an ingress message can
set it during the run, such as a restock. Models that used current_cap as a
level, such as wear, use level instead.
A container with no current_cap has no limit, and max_cap defaults to
current_cap. Use math.inf or float("inf") for no limit.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The internal identifier (e.g., 'stock'). |
required |
current_cap
|
Optional[float]
|
The capacity at simulation start. None, the default, is no
limit, or |
None
|
max_cap
|
Optional[float]
|
The largest capacity a live update can set. None, the default,
is |
None
|
level
|
float
|
The amount held at simulation start. Defaults to 0. |
0.0
|
type
|
Optional[Type[ContainerBase]]
|
A subclass of |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a store already has this name, or |
Source code in src/desflow/core/context.py
add_egress(provider, when=None, batch_size=None, flush_interval=None)
Registers an asynchronous egress provider for data exfiltration.
Every registered provider receives every record, so attaching a stream sink and
a lake sink writes the same dataset to both in one pass. Pass when to send
only some records to a provider, which reads the same way as the path_router
that ParquetStorageEgress already accepts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
provider
|
Any
|
An initialized connector subclassing |
required |
when
|
Union[Callable[[dict], bool], str, None]
|
Optional predicate taking one record and returning True to send it to
this provider. None sends every record, which is the usual case.
|
None
|
batch_size
|
Optional[int]
|
Records this provider buffers before flushing. None uses the
context default from |
None
|
flush_interval
|
Optional[float]
|
Simulation seconds this provider waits before forcing a
flush. None uses the context default. Ignored when |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Each provider has its own buffer and its own queue of batches. A record that several providers take is stored once, and stays in memory until the last provider that holds it has written it.
Example
Source code in src/desflow/core/context.py
add_ingress(provider)
Registers an asynchronous ingress provider to listen for external mutations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
provider
|
Any
|
An initialized connector subclassing |
required |
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Source code in src/desflow/core/context.py
add_process(func, **kwargs)
Registers a generator function to start when the run starts.
This is the builder form of spawn, for a process that is not an arrival or
telemetry loop, such as a drift engine or a scripted scenario. It is how a
YAML blueprint starts the processes it references with !python.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func
|
Callable
|
A generator function. It is called as |
required |
**kwargs
|
Any
|
Extra keyword arguments passed to |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Example
Source code in src/desflow/core/context.py
add_resource(name, current_cap=None, max_cap=None, queue=DEFAULT_QUEUE_ORDER, preempt=False, queue_limit=None, type=None)
Registers a discrete logical resource capable of processing tasks.
Resources act as the primary gating mechanism in the simulation, blocking concurrent processes to prevent physical collisions (e.g., restricting the number of slabs inside a mill stand).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The internal identifier (e.g., 'mill_structural'). |
required |
current_cap
|
Optional[Union[int, float]]
|
The active physical token capacity at simulation start, a
whole number or |
None
|
max_cap
|
Optional[Union[int, float]]
|
The absolute physical maximum bound of tokens allowed. Defaults
to |
None
|
queue
|
str
|
The order waiting tasks are served in: |
DEFAULT_QUEUE_ORDER
|
preempt
|
bool
|
When True, a task of a lower priority value takes a unit from
a task of a higher one when every unit is busy, as SimPy's
|
False
|
queue_limit
|
Optional[int]
|
The most requests that may wait, or None for no limit. A
task that finds that many waiting balks; a request made in the
low-level API raises |
None
|
type
|
Optional[Type[ResourceBase]]
|
A subclass of |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/core/context.py
534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 | |
add_scenario(steps)
Adds scenario steps: changes and records at set simulation times.
Each step is a mapping: at, seconds or a duration from the start of the
run, and a path and value to set in the registry, a record to publish
as a ground-truth label, an interrupt ({"task": ..., "where": ...,
"cause": ...}), or more than one. A step applies before every other event
at its time, so an arrival gap drawn at that moment uses a new rate; a step
at or after until never applies. Steps at equal times apply in the order
given.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
steps
|
List[Mapping[str, Any]]
|
The steps. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a step is malformed. A path the model does not have is refused when the run starts. |
Source code in src/desflow/core/context.py
add_service(name, dist, *, rate=None, mean=None, std=None, low=None, value=None, high=None, mode=None, shape=None, scale=None, values=None, weights=None, edges=None, counts=None)
Registers a statistical distribution defining a process service duration.
The parameters are keyword-only, so their order never changes the meaning.
Give only those dist takes; desflow.core.distributions.DISTRIBUTIONS
lists each family's parameters and rules.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The internal identifier (e.g., 'packing'). |
required |
dist
|
str
|
The family: exponential, normal, lognormal, truncated_normal, constant, uniform, triangular, gamma, weibull or empirical. |
required |
rate
|
Optional[float]
|
Events per second, for exponential. |
None
|
mean
|
Optional[float]
|
The mean, for normal, lognormal and truncated_normal. |
None
|
std
|
Optional[float]
|
The standard deviation, for the same three (0 when not given). |
None
|
low
|
Optional[float]
|
The lowest value, for truncated_normal (0 when not given), uniform and triangular. |
None
|
value
|
Optional[float]
|
The time a constant distribution always gives. |
None
|
high
|
Optional[float]
|
The highest value, for uniform and triangular. |
None
|
mode
|
Optional[float]
|
The most likely value, for triangular. |
None
|
shape
|
Optional[float]
|
The shape, for gamma and weibull. |
None
|
scale
|
Optional[float]
|
The scale, for gamma and weibull. |
None
|
values
|
Optional[List[float]]
|
The values an empirical draws from, taken at load. |
None
|
weights
|
Optional[List[float]]
|
One weight per empirical value. |
None
|
edges
|
Optional[List[float]]
|
The bin edges of an empirical histogram. |
None
|
counts
|
Optional[List[float]]
|
One count per empirical histogram bin. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the family is unknown, or a parameter is missing, out of range or not one the family takes. |
Source code in src/desflow/core/context.py
add_signal(name, once=False, on=None)
Declares a signal that tasks fire and wait for.
A task fires it with signal=name when it finishes, and fails it when the
task fails; a task waits for it with wait={"signal": name}; a process
reaches it with context.signal(name).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The signal's name. |
required |
once
|
bool
|
When True the signal fires once, as SimPy's |
False
|
on
|
Optional[List[Mapping[str, Any]]]
|
Actions run when the signal fires, each a mapping with one of |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Source code in src/desflow/core/context.py
add_step(at, path=None, value=None, record=None, interrupt=None)
Adds one scenario step; add_scenario describes the steps.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
at
|
Union[float, str]
|
Seconds, or a duration such as |
required |
path
|
Optional[str]
|
A registry path to set, with |
None
|
value
|
Any
|
The value to set. |
None
|
record
|
Optional[Mapping[str, Any]]
|
A ground-truth record to publish. |
None
|
interrupt
|
Optional[Mapping[str, Any]]
|
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Source code in src/desflow/core/context.py
add_store(name, current_cap=None, max_cap=None, priority=False, items=None, type=None)
Registers a store: a place that holds items, which tasks put and get.
A task puts its finished payload as an item and gets the next item,
waiting when the store is full or empty. A fifo store, the default,
returns the oldest item; a priority store returns the item whose task
gave the lowest put priority first, and the oldest among equal
priorities.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The internal identifier (e.g., 'parcels'). |
required |
current_cap
|
Optional[float]
|
The number of items it can hold at simulation start, a
whole number or infinity. None, the default, is no limit, or
|
None
|
max_cap
|
Optional[float]
|
The largest capacity a live update can set. None, the default,
is |
None
|
priority
|
bool
|
When True, items are returned by the |
False
|
items
|
Optional[List[Any]]
|
The items it holds when the run starts, in the order a get
returns them. A priority store takes |
None
|
type
|
Optional[Type[StoreBase]]
|
A subclass of |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a container already has this name, a capacity is not a
whole number or infinity, there are more items than |
Source code in src/desflow/core/context.py
add_variable(name, value)
Registers a dynamic, untyped variable accessible via the central Registry.
Variables act as arbitrary state payloads that can be targeted by the Control Plane. For example, storing a JSON dictionary defining the velocity and frequency of a gradual wear injection loop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The internal identifier (e.g., 'velocity_structural'). |
required |
value
|
Any
|
The initial payload (supports primitives, lists, and dicts). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Source code in src/desflow/core/context.py
advance(until)
Runs a started run up to until, then returns with the run still open.
Call it again to run the next segment, or finish() to run to the end.
Segments give the same records as one run, because nothing is flushed at a
segment boundary. Between segments the model can be read, changed with
context.env, and stepped with step(). advance() never moves the
wall-clock anchor, so a paced run that waits between segments is behind
when it goes on; call context.env.resume_pacing() first to pace again from
now.
The run stops at the earlier of until and its own end: start()'s until,
the stop or the end of the drain. An until that is an event, such as
context.signal("gate").wait(), runs to that event and returns its value;
one that has been processed already returns at once.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
until
|
Union[float, str, Event]
|
A time in simulation seconds or as a duration such as |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
None when the run stops at a time, the event's value for an event,
and the value |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If the run has not started or is closed, or it has ended
( |
ValueError
|
If |
SignalFailed
|
If the signal that |
Source code in src/desflow/core/context.py
analyze(verify=False, *, until=None, seed=None, replications=None, target=0.85)
Explains the model: its graph, and the queueing estimates that apply to it.
The graph and the analysis come from what the builder declared, before
start(), and leave the context as it was: it is not started, and a run
after it gives the same records as a run without it. desflow explain
prints the same Explanation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
verify
|
bool
|
Also run the model unpaced and compare its statistics with the analysis. The run uses a new copy of the model, built again from the blueprint or the Python file it came from. |
False
|
until
|
Union[float, str, None]
|
The end of the verifying run, in seconds or as a duration; the
blueprint's |
None
|
seed
|
Optional[int]
|
The verifying run's seed; the model's |
None
|
replications
|
Optional[int]
|
Verify with this many runs, 2 or more, and an interval for each measure, instead of one run. |
None
|
target
|
float
|
The utilisation the tuning hints aim for, above 0 and at most 10; at 1 or above the hints are for deliberate overload. |
0.85
|
Returns:
| Name | Type | Description |
|---|---|---|
Explanation |
Explanation
|
The graph, the estimates per arrival, task, resource, store
and container, the bottleneck, the hints, and with |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the model names a task or another part that is not
added, as |
Source code in src/desflow/core/context.py
arrival_loop(arrival_id)
Registers an infinite generator loop to automatically start on boot.
The wrapped function will be injected with the SimulationContext
instance at runtime, granting it access to context.wait_for_arrival().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
arrival_id
|
str
|
The ID of the configured arrival rate to dynamically sample. |
required |
Source code in src/desflow/core/context.py
call(task, *args, entity=None, upstream=None, **kwargs)
Starts a task as a child flow and returns its process.
The task runs with its own instance id, and so do the tasks its next
leads to. The process ends when that flow ends: its value is the last
task's finished payload, and it fails with TaskFailed when a task of the
flow fails and no on_fail handles it. A flow started this way is not
counted as the entity's flow finishing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task
|
str
|
The task's name. |
required |
*args
|
Any
|
Positional arguments its function takes after the task id. |
()
|
entity
|
Optional[Entity]
|
The entity, such as the caller's own; a new one when None. |
None
|
upstream
|
Optional[Upstream]
|
What the task receives as |
None
|
**kwargs
|
Any
|
Keyword arguments its function takes. |
{}
|
Returns:
| Type | Description |
|---|---|
Process
|
simpy.Process: The child flow. A failure nobody waits for does not stop the run. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called before the run starts. |
ValueError
|
If no task has that name. |
Source code in src/desflow/core/context.py
close()
Ends a started run: publishes the last statistics, flushes and stops the egress connectors, then stops the ingress connectors. Closing twice does nothing.
Source code in src/desflow/core/context.py
compile_parameters()
Compiles the builder state into the SimParameter that run() registers.
Useful before the run, to see which registry paths a configuration creates or to compare two configurations.
Returns:
| Name | Type | Description |
|---|---|---|
SimParameter |
SimParameter
|
The parameters, sharing their config objects with the builder. |
Source code in src/desflow/core/context.py
finish()
Runs a started run to its end, then closes it, as the rest of run() does.
The end is until, the stop or the end of the drain that start() set
up. After one or more advance() or step() calls it runs on from where
they stopped. When advance() already reached the end, it returns at once
with the end's value: None for until or a drain, the stop's value for a
stop, and a failed stop's exception raised again. When advance() or step()
has brought the clock to until or past it and there is no stop or drain,
it returns None at once, as a run that ended at until does. When step()
brought the clock to exactly until, the records hold one run's records
(which end with the final statistics) plus those of the events at until
that step() processed, as in SimPy; a statistics reading due at until
is not published again, and the close adds only the values that changed
after it. After steps past until, the run closes where the steps stopped,
with the final statistics at that time.
close() runs even when the run stops with an error or Ctrl+C.
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The value of the event that stopped the run, as for |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If the run has not started. |
SignalFailed
|
If the signal that |
Source code in src/desflow/core/context.py
from_yaml(path, preview=None)
classmethod
Builds a context from a YAML blueprint file.
The blueprint is validated before anything is built, and every error names the file and line. The context is built through the same builder methods a Python script calls, so the result is the same simulation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Union[str, Path]
|
The blueprint file. |
required |
preview
|
Optional[Preview]
|
The preview |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The built context. Its |
Raises:
| Type | Description |
|---|---|
BlueprintError
|
If the blueprint is invalid. |
Source code in src/desflow/core/context.py
get_container(container_id)
Retrieves an active DynamicContainer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
container_id
|
str
|
The container's name. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
DynamicContainer |
DynamicContainer
|
The container. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called during the Builder Phase. |
Source code in src/desflow/core/context.py
get_resource(resource_id)
Returns the resource a running model built: a DynamicResource, or the
ResourceBase subclass its type names.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resource_id
|
str
|
The resource's name, as given to |
required |
Returns:
| Name | Type | Description |
|---|---|---|
ResourceBase |
ResourceBase
|
The resource, for raw requests and readings. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called before the run has started. |
Source code in src/desflow/core/context.py
get_store(store_id)
Retrieves an active DynamicStore.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
store_id
|
str
|
The store's name. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
DynamicStore |
DynamicStore
|
The store. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called during the Builder Phase. |
Source code in src/desflow/core/context.py
interrupt(task, where=None, cause=None)
Interrupts every running instance of a task whose condition holds.
Each instance receives SimPy's Interrupt with cause, publishes
interrupted and follows its task's on_interrupt. The caller's own
instance is never interrupted, and an instance that has finished is not
running.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task
|
str
|
The task's name. |
required |
where
|
Union[str, Expression, None]
|
A condition read for each instance, such as |
None
|
cause
|
Any
|
What the instances receive as the interrupt's cause, published
as the record's |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
How many instances were interrupted. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called before the run starts. |
Source code in src/desflow/core/context.py
next_instance(task)
Returns the next instance id of a task, counting from 0.
Each task has one counter, which next uses when it starts the task. An
arrival loop that spawns a task by hand takes the id from here too, so a
task started both by an arrival and by next never repeats an id.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task
|
str
|
The task's name. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
The task's next instance id. |
Source code in src/desflow/core/context.py
peek()
Returns the time of the next model event of a started run, or infinity if none.
The events of desflow's own, such as a statistics reading or a flush timer, are not counted.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If the run has not started. |
Source code in src/desflow/core/context.py
publish(metric_name, value)
Publishes a telemetry value, named <sim_id>.<metric_name>, to the sinks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
metric_name
|
str
|
The value's name, after the sim id, such as |
required |
value
|
Any
|
The value, which must be a JSON value. |
required |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called before the run has started. |
Source code in src/desflow/core/context.py
run(until=None, stop=None, drain=None, on_unhandled_failure=None)
Compiles the defined infrastructure architecture and triggers the simulation loop.
run() is start(), then SimPy's env.run until the end, then close(),
which runs even when the run stops with an error or Ctrl+C.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
until
|
Any
|
The absolute simulation termination time. Can be a numeric float
(representing base seconds) or a human-readable string parsed by
the utility module (e.g., "1 week", "8 hours"). If None, a context
built by |
None
|
stop
|
Any
|
Ends the run before |
None
|
drain
|
Any
|
At |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The value of the event that stopped the run, as SimPy's |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a task's |
SignalFailed
|
If the signal that |
Source code in src/desflow/core/context.py
signal(name)
Returns a declared signal, to fire, fail or wait for from a process.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
A name |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Signal |
Signal
|
The signal: |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called before the run starts. |
Source code in src/desflow/core/context.py
spawn(process, delay=None)
Starts a generator as a SimPy process, such as a task instance.
This acts as the "escape hatch" for complex logic (like physical slab
tracking) that outgrows the strict abstraction of the @task decorator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
process
|
Any
|
A generator, such as |
required |
delay
|
Optional[float]
|
Seconds to wait before starting it, as SimPy's |
None
|
Returns:
| Type | Description |
|---|---|
Process
|
simpy.Process: The process. Yield it to wait until it ends; its value is
what the generator returns, and |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/core/context.py
start(until=None, stop=None, drain=None, *, seed=None, overrides=None, stats=None, egress=True, ingress=True, run_settings=True, on_unhandled_failure=None)
Builds and starts everything run() does before the clock moves.
It checks the tasks, calls the blueprint's run.before functions, builds the
environment, the sampler, the statistics, the parameters, the connectors,
the resources, stores and containers, and starts every loop and process.
Then step() processes one event at a time and peek() reads the time of
the next, as SimPy's env.step() and env.peek() do; close() ends the
run. A context built by from_yaml takes its run section's values.
Replications and experiments start each run here too, with the arguments after
drain, and finish it with finish(), so a replication builds and ends
exactly as run() does, run.stop and run.drain included.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
until
|
Any
|
As for |
None
|
stop
|
Any
|
As for |
None
|
drain
|
Any
|
As for |
None
|
seed
|
Optional[int]
|
The run's seed, in place of |
None
|
overrides
|
Optional[Mapping[str, Any]]
|
Registry paths and the values they take before the clock starts, applied as a live update is, except that a value the registry would refuse or clamp raises. |
None
|
stats
|
Optional[str]
|
The lowest statistics level the run keeps; a higher level set
with |
None
|
egress
|
bool
|
False builds the run without its sinks and without the statistics timer, as a replication runs: nothing receives its records, and its statistics are read at the end. |
True
|
ingress
|
bool
|
False builds the run without its ingress, so no connector listens for parameter changes. |
True
|
run_settings
|
bool
|
False leaves out a blueprint's |
True
|
Returns:
| Name | Type | Description |
|---|---|---|
DynamicRealtimeEnvironment |
DynamicRealtimeEnvironment
|
The environment, also |
Raises:
| Type | Description |
|---|---|
ValueError
|
As for |
Source code in src/desflow/core/context.py
2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 | |
step()
Processes the next model event of a started run, as SimPy's env.step() does.
The events of desflow's own that are due before it, such as a statistics reading or a flush timer, run in the same call and are not counted.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If the run has not started. |
EmptySchedule
|
If no model event is left. |
Source code in src/desflow/core/context.py
stop_arrivals()
Stops every arrival loop, so no new entity arrives; the entities already in
the model carry on. A drain calls it at until.
Source code in src/desflow/core/context.py
subscribe(event)
Asks to be interrupted when event happens, as SimPy's subscribe_at.
The calling process keeps running; when event fires it is interrupted one
urgent step later with the cause {"event": name, "value": value}, where
name is the process's name for a process and None for any other event.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event
|
Any
|
Any SimPy event, a process included. |
required |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called outside a process, or |
Source code in src/desflow/core/context.py
task(service_id=None, resource_id=None, *, name=None, priority=None, get=None, put=None, next=None, set=None, patience=None, balk=None, wait=None, signal=None, fail=None, fail_reason=None, on_fail=None, keep=None, preempt=True, on_interrupt='resume', on_interrupt_next=None, interrupt=None, spawn=None, subscribe=None, on=None)
Transforms a standard Python function into an event-driven SimPy task.
This decorator completely abstracts the boilerplate of resource acquisition, clock synchronization, and standardized telemetry lineage.
Lifecycle Execution:
1. Checks balk; when it holds, emits balked and stops.
2. Emits a queued event, when the task waits for anything: its wait, a
resource or a get.
3. Waits for its wait, then for a unit of each resource in resource_id,
in the order listed and each in its queue order, then for its get;
patience counts the time spent waiting for all three, and pauses
while the task is served.
4. Emits a started event.
5. Executes the wrapped user function to extract custom payload data.
6. Samples the live service_id distribution and yields the timeout.
7. Adds its put, waiting for room, and checks fail.
8. Emits its finished event with the user's payload, or failed.
9. Fires its signal (or fails it), sends its interrupt action, applies
set to the entity's attributes and starts the next task, or
on_fail.
Every event carries the entity's key as its record key, and the task's name
and instance id as the record's task and instance fields.
A task with a service and no resource is a timed step: no queued event, so
its wait is 0. A task with no service, resource or get takes no time: when
spawned it emits the payload at once (after its put, if it has one), with
no queued or started event.
Call the task as task(task_id, ...). Pass entity= with the entity
wait_for_arrival returned; without one the task starts an entity of its
own, keyed by its name and a counter. The function receives entity,
upstream (what the previous task handed on, or None) and item (what a
store get took) by keyword, each only when it declares that parameter, so
existing functions keep working. A task started by next is called with its
own instance id and the same other arguments as the task before it. A
function that declares wait receives the wait's branches, each a mapping
of done, ok and value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service_id
|
Optional[str]
|
The ID of the configured distribution dictating the execution time. |
None
|
resource_id
|
Union[str, List[str], None]
|
The resource the task holds while it works, or a list of them, acquired in the order listed while the earlier ones are held. |
None
|
name
|
Optional[str]
|
The task's name in records, statistics and |
None
|
priority
|
Union[float, str, Callable[[Any, Dict[str, Any]], float], None]
|
For a resource whose queue order is |
None
|
get
|
Union[Transfer, Mapping[str, Any], None]
|
What the task takes before it starts: |
None
|
put
|
Union[Transfer, Mapping[str, Any], None]
|
What the task adds when it finishes, in the same form; a store receives the finished payload. |
None
|
next
|
NextSpec
|
The task that follows: a task name, |
None
|
set
|
Optional[Mapping[str, Any]]
|
Attribute values applied to the entity when the task finishes. |
None
|
patience
|
Union[float, str, Callable[[Any, Dict[str, Any]], float], None]
|
How long the task may wait, for its |
None
|
balk
|
Optional[str]
|
A condition, such as |
None
|
wait
|
Optional[Mapping[str, Any]]
|
What the task waits for before it queues for its resource: one
branch, |
None
|
signal
|
Optional[str]
|
A signal, added with |
None
|
fail
|
Union[str, Callable[[Any, Any], bool], None]
|
A condition checked after the service, such as
|
None
|
fail_reason
|
Optional[str]
|
The |
None
|
on_fail
|
Optional[str]
|
The task a failed instance goes to. Without it, a failure ends the entity's flow. A function that raises fails the task the same way, with the error as the reason. |
None
|
keep
|
Optional[List[str]]
|
Resources in |
None
|
preempt
|
bool
|
On a resource added with |
True
|
on_interrupt
|
str
|
What an interrupt or a preemption does: |
'resume'
|
on_interrupt_next
|
Optional[str]
|
The task an abandoned instance goes to. |
None
|
interrupt
|
Optional[Mapping[str, Any]]
|
An action applied when the task finishes:
|
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If a resource is given without a service, a priority without
a resource, the name is taken, or |
Returns:
| Name | Type | Description |
|---|---|---|
Callable |
Callable
|
A decorator that wraps the target function into a SimPy generator. |
Example
Source code in src/desflow/core/context.py
996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 | |
telemetry_loop(interval)
Registers an infinite background daemon loop for periodic data publishing.
This is primarily used for observing and extracting continuous state (like wear containers) at strict intervals.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
interval
|
float
|
The simulation time (in seconds) to wait between iterations. |
required |
Source code in src/desflow/core/context.py
wait_for_arrival(arrival_id)
Returns the wait for the next arrival: a timeout from the live distribution, or for a replayed arrival the wait for its next row.
The timeout's value is the new Entity, so an arrival loop writes
entity = yield context.wait_for_arrival("orders") and passes
entity=entity to the task it spawns.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
arrival_id
|
str
|
The arrival's name. |
required |
Returns:
| Type | Description |
|---|---|
Event
|
simpy.Event: A timeout for the gap to the next arrival, or the wait for
the next replayed row, which succeeds with the new |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If called during the Builder Phase (before |
Source code in src/desflow/core/context.py
with_batching(batch_size, flush_interval, max_queued_batches=2000, drain_stall_seconds=30.0)
Configures the internal egress buffering strategy to mitigate I/O lock contention.
batch_size and flush_interval are defaults. A provider that names either
on add_egress uses its own instead, which is how a stream sink and a lake
sink share one run.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
batch_size
|
int
|
The maximum number of standard events to hold in memory before forcing an asynchronous flush to the egress thread. |
required |
flush_interval
|
float
|
The maximum simulation time (in seconds) to wait before forcefully flushing the buffer, regardless of capacity. |
required |
max_queued_batches
|
int
|
Upper bound on batches waiting for the egress threads. A full queue blocks the producer, so a sink that cannot keep up slows the simulation instead of building a backlog that teardown would then have to discard. |
2000
|
drain_stall_seconds
|
float
|
How long teardown keeps waiting once the queue has stopped shrinking. While it keeps draining the wait is unbounded, so a slow sink finishes rather than losing its tail. |
30.0
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Source code in src/desflow/core/context.py
with_stats(level='basic', interval=DEFAULT_STATS_INTERVAL)
Switches on the built-in statistics, published as telemetry.
Each level publishes what the levels before it publish, and more: minimal
the readings of every resource, store and container right now; basic
finished counts and time-weighted averages; typical the mean and maximum
wait and service time of each task; all their 50th, 95th and 99th
percentiles. Names look like <sim_id>.stats.tasks.<task>.wait_mean.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
level
|
str
|
|
'basic'
|
interval
|
float
|
Simulation seconds between publications; they are also published once at the end of the run, except the values unchanged since a reading at that same time. Defaults to 10. |
DEFAULT_STATS_INTERVAL
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The current instance for method chaining. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the level is unknown or the interval is not positive. |
Source code in src/desflow/core/context.py
Task Flow
A task's flow is built from the classes below. The signals and failures here are shared between tasks.
Signal(env, name='signal', once=False)
A named event that can fire many times, each firing waking every waiter.
Attributes:
| Name | Type | Description |
|---|---|---|
env |
The environment. |
|
name |
str
|
The signal's name. |
count |
int
|
How many times it has fired or failed. |
ok |
Optional[bool]
|
Whether the last firing succeeded; None before the first. |
value |
Any
|
The value of the last firing, or the failure reason. |
Creates a signal that has not fired yet.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Environment
|
The environment. |
required |
name
|
str
|
Its name, as a blueprint's |
'signal'
|
once
|
bool
|
When True the signal fires once, as SimPy's |
False
|
Source code in src/desflow/core/signals.py
Functions
attach(callbacks)
Keeps callbacks as the signal's on: callbacks and adds them to the
current pending event, so they run when the signal next fires, before any
waiter, as SimPy runs a callback added before the waiters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callbacks
|
List[Callable[[Any], None]]
|
Functions called with the fired event, copied onto each new pending event so they run on every firing. |
required |
Source code in src/desflow/core/signals.py
fail(reason, urgent=False, after=0.0)
Fails the signal: every waiter receives SignalFailed with the reason.
A failure nobody waits for is not an error: the event is defused, so the run goes on.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reason
|
Any
|
Why it failed, such as a task's failure reason. |
required |
urgent
|
bool
|
As for |
False
|
after
|
float
|
As for |
0.0
|
Raises:
| Type | Description |
|---|---|
ValueError
|
As for |
Source code in src/desflow/core/signals.py
fire(value=None, urgent=False, after=0.0)
Fires the signal: every waiter resumes with value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
Any
|
What the waiters receive, such as a task's finished payload. |
None
|
urgent
|
bool
|
When True the waiters resume before the other events at their
time, as SimPy's |
False
|
after
|
float
|
Simulation seconds from now at which the waiters resume. The signal counts as fired now, so a second firing is refused. Needs a one-shot signal. |
0.0
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/core/signals.py
wait()
Returns the event of the next firing, for a process to yield.
Returns:
| Type | Description |
|---|---|
Event
|
simpy.Event: Succeeds with the firing's value, or fails with
|
SignalFailed(signal, reason)
Bases: Exception
The exception a waiter of a failed signal receives.
Attributes:
| Name | Type | Description |
|---|---|---|
signal |
str
|
The signal's name. |
reason |
Any
|
What |
Source code in src/desflow/core/signals.py
TaskFailed(task, reason)
Bases: Exception
A task instance failed, and no on_fail handled it.
A child flow, started by a wait branch or context.call, fails with it, so
the task waiting for the flow fails too.
Attributes:
| Name | Type | Description |
|---|---|---|
task |
str
|
The task that failed. |
reason |
Any
|
Why: its |
Source code in src/desflow/core/task.py
Entity(key, arrival, attributes=dict(), held=dict(), entered=False, instances=None, spawned_by=None, ends_pending=0, ending=False, last_ended=None)
dataclass
One thing moving through the model, such as an order, a slab or a session.
Each arrival starts a new entity. Its key is the record key of every event it causes, in every task it passes through.
Attributes:
| Name | Type | Description |
|---|---|---|
key |
str
|
The record key: the arrival name and a counter, such as
|
arrival |
str
|
The arrival that started it, which names its flow in the statistics. |
attributes |
Dict[str, Any]
|
State kept across tasks: set by the arrival's
|
held |
Dict[str, Any]
|
Resource units a task kept for the next task with
|
Attributes
left
property
True once _entity_left has run: the flow has ended and nothing is
pending.
Upstream(task_id, key, attributes, payload=None)
dataclass
What a finished task hands the task routed after it.
Attributes:
| Name | Type | Description |
|---|---|---|
task_id |
Any
|
The instance id of the finished task. |
key |
str
|
The entity's key. |
attributes |
Dict[str, Any]
|
The entity's attributes, after the finished
task's |
payload |
Any
|
The finished task's payload, read in conditions as
|
Attributes
fields
property
The payload when it is a mapping, and an empty mapping otherwise.
Router(rules, rng=None)
Chooses the task that follows a finished one.
The rules take one of these forms, in every API:
- a task name: always that task;
{"when": [{"condition": "...", "task": "..."}, ...], "default": "..."}: the first rule whose condition holds wins, anddefault, if given, catches the rest; with no default and no rule holding, the flow ends;{"weights": {"ship": 9, "rework": 1}}: a task drawn with these weights from the routing task's own random sequence, so seeded runs repeat;- a function taking the
Upstreamand returning a task name or None; - None: the flow ends.
A route back to an earlier task (rework) is allowed.
Attributes:
| Name | Type | Description |
|---|---|---|
tasks |
Optional[Set[str]]
|
Every task the rules can name, so a loader can check they exist; None for a function. |
rng |
Optional[Generator]
|
The generator weights draw from. The builder sets it when the run starts. |
Checks the rules and parses their conditions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rules
|
NextSpec
|
The rules, in one of the forms above. |
required |
rng
|
Optional[Generator]
|
The generator weights draw from, such as
|
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If the rules are malformed or a condition is refused. |
Source code in src/desflow/core/flow.py
Attributes
conditions
property
The parsed conditions, in rule order.
probabilities
property
The share of finished instances each next task receives, where the rules fix it: 1.0 for a task name, the normalised weights for weights, and an empty mapping when the flow ends. None for when rules or a function, whose split depends on the run.
weighted
property
True when the rules are weights, which draw from rng.
Functions
choose(upstream, env=None, levels=None, item=None, scope=None)
Returns the next task's name, or None when the flow ends.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
upstream
|
Upstream
|
What the finished task hands on. |
required |
env
|
Any
|
The environment, for |
None
|
levels
|
Optional[Mapping[str, Any]]
|
Stores and containers by name, for |
None
|
item
|
Any
|
The item the finished task's store get took, for |
None
|
scope
|
Optional[Scope]
|
What the conditions read, when the caller has built it; it replaces the one built from the arguments above. |
None
|
Returns:
| Type | Description |
|---|---|
Optional[str]
|
Optional[str]: The next task, or None. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a condition reads a name that is missing. |
Source code in src/desflow/core/flow.py
Transfer(container=None, store=None, amount=1.0, priority=0.0, filter=None, wait=True)
What a task takes from, or adds to, a store or a container.
A container transfer moves an amount: a number, an expression such as
payload.quantity, or in Python a function of the task id and the upstream
payload. A store transfer moves one item: a get takes the next item, and a
put adds the task's finished payload.
A put with wait=False is made and not waited for: the task goes on at once, as
a SimPy process that calls store.put(item) without yield. The put stays in the
queue, completes in first in, first out order whenever there is room, and is
never cancelled by an interrupt, a patience or an abandon. It publishes no
record, a drain does not wait for it, and one still queued when the run ends is
dropped.
Attributes:
| Name | Type | Description |
|---|---|---|
container |
Optional[str]
|
The container's name. |
store |
Optional[str]
|
The store's name. |
amount |
The amount, for a container. |
|
wait |
bool
|
False for a put the task does not wait for. |
Checks the transfer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
container
|
Optional[str]
|
The container's name. |
None
|
store
|
Optional[str]
|
The store's name, instead of a container. |
None
|
amount
|
Amount
|
For a container, a number, an expression or a function. |
1.0
|
priority
|
Any
|
For a put into a priority store, any value SimPy can order: a
number, a string, a tuple, or an expression or function that
returns one, lowest served first. A string is an expression. It is
kept as given, so |
0.0
|
filter
|
Union[str, Callable[[Any], bool], None]
|
For a store get, which item it takes: a condition such as
|
None
|
wait
|
bool
|
False for a put the task does not wait for (see above). Defaults to True. |
True
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If not exactly one of container and store is given, an
amount is given for a store, a priority is given for a
container, the amount is malformed, the priority is an expression
that is malformed, or |
Source code in src/desflow/core/flow.py
Attributes
expressions
property
Every expression the transfer holds, so a loader can check its names.
target
property
The name of the store or container.
Functions
amount_for(task_id, upstream, scope)
Returns the amount for one task instance.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task_id
|
Any
|
The instance id. |
required |
upstream
|
Optional[Upstream]
|
What the previous task handed on. |
required |
scope
|
Scope
|
What an expression's names read. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
float |
float
|
The amount. |
Source code in src/desflow/core/flow.py
filter_for(scope)
Returns the function a store get offers each item to, or None for any item.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scope
|
Scope
|
What the condition's other names read, taken when the get is made;
|
required |
Returns:
| Type | Description |
|---|---|
Optional[Callable[[Any], bool]]
|
Optional[Callable[[Any], bool]]: True for an item the get may take. |
Source code in src/desflow/core/flow.py
of(spec)
classmethod
Returns a transfer from a mapping such as {"container": "stock", "amount": 2}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
spec
|
Union[Transfer, Mapping[str, Any]]
|
A mapping with |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Transfer |
Transfer
|
The transfer. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the mapping has other keys or is malformed. |
Source code in src/desflow/core/flow.py
priority_for(task_id, upstream, scope)
Returns the priority for one put into a priority store, lowest first.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task_id
|
Any
|
The instance id. |
required |
upstream
|
Optional[Upstream]
|
What the previous task handed on. |
required |
scope
|
Scope
|
What an expression's names read. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The priority as it was given or computed, not changed to a number. |
Source code in src/desflow/core/flow.py
Loading a Model
SimulationContext.from_yaml is the usual entry point for a blueprint. from_yaml calls build, which returns the context and the run section. load_context loads a blueprint or a builder Python file from a folder, local or in object storage, in the way desflow run loads a blueprint. desflow run --sample and --stdout pass a Preview to these functions.
load_context(folder, entry, preview=None)
Builds the simulation that entry in folder defines.
An entry ending in .py is a builder file: it is imported under a new module
name each call, with folder first on sys.path during the import, so its
imports resolve from its folder and its if __name__ == "__main__": block does
not run. Its DESFLOW_APP names the context to use; without it, the file must
define exactly one SimulationContext at module level. Any other entry is a
YAML blueprint, built by SimulationContext.from_yaml.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
folder
|
Union[str, Path]
|
A local folder, or a URI that |
required |
entry
|
str
|
The entry file's name inside |
required |
preview
|
Optional[Preview]
|
The preview |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
SimulationContext |
SimulationContext
|
The built context, not yet run. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a Python entry defines no context, or several with no
|
ImportError
|
If |
BlueprintError
|
If a blueprint is invalid. |
Source code in src/desflow/loading.py
build(path, preview=None)
Reads, validates and builds a blueprint file.
The structure, keys and names of the whole file are validated with or without
a preview. A preview builds none of the connectors it replaces or skips, so it
checks only their type; their config is checked only by a run that builds
them, such as desflow run FILE without --sample or --stdout.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Union[str, Path]
|
The YAML file. |
required |
preview
|
Optional[Preview]
|
Replace the file's egress with a JSON console, and for a sample leave out its ingress; None builds the file as written. |
None
|
Returns:
| Type | Description |
|---|---|
Tuple[SimulationContext, Run]
|
The built context, and the |
Raises:
| Type | Description |
|---|---|
BlueprintError
|
If the file is invalid. The message names the file and line. |
Source code in src/desflow/blueprint/build.py
Preview(sample=None)
dataclass
A preview run of a blueprint, as desflow run --sample and --stdout make.
Every egress entry is replaced by a JSON console on stdout, and a second one for
labels when the file has a when: labels entry, so no sink module the file
names by a short type is imported, and no sink is written to.
Attributes:
| Name | Type | Description |
|---|---|---|
sample |
Optional[int]
|
Print this many events, unpaced and with no ingress, then stop.
None, for |
BlueprintError
Bases: ValueError
A blueprint that cannot be built. The message names the file and line.
Replications
replicate(model, n, *, seed, until=None, warmup=None, measures=None, level=0.95, workers=1, overrides=None, stop=None, drain=None)
Runs a model n times with spawned seeds and summarises its statistics.
Each replication runs unpaced, with no ingress and no egress, and reads the
built-in statistics at its end. When the model has sinks (a blueprint's
egress, or a builder's add_egress), they receive the summary as telemetry
once, at the end: each replication's value and the mean, half-width and count of
each measure.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
Model
|
A YAML blueprint path, a builder factory (a function with no
arguments returning a new |
required |
n
|
int
|
How many replications, 2 or more. |
required |
seed
|
Optional[int]
|
The seed the replications' seeds are spawned from. None draws one and logs it. |
required |
until
|
Union[float, str, None]
|
Where each replication's arrivals stop, in seconds or as a duration
such as |
None
|
warmup
|
Union[float, str, None]
|
The end of each replication's warm-up; None keeps the model's own. |
None
|
measures
|
Optional[Sequence[str]]
|
Statistic names without the sim id, such as
|
None
|
level
|
float
|
The confidence level of the half-widths. Defaults to 0.95. |
0.95
|
workers
|
int
|
Replications run at once, each in its own process. Above 1 the model must be a path or a function defined at the top of a module. |
1
|
overrides
|
Optional[Mapping[str, Any]]
|
Registry paths and the values they take in every replication. |
None
|
stop
|
Any
|
Ends each replication before |
None
|
drain
|
Any
|
At |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
Replications |
Replications
|
Every value with its seed, and the summary. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/analysis/runner.py
182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 | |
Replications(sim_id, seeds, measures, values, level=0.95, streams=list())
dataclass
The result of replicate: every value with its seed, and a summary per measure.
Attributes:
| Name | Type | Description |
|---|---|---|
sim_id |
str
|
The model's sim id. |
seeds |
List[int]
|
The seed of each replication, in order. Replication |
measures |
List[str]
|
The statistics summarised, such as
|
values |
Dict[str, List[float]]
|
Each measure's value in each replication. |
level |
float
|
The confidence level of the half-widths. |
streams |
List[List[str]]
|
The random streams each replication drew from. |
Attributes
summary
property
One row per measure: measure, n, mean, half_width and level.
table
property
One row per replication and measure: replication (from 1), seed,
measure and value.
Functions
summary_text()
The summary as a table: each measure's n, mean and half-width.
Source code in src/desflow/analysis/runner.py
telemetry()
The telemetry records the sinks receive, as (path_id, value) pairs.
One per measure and replication, <sim_id>.stats.<measure>.replication.<i>,
then the mean, half-width and count across replications,
<sim_id>.stats.<measure>.mean, .half_width and .n.
Source code in src/desflow/analysis/runner.py
LowLevelModel(function, sim_id)
dataclass
A model written with the low-level API, for replicate and run_experiment.
Attributes:
| Name | Type | Description |
|---|---|---|
function |
Callable
|
Called as |
sim_id |
str
|
The sim id the parameters are registered under, which the statistics' names start with. |
interval(values, level=0.95)
Returns the mean of values and the Student t half-width of its interval.
The half-width is t × s / √n, where s is the sample standard deviation and
t the quantile at (1 + level) / 2 with n - 1 degrees of freedom.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
values
|
Sequence[float]
|
Independent values, such as one statistic from each replication. |
required |
level
|
float
|
The confidence level, between 0 and 1. Defaults to 0.95. |
0.95
|
Returns:
| Name | Type | Description |
|---|---|---|
Interval |
Interval
|
The mean and its half-width. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If there are fewer than 2 values or the level is not between 0 and 1. |
Source code in src/desflow/analysis/intervals.py
Interval(n, mean, std, half_width, level)
dataclass
A mean and the half-width of its confidence interval.
Attributes:
| Name | Type | Description |
|---|---|---|
n |
int
|
How many values the mean is of. |
mean |
float
|
Their mean. |
std |
float
|
Their sample standard deviation. |
half_width |
float
|
The Student t half-width at |
level |
float
|
The confidence level, such as 0.95. |
Attributes
high
property
The upper bound, mean + half_width.
low
property
The lower bound, mean - half_width.
t_quantile(p, df)
Returns the value a Student t variable is at most with probability p.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
p
|
float
|
The probability, between 0 and 1. |
required |
df
|
float
|
The degrees of freedom, above 0. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
float |
float
|
The quantile, such as 2.262 for |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/analysis/intervals.py
Experiments
run_experiment(spec, *, raise_on_failure=True, output=None)
Runs an experiment: its variants and sweep, replications, comparisons and checks.
Every artifact is written to the experiment's output folder before anything is
raised: experiment.json (settings and evidence), results.csv, summary.csv,
comparisons.csv, checks.csv and report.md.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
spec
|
Union[str, Path, Mapping[str, Any], Any]
|
An experiment file, a mapping of the same settings, or an
|
required |
raise_on_failure
|
bool
|
When True, a failed check raises |
True
|
output
|
Union[str, Path, None]
|
The folder to write the artifacts to, in place of the experiment's
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
ExperimentResult |
Any
|
What the experiment found. |
Raises:
| Type | Description |
|---|---|
BlueprintError
|
If the experiment file is invalid, or the model or observed file it names does not exist, naming the file and line. |
ValueError
|
If the settings are invalid, a variant sets a path the model does not have, or the observed values do not match the measures; for settings given as a mapping, also a missing model or observed file. |
CheckFailed
|
If a check failed and |
Source code in src/desflow/analysis/runner.py
431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 | |
ExperimentSpec
Bases: _Model
A whole experiment file, or the same settings given in Python.
model is a blueprint path, a builder factory (a function with no arguments
returning a new SimulationContext), a LowLevelModel, or in a file a mapping
of a low-level function and its sim_id. Relative paths are read from the
experiment file's folder. until is where each replication's arrivals stop;
stop and drain end each replication as SimulationContext.run does, and
when they are not set a blueprint's own run.stop and run.drain apply.
load_experiment(path)
Reads and validates an experiment file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Union[str, Path]
|
The experiment file. |
required |
Returns:
| Type | Description |
|---|---|
tuple[ExperimentSpec, SourceMap]
|
The settings, and the map from each key to its line. |
Raises:
| Type | Description |
|---|---|
BlueprintError
|
If the file is invalid; the message names the file and line. |
Source code in src/desflow/analysis/spec.py
ExperimentResult(spec, output, designs, results, summary, comparisons, checks, observed, report)
dataclass
What an experiment found, as run_experiment returns it and its artifacts hold it.
Attributes:
| Name | Type | Description |
|---|---|---|
spec |
ExperimentSpec
|
The experiment's settings. |
output |
Path
|
The folder the artifacts were written to. |
designs |
List[Design]
|
Each variant at each sweep point, with its seeds. |
results |
List[Dict]
|
One row per design, replication and statistic. |
summary |
List[Dict]
|
One row per design and statistic. |
comparisons |
List[Dict]
|
One row per compared pair, point and measure. |
checks |
List[Dict]
|
One row per check, design and bound. |
observed |
List[Dict]
|
Each observed value beside its simulated interval. |
report |
str
|
The Markdown report. |
Attributes
failed
property
The check rows that failed.
CheckFailed(result)
Bases: Exception
An experiment's check failed. The artifacts are written before it is raised.
Attributes:
| Name | Type | Description |
|---|---|---|
result |
ExperimentResult
|
The whole result, failed checks included. |
Source code in src/desflow/analysis/results.py
Input Fitting
fit_column(values, *, unit='s', candidates=None, lower=None, upper=None, bins=None, min_count=20, tolerance=0.25, drop_invalid=False)
Fits a column of measured times to desflow's distribution families.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
values
|
Sequence[Any]
|
The data, as numbers or text; an empty cell or None is missing. |
required |
unit
|
str
|
The data's unit: |
's'
|
candidates
|
Optional[Sequence[str]]
|
The families to try; None tries every family. |
None
|
lower
|
Optional[float]
|
The lowest plausible value, in the data's unit; a value below it is invalid, and a family that can draw below it is flagged. |
None
|
upper
|
Optional[float]
|
The highest plausible value, in the same way. |
None
|
bins
|
Optional[int]
|
Offer the empirical distribution as a histogram with this many bins, rather than as the values themselves. |
None
|
min_count
|
int
|
The fewest values a family is fitted to. Defaults to 20. |
20
|
tolerance
|
float
|
How far a fitted family's mean and standard deviation may be from the data's, as a share of the data's. Defaults to 0.25. |
0.25
|
drop_invalid
|
bool
|
Leave invalid values out rather than raising. |
False
|
Returns:
| Name | Type | Description |
|---|---|---|
FitResult |
FitResult
|
Every candidate's outcome, and the best. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the unit or a candidate is unknown, or a value is invalid
and |
ImportError
|
If the fit extra is not installed. |
Source code in src/desflow/analysis/fit.py
380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 | |
FitResult(candidates, values, unit='s', rows=0, missing=0, invalid=0, lower=None, upper=None)
dataclass
The outcome of fitting one column: every candidate, the best and the data.
Attributes:
| Name | Type | Description |
|---|---|---|
candidates |
List[Candidate]
|
One per family tried, empirical last. |
values |
List[float]
|
The values fitted, in seconds. |
unit |
str
|
The unit the data was written in. |
rows |
int
|
Rows read. |
missing |
int
|
Empty cells, left out. |
invalid |
int
|
Values that are not numbers, not finite, or outside the
bounds; left out with |
lower |
Optional[float]
|
The lowest plausible value, in seconds, or None. |
upper |
Optional[float]
|
The highest plausible value, in seconds, or None. |
Attributes
best
property
The fitted family with the lowest AIC that cannot draw outside the bounds;
constant when every value is the same. When no family within the bounds
fitted, the empirical distribution of the data, which stays within them;
then the fitted family with the lowest AIC; None when nothing is usable.
table
property
One row per candidate, with its status, reason, parameters and diagnostics, and whether it is the best.
Functions
plot(path, kind=None)
Draws the data against values generated from a candidate, as a PNG.
The left panel puts the histogram of the data over the histogram of
generated values; the right is a Q-Q plot, each sorted data value against
the generated value at the same quantile. The generated values come from
desflow's Sampler, as a run draws them, with the same smallest delay.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Union[str, Path]
|
The image to write. |
required |
kind
|
Optional[str]
|
The family; None takes the best. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
Path |
Path
|
The image written. |
Source code in src/desflow/analysis/fit.py
settings(kind=None)
Returns a usable candidate's settings, for add_service(name, **settings).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kind
|
Optional[str]
|
The family; None takes the best. |
None
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict[str, Any]: |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no candidate is usable, or |
Source code in src/desflow/analysis/fit.py
text()
The table as text: one line per candidate, the best marked.
Source code in src/desflow/analysis/fit.py
yaml(kind=None, source='')
The settings as a YAML mapping a blueprint loads with !include.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kind
|
Optional[str]
|
The family; None takes the best. |
None
|
source
|
str
|
Where the data came from, for the comment at the top. |
''
|
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
A comment line, then |
Source code in src/desflow/analysis/fit.py
Candidate(family, status, reason='', parameters=dict(), log_likelihood=None, aic=None, ks_statistic=None, ks_p_value=None, bounds='')
dataclass
One family's outcome: its parameters, diagnostics and status.
Attributes:
| Name | Type | Description |
|---|---|---|
family |
str
|
The family, as |
status |
str
|
|
reason |
str
|
Why it was rejected, did not fit or was skipped; empty when fitted. |
parameters |
Dict[str, Any]
|
desflow's parameters, in seconds. |
log_likelihood |
Optional[float]
|
The log-likelihood of the data. |
aic |
Optional[float]
|
Akaike's information criterion, 2k minus twice the log-likelihood, where k counts the fitted parameters. |
ks_statistic |
Optional[float]
|
The Kolmogorov-Smirnov distance between the data and the fitted family. |
ks_p_value |
Optional[float]
|
Its p-value. It is approximate, because the parameters were estimated from the same data. |
bounds |
str
|
Where the family can draw outside |
Attributes
usable
property
True when the family was fitted and its result passed every check.
Explaining a Model
SimulationContext.analyze is the usual entry point, and desflow explain prints the same Explanation. build_graph, analyze and verify are the steps it takes, and render_terminal and to_json print the result. The JSON form is experimental through 1.x.
Explanation
Bases: _Model
A model's graph and its analysis.
Attributes:
| Name | Type | Description |
|---|---|---|
schema_version |
Literal['0']
|
The version of this shape, |
sim_id |
str
|
The model's sim id. |
target |
float
|
The utilisation the hints aim for. |
graph |
Graph
|
The graph. |
arrivals |
List[ArrivalResult]
|
Each arrival. |
tasks |
List[TaskResult]
|
Each task. |
resources |
List[ResourceResult]
|
Each resource. |
holders |
List[HolderResult]
|
Each store and container. |
bottleneck |
Optional[str]
|
The resource with the highest utilisation the formulas cover. |
bottleneck_reason |
Optional[str]
|
Why no bottleneck can be named, when none is. |
hints |
List[Hint]
|
Changes that bring each resource to |
verification |
Optional[Verification]
|
What |
build_graph(context)
Builds the graph of a context from what it declares, without starting it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
context
|
SimulationContext
|
A builder or YAML context, before or after a run. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Graph |
Graph
|
Its arrivals, tasks, resources, stores, containers, opaque code and the exit, with the edges between them. |
Source code in src/desflow/explain/graph.py
analyze(graph, target=0.85)
Analyses a graph: rates by flow balance, then each resource's load and wait.
A value is given only where the formulas cover it (support.py); everywhere
else it is None, with the reason.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
graph
|
Graph
|
The graph |
required |
target
|
float
|
The utilisation the hints aim for, above 0 and at most 10. |
0.85
|
Returns:
| Name | Type | Description |
|---|---|---|
Explanation |
Explanation
|
The graph and its analysis. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/explain/analysis.py
verify(model, explanation, until=None, seed=None, replications=None)
Runs the model unpaced and adds the comparison to the explanation.
Each run is a replication's run (run_once): no ingress, no egress, statistics
at typical. It compares, per resource, utilization_avg / 100 with ρ and
queue_length_avg with Lq; per task with one resource, wait_mean with Wq; per
arrival, flows.<arrival>.finished over the time after the warm-up with λ.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
Any
|
A blueprint path, or a function that builds the context again. |
required |
explanation
|
Explanation
|
The analysis to compare with. |
required |
until
|
Union[float, str, None]
|
The end of each run; the blueprint's |
None
|
seed
|
Optional[int]
|
The run's seed, or the seed the replications' seeds come from; the
model's |
None
|
replications
|
Optional[int]
|
The number of runs, 2 or more; None for one run. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
Explanation |
Explanation
|
A copy of |
Raises:
| Type | Description |
|---|---|
ValueError
|
If there is no end time, the end is not after the warm-up, or
|
Source code in src/desflow/explain/verify.py
render_terminal(explanation, width=None)
Returns the explanation as plain text: the graph as a tree from each arrival,
the resources, the units each allows, the bottleneck, what is not estimated and
why, the hints, and with --verify the comparison.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
explanation
|
Explanation
|
What |
required |
width
|
Optional[int]
|
The width in characters; the terminal's, or 80, when None. |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
The text, ending in a newline. |
Source code in src/desflow/explain/render.py
to_json(explanation)
Returns the explanation as JSON, with schema_version "0".
The shape is experimental through 1.x; explain.schema.json, beside this
module, is its schema.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
explanation
|
Explanation
|
What |
required |
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
The JSON text. |
Source code in src/desflow/explain/render.py
Environment
DynamicRealtimeEnvironment(initial_time=0, factor=1.0, strict=True, logical_start_time=None, go_live_at=None, timezone=None, on_unhandled_failure='stop')
Bases: RealtimeEnvironment, RegistryMixIn, IngressMixIn, EgressMixIn, StatisticsMixIn
The core simulation engine for desflow.
This environment extends SimPy's RealtimeEnvironment by incorporating
a centralized SimulationRegistry for dynamic state updates, and MixIns
for managing high-throughput asynchronous I/O with external systems.
Attributes:
| Name | Type | Description |
|---|---|---|
start_datetime |
datetime
|
The real-world clock time when the simulation started, with its time zone. |
zone |
tzinfo
|
The time zone timestamps are written in. |
Initializes the real-time simulation environment.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
initial_time
|
float
|
The initial simulation time. Defaults to 0. |
0
|
factor
|
float
|
The real-time factor (e.g., 1.0 = 1 sim second per real second). Defaults to 1.0. |
1.0
|
strict
|
bool
|
If True, a paced step raises |
True
|
logical_start_time
|
datetime
|
Overrides the environment's base clock. Crucial for historical backfilling (e.g., generating data from last week). |
None
|
go_live_at
|
datetime
|
Logical instant at which pacing switches to
real time. Until then the run uses |
None
|
timezone
|
str
|
An IANA time zone name, such as
|
None
|
on_unhandled_failure
|
str
|
|
'stop'
|
The clock runs in UTC, so a daylight-saving change in the zone never adds or drops simulation time. An instant at or before the start of the run makes the whole run real time, which is the degenerate case of the same rule.
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/core/environment.py
878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 | |
Functions
setup_ingress(providers)
Initializes the ingress queues and starts the background listener threads.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
providers
|
List[BaseIngress]
|
A list of initialized ingress connector instances. |
required |
Source code in src/desflow/core/environment.py
setup_egress(providers, batch_size=500, flush_interval=1.0, lag_monitor_interval='auto', max_queued_batches=2000, drain_stall_seconds=30.0, predicates=None, batch_sizes=None, flush_intervals=None, labels=None, warmup=None)
Initializes the egress buffers and starts the background publisher threads.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
providers
|
List[BaseEgress]
|
A list of initialized egress connector instances. |
required |
batch_size
|
int
|
The maximum number of events to buffer before flushing. Defaults to 500. |
500
|
flush_interval
|
float
|
Maximum simulation seconds to wait before
forcing a flush. Defaults to 1.0. Ignored when |
1.0
|
lag_monitor_interval
|
float
|
Simulation seconds between
|
'auto'
|
max_queued_batches
|
int
|
Upper bound on batches waiting for the egress threads. A full queue blocks the producer, so a sink that cannot keep up slows the simulation rather than accumulating a backlog that teardown would have to discard. Defaults to 2000. |
2000
|
drain_stall_seconds
|
float
|
At teardown every queue is drained until empty. The wait is abandoned only if a queue stops shrinking for this long, so a slow sink finishes rather than losing its tail. Defaults to 30.0. |
30.0
|
predicates
|
List
|
One entry per provider, aligned by position. A callable takes a record and returns True to send it to that provider. None means that provider receives every record. Defaults to None, which is every provider receiving everything. |
None
|
batch_sizes
|
List
|
One entry per provider, aligned by position.
None falls back to |
None
|
flush_intervals
|
List
|
One entry per provider, aligned by
position. None falls back to |
None
|
labels
|
List
|
One entry per provider, aligned by position.
True makes it a labels sink, which receives the ground-truth records
from |
None
|
warmup
|
float
|
The length of the warm-up, in simulation seconds from now, the start of the run when the egress is set up with the model. No provider receives a record stamped before it ends, and the built-in statistics are reset then. Defaults to None, no warm-up. |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If a per-provider list has the wrong length. |
Each provider gets its own queue. They previously shared one, so several providers competed for batches instead of each receiving them, and a run with two sinks wrote part of the data to each with no error and exit code 0.
Source code in src/desflow/core/environment.py
235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 | |
publish_telemetry(path_id, value)
Publishes a low-volume telemetry metric (e.g., utilization, queue length).
Telemetry shares the main event buffer to ensure efficient batching for high-throughput file storage (like Parquet or S3). Nothing happens when no egress is set up.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path_id
|
str
|
The value's name, written as the record's |
required |
value
|
Any
|
The value, made a JSON value on the way out, as an event's is. |
required |
Source code in src/desflow/core/environment.py
publish_event(event_key, value, task=None, instance=None)
Buffers a high-volume discrete event (e.g., a task starting or finishing).
Events are buffered and flushed either when batch_size is reached
or when flush_interval occurs to optimize external I/O throughput.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_key
|
str
|
The record key: the entity's key, such as |
required |
value
|
Any
|
A dictionary or a Pydantic model containing the event payload. It is made JSON-serializable on the way out, so a nested model is flattened and a datetime becomes an ISO string, exactly as before. |
required |
task
|
str
|
The name of the task the event belongs to, written
as the record's |
None
|
instance
|
Any
|
The task's instance id, written as the record's
|
None
|
Source code in src/desflow/core/environment.py
watch_changes(callback, queues=False)
Calls callback after every change a shared condition can read.
A change is a store or container level, a resource's capacity or use, a
registry update or a signal firing. A wait on a condition, such as
{condition: "level.stock >= 5"}, watches this way and checks its condition
again after each one, as a SimPy loop over a fresh "changed" event would.
A change of the length of a store's or container's put_queue or
get_queue is not one of these: it calls only the functions added with
queues=True, so a condition that does not read a queue is not checked
again for it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback
|
Callable[[], None]
|
A function of no arguments. |
required |
queues
|
bool
|
True to call it also after a change of the length of a
store's or container's put or get queue. A condition that reads
|
False
|
Source code in src/desflow/core/environment.py
unwatch_changes(callback)
Stops calling a function that watch_changes added; does nothing if it is gone.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
callback
|
Callable[[], None]
|
The function |
required |
Source code in src/desflow/core/environment.py
notify_queue_change()
Calls the functions watch_changes added with queues=True. A store or
container calls it after the length of its put_queue or get_queue
changes; it does not change the holder's statistics.
Source code in src/desflow/core/environment.py
resume_pacing()
Restarts pacing from the current simulation time and the wall clock now.
After a pause, such as a debugger stop or a wait between advance() calls,
the wall clock is ahead of the simulation. The next paced event would run at
once, or raise under strict. Calling this makes the next event wait factor
real seconds for each simulation second after now. SimPy's sync() moves
only the wall-clock anchor, so it is not enough on its own once now has
moved. At factor 0 the call changes nothing a run reads.
Source code in src/desflow/core/environment.py
teardown()
Gracefully terminates the environment.
Ensures that any remaining data in event buffers is flushed to the
egress connectors, and that background asyncio threads for both
ingress and egress are cleanly stopped. Should be called in a finally block.
Source code in src/desflow/core/environment.py
Registry and Parameters
SimulationRegistry(env)
A central registry that maps dot-notation paths to dynamic simulation parameters.
The Registry acts as the single source of truth for the simulation's state. It allows external streams (like Kafka or Redis) to update parameters while the simulation runs, and passes each change to the SimPy processes that use it.
Attributes:
| Name | Type | Description |
|---|---|---|
env |
Environment
|
The active SimPy environment. |
Source code in src/desflow/core/registry.py
Functions
configs()
Returns every configuration object by its path, such as Shop.service.packing.
An experiment records them as the model's compiled parameters. The objects are the live ones, so read them rather than change them.
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict[str, Any]: Path to |
Source code in src/desflow/core/registry.py
get(path)
Retrieve the DynamicValue object at a specific path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The dot-notation path (e.g., 'Shop.arrival.orders.rate'). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
DynamicValue |
DynamicValue
|
The wrapper object for the requested parameter. |
Raises:
| Type | Description |
|---|---|
KeyError
|
If the path does not exist in the registry. |
Source code in src/desflow/core/registry.py
get_config(path)
Retrieve the live configuration object (e.g., DistributionConfig).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The dot-notation path (e.g., 'Shop.service.packing'). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The synchronized configuration object. |
Raises:
| Type | Description |
|---|---|
KeyError
|
If the config path does not exist in the registry. |
Source code in src/desflow/core/registry.py
paths()
Returns every value path, such as Shop.resources.packers.current_cap.
Returns:
| Type | Description |
|---|---|
List[str]
|
List[str]: The paths |
Source code in src/desflow/core/registry.py
register_container(sim_id, name, cap_config)
Registers a container's capacity paths and its .level path.
A current_cap left as None is the max_cap if that is given and no limit
(infinity) if not, and a max_cap left as None is the current_cap. The
config is updated to the values in effect.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sim_id
|
str
|
The simulation ID. |
required |
name
|
str
|
The container's name. |
required |
cap_config
|
CapacityConfig
|
Its capacity and starting level. |
required |
Source code in src/desflow/core/registry.py
register_resource(sim_id, name, cap_config)
Registers a resource's capacity paths, filling in what the config leaves out.
A current_cap left as None is 1, and a max_cap left as None is the
current_cap. The config is updated to the values in effect.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sim_id
|
str
|
The simulation ID. |
required |
name
|
str
|
The resource's name. |
required |
cap_config
|
CapacityConfig
|
Its capacity, which may leave values as None. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If only |
Source code in src/desflow/core/registry.py
register_sim_parameter(param)
Takes a SimParameter instance and flattens it into the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
param
|
SimParameter
|
The initial state schema to register. |
required |
Source code in src/desflow/core/registry.py
register_store(sim_id, name, cap_config)
Registers a store's capacity paths, filling in what the config leaves out.
A current_cap left as None is the max_cap if that is given and no limit
(infinity) if not, and a max_cap left as None is the current_cap. The
config is updated to the values in effect.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sim_id
|
str
|
The simulation ID. |
required |
name
|
str
|
The store's name. |
required |
cap_config
|
CapacityConfig
|
Its capacity, which may leave values as None. |
required |
Source code in src/desflow/core/registry.py
set_exactly(path, value)
Sets a path to exactly value, raising where update would log and go on.
Replications and experiments set their variants' values this way before the clock starts, so a misspelt path or a refused value stops the experiment rather than leaving the model's own value in place.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The dot-notation path. |
required |
value
|
Any
|
The value it must hold. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the path does not exist, or the registry refuses, casts to
another value or clamps the value, or a |
Source code in src/desflow/core/registry.py
update(path, new_value)
Update a value safely and synchronize its parent configuration object.
Includes dynamic type casting to protect the simulation from crashing if external systems send improperly typed data (e.g., sending the string "5" instead of the integer 5).
A capacity path (current_cap or max_cap) takes inf and the text "inf"
as no limit, whatever type it held.
A distribution parameter outside its range (for example a negative std) is
refused with an error log line, as a value of the wrong type is, and the old
value is kept. A current_cap outside [0, max_cap] is clamped to that range
with a warning log line, also when a lower max_cap arrives, and so is a
container level outside [0, current_cap]. A path the model does not have,
such as a parameter the distribution's family does not take, is refused with a
warning log line.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The dot-notation path to update. |
required |
new_value
|
Any
|
The new value to apply. |
required |
Source code in src/desflow/core/registry.py
167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 | |
SimParameter(sim_id, arrival=dict(), service=dict(), resources=dict(), containers=dict(), stores=dict(), breakdowns=dict(), variables=dict())
dataclass
The master schema representing the state of a specific simulation entity (like a production line).
This object is registered with the SimulationRegistry, which flattens
the nested dictionaries into dot-notation paths (e.g., 'Shop.arrival.orders.rate').
Attributes:
| Name | Type | Description |
|---|---|---|
sim_id |
str
|
The unique prefix for this group of parameters (e.g., 'Shop'). |
arrival |
Dict[str, DistributionConfig]
|
Configurations for arrival generation rates. |
service |
Dict[str, DistributionConfig]
|
Configurations for process task durations. |
resources |
Dict[str, CapacityConfig]
|
Configurations for standard SimPy Resources. |
containers |
Dict[str, CapacityConfig]
|
Configurations for continuous SimPy Containers.
A |
stores |
Dict[str, CapacityConfig]
|
Configurations for discrete SimPy Stores. |
breakdowns |
Dict[str, BreakdownConfig]
|
How each named resource breaks
down, registered at |
variables |
Dict[str, Any]
|
Flexible user-defined state variables (e.g., int, float, bool, str) used to drive custom logic, track string-based states, or act as external control dials. |
CapacityConfig(current_cap=None, max_cap=None)
dataclass
Configuration for the capacity of a simulated resource, container, or store.
A value left as None is filled in when the holder registers. A resource's
current_cap is 1, and max_cap is current_cap. A store's or container's
current_cap is max_cap when that is given and no limit if not, and its
max_cap is current_cap. float("inf") also means no limit.
Attributes:
| Name | Type | Description |
|---|---|---|
current_cap |
Optional[Union[int, float]]
|
The currently active capacity. |
max_cap |
Optional[Union[int, float]]
|
The absolute physical maximum capacity. |
DistributionConfig(dist, rate=None, mean=None, std=None, low=None, value=None, high=None, mode=None, shape=None, scale=None, values=None, weights=None, edges=None, counts=None)
dataclass
Configuration for a statistical distribution used in the simulation.
This config is dynamically updatable. For example, you can change the rate
of an 'exponential' distribution via Kafka, and the Sampler will instantly
use the new rate for the next event. The family (dist) is fixed at load, and
only its parameters change live.
Numbers are stored as floats, so a live update to 0.5 is not cast to an integer.
The family and its parameters are checked when the config is built, against the
registry of distributions in desflow.core.distributions.
Attributes:
| Name | Type | Description |
|---|---|---|
dist |
str
|
The family: exponential, normal, lognormal, truncated_normal, constant, uniform, triangular, gamma, weibull or empirical. |
rate |
Optional[float]
|
Events per second, for exponential. Greater than 0. |
mean |
Optional[float]
|
The mean, for normal, lognormal and truncated_normal. Greater than 0 for lognormal. |
std |
Optional[float]
|
The standard deviation, 0 or more; 0 (or no value) gives the mean every time. |
low |
Optional[float]
|
The lowest value: of a truncated normal (0 by default, and a draw below it is drawn again), a uniform or a triangular. |
value |
Optional[float]
|
The time a constant distribution always gives. 0 or more. |
high |
Optional[float]
|
The highest value of a uniform or a triangular. |
mode |
Optional[float]
|
The most likely value of a triangular. |
shape |
Optional[float]
|
The shape of a gamma or a weibull. Greater than 0. |
scale |
Optional[float]
|
The scale of a gamma or a weibull. Greater than 0. |
values |
Optional[List[float]]
|
The values an empirical draws from. |
weights |
Optional[List[float]]
|
One weight per empirical value. |
edges |
Optional[List[float]]
|
The bin edges of an empirical histogram. |
counts |
Optional[List[float]]
|
One count per empirical histogram bin. |
stream |
Optional[str]
|
The name of the random sequence it draws from: its
registry path without the sim id, such as |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the family is unknown, a parameter it does not take is given, or a parameter it takes is missing or out of range. |
ContainerConfig(current_cap=None, max_cap=None, level=0)
dataclass
Bases: CapacityConfig
Configuration for a container: its capacity and the level it starts at.
current_cap is the container's capacity, its largest level, as it is a
resource's capacity. The level has a registry path of its own, .level, so a
scenario or an ingress message can set it during a run, such as a restock.
Attributes:
| Name | Type | Description |
|---|---|---|
current_cap |
Union[int, float]
|
The capacity: the largest level. |
max_cap |
Union[int, float]
|
The largest capacity a live update can set. |
level |
Union[int, float]
|
The amount held when the run starts, kept as the
type given, as |
BreakdownConfig(time_to_failure, repair, work='interrupt')
dataclass
How a resource breaks down: the time to each failure, the repair time, and what happens to the work in progress.
The failure clock runs on calendar time: the next failure is drawn when a repair
ends, busy or idle. While down, the resource's capacity is 0; its current_cap
keeps the planned value, which a change during the repair sets for its return.
Attributes:
| Name | Type | Description |
|---|---|---|
time_to_failure |
DistributionConfig
|
The time from a repair, or the start, to the next failure. |
repair |
DistributionConfig
|
How long a repair takes. |
work |
str
|
|
Data Payloads
TelemetryPayload
Bases: BaseStreamPayload
Schema for low-volume, single-metric telemetry updates.
This payload is used for publishing continuous system state variables, such as resource utilization, queue lengths, or system lag, typically used for real-time dashboards.
Attributes:
| Name | Type | Description |
|---|---|---|
stream_type |
Literal['telemetry']
|
Hardcoded to "telemetry" for downstream routing. |
path_id |
str
|
The dot-notation path of the metric (e.g., 'Shop.packers.utilization'). |
value |
Any
|
The scalar value of the metric at the given simulation time. |
EventPayload
Bases: BaseStreamPayload
Schema for high-volume discrete simulation events.
This payload tracks state transitions of specific entities (like tasks or orders) throughout
the simulation lifecycle. The key attribute allows message brokers like Kafka to maintain
strict chronological ordering for specific tasks.
Attributes:
| Name | Type | Description |
|---|---|---|
stream_type |
Literal['event']
|
Hardcoded to "event" for downstream routing. |
key |
str
|
The entity key, the record and partition key (e.g., 'orders-1'). |
task |
Optional[str]
|
The name of the task the event belongs to, or None for an event published outside a task. |
instance |
Optional[Any]
|
The task's instance id, or None. |
value |
Dict[str, Any]
|
A dictionary containing the event's detailed payload. |
LabelPayload
Bases: BaseStreamPayload
Schema for ground-truth records: what a scenario step changed, and why.
They are sent only to sinks with when: labels, which receive nothing else, so
data and labels always reach separate sinks.
Attributes:
| Name | Type | Description |
|---|---|---|
stream_type |
Literal['label']
|
Hardcoded to "label" for downstream routing. |
path |
Optional[str]
|
The registry path the step changed, or None for a step
with only a |
old |
Any
|
The value before the step, or None. |
new |
Any
|
The value after the step, or None. |
label |
Dict[str, Any]
|
The step's |
BaseStreamPayload
Bases: BaseModel
Base schema for all egress data emitted by the simulation engine.
This schema guarantees that all outgoing data streams share a common temporal context, allowing downstream systems to accurately synchronize simulation time with real-world time.
Attributes:
| Name | Type | Description |
|---|---|---|
stream_type |
Literal['telemetry', 'event', 'label']
|
An identifier indicating the nature of the payload. |
sim_ts |
float
|
The simulation clock time (in seconds) when the payload was generated. |
timestamp |
str
|
The real-world ISO 8601 timestamp indicating when the payload was generated. |
Utilities
Sampler(rng=None, *, seed=None)
Generates random numbers based on live DistributionConfig objects.
The Sampler evaluates the configuration at the exact moment it is called. This allows the simulation to dynamically respond to external parameter changes.
Each distribution draws from a random sequence of its own, made from the run's
seed and the distribution's registry path without the sim id (arrival.orders,
service.packing): SeedSequence(seed, spawn_key=(CRC32 of the path,)). Adding
an arrival therefore leaves every other distribution's draws unchanged, so two
versions of a model run with one seed draw the same numbers for every
distribution they share (common random numbers).
Attributes:
| Name | Type | Description |
|---|---|---|
rng |
Optional[Generator]
|
The run's main generator, for draws in your own code. Distributions do not draw from it, so adding one does not change these draws either. None when nothing is drawn at random. |
seed |
Optional[int]
|
The seed the sequences come from, or None. |
Initializes the Sampler.
With neither rng nor seed, nothing is drawn at random: each sample is
the distribution's mean (for truncated_normal, its mean or low, whichever
is higher).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rng
|
Optional[Generator]
|
A seeded NumPy generator, such as |
None
|
seed
|
Optional[int]
|
The run's seed, instead of |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If both |
Source code in src/desflow/core/sampler.py
Functions
sample(config, min_delay=MIN_DELAY)
Draws a random sample using the current parameters in the provided config.
A config the registry has registered draws from its own sequence; one built
by hand and never registered draws from rng.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
DistributionConfig
|
The live configuration object retrieved from the Registry. |
required |
min_delay
|
float
|
The smallest value returned. |
MIN_DELAY
|
Returns:
| Name | Type | Description |
|---|---|---|
float |
float
|
A sampled float representing a time duration (e.g., arrival gap or service time).
Returns |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the |
Source code in src/desflow/core/sampler.py
time_to_seconds(time_str)
Converts a human-readable duration into seconds.
Examples:
"1 week" -> 604800.0 "2.5 hours" -> 9000.0 "30 mins" -> 1800.0
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
time_str
|
str
|
A number and a unit, such as |
required |
Returns:
| Name | Type | Description |
|---|---|---|
float |
float
|
The seconds. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the text is not a number and a known unit. |
Source code in src/desflow/utils.py
Resources
ResourceBase(env, sim_id, resource_id, queue=DEFAULT_QUEUE_ORDER, preempt=False, queue_limit=None)
Bases: BaseDynamicResource
A resource of units that tasks request, with a capacity that follows the registry.
It is DynamicResource built on SimPy's hooks, so a subclass changes it as a
subclass of SimPy's Resource does: set PutQueue or GetQueue, set request
or release to a BoundClass of a request or release class, or override
_do_put(request) and _do_get(release). Use a subclass with type= in
add_resource or type: !python module.Class in a blueprint.
The constructor is DynamicResource's, because a resource reads its capacity
from the registry. A subclass that defines __init__ takes *args, **kwargs and
passes them on.
Grants follow SimPy's Resource: a request is granted when it is made if a unit
is free and nothing waits ahead of it, and a release grants the head of the
queue one event later, when its release event is processed; a release that
gives back nothing tries the head too. Equal-time grants therefore happen in
the order SimPy gives them.
Waiting requests are served in the resource's queue order: fifo (the
default) serves the oldest first, lifo the newest first, and priority by
the request's key, (priority, time, not preempt), as SimPy's
PriorityResource does: the lowest priority value first, then the earliest,
then one that may preempt. A sequence number breaks the remaining ties, so
such requests are served first in, first out.
Attributes:
| Name | Type | Description |
|---|---|---|
env |
DynamicRealtimeEnvironment
|
The active simulation environment. |
sim_id |
str
|
The prefix ID (e.g., 'Shop'). |
resource_id |
str
|
The specific resource name (e.g., 'packers'). |
queue_order |
str
|
|
PutQueue |
Any
|
The type of |
GetQueue |
Any
|
The type of |
put_queue |
The waiting requests in service order. |
|
queue |
An alias of |
|
get_queue |
The releases being served, empty between events. |
|
users |
List[DynamicResourceRequest]
|
The granted requests, oldest first. |
count |
int
|
How many users there are; the same as |
request |
Makes a |
|
release |
Makes a |
Initializes the resource and binds it to the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Environment
|
The SimPy environment (must include the RegistryMixIn). |
required |
sim_id
|
str
|
The parent simulation ID. |
required |
resource_id
|
str
|
The name of the resource. |
required |
queue
|
str
|
The queue order: |
DEFAULT_QUEUE_ORDER
|
preempt
|
bool
|
When True, a request that finds every unit busy takes
one from the user SimPy's |
False
|
queue_limit
|
Optional[int]
|
The most requests that may wait, or None for
no limit. A request made while that many wait raises |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/resources/resource.py
Attributes
request = BoundClass(DynamicResourceRequest)
class-attribute
instance-attribute
Requests one unit: request(priority=0, preempt=True) returns a
DynamicResourceRequest, an event that succeeds when a unit is granted. Lower
priority values are served first under queue="priority"; preempt lets the
request take a unit from a user of a higher value on a resource built with
preempt=True. Set it to BoundClass(MyRequest) to use a subclass.
release = BoundClass(DynamicRelease)
class-attribute
instance-attribute
Gives a granted unit back: release(request) returns a DynamicRelease,
triggered at once. Releasing a request twice, or one never granted, gives back
nothing; the queue is still tried when the release is processed. With no request,
the oldest unit held is given back. While units are busy above a lowered
capacity, a release brings the busy count down and grants nothing until it is
within the capacity. Set it to BoundClass(MyRelease) to use a subclass.
in_use
property
Units currently held by tasks, including units above a lowered capacity.
count
property
Number of users currently using the resource, as SimPy's Resource.count;
the same as in_use.
capacity
property
The current active capacity of the resource.
This value reflects the live state from the Registry and may change during the simulation.
DynamicResource(env, sim_id, resource_id, queue=DEFAULT_QUEUE_ORDER, preempt=False, queue_limit=None)
Bases: ResourceBase
A SimPy Resource with a dynamically adjustable capacity.
Unlike a standard simpy.Resource where capacity is fixed at creation,
the DynamicResource listens to a SimulationRegistry path (e.g., 'Shop.resources.packers.current_cap').
When the registry updates the capacity, the resource adjusts at once, granting
waiting requests if the capacity rises. It is ResourceBase as built, and the
class to subclass for a resource of your own is ResourceBase.
Source code in src/desflow/resources/resource.py
ContainerBase
Bases: _Notifier, Container
A SimPy Container that tells DynamicContainer about every change, the class
to subclass for a container of your own.
Subclass it as you subclass simpy.Container: set PutQueue or GetQueue, set
put or get to a BoundClass of an event class, or override _do_put(event)
and _do_get(event). Use the subclass with type= in add_container or type:
!python module.Class in a blueprint. The constructor is SimPy's,
(env, capacity=inf, init=0); a subclass that defines __init__ passes its
arguments on.
A change of the level made by _do_put or _do_get is reported once to the
statistics and to the conditions that read the container, whether the override
calls super() or not, and whatever it returns. A change of the length of
put_queue or get_queue is reported to the conditions that read putters. or
getters.; a queue class must support len() and remove().
Attributes:
| Name | Type | Description |
|---|---|---|
level |
The amount held. |
|
put_queue |
The waiting puts, SimPy's queue. |
|
get_queue |
The waiting gets, SimPy's queue. |
|
put |
Makes a |
|
get |
Makes a |
Attributes
put = BoundClass(DynamicContainerPut)
class-attribute
instance-attribute
get = BoundClass(DynamicContainerGet)
class-attribute
instance-attribute
DynamicContainer(env, sim_id, container_id, init=None, type=None)
Bases: BaseDynamicResource
A wrapper for SimPy Container with dynamic capacity updates.
Useful for modeling continuous bulk materials (fluids, gases, battery charge) and
stock. Unlike a standard simpy.Container, the capacity of this container listens
to a SimulationRegistry path and can be mutated at runtime.
current_cap is the capacity, the largest level, as it is for a resource. The
level has a registry path of its own, <sim_id>.containers.<name>.level: the
registry follows every put and get, and a live update to the path sets the level,
such as a restock, waking the tasks it lets through.
Capacity Shrinkage Behavior:
If the capacity is dynamically shrunk below the current level of the container,
the material is NOT destroyed. The container will temporarily exist in an
"overflow" state. All pending and future put requests will be strictly
blocked until downstream processes get enough material to drain the level
back below the new, smaller capacity.
Attributes:
| Name | Type | Description |
|---|---|---|
env |
Environment
|
The active SimPy simulation environment. |
sim_id |
str
|
The parent simulation ID prefix. |
obj_id |
str
|
The specific container name/identifier. |
Initializes the DynamicContainer and binds its capacity to the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Environment
|
The SimPy environment (must include Registry interface). |
required |
sim_id
|
str
|
The parent simulation ID. |
required |
container_id
|
str
|
The unique name of this container. |
required |
init
|
float
|
The amount in the container at simulation start.
None, the default, takes the registry's |
None
|
type
|
Type[ContainerBase]
|
A subclass of |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/resources/container.py
Attributes
level
property
float: The current amount of bulk material stored inside the container.
capacity
property
float: The current active maximum capacity of the container. This value reflects the live state from the Registry.
putters_waiting
property
int: Puts waiting for room. A put granted at once is not counted,
although SimPy removes it from put_queue only after the grant.
getters_waiting
property
int: Gets waiting for an amount. A get granted at once is not counted,
although SimPy removes it from get_queue only after the grant.
Functions
put(*args, **kwargs)
Request to put a specific amount of material into the container.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*args
|
Any
|
The amount of material to add. A custom |
()
|
**kwargs
|
Any
|
Keyword arguments, passed to the container's |
{}
|
Returns:
| Type | Description |
|---|---|
Event
|
simpy.events.Event: A SimPy event that triggers when capacity is available. |
Source code in src/desflow/resources/container.py
get(*args, **kwargs)
Request to get a specific amount of material from the container.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*args
|
Any
|
The amount of material to withdraw. A custom |
()
|
**kwargs
|
Any
|
Keyword arguments, passed to the container's |
{}
|
Returns:
| Type | Description |
|---|---|
ContainerGet
|
simpy.resources.container.ContainerGet: A SimPy event that triggers when enough material is available. |
Source code in src/desflow/resources/container.py
StoreBase
Bases: _Notifier, Store
A SimPy Store that tells DynamicStore about every change, the class to
subclass for a store of your own.
Subclass it as you subclass simpy.Store: set PutQueue or GetQueue, set
put or get to a BoundClass of an event class, or override _do_put(event)
and _do_get(event). Use the subclass with type= in add_store or type:
!python module.Class in a blueprint. The constructor is SimPy's,
(env, capacity=inf); a subclass that defines __init__ passes its arguments
on.
A change of the number of items made by _do_put or _do_get is reported
once to the statistics and to the conditions that read the store, whether the
override calls super() or not. A change of the length of put_queue or
get_queue is reported to the conditions that read putters. or getters.
(and count, items); a queue class must support len() and remove().
Attributes:
| Name | Type | Description |
|---|---|---|
items |
The items held, in the order a get returns them. |
|
put_queue |
The waiting puts, SimPy's queue. |
|
get_queue |
The waiting gets, SimPy's queue. |
|
put |
Makes a |
|
get |
Makes a |
Attributes
put = BoundClass(DynamicStorePut)
class-attribute
instance-attribute
get = BoundClass(DynamicStoreGet)
class-attribute
instance-attribute
DynamicStore(env, sim_id, store_id, priority=False, items=None, type=None)
Bases: BaseDynamicResource
A wrapper for SimPy FilterStore and PriorityStore with dynamic capacity updates.
Useful for modeling buffers, queues, or staging areas that hold discrete,
heterogeneous items. By setting priority=True, the store will automatically
sort incoming items (which must be sortable or wrapped in simpy.PriorityItem)
so high-priority items are retrieved first.
Capacity Shrinkage Behavior:
If the capacity is dynamically shrunk below the current number of items
in the store, the existing items are NOT destroyed. The store will temporarily
exist in an "over-capacity" state. All pending and future put requests
will be blocked until downstream processes get enough items to bring the
total item count below the new capacity.
Attributes:
| Name | Type | Description |
|---|---|---|
env |
Environment
|
The active SimPy simulation environment. |
sim_id |
str
|
The parent simulation ID prefix. |
obj_id |
str
|
The specific store name/identifier. |
Initializes the DynamicStore and binds its capacity to the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Environment
|
The SimPy environment (must include Registry interface). |
required |
sim_id
|
str
|
The parent simulation ID. |
required |
store_id
|
str
|
The unique name of this store. |
required |
priority
|
bool
|
If True, uses |
False
|
items
|
Iterable
|
The items it holds when the run starts, in
the order a get returns them. A priority store takes |
None
|
type
|
Type[StoreBase]
|
A subclass of |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If there are more items than the capacity, a priority
store's item is not a |
Source code in src/desflow/resources/store.py
Attributes
items
property
list: The actual list of distinct item objects currently residing in the store.
capacity
property
int: The current active maximum slot capacity of the store. This value reflects the live state from the Registry.
putters_waiting
property
int: Puts waiting for room. A put granted at once is not counted,
although SimPy removes it from put_queue only after the grant.
getters_waiting
property
int: Gets waiting for an item. A get granted at once is not counted,
although SimPy removes it from get_queue only after the grant.
Functions
put(*args, **kwargs)
Request to put a specific distinct item into the store.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*args
|
Any
|
The item to place into the store. A custom |
()
|
**kwargs
|
Any
|
Keyword arguments, passed to the store's |
{}
|
Returns:
| Type | Description |
|---|---|
Event
|
simpy.events.Event: A SimPy event that triggers when a slot is available. |
Source code in src/desflow/resources/store.py
get(*args, **kwargs)
Request to get the next available item from the store.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*args
|
Any
|
The filter, if any. A custom |
()
|
**kwargs
|
Any
|
Keyword arguments, passed to the store's |
{}
|
Returns:
| Type | Description |
|---|---|
StoreGet
|
simpy.resources.store.StoreGet: A SimPy event that yields the
requested item object when available. Its |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a priority store is given a filter: SimPy has no store that both ranks and filters. |
Source code in src/desflow/resources/store.py
Breakdowns, Schedules and Replay
The builder and YAML start these SimPy processes and readers for you. Low-level code starts them itself.
run_breakdowns(env, resource, sampler, work='interrupt')
Breaks a resource down and repairs it, for as long as the run lasts.
The time to failure and the repair time are read from the registry, at
<sim_id>.breakdown.<resource>.time_to_failure and .repair, which a
SimParameter's breakdowns registers, so they can change during the run. Each
failure publishes an event with the resource's name as its key and
{"status": "down", "reason": "failure"}, and each repair one with
{"status": "repaired", "reason": "failure"}; neither has a task.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Any
|
The environment. |
required |
resource
|
ResourceBase
|
The resource that breaks down. |
required |
sampler
|
Sampler
|
The sampler the times are drawn with. |
required |
work
|
str
|
|
'interrupt'
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/core/breakdowns.py
run_scenario(env, steps, until=None, interrupt=None)
Applies each step at its time, in time order and, at equal times, in the order
given. A step's at counts from when this process starts, the start of the run
when it is started with the model, so it does not depend on initial_time.
A step with a path and value updates the registry; one with a record
publishes a ground-truth record of the change (env.publish_label); one with an
interrupt, {task, where, cause}, calls interrupt(task, where=..., cause=...),
which the builder and YAML pass as context.interrupt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Any
|
The environment. |
required |
steps
|
Iterable[Mapping[str, Any]]
|
The steps, each a mapping as |
required |
until
|
Optional[float]
|
The run's end, a reading of the clock; steps due at or after it are
dropped, as an event due at exactly |
None
|
interrupt
|
Optional[Callable[..., Any]]
|
What an |
None
|
Source code in src/desflow/core/scenario.py
read_trace(source, time='time', unit='s', fields='all', on_invalid='error', start=None, initial_time=0.0)
Reads and checks the rows of a trace, in their order.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
source
|
Union[str, Path, Iterable[Mapping[str, Any]]]
|
A CSV file, a Parquet file (which needs the |
required |
time
|
str
|
The column that holds each row's time. |
'time'
|
unit
|
str
|
|
's'
|
fields
|
Union[str, Sequence[str]]
|
The columns each arrival carries as attributes, or |
'all'
|
on_invalid
|
str
|
|
'error'
|
start
|
Optional[datetime]
|
The date and time of simulation time 0, as for every record's
|
None
|
initial_time
|
float
|
The simulation clock's start. |
0.0
|
Returns:
| Type | Description |
|---|---|
List[Row]
|
List[Row]: The rows, each with its simulation time. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the unit or |
ImportError
|
If a Parquet file is read without pyarrow. |
Source code in src/desflow/core/replay.py
Admin and Infrastructure
KafkaAdminConnector(bootstrap_servers, max_tasks=100, **kwargs)
Unified Kafka Admin and Monitoring Connector.
This connector acts as a management layer for the simulation's Kafka
infrastructure. It handles synchronous administrative tasks (topic creation)
using kafka-python and asynchronous data operations (sending config,
collecting telemetry/events) using aiokafka.
It maintains an internal state of simulation vitals and task lifecycles by consuming from the simulation's egress topics.
Attributes:
| Name | Type | Description |
|---|---|---|
bootstrap_servers |
str
|
Kafka broker addresses. |
max_tasks |
int
|
The maximum number of task records to keep per service in memory to prevent unbounded growth. |
kwargs |
dict
|
Additional arguments passed to Kafka clients. |
Initialize the connector with broker settings and state limits.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bootstrap_servers
|
str
|
Kafka broker addresses. |
required |
max_tasks
|
int
|
The maximum number of task records to keep per service in memory (rolling window). |
100
|
**kwargs
|
Any
|
Additional arguments passed to Kafka clients (e.g., security settings). |
{}
|
Source code in src/desflow/connectors/admin/kafka.py
Functions
create_topics(topics_config)
Creates Kafka topics required for the simulation.
This is a synchronous call to ensure that all necessary infrastructure (config, telemetry, and event topics) is ready before the simulation environment starts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topics_config
|
List[Dict[str, Any]]
|
A list of topic configurations. Each dict should contain 'name', and optionally 'partitions' and 'replication'. |
required |
Source code in src/desflow/connectors/admin/kafka.py
send_config(topic, path, value)
async
Sends a surgical parameter update to a simulation config topic.
This method acts as an external controller, allowing users to
dynamically update registry paths (e.g., arrival rates or resource
capacities) while the simulation is running. The message is the JSON
object {"path": path, "value": value}, the format KafkaIngress reads.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The Kafka topic the simulation is listening to for config. |
required |
path
|
str
|
The registry dot-notation path (e.g., 'Shop.resources.packers.max_cap'). |
required |
value
|
Any
|
The new value to be applied to the path. |
required |
Source code in src/desflow/connectors/admin/kafka.py
collect_data(topics, auto_offset_reset='latest')
async
Async loop to consume from telemetry and event topics.
This loop continuously listens to the simulation's egress and updates
the connector's internal _vitals and _state attributes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topics
|
List[str]
|
List of topics to consume from (telemetry and events). |
required |
auto_offset_reset
|
str
|
Where to start consuming if no offset is committed. |
'latest'
|
Source code in src/desflow/connectors/admin/kafka.py
get_state()
Returns the aggregated event state for task lifecycles.
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
A nested dictionary: sim_id -> service -> task_id -> {status: timestamp}. |
get_vitals()
Returns the latest system telemetry metrics.
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
A dictionary of path_id to latest value (e.g., utilization, queue length). |
Connectors (Ingress)
BaseIngress
Base class for all ingress providers in the simulation.
Ingress providers act as asynchronous listeners that bridge external data sources (such as Kafka topics, Redis channels, or local schedules) with the simulation's internal state. They are responsible for fetching updates and placing them into a thread-safe queue for the registry to process.
Functions
run(ingress_queue)
async
Listens to an external source and pushes updates to the ingress queue.
This method should be implemented as an asynchronous loop that waits for external signals or data and converts them into (path, value) tuples suitable for the SimulationRegistry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ingress_queue
|
Queue
|
A thread-safe queue used to transmit updates to the main simulation environment. |
required |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If the subclass does not override this method. |
Source code in src/desflow/connectors/ingress/base.py
KafkaIngress(topic, bootstrap_servers, topic_deserializers=None, default_deserializer=None, **kwargs)
Bases: BaseIngress
Resilient Kafka consumer for dynamic simulation configuration updates.
This connector subscribes to a specified Kafka topic and decodes incoming messages into state updates for the SimulationRegistry. It supports pluggable deserialization (Avro/JSON) while remaining 100% backward compatible.
The expected resulting dictionary format must contain a 'path' string and a non-null 'value' of any serializable type. A message that cannot be decoded or does not have both is logged as a warning and skipped.
Attributes:
| Name | Type | Description |
|---|---|---|
topic |
str
|
The Kafka topic name used for configuration signals. |
bootstrap_servers |
str
|
Comma-separated string of Kafka broker addresses. |
topic_deserializers |
Optional[Dict[str, MessageDeserializer]]
|
Mapping of topics to specific deserializers. |
default_deserializer |
Optional[MessageDeserializer]
|
Fallback deserializer. Defaults to |
kwargs |
dict
|
Additional configuration parameters for the AIOKafkaConsumer. |
Initializes the KafkaIngress with connection and subscription details.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The Kafka topic to subscribe to. |
required |
bootstrap_servers
|
str
|
Kafka broker address or list of addresses. |
required |
topic_deserializers
|
Optional[Dict[str, MessageDeserializer]]
|
Mapping of topics to their specific deserializers. |
None
|
default_deserializer
|
Optional[MessageDeserializer]
|
Fallback deserializer if the topic isn't explicitly mapped. |
None
|
**kwargs
|
Any
|
Arbitrary keyword arguments passed to the underlying AIOKafkaConsumer. |
{}
|
Source code in src/desflow/connectors/ingress/kafka.py
Functions
run(ingress_queue)
async
Consumes parameter changes from Kafka and puts each on the ingress queue.
Each message is read as JSON with path and value; a message that cannot be
read is logged and skipped.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ingress_queue
|
Queue
|
The queue the environment applies changes from, as
|
required |
Source code in src/desflow/connectors/ingress/kafka.py
LocalIngress(schedule)
Bases: BaseIngress
Ingress provider for deterministic, time-scheduled parameter updates.
This connector is primarily used for testing, benchmarking, or local simulation runs where parameter changes need to occur at specific wall-clock intervals without requiring an external message broker like Kafka. It feeds a pre-defined sequence of updates into the simulation registry based on relative delays.
Attributes:
| Name | Type | Description |
|---|---|---|
schedule |
List[Tuple[float, str, Any]]
|
A chronologically sorted list of updates, where each tuple contains (delay_seconds, path, value). |
Initializes the LocalIngress and sorts the provided schedule.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
schedule
|
List[Tuple[float, str, Any]]
|
A list of tuples representing scheduled updates. Format: [(delay_from_start, "registry_path", new_value), ...]. The list is automatically sorted by the delay time. |
required |
Source code in src/desflow/connectors/ingress/local.py
Functions
run(ingress_queue)
async
Processes the schedule and pushes updates to the queue at the correct wall-clock times.
This method calculates the relative sleep intervals between
scheduled events to ensure that updates are placed in the
ingress queue precisely when requested. Because it uses
asyncio.sleep, it does not block the main simulation execution.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ingress_queue
|
Queue
|
A thread-safe queue used to transmit the (path, value) updates to the SimulationRegistry. |
required |
Source code in src/desflow/connectors/ingress/local.py
PostgresIngress(connection_dsn, table_name='simulation_params', poll_interval=2.0, **kwargs)
Bases: BaseIngress
Asynchronous ingress provider for polling updates from a PostgreSQL database.
This connector polls a specified configuration table for rows where 'is_applied' is FALSE. It parses the parameter path and value, forwards them to the simulation registry, and then updates the row to mark it as applied. A row with a null path or value is logged as a warning, skipped and marked as applied, so it is not read again.
Initializes the PostgresIngress.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
connection_dsn
|
str
|
PostgreSQL connection string (DSN). |
required |
table_name
|
str
|
Table to poll for parameter updates. Must have columns: id (serial), path (text), value (text/json), is_applied (boolean). |
'simulation_params'
|
poll_interval
|
float
|
Seconds to wait between database polls. |
2.0
|
**kwargs
|
Any
|
Additional connection pool arguments for asyncpg. |
{}
|
Source code in src/desflow/connectors/ingress/postgres.py
Functions
run(ingress_queue)
async
Main execution loop fetching dynamic parameter updates from PostgreSQL.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ingress_queue
|
Queue
|
A thread-safe queue used to transmit (path, value) tuples to the SimulationRegistry. |
required |
Source code in src/desflow/connectors/ingress/postgres.py
91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | |
RedisIngress(url, channel_name='params', poll_interval=0.5, **kwargs)
Bases: BaseIngress
Asynchronous ingress provider for dynamic parameter updates via Redis Pub/Sub.
This connector utilizes redis.asyncio to listen on a specified Redis channel
for parameter updates, parsing them from JSON, and pushing them to the simulation's
internal update queue. Each message is a JSON object with path and value,
such as {"path": "Shop.arrival.orders.rate", "value": 10.0}; a message that
cannot be parsed or does not have both is logged as a warning and skipped.
Attributes:
| Name | Type | Description |
|---|---|---|
url |
str
|
The connection URL for the Redis server. |
channel_name |
str
|
The Redis Pub/Sub channel to subscribe to. |
poll_interval |
float
|
Sleep interval when no messages are found. |
client |
Redis | None
|
The underlying redis-py asynchronous client instance. |
pubsub |
PubSub | None
|
The active Pub/Sub subscription instance. |
**kwargs |
PubSub | None
|
Additional configuration dictionary passed to |
Initializes the RedisIngress with connection and channel settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
Redis connection URL (e.g., redis://127.0.0.1:6379/0). |
required |
channel_name
|
str
|
Redis Pub/Sub channel to subscribe to. |
'params'
|
poll_interval
|
float
|
Sleep interval when no messages are found. |
0.5
|
**kwargs
|
Any
|
Additional configuration dictionary passed to |
{}
|
Source code in src/desflow/connectors/ingress/redis.py
Functions
run(update_queue)
async
The main execution loop for subscribing to Redis and receiving updates.
Continuously polls the Pub/Sub channel for incoming JSON messages containing
path and value, placing successfully parsed updates onto the
internal update queue.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
update_queue
|
Queue
|
A thread-safe queue where incoming parameters are pushed so they can be applied within the SimPy context. |
required |
Source code in src/desflow/connectors/ingress/redis.py
Deserializers
MessageDeserializer
Bases: Protocol
Protocol defining the interface for all Kafka ingress deserializers.
Any custom deserializer passed to KafkaIngress must implement the
deserialize method according to this signature.
Functions
deserialize(topic, payload)
Deserializes a raw byte payload from Kafka into a Python object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The Kafka topic from which the message was consumed. |
required |
payload
|
bytes
|
The raw binary payload of the Kafka message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The deserialized data, typically a dictionary containing a 'path' and 'value'. |
Source code in src/desflow/connectors/ingress/kafka.py
JsonDeserializer
Default fallback deserializer providing backward compatibility via standard JSON.
This deserializer assumes the incoming payload is a UTF-8 encoded JSON string.
Functions
deserialize(topic, payload)
Deserializes a JSON byte string.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The Kafka topic from which the message was consumed (unused). |
required |
payload
|
bytes
|
The UTF-8 encoded JSON byte string. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The parsed JSON data as a Python dictionary. |
Source code in src/desflow/connectors/ingress/kafka.py
ConfluentAvroDeserializer(registry_url, *, auth=UNSET, verify=UNSET, cert=UNSET, timeout=UNSET, headers=UNSET, proxy=UNSET, max_retries=UNSET, **registry_kwargs)
Lazy-loaded deserializer for Confluent Schema Registry using fastavro and httpx.
This class converts Avro-encoded byte strings back into Python dictionaries.
It fetches the writer schema from the registry using the schema id (or GUID)
in the message's prefix, once for each id, so no schema_str is required
during initialization.
Initializes the Confluent Avro deserializer. It makes no request to the registry.
UNSET as a default, in the signature or the Default column of the API page,
means the argument was not given; each argument's description says its real
default.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry_url
|
str
|
The base URL of the Confluent Schema Registry. Several URLs separated by commas are tried in order. |
required |
auth
|
Optional[Tuple[str, str]]
|
A username and password for basic
authentication, for example |
UNSET
|
verify
|
bool | str | SSLContext
|
How to check the registry's certificate: True (the default), False, the path of a CA file or directory, or an SSL context. |
UNSET
|
cert
|
str | tuple | None
|
A client certificate file, or a tuple of the certificate file, the key file and the key password. The default is None. |
UNSET
|
timeout
|
Optional[float]
|
Seconds each step of a request to the registry (connecting, sending, reading) may take. The default is 10.0. None waits without a limit. |
UNSET
|
headers
|
Optional[Mapping[str, str]]
|
Headers sent with every request,
for example |
UNSET
|
proxy
|
Optional[str]
|
The URL of a proxy. The default is None. |
UNSET
|
max_retries
|
int
|
How many times a request is repeated after a transport error or a status of 408, 429, 500, 502, 503 or 504. The default is 3. |
UNSET
|
**registry_kwargs
|
Any
|
Deprecated confluent-kafka keys, such as
|
{}
|
Raises:
| Type | Description |
|---|---|
ImportError
|
If the 'fastavro' or 'httpx' package is not installed. |
ValueError
|
If a deprecated key is not accepted, or is given together with its named argument. |
Source code in src/desflow/connectors/ingress/kafka.py
79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | |
Functions
deserialize(topic, payload)
Deserializes an Avro-encoded byte string.
The first message with a schema id makes a request to the registry for the writer schema.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The Kafka topic from which the message was consumed. |
required |
payload
|
bytes
|
The Avro-encoded binary payload containing the schema ID. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The deserialized data as a Python dictionary. |
Raises:
| Type | Description |
|---|---|
SerializationError
|
If the payload has no valid prefix, the writer schema is not Avro, or the schema has Data Contract read rules. |
SchemaRegistryError
|
If the registry has no schema for the id. |
TransportError
|
If no registry URL could be reached. |
Source code in src/desflow/connectors/ingress/kafka.py
GlueAvroDeserializer(registry_name, **boto3_kwargs)
Lazy-loaded deserializer for AWS Glue Schema Registry.
This class converts Avro-encoded byte strings back into Python dictionaries, integrating directly with AWS Glue Schema Registry. It resolves schemas dynamically based on the metadata embedded in the AWS Glue message payload.
Initializes the AWS Glue Avro deserializer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry_name
|
str
|
The name of the AWS Glue Schema Registry. |
required |
**boto3_kwargs
|
Any
|
Additional arguments passed directly to the |
{}
|
Raises:
| Type | Description |
|---|---|
ImportError
|
If the 'aws-glue-schema-registry' or 'boto3' packages are not installed. |
Source code in src/desflow/connectors/ingress/kafka.py
Functions
deserialize(topic, payload)
Deserializes an Avro-encoded byte string using AWS Glue.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The Kafka topic from which the message was consumed. |
required |
payload
|
bytes
|
The Avro-encoded binary payload. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Any |
Any
|
The raw data dictionary extracted from the AWS DataAndSchema object. |
Source code in src/desflow/connectors/ingress/kafka.py
Connectors (Egress)
BaseEgress
Base class for all egress providers in the simulation.
Egress providers act as asynchronous bridges that consume processed simulation data (telemetry and events) from a thread-safe internal queue and transmit it to external destinations such as Kafka, databases, or the console.
Functions
run(egress_queue)
async
Listens to the internal queue and pushes data to an external sink.
This method should contain an asynchronous loop that polls the provided queue and handles the networking/I/O logic specific to the destination system.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing batches of dictionaries to be exported. |
required |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If the subclass does not override this method. |
Source code in src/desflow/connectors/egress/base.py
KafkaEgress(bootstrap_servers, telemetry_topic='sim-telemetry', event_topic='sim-events', topic_router=None, topic_serializers=None, default_serializer=None, include_stream_type=False, **kwargs)
Bases: BaseEgress
High-throughput Kafka producer for simulation telemetry and events.
This connector utilizes aiokafka for asynchronous I/O and orjson for fast
serialization. It implements a resilient connection loop with exponential
backoff.
By default, data is routed to the telemetry_topic or event_topic based on
its stream_type. If a topic_router callable is provided, topic selection
is delegated to that function instead, allowing for advanced multiplexing
(e.g., splitting ML vs. UI events).
Without a topic_router, the event and telemetry topics are created at start
when they do not exist, with one partition and a replication factor of one.
With a router the topics are left to the caller, because only the router knows
which ones it uses.
Note
By default stream_type is treated as routing metadata and is left out of
the published message, because the topic already identifies the stream.
ParquetStorageEgress keeps stream_type as a column, so a downstream
filter written against Parquet files matches nothing when it is pointed at
a Kafka topic. Set include_stream_type=True to keep the field in the
message and make both paths carry it.
Attributes:
| Name | Type | Description |
|---|---|---|
bootstrap_servers |
str
|
Comma-separated list of Kafka brokers. |
telemetry_topic |
str
|
The default topic name for telemetry data. |
event_topic |
str
|
The default topic name for lifecycle events. |
topic_router |
Callable
|
Optional external logic to determine the topic. |
topic_serializers |
Optional[Dict[str, MessageSerializer]]
|
Optional mapping of topics to specific serializers. |
default_serializer |
Optional[MessageSerializer]
|
Fallback serializer if a topic is not in |
include_stream_type |
bool
|
Whether the published message keeps the |
producer_config |
dict
|
Configuration dictionary passed to AIOKafkaProducer. |
Examples:
Defining a custom topic router to split machine learning events from standard telemetry:
Initializes the KafkaEgress with topic and connection settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bootstrap_servers
|
str
|
Comma-separated list of Kafka brokers. |
required |
telemetry_topic
|
str
|
Destination topic for telemetry stream. |
'sim-telemetry'
|
event_topic
|
str
|
Destination topic for event stream. |
'sim-events'
|
topic_router
|
Optional[Callable[[dict], str]]
|
Optional callable to dynamically route payloads to specific topics. |
None
|
topic_serializers
|
Optional[Dict[str, MessageSerializer]]
|
Mapping of target topics to their specific |
None
|
default_serializer
|
Optional[MessageSerializer]
|
The fallback serializer to use if a topic lacks a specific mapping. Defaults to |
None
|
include_stream_type
|
bool
|
Keep |
False
|
**kwargs
|
Any
|
Additional overrides for the AIOKafkaProducer configuration. |
{}
|
Source code in src/desflow/connectors/egress/kafka.py
Functions
run(egress_queue)
async
The main execution loop that consumes the egress queue and publishes to Kafka.
This method maintains a persistent connection to Kafka. If the connection
is lost, it implements an exponential backoff retry strategy. It polls
the internal egress_queue for batches of data, determines the target
topic based on the 'stream_type', and performs asynchronous sends using
the configured serializers.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing lists of dictionaries or Pydantic models generated by the EgressMixIn. |
required |
Note
The loop exits gracefully upon receiving an asyncio.CancelledError,
ensuring the Kafka producer is stopped correctly.
Source code in src/desflow/connectors/egress/kafka.py
422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 | |
ConsoleEgress(format='log')
Bases: BaseEgress
Local egress provider that outputs simulation data to the console.
This connector serves as a debugging and local development tool. The log
format logs each batch, distinguishing telemetry metrics from discrete events
by prefixes. The json format writes each record to stdout as one JSON line,
with its stream_type, for a pipe such as jq.
Chooses how records are written.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
format
|
Literal['log', 'json']
|
|
'log'
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Source code in src/desflow/connectors/egress/local.py
Functions
run(egress_queue)
async
Continuously polls the egress queue and writes each batch.
This method implements an asynchronous polling loop. It extracts batches
from the internal queue. With the json format it writes each batch to
stdout in one write, one JSON line per record as published. With the
log format it writes each batch with one log call at INFO
level, one line per record, each line starting with a prefix:
- [TEL]: Telemetry data (e.g., resource utilization, queue lengths).
- [EVT]: Event data (e.g., task lifecycle transitions).
- [LBL]: Ground-truth records, on a sink added with when: labels.
One call per batch rather than per record costs far less (issue #24), so the logging format's prefix, such as the time and level, appears on the first line of each batch only.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing lists of dictionaries (batches) generated by the EgressMixIn. |
required |
Note
If the queue is empty, the method yields control to the asyncio event loop for a short duration to prevent CPU pinning.
Source code in src/desflow/connectors/egress/local.py
PostgresEgress(connection_dsn, table_name=None, upsert_keys=None, tables=None, **kwargs)
Bases: BaseEgress
Asynchronous egress provider for persisting simulation data to PostgreSQL.
This connector handles long-term storage of simulation results or continuous generation of relational CDC data, performing bulk inserts of batched data into a specified PostgreSQL table using asyncpg.
By default a record whose key already exists is skipped (ON CONFLICT DO
NOTHING). Name the key columns in upsert_keys to update the existing row
instead, which a simulation needs when it changes rows it wrote earlier, such
as an order moving from processing to shipped.
An ISO time string bound for a timestamp or date column is converted to a
datetime or date first, because asyncpg rejects the string itself.
Tables named in tables are created at start when they do not exist, so a run
needs no separate schema step.
An event's row also carries the record's key as event_id, and its task and
instance, unless the payload has fields of those names. A ground-truth record,
sent to a sink added with when: labels, is written as sim_ts, timestamp,
path, old, new and label, the last three as JSON text.
Examples:
Updating orders as their status changes:
Creating the table at start, in the shape a YAML file gives:
Initializes the PostgresEgress with connection details.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
connection_dsn
|
str
|
PostgreSQL connection string (DSN). |
required |
table_name
|
str | None
|
Target table for simulation records. Defaults to the one
table in |
None
|
upsert_keys
|
list[str] | None
|
Optional key columns. When given, a record whose key already exists updates that row's other columns, and the columns need a unique index or constraint. When None, such a record is skipped. |
None
|
tables
|
dict[str, Any] | None
|
Optional tables to create at start when they do not exist. Each
table name maps to |
None
|
**kwargs
|
Any
|
Additional connection pool arguments for asyncpg. |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If a |
Source code in src/desflow/connectors/egress/postgres.py
Functions
run(egress_queue)
async
Main execution loop for PostgreSQL data persistence.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing batches of simulation data. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a column in |
Source code in src/desflow/connectors/egress/postgres.py
308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 | |
RedisEgress(url, stream_name='simulation_data', **kwargs)
Bases: BaseEgress
High-throughput asynchronous egress provider for streaming simulation data to Redis.
This connector utilizes redis.asyncio for asynchronous I/O and publishes batched
simulation results (events and telemetry) to Redis Streams (XADD) for real-time consumers.
By default, it uses a unified stream, but supports dynamic stream multiplexing
by reading the __stream__ key from the data payloads.
Attributes:
| Name | Type | Description |
|---|---|---|
url |
str
|
The connection URL for the Redis server (e.g., redis://127.0.0.1:6379/0). |
stream_name |
str
|
The default target Redis Stream name. |
client |
Redis | None
|
The underlying redis-py asynchronous client instance. |
**kwargs |
Redis | None
|
Additional configuration dictionary passed to |
Initializes the RedisEgress with topic and connection settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
Redis connection URL (e.g., redis://127.0.0.1:6379/0). |
required |
stream_name
|
str
|
Target Redis Stream name. |
'simulation_data'
|
**kwargs
|
Any
|
Additional arguments for |
{}
|
Source code in src/desflow/connectors/egress/redis.py
Functions
run(egress_queue)
async
The main execution loop that consumes the egress queue and publishes to Redis.
This method maintains a connection to Redis and uses pipeline (XADD)
commands to insert batches of data efficiently.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing lists of dictionaries or Pydantic models generated by the EgressMixIn. |
required |
Source code in src/desflow/connectors/egress/redis.py
JsonlStorageEgress(default_path=None, filesystem=None, path_router=None)
Bases: BaseEgress
High-throughput JSONL batch writer for local file systems or object storage.
This connector uses PyArrow's Virtual File System (VFS) to write data to local disks, S3-compatible storage such as AWS S3 or SeaweedFS, Google Cloud Storage or Azure Blob Storage. It writes a uniquely named file for every batch, so no two writes go to the same file and a file, once written, is never changed.
By default, events are appended to the default_path as flat rows: each
event's value mapping is unpacked into top-level fields, and telemetry is left
out. If a path_router callable is provided, destination logic is delegated to
that function, allowing for advanced multiplexing (e.g., splitting logs vs.
errors) or dynamically dropping specific records by returning None. Records
are then written as the router leaves them.
Attributes:
| Name | Type | Description |
|---|---|---|
default_path |
Optional[str]
|
The fallback destination path if no router is provided. |
filesystem |
Optional[Any]
|
A PyArrow |
path_router |
Optional[Callable]
|
Optional logic to dynamically route or drop payloads. |
Examples:
Routing errors to a specific file and dropping meaningless metrics:
Initializes the JsonlStorageEgress with routing and VFS settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
default_path
|
Optional[str]
|
The target file path prefix (e.g., "data/logs.jsonl"),
used when no router is given. Only events are written there, with
their |
None
|
filesystem
|
Optional[Any]
|
An instantiated PyArrow FileSystem (e.g., |
None
|
path_router
|
Optional[Callable[[dict], Optional[str]]]
|
A function taking a dict payload and returning a string path, or None to drop the record. |
None
|
Source code in src/desflow/connectors/egress/storage.py
Functions
run(egress_queue)
async
The main execution loop that consumes the egress queue and writes JSONL chunks.
This loop continuously polls the internal egress_queue for data batches.
When a batch is received, it offloads the synchronous PyArrow file I/O operations
to a background thread to prevent blocking the asyncio event loop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing batches of dictionaries generated by the environment's EgressMixIn. |
required |
Raises:
| Type | Description |
|---|---|
ImportError
|
If the 'pyarrow' package is not installed. |
Note
The loop exits gracefully upon receiving an asyncio.CancelledError,
ensuring background file writers have completed before shutting down.
Source code in src/desflow/connectors/egress/storage.py
ParquetStorageEgress(default_path=None, filesystem=None, path_router=None)
Bases: BaseEgress
Columnar Parquet writer for local disks or object storage.
This connector buffers simulation records and converts them into heavily
compressed Parquet tables using pyarrow. It natively supports writing to
local disks, S3-compatible object stores (AWS S3, SeaweedFS), Google Cloud Storage
or Azure Blob Storage via PyArrow's VFS.
To support massive parallel processing (e.g., Athena, Databricks), it implements file rotation (chunking), creating a uniquely named Parquet part-file for every flush of the buffer. A file's columns come from every row of its batch; a column a later batch adds widens the path's schema, and each batch is cast to it, so a column keeps its type across chunks.
Without a path_router, events are written to default_path as flat rows:
each event's value mapping becomes columns, and telemetry is left out. With a
router, records are written as the router leaves them.
Attributes:
| Name | Type | Description |
|---|---|---|
default_path |
Optional[str]
|
The fallback destination path. |
filesystem |
Optional[Any]
|
A PyArrow |
path_router |
Optional[Callable]
|
Optional logic to dynamically route or drop payloads. |
schemas |
Dict[str, Any]
|
Internal registry of inferred PyArrow schemas per file path. |
Examples:
Splitting simulation events and telemetry into distinct Parquet datasets directly to S3:
Initializes the ParquetStorageEgress with routing and VFS settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
default_path
|
Optional[str]
|
The target file path prefix (e.g., "data/events.parquet"),
used when no router is given. Only events are written there, with
their |
None
|
filesystem
|
Optional[Any]
|
An instantiated PyArrow FileSystem (e.g., |
None
|
path_router
|
Optional[Callable[[dict], Optional[str]]]
|
A function taking a dict payload and returning a string path, or None to drop the record. |
None
|
Source code in src/desflow/connectors/egress/storage.py
Functions
run(egress_queue)
async
The main execution loop that consumes the egress queue and writes Parquet chunks.
This loop continuously polls the internal egress_queue for data batches.
When a batch is received, it offloads the synchronous PyArrow table conversion
and file I/O operations to a background thread to prevent blocking the asyncio loop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing batches of dictionaries generated by the environment's EgressMixIn. |
required |
Raises:
| Type | Description |
|---|---|
ImportError
|
If the 'pyarrow' package is not installed. |
Note
The loop exits gracefully upon receiving an asyncio.CancelledError,
ensuring the final Parquet table has been fully flushed to disk.
Source code in src/desflow/connectors/egress/storage.py
IcebergStorageEgress(catalog, default_table=None, table_router=None, schemas=None, locations=None, upsert_keys=None)
Bases: BaseEgress
Apache Iceberg table writer for lakehouse ingestion patterns.
ParquetStorageEgress writes files and leaves registration to the caller. This
connector appends straight into an Iceberg table instead, so one run can feed a
stream sink and a queryable table in the same pass. Each flush of a provider's
buffer becomes exactly one Iceberg commit, which is why the buffer size matters:
every commit writes a manifest, a manifest list and a new metadata.json, and
query planning degrades as the snapshot count grows. Give this provider a large
batch_size on add_egress so the run produces a handful of snapshots rather
than hundreds.
Without a table_router, events are written to default_table as flat rows:
each event's value mapping becomes columns, and telemetry is left out. With a
router, records are written as the router leaves them.
By default every record is appended. Name a table's key columns in
upsert_keys to write it with PyIceberg's upsert instead: a record whose key
exists updates that row, and a new key is inserted, so rerunning a seeded
simulation over the same window does not duplicate rows. An upsert reads the
matching rows before writing, so it is slower than an append, and it never
deletes rows. It replaces the whole row, so every record must carry every column: a
missing column is written as null. Within one flush, the last record for a key wins. An upsert
flush is still one commit, but it can add up to three snapshots: an overwrite
and two appends.
The catalog is supplied by the caller, either as a built Catalog or as a
mapping of PyIceberg catalog properties that is passed to load_catalog. The
supported catalogs, which the iceberg extra installs, each with the test that
writes through it, are:
- Iceberg REST, in the integration tests against the odctl
catalogprofile; - SQL on SQLite, in the unit tests;
- SQL on PostgreSQL, in the integration tests against the odctl
postgresprofile; - Glue, in the integration tests against Floci, an AWS emulator.
A table without a schema in schemas is created with the columns of every row of
its first batch, and a column a later batch adds is added to the table, as the
Parquet writer widens its schema. Inference reads an ISO timestamp as a string,
so a table whose consumer expects a real timestamp needs an explicit schema. An
ISO string bound for a timestamp or date column of the table is converted to a
datetime or date before the write, because PyArrow rejects the string itself.
Attributes:
| Name | Type | Description |
|---|---|---|
catalog |
Any
|
An instantiated pyiceberg |
catalog_properties |
Optional[Dict[str, Any]]
|
The PyIceberg catalog
properties given instead of a |
default_table |
Optional[str]
|
Fallback |
table_router |
Optional[Callable]
|
Logic returning |
schemas |
Dict[str, Any]
|
PyArrow schema per table identifier. Seeded with any explicit schemas, then filled in from the tables that get created. |
locations |
Dict[str, str]
|
Explicit storage location per table identifier. |
upsert_keys |
Dict[str, list]
|
Key columns per table identifier, for the
tables written with |
Examples:
Splitting events and telemetry into two tables on a REST catalog:
Initializes the IcebergStorageEgress with a catalog and routing settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
catalog
|
Any
|
An instantiated pyiceberg |
required |
default_table
|
Optional[str]
|
The target |
None
|
table_router
|
Optional[Callable[[dict], Optional[str]]]
|
A function taking a dict payload and returning a
|
None
|
schemas
|
Optional[Dict[str, Any]]
|
Optional PyArrow schema per table identifier. A table named here is created with that schema and every batch is cast to it, instead of the schema being inferred from the first batch. A schema can also be a mapping of column name to an Iceberg type name: string, int, long, float, double, boolean, timestamp, timestamptz, date or binary. |
None
|
locations
|
Optional[Dict[str, str]]
|
Optional storage location per table identifier. Pin this when a
consumer reads the table by path, because a catalog with
|
None
|
upsert_keys
|
Optional[Dict[str, list]]
|
Optional key columns per table identifier. A table named
here is written with |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If neither |
ImportError
|
If |
Source code in src/desflow/connectors/egress/iceberg.py
191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 | |
Functions
run(egress_queue)
async
The main execution loop that consumes the egress queue and commits to Iceberg.
This loop continuously polls the internal egress_queue for data batches.
When a batch is received, it offloads the synchronous PyArrow conversion and
the Iceberg commit to a background thread, so the asyncio loop is not blocked
while the catalog is written.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
egress_queue
|
Queue
|
A thread-safe queue containing batches of dictionaries generated by the environment's EgressMixIn. |
required |
Raises:
| Type | Description |
|---|---|
ImportError
|
If the 'pyarrow' package is not installed. |
Note
The loop exits gracefully upon receiving an asyncio.CancelledError, so
the final commit has completed before shutdown returns.
Source code in src/desflow/connectors/egress/iceberg.py
Serializers
MessageSerializer
Bases: Protocol
Protocol defining the interface for all Kafka egress serializers.
Any custom serializer passed to KafkaEgress must implement the
serialize method according to this signature.
Functions
serialize(topic, data)
Serializes the given payload into a byte string for Kafka transmission.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The destination Kafka topic. This is often required by schema registries (like Avro) to resolve the correct schema subject, even if simpler serializers (like JSON) choose to ignore it. |
required |
data
|
Any
|
The payload to serialize. This is typically a raw dictionary or a Pydantic model instance. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bytes |
bytes
|
The fully encoded byte string ready to be published to Kafka. |
Source code in src/desflow/connectors/egress/kafka.py
JsonSerializer
Default fallback serializer providing backward compatibility via orjson.
This serializer converts dictionaries or Pydantic models into standard JSON byte strings.
Functions
serialize(topic, data)
Serializes the given data to a JSON byte string.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The target Kafka topic (unused by this serializer but required by protocol). |
required |
data
|
Any
|
The payload to serialize (dictionary or Pydantic model). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bytes |
bytes
|
The JSON-encoded byte string. |
Source code in src/desflow/connectors/egress/kafka.py
ConfluentAvroSerializer(registry_url, schema_str, *, auth=UNSET, verify=UNSET, cert=UNSET, timeout=UNSET, headers=UNSET, proxy=UNSET, max_retries=UNSET, **registry_kwargs)
Lazy-loaded serializer for Confluent Schema Registry using fastavro and httpx.
This class converts simulation data into Avro-encoded byte strings in the
Confluent wire format. It registers the schema under the subject
<topic>-value on the first record of each topic, and reuses the schema id
after that.
Initializes the Confluent Avro serializer. It makes no request to the registry.
UNSET as a default, in the signature or the Default column of the API page,
means the argument was not given; each argument's description says its real
default.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry_url
|
str
|
The base URL of the Confluent Schema Registry. Several URLs separated by commas are tried in order. |
required |
schema_str
|
str
|
The Avro schema defined as a JSON string (typically
generated from a dataclasses-avroschema |
required |
auth
|
Optional[Tuple[str, str]]
|
A username and password for basic
authentication, for example |
UNSET
|
verify
|
bool | str | SSLContext
|
How to check the registry's certificate: True (the default), False, the path of a CA file or directory, or an SSL context. |
UNSET
|
cert
|
str | tuple | None
|
A client certificate file, or a tuple of the certificate file, the key file and the key password. The default is None. |
UNSET
|
timeout
|
Optional[float]
|
Seconds each step of a request to the registry (connecting, sending, reading) may take. The default is 10.0. None waits without a limit. |
UNSET
|
headers
|
Optional[Mapping[str, str]]
|
Headers sent with every request,
for example |
UNSET
|
proxy
|
Optional[str]
|
The URL of a proxy. The default is None. |
UNSET
|
max_retries
|
int
|
How many times a request is repeated after a transport error or a status of 408, 429, 500, 502, 503 or 504. The default is 3. |
UNSET
|
**registry_kwargs
|
Any
|
Deprecated confluent-kafka keys, such as
|
{}
|
Raises:
| Type | Description |
|---|---|
ImportError
|
If the 'fastavro' or 'httpx' package is not installed. |
ValueError
|
If a deprecated key is not accepted, or is given together with its named argument. |
Source code in src/desflow/connectors/egress/kafka.py
97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | |
Functions
serialize(topic, data)
Serializes the given data into an Avro byte string.
The first call for a topic registers the schema under <topic>-value, which
makes a request to the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The target Kafka topic, used for schema subject naming. |
required |
data
|
Any
|
The payload to serialize (dictionary or Pydantic model). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bytes |
bytes
|
The Avro-encoded byte string, with the Confluent prefix. |
Raises:
| Type | Description |
|---|---|
SchemaRegistryError
|
If the registry refuses the schema. |
TransportError
|
If no registry URL could be reached. |
Source code in src/desflow/connectors/egress/kafka.py
GlueAvroSerializer(registry_name, schema_str, **boto3_kwargs)
Lazy-loaded serializer for AWS Glue Schema Registry using boto3.
This class converts simulation data into Avro-encoded byte strings, integrating directly with AWS Glue Schema Registry for schema validation.
Initializes the AWS Glue Avro serializer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry_name
|
str
|
The name of the AWS Glue Schema Registry. |
required |
schema_str
|
str
|
The Avro schema defined as a JSON string (typically
generated from a dataclasses-avroschema |
required |
**boto3_kwargs
|
Any
|
Additional arguments passed directly to the |
{}
|
Raises:
| Type | Description |
|---|---|
ImportError
|
If the 'aws-glue-schema-registry' or 'boto3' packages are not installed. |
Source code in src/desflow/connectors/egress/kafka.py
Functions
serialize(topic, data)
Serializes the given data into an Avro byte string using AWS Glue.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic
|
str
|
The target Kafka topic. |
required |
data
|
Any
|
The payload to serialize (dictionary or Pydantic model). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bytes |
bytes
|
The Avro-encoded byte string. |
Source code in src/desflow/connectors/egress/kafka.py
SchemaRegistryError(http_status, error_code, message)
Bases: RuntimeError
A schema registry answered a request with an error status.
Attributes:
| Name | Type | Description |
|---|---|---|
http_status |
int
|
The HTTP status of the response. |
error_code |
int
|
The registry's own error code, or -1 when the body of the response held none. |
message |
str
|
The registry's message, or the raw body when it was not JSON. |
Source code in src/desflow/connectors/_schema_registry/errors.py
SerializationError
Bases: ValueError
A payload or a schema cannot be written or read.
Raised for data too short to hold the registry's framing, a magic byte that is not known, a schema of a format the class does not handle, and a schema with rules that desflow does not run.