API Reference

Simulation Context

SimulationContext(sim_id, factor=1.0, random_seed=None, logical_start_time=None, go_live_at=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. Centralized Switchboard: 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: By exposing a central Sampler driven by a seeded NumPy generator, all background arrivals, decorators, and user-defined physical generators share the same random state, ensuring 100% reproducible Digital Twin executions.

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 factor is replaced by real-time pacing, so one run can backfill history and then tail live.
random_seed	Optional[int]	The deterministic seed for the NumPy RNG.
sampler	Optional[Sampler]	The centralized Random Number Generator engine. Accessible to user-defined raw generators only after .run() is called.

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 DynamicRealtimeEnvironment. Crucial for historical backfilling (e.g., generating data from last week).	None
go_live_at	Optional[datetime]	Optional logical instant at which the run stops using factor and switches to real time, forwarded to DynamicRealtimeEnvironment. With factor=0.0 and a backdated logical_start_time, the history up to this instant is generated as fast as the machine allows and the rest arrives in real time, so a backfill and a live tail take one run.	None

Source code in src/dynamic_des/core/context.py

def __init__(
    self,
    sim_id: str,
    factor: float = 1.0,
    random_seed: Optional[int] = None,
    logical_start_time: Optional[datetime] = None,
    go_live_at: Optional[datetime] = None,
):
    """
    Initializes the context builder with a designated namespace and temporal factor.

    Args:
        sim_id: The unique prefix for this group of parameters. All generated
            telemetry and event paths will be prefixed with this ID.
        factor: The real-time synchronization factor. Defaults to 1.0.
        random_seed: Optional seed for deterministic execution.
        logical_start_time: Optional override for the environment's base clock,
            forwarded to `DynamicRealtimeEnvironment`. Crucial for historical
            backfilling (e.g., generating data from last week).
        go_live_at: Optional logical instant at which the run stops using `factor`
            and switches to real time, forwarded to `DynamicRealtimeEnvironment`.
            With `factor=0.0` and a backdated `logical_start_time`, the history up
            to this instant is generated as fast as the machine allows and the rest
            arrives in real time, so a backfill and a live tail take one run.
    """
    self.sim_id = sim_id
    self.factor = factor
    self.random_seed = random_seed
    self.logical_start_time = logical_start_time
    self.go_live_at = go_live_at

    # Builder State (Pre-Compilation)
    self._ingress_providers: List[Any] = []
    self._egress_providers: List[Any] = []
    # Aligned with _egress_providers by position. None means the provider
    # receives every record.
    self._egress_predicates: List[Optional[Callable[[dict], bool]]] = []
    self._batch_size = 500
    self._flush_interval = 1.0
    self._max_queued_batches = 2000
    self._drain_stall_seconds = 30.0

    self._resources_config: Dict[str, CapacityConfig] = {}
    self._containers_config: Dict[str, CapacityConfig] = {}
    self._services_config: Dict[str, DistributionConfig] = {}
    self._arrivals_config: Dict[str, DistributionConfig] = {}
    self._variables_config: Dict[str, Any] = {}

    # Runtime State (Post-Compilation)
    self._env: Optional[DynamicRealtimeEnvironment] = None
    self.sampler: Optional[Sampler] = None
    self._resources_map: Dict[str, DynamicResource] = {}
    self._startup_loops: List[Tuple[Callable, Optional[str]]] = []

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

Functions

add_arrival(name, dist, rate=0.0, mean=0.0)

Registers a statistical distribution defining the inter-arrival times of entities.

Parameters:

Name	Type	Description	Default
name	str	The internal identifier (e.g., 'structural').	required
dist	Literal['exponential', 'normal', 'lognormal']	The distribution type ('exponential', 'normal', 'lognormal').	required
rate	float	The rate (lambda) for exponential distributions.	0.0
mean	float	The mean (mu) for normal distributions.	0.0

Returns:

Name	Type	Description
SimulationContext	SimulationContext	The current instance for method chaining.

Source code in src/dynamic_des/core/context.py

def add_arrival(
    self,
    name: str,
    dist: Literal["exponential", "normal", "lognormal"],
    rate: float = 0.0,
    mean: float = 0.0,
) -> "SimulationContext":
    """
    Registers a statistical distribution defining the inter-arrival times of entities.

    Args:
        name: The internal identifier (e.g., 'structural').
        dist: The distribution type ('exponential', 'normal', 'lognormal').
        rate: The rate (lambda) for exponential distributions.
        mean: The mean (mu) for normal distributions.

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._arrivals_config[name] = DistributionConfig(
        dist=dist, rate=rate, mean=mean
    )
    return self

add_container(name, current_cap, max_cap)

Registers a continuous state container representing fluid or gradual levels.

Unlike Resources, Containers map to floating-point metrics. They are specifically utilized for modeling continuous physical states such as concept drift (wear and tear), fluid levels, or thermal degradation.

Parameters:

Name	Type	Description	Default
name	str	The internal identifier (e.g., 'wear_structural').	required
current_cap	float	The starting floating-point level (e.g., 0.001 for a new machine).	required
max_cap	float	The absolute maximum ceiling (e.g., 100.0 for catastrophic failure).	required

Returns:

Name	Type	Description
SimulationContext	SimulationContext	The current instance for method chaining.

Source code in src/dynamic_des/core/context.py

def add_container(
    self, name: str, current_cap: float, max_cap: float
) -> "SimulationContext":
    """
    Registers a continuous state container representing fluid or gradual levels.

    Unlike Resources, Containers map to floating-point metrics. They are
    specifically utilized for modeling continuous physical states such as
    concept drift (wear and tear), fluid levels, or thermal degradation.

    Args:
        name: The internal identifier (e.g., 'wear_structural').
        current_cap: The starting floating-point level (e.g., 0.001 for a new machine).
        max_cap: The absolute maximum ceiling (e.g., 100.0 for catastrophic failure).

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._containers_config[name] = CapacityConfig(
        current_cap=current_cap, max_cap=max_cap
    )
    return self

add_egress(provider, when=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 BaseEgress (e.g., KafkaEgress or ParquetStorageEgress).	required
when	Optional[Callable[[dict], bool]]	Optional predicate taking one record and returning True to send it to this provider. None sends every record, which is the usual case.	None

Returns:

Name	Type	Description
SimulationContext	SimulationContext	The current instance for method chaining.

Example

# Same dataset to both sinks, in one run
app.add_egress(kafka).add_egress(parquet)

# Or split it: hot tail to Kafka, cold history to Parquet
app.add_egress(kafka, when=lambda r: r["timestamp"] >= hot_from)
app.add_egress(parquet, when=lambda r: r["timestamp"] < hot_from)

Source code in src/dynamic_des/core/context.py

def add_egress(
    self,
    provider: Any,
    when: Optional[Callable[[dict], bool]] = None,
) -> "SimulationContext":
    """
    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.

    Args:
        provider: An initialized connector subclassing `BaseEgress`
            (e.g., `KafkaEgress` or `ParquetStorageEgress`).
        when: Optional predicate taking one record and returning True to send it to
            this provider. None sends every record, which is the usual case.

    Returns:
        SimulationContext: The current instance for method chaining.

    Example:
        ```python
        # Same dataset to both sinks, in one run
        app.add_egress(kafka).add_egress(parquet)

        # Or split it: hot tail to Kafka, cold history to Parquet
        app.add_egress(kafka, when=lambda r: r["timestamp"] >= hot_from)
        app.add_egress(parquet, when=lambda r: r["timestamp"] < hot_from)
        ```
    """
    self._egress_providers.append(provider)
    self._egress_predicates.append(when)
    return self

add_ingress(provider)

Registers an asynchronous ingress provider to listen for external mutations.

Parameters:

Name	Type	Description	Default
provider	Any	An initialized connector subclassing BaseIngress (e.g., KafkaIngress).	required

Returns:

Name	Type	Description
SimulationContext	SimulationContext	The current instance for method chaining.

Source code in src/dynamic_des/core/context.py

def add_ingress(self, provider: Any) -> "SimulationContext":
    """
    Registers an asynchronous ingress provider to listen for external mutations.

    Args:
        provider: An initialized connector subclassing `BaseIngress`
            (e.g., `KafkaIngress`).

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._ingress_providers.append(provider)
    return self

add_resource(name, current_cap, max_cap)

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	int	The active physical token capacity at simulation start.	required
max_cap	int	The absolute physical maximum bound of tokens allowed.	required

Returns:

Name	Type	Description
SimulationContext	SimulationContext	The current instance for method chaining.

Source code in src/dynamic_des/core/context.py

def add_resource(
    self, name: str, current_cap: int, max_cap: int
) -> "SimulationContext":
    """
    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).

    Args:
        name: The internal identifier (e.g., 'mill_structural').
        current_cap: The active physical token capacity at simulation start.
        max_cap: The absolute physical maximum bound of tokens allowed.

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._resources_config[name] = CapacityConfig(
        current_cap=current_cap, max_cap=max_cap
    )
    return self

add_service(name, dist, mean=0.0, std=0.0, rate=0.0)

Registers a statistical distribution defining a process service duration.

Parameters:

Name	Type	Description	Default
name	str	The internal identifier (e.g., 'pass_roughing').	required
dist	Literal['exponential', 'normal', 'lognormal']	The distribution family ('normal', 'exponential', 'lognormal').	required
mean	float	The mean (mu) for normal/lognormal distributions.	0.0
std	float	The standard deviation (sigma) for normal/lognormal distributions.	0.0
rate	float	The rate (lambda) for exponential distributions.	0.0

Returns:

Name	Type	Description
SimulationContext	SimulationContext	The current instance for method chaining.

Source code in src/dynamic_des/core/context.py

def add_service(
    self,
    name: str,
    dist: Literal["exponential", "normal", "lognormal"],
    mean: float = 0.0,
    std: float = 0.0,
    rate: float = 0.0,
) -> "SimulationContext":
    """
    Registers a statistical distribution defining a process service duration.

    Args:
        name: The internal identifier (e.g., 'pass_roughing').
        dist: The distribution family ('normal', 'exponential', 'lognormal').
        mean: The mean (mu) for normal/lognormal distributions.
        std: The standard deviation (sigma) for normal/lognormal distributions.
        rate: The rate (lambda) for exponential distributions.

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._services_config[name] = DistributionConfig(
        dist=dist, mean=mean, std=std, rate=rate
    )
    return self

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/dynamic_des/core/context.py

def add_variable(self, name: str, value: Any) -> "SimulationContext":
    """
    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.

    Args:
        name: The internal identifier (e.g., 'velocity_structural').
        value: The initial payload (supports primitives, lists, and dicts).

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._variables_config[name] = value
    return self

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/dynamic_des/core/context.py

def arrival_loop(self, arrival_id: str) -> Callable:
    """
    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()`.

    Args:
        arrival_id: The ID of the configured arrival rate to dynamically sample.
    """

    def decorator(user_func):
        self._startup_loops.append((user_func, arrival_id))
        return user_func

    return decorator

get_resource(resource_id)

Retrieves an active SimPy DynamicResource instance for raw lock management.

Raises:

Type	Description
RuntimeError	If called during the Builder Phase.

Source code in src/dynamic_des/core/context.py

def get_resource(self, resource_id: str) -> DynamicResource:
    """
    Retrieves an active SimPy `DynamicResource` instance for raw lock management.

    Raises:
        RuntimeError: If called during the Builder Phase.
    """
    if not self._env:
        raise RuntimeError("Resources are not instantiated until context is run.")
    return self._resources_map[resource_id]

publish(metric_name, value)

Publishes a scalar telemetry metric to the non-blocking egress buffer.

Source code in src/dynamic_des/core/context.py

def publish(self, metric_name: str, value: Any) -> None:
    """
    Publishes a scalar telemetry metric to the non-blocking egress buffer.
    """
    if not self._env:
        raise RuntimeError("Cannot publish telemetry before context is run.")
    self._env.publish_telemetry(f"{self.sim_id}.{metric_name}", value)

run(until=None)

Compiles the defined infrastructure architecture and triggers the simulation loop.

Phase 1: Instantiates the foundational DynamicRealtimeEnvironment and Sampler. Phase 2: Compiles all builder dictionaries into a monolithic SimParameter object and registers it with the Switchboard Registry. Phase 3: Connects and boots the asynchronous Kafka/Redis background threads. Phase 4: Hydrates physical SimPy limits based on the registry boundaries. Phase 5: Spawns all registered @arrival_loop and @telemetry_loop generators. Phase 6: Relinquishes the main thread to the SimPy environment execution clock.

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, the simulation runs infinitely.	None

Source code in src/dynamic_des/core/context.py

def run(self, until: Any = None) -> None:
    """
    Compiles the defined infrastructure architecture and triggers the simulation loop.

    Phase 1: Instantiates the foundational `DynamicRealtimeEnvironment` and `Sampler`.
    Phase 2: Compiles all builder dictionaries into a monolithic `SimParameter`
             object and registers it with the Switchboard Registry.
    Phase 3: Connects and boots the asynchronous Kafka/Redis background threads.
    Phase 4: Hydrates physical SimPy limits based on the registry boundaries.
    Phase 5: Spawns all registered `@arrival_loop` and `@telemetry_loop` generators.
    Phase 6: Relinquishes the main thread to the SimPy environment execution clock.

    Args:
        until: 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, the
            simulation runs infinitely.
    """
    logger.info(f"Building SimulationContext for '{self.sim_id}'...")

    env_kwargs: Dict[str, Any] = {"factor": self.factor}
    if self.logical_start_time is not None:
        env_kwargs["logical_start_time"] = self.logical_start_time
    if self.go_live_at is not None:
        env_kwargs["go_live_at"] = self.go_live_at
    self._env = DynamicRealtimeEnvironment(**env_kwargs)

    # Initializes RNG with the deterministic seed
    self.sampler = Sampler(rng=np.random.default_rng(self.random_seed))

    # Compile Master Parameter Object
    params = SimParameter(
        sim_id=self.sim_id,
        resources=self._resources_config,
        containers=self._containers_config,
        service=self._services_config,
        arrival=self._arrivals_config,
        variables=self._variables_config,
    )
    self._env.registry.register_sim_parameter(params)

    # Connect External Infrastructure
    if self._ingress_providers:
        self._env.setup_ingress(self._ingress_providers)

    if self._egress_providers:
        self._env.setup_egress(
            self._egress_providers,
            batch_size=self._batch_size,
            flush_interval=self._flush_interval,
            max_queued_batches=self._max_queued_batches,
            drain_stall_seconds=self._drain_stall_seconds,
            predicates=self._egress_predicates,
        )

    # Hydrate Physical SimPy Resources
    for res_id in self._resources_config.keys():
        self._resources_map[res_id] = DynamicResource(
            self._env, self.sim_id, res_id
        )

    # Boot Registered Daemon Generators
    for loop_func, meta in self._startup_loops:
        if meta:
            # Arrival loops require context injection
            self._env.process(loop_func(self))
        else:
            # Telemetry loops run blindly
            self._env.process(loop_func())

    try:
        logger.info("Simulation engine started.")
        if isinstance(until, str):
            from dynamic_des.utils import time_to_seconds

            until = time_to_seconds(until)

        self._env.run(until=until)
    finally:
        self._env.teardown()

spawn(process)

Registers a new asynchronous generator with the active SimPy event loop.

This acts as the "escape hatch" for complex logic (like physical slab tracking) that outgrows the strict abstraction of the @task decorator.

Source code in src/dynamic_des/core/context.py

def spawn(self, process: Any) -> None:
    """
    Registers a new asynchronous generator with the active SimPy event loop.

    This acts as the "escape hatch" for complex logic (like physical slab
    tracking) that outgrows the strict abstraction of the `@task` decorator.
    """
    if not self._env:
        raise RuntimeError("Cannot spawn processes before context is run.")
    self._env.process(process)

task(service_id, resource_id)

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. Emits a strict queued event payload to the telemetry stream. 2. Blocks asynchronous execution until the specified resource_id token is acquired. 3. Emits a strict started event payload upon acquisition. 4. Executes the wrapped user function to extract custom payload data. 5. Samples the live service_id distribution and yields the temporal timeout. 6. Emits a strict finished event containing the user's custom payload.

Parameters:

Name	Type	Description	Default
service_id	str	The ID of the configured distribution dictating the execution time.	required
resource_id	str	The ID of the configured resource this task exclusively requires.	required

Returns:

Name	Type	Description
Callable	Callable	A decorator that wraps the target function into a SimPy generator.

Example

@app.task(service_id="milling", resource_id="lathe")
def process_part(task_id: int):
    return {"status": "success", "part_id": task_id}

Source code in src/dynamic_des/core/context.py

def task(self, service_id: str, resource_id: str) -> Callable:
    """
    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. Emits a strict `queued` event payload to the telemetry stream.
    2. Blocks asynchronous execution until the specified `resource_id` token is acquired.
    3. Emits a strict `started` event payload upon acquisition.
    4. Executes the wrapped user function to extract custom payload data.
    5. Samples the live `service_id` distribution and yields the temporal timeout.
    6. Emits a strict `finished` event containing the user's custom payload.

    Args:
        service_id: The ID of the configured distribution dictating the execution time.
        resource_id: The ID of the configured resource this task exclusively requires.

    Returns:
        Callable: A decorator that wraps the target function into a SimPy generator.

    Example:
        ```python
        @app.task(service_id="milling", resource_id="lathe")
        def process_part(task_id: int):
            return {"status": "success", "part_id": task_id}
        ```
    """

    def decorator(user_func):
        def wrapper(task_id: int, *args, **kwargs):
            if not self._env or not self.sampler:
                raise RuntimeError("Simulation has not been built yet.")

            res = self._resources_map[resource_id]
            task_key = f"task-{task_id}"
            path_id = f"{self.sim_id}.service.{service_id}"

            self._env.publish_event(
                task_key, {"path_id": path_id, "status": "queued"}
            )

            with res.request() as req:
                yield req
                self._env.publish_event(
                    task_key, {"path_id": path_id, "status": "started"}
                )

                payload = user_func(task_id, *args, **kwargs)

                service_cfg = self._env.registry.get_config(path_id)
                yield self._env.timeout(self.sampler.sample(service_cfg))

                self._env.publish_event(task_key, payload)

        return wrapper

    return decorator

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/dynamic_des/core/context.py

def telemetry_loop(self, interval: float) -> Callable:
    """
    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.

    Args:
        interval: The simulation time (in seconds) to wait between iterations.
    """

    def decorator(user_func):
        def wrapper():
            if not self._env:
                raise RuntimeError("Simulation has not been built yet.")
            while True:
                user_func(self)
                yield self._env.timeout(interval)

        self._startup_loops.append((wrapper, None))
        return user_func

    return decorator

wait_for_arrival(arrival_id)

Yields a dynamic SimPy timeout based on the live arrival distribution.

Raises:

Type	Description
RuntimeError	If called during the Builder Phase (before .run()).

Source code in src/dynamic_des/core/context.py

def wait_for_arrival(self, arrival_id: str) -> Any:
    """
    Yields a dynamic SimPy timeout based on the live arrival distribution.

    Raises:
        RuntimeError: If called during the Builder Phase (before `.run()`).
    """
    if not self._env or not self.sampler:
        raise RuntimeError("Cannot sample arrival before context is run.")

    cfg = self._env.registry.get_config(f"{self.sim_id}.arrival.{arrival_id}")
    return self._env.timeout(self.sampler.sample(cfg))

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.

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/dynamic_des/core/context.py

def with_batching(
    self,
    batch_size: int,
    flush_interval: float,
    max_queued_batches: int = 2000,
    drain_stall_seconds: float = 30.0,
) -> "SimulationContext":
    """
    Configures the internal egress buffering strategy to mitigate I/O lock contention.

    Args:
        batch_size: The maximum number of standard events to hold in memory
            before forcing an asynchronous flush to the egress thread.
        flush_interval: The maximum simulation time (in seconds) to wait
            before forcefully flushing the buffer, regardless of capacity.
        max_queued_batches: 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.
        drain_stall_seconds: 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.

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._batch_size = batch_size
    self._flush_interval = flush_interval
    self._max_queued_batches = max_queued_batches
    self._drain_stall_seconds = drain_stall_seconds
    return self

Environment

DynamicRealtimeEnvironment(initial_time=0, factor=1.0, strict=False, logical_start_time=None, go_live_at=None)

Bases: RealtimeEnvironment, RegistryMixIn, IngressMixIn, EgressMixIn

The core simulation engine for dynamic-des.

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.

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, raises RuntimeError if simulation falls too far behind real time. Defaults to False.	False
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 factor, which is usually 0.0 so a backdated history is produced as fast as the machine allows. From that instant one simulated second takes one real second, so a backfill and a live tail run in one process instead of two.	None

go_live_at is read against the same clock as logical_start_time, so both must be naive or both timezone-aware. 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.

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

def __init__(
    self,
    initial_time=0,
    factor=1.0,
    strict=False,
    logical_start_time: Optional[datetime] = None,
    go_live_at: Optional[datetime] = None,
):
    """
    Initializes the real-time simulation environment.

    Args:
        initial_time (float, optional): The initial simulation time. Defaults to 0.
        factor (float, optional): The real-time factor (e.g., 1.0 = 1 sim second per real second). Defaults to 1.0.
        strict (bool, optional): If True, raises RuntimeError if simulation falls too far behind real time. Defaults to False.
        logical_start_time (datetime, optional): Overrides the environment's base clock.
            Crucial for historical backfilling (e.g., generating data from last week).
        go_live_at (datetime, optional): Logical instant at which pacing switches to
            real time. Until then the run uses `factor`, which is usually 0.0 so a
            backdated history is produced as fast as the machine allows. From that
            instant one simulated second takes one real second, so a backfill and a
            live tail run in one process instead of two.

    `go_live_at` is read against the same clock as `logical_start_time`, so both must
    be naive or both timezone-aware. 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.
    """
    # Inject the custom time, or default to the exact moment the script executes
    self.start_datetime = logical_start_time or datetime.now()

    super().__init__(initial_time=initial_time, factor=factor, strict=strict)

    self._go_live_sim_time: Optional[float] = None
    if go_live_at is not None:
        self._go_live_sim_time = self._resolve_go_live_time(
            go_live_at, float(initial_time)
        )

    self.setup_registry()

Registry & Parameters

SimulationRegistry(env)

A centralized 'Switchboard' 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 on the fly, seamlessly synchronizing them with the underlying SimPy processes.

Attributes:

Name	Type	Description
env	Environment	The active SimPy environment.

Source code in src/dynamic_des/core/registry.py

def __init__(self, env: Environment):
    self.env = env
    self._values: Dict[str, DynamicValue] = {}
    self._configs: Dict[str, Any] = {}

Functions

get(path)

Retrieve the DynamicValue object at a specific path.

Parameters:

Name	Type	Description	Default
path	str	The dot-notation path (e.g., 'Line_A.arrival.standard.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/dynamic_des/core/registry.py

def get(self, path: str) -> DynamicValue:
    """
    Retrieve the `DynamicValue` object at a specific path.

    Args:
        path (str): The dot-notation path (e.g., 'Line_A.arrival.standard.rate').

    Returns:
        DynamicValue: The wrapper object for the requested parameter.

    Raises:
        KeyError: If the path does not exist in the registry.
    """
    if path not in self._values:
        raise KeyError(f"Path '{path}' not found in Simulation Registry.")
    return self._values[path]

get_config(path)

Retrieve the live configuration object (e.g., DistributionConfig).

Parameters:

Name	Type	Description	Default
path	str	The dot-notation path (e.g., 'Line_A.service.milling').	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/dynamic_des/core/registry.py

def get_config(self, path: str) -> Any:
    """
    Retrieve the live configuration object (e.g., `DistributionConfig`).

    Args:
        path (str): The dot-notation path (e.g., 'Line_A.service.milling').

    Returns:
        Any: The synchronized configuration object.

    Raises:
        KeyError: If the config path does not exist in the registry.
    """
    if path not in self._configs:
        raise KeyError(f"Config path '{path}' not found in Simulation Registry.")
    return self._configs[path]

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/dynamic_des/core/registry.py

def register_sim_parameter(self, param: SimParameter):
    """
    Takes a `SimParameter` instance and flattens it into the registry.

    Args:
        param (SimParameter): The initial state schema to register.
    """
    prefix = param.sim_id

    # Register Arrivals
    for name, dist_config in param.arrival.items():
        path = f"{prefix}.arrival.{name}"
        self._configs[path] = dist_config
        self._register_dist(path, dist_config)

    # Register Service Steps
    for name, dist_config in param.service.items():
        path = f"{prefix}.service.{name}"
        self._configs[path] = dist_config
        self._register_dist(path, dist_config)

    # Register Resources
    for name, cap_config in param.resources.items():
        path = f"{prefix}.resources.{name}"
        self._configs[path] = cap_config
        self._register_cap(path, cap_config)

    # Register Containers
    for name, cap_config in param.containers.items():
        path = f"{prefix}.containers.{name}"
        self._configs[path] = cap_config
        self._register_cap(path, cap_config)

    # Register Stores
    for name, cap_config in param.stores.items():
        path = f"{prefix}.stores.{name}"
        self._configs[path] = cap_config
        self._register_cap(path, cap_config)

    # Register Custom Variables
    self._configs[f"{prefix}.variables"] = param.variables
    for name, value in param.variables.items():
        path = f"{prefix}.variables.{name}"
        self._values[path] = DynamicValue(self.env, path, value)

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

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/dynamic_des/core/registry.py

def update(self, path: str, new_value: Any):
    """
    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).

    Args:
        path (str): The dot-notation path to update.
        new_value (Any): The new value to apply.
    """
    if path in self._values:
        current_val = self._values[path].value
        validated_value = new_value

        # Type Validation & Casting
        if current_val is not None:
            expected_type = type(current_val)
            # If types don't match, attempt a safe cast (e.g., "5" -> 5)
            if not isinstance(new_value, expected_type):
                try:
                    validated_value = expected_type(new_value)
                    logger.debug(
                        f"Cast '{new_value}' to {expected_type.__name__} for '{path}'"
                    )
                except (ValueError, TypeError):
                    logger.error(
                        f"Type mismatch for '{path}'. Expected {expected_type.__name__}, "
                        f"got {type(new_value).__name__} with value '{new_value}'. Update ignored."
                    )
                    return  # Exit early to prevent crashing the simulation

        # Update the DynamicValue (triggers SimPy events)
        self._values[path].update(validated_value)

        # Synchronize the attribute on the original config object
        if "." in path:
            parent_path, attr = path.rsplit(".", 1)
            if parent_path in self._configs:
                parent_obj = self._configs[parent_path]
                # Check if the parent is a dictionary (like `variables`)
                if isinstance(parent_obj, dict):
                    parent_obj[attr] = validated_value
                else:
                    setattr(parent_obj, attr, validated_value)
    else:
        logger.warning(f"Attempted to update non-existent path: {path}")

SimParameter(sim_id, arrival=dict(), service=dict(), resources=dict(), containers=dict(), stores=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., 'Line_A.arrival.standard.rate').

Attributes:

Name	Type	Description
sim_id	str	The unique prefix for this group of parameters (e.g., 'Line_A').
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.
stores	Dict[str, CapacityConfig]	Configurations for discrete SimPy Stores.
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, max_cap) dataclass

Configuration for the capacity of a simulated resource, container, or store.

Attributes:

Name	Type	Description
current_cap	Union[int, float]	The currently active capacity.
max_cap	Union[int, float]	The absolute physical maximum capacity.

DistributionConfig(dist, rate=None, mean=None, std=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.

Attributes:

Name	Type	Description
dist	str	The type of distribution (e.g., 'exponential', 'normal', 'uniform').
rate	Optional[float]	The rate parameter (lambda) for exponential distributions.
mean	Optional[float]	The mean (mu) for normal distributions.
std	Optional[float]	The standard deviation (sigma) for normal distributions.

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., 'Line_A.lathe.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 parts) 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	A unique identifier/partition key for the event (e.g., 'task-001').
value	Dict[str, Any]	A dictionary containing the event's detailed payload.

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']	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)

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.

Attributes:

Name	Type	Description
rng	Generator	The NumPy random number generator instance.

Initializes the Sampler.

Parameters:

Name	Type	Description	Default
rng	Generator	A seeded NumPy random generator for reproducible runs. If None, a default unseeded generator is used.	None

Source code in src/dynamic_des/core/sampler.py

def __init__(self, rng: Optional[np.random.Generator] = None):
    """
    Initializes the Sampler.

    Args:
        rng (np.random.Generator, optional): A seeded NumPy random generator for reproducible runs.
            If None, a default unseeded generator is used.
    """
    self.rng = rng

Functions

sample(config, min_delay=1e-05)

Draws a random sample using the current parameters in the provided config.

Parameters:

Name	Type	Description	Default
config	DistributionConfig	The live configuration object retrieved from the Registry.	required

Returns:

Name	Type	Description
float	float	A sampled float representing a time duration (e.g., arrival gap or service time). Returns 0.0 if the resulting sample is negative.

Raises:

Type	Description
ValueError	If the dist type string is not supported.

Source code in src/dynamic_des/core/sampler.py

def sample(self, config: DistributionConfig, min_delay: float = 0.00001) -> float:
    """
    Draws a random sample using the current parameters in the provided config.

    Args:
        config (DistributionConfig): The live configuration object retrieved from the Registry.

    Returns:
        float: A sampled float representing a time duration (e.g., arrival gap or service time).
               Returns 0.0 if the resulting sample is negative.

    Raises:
        ValueError: If the `dist` type string is not supported.
    """

    if config.dist == "exponential":
        scale = 1.0 / config.rate if config.rate and config.rate > 0 else 1.0
        if self.rng is None:
            return max(min_delay, scale)
        return max(min_delay, self.rng.exponential(scale))

    if config.dist == "normal":
        m = config.mean or 0.0
        if self.rng is None:
            return max(min_delay, m)
        val = self.rng.normal(m, config.std or 0.0)
        return max(min_delay, val)

    if config.dist == "lognormal":
        m, s = config.mean or 1.0, config.std or 0.1
        if self.rng is None:
            return max(min_delay, m)

        # mu/sigma conversion
        mu = np.log(m**2 / np.sqrt(s**2 + m**2))
        sigma = np.sqrt(np.log(1 + (s**2 / m**2)))
        return max(min_delay, self.rng.lognormal(mu, sigma))

    return min_delay

Resources

DynamicResource(env, sim_id, resource_id)

Bases: BaseDynamicResource

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., 'Line_A.lathe.current_cap'). When the registry updates the capacity, the resource automatically adjusts, triggering pending requests if capacity increases.

Attributes:

Name	Type	Description
env	DynamicRealtimeEnvironment	The active simulation environment.
sim_id	str	The prefix ID (e.g., 'Line_A').
resource_id	str	The specific resource name (e.g., 'lathe').

Initializes the DynamicResource 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

Source code in src/dynamic_des/resources/resource.py

def __init__(self, env: Environment, sim_id: str, resource_id: str):
    """
    Initializes the DynamicResource and binds it to the registry.

    Args:
        env (Environment): The SimPy environment (must include the RegistryMixIn).
        sim_id (str): The parent simulation ID.
        resource_id (str): The name of the resource.
    """
    super().__init__(env, sim_id, resource_id, "resources")

    max_cap = int(self._max_cap_val.value)
    # Prevent initialization out-of-bounds
    self._capacity = max(0, min(int(self._current_cap_val.value), max_cap))

    self.queue = PriorityStore(env)
    self.pool = Container(env, init=self._capacity, capacity=max_cap)

    self._request_event = self.env.event()
    self.env.process(self._dispatcher())

Attributes

in_use property

Currently occupied slots (Total Capacity - Idle Tokens).

capacity property

The current active capacity of the resource.

This value reflects the live state from the Registry and may change during the simulation.

Functions

request(priority=1)

Returns a context-manager enabled request.

Source code in src/dynamic_des/resources/resource.py

def request(self, priority: int = 1):
    """Returns a context-manager enabled request."""
    return DynamicResourceRequest(self, priority)

release()

Return a token to the pool.

Source code in src/dynamic_des/resources/resource.py

def release(self):
    """Return a token to the pool."""
    return self.pool.put(1)

DynamicContainer(env, sim_id, container_id, init=0)

Bases: BaseDynamicResource

A wrapper for SimPy Container with dynamic capacity updates.

Useful for modeling continuous bulk materials (fluids, gases, battery charge). Unlike a standard simpy.Container, the capacity of this container listens to a SimulationRegistry path and can be mutated at runtime.

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 initial amount of material in the container at simulation start. Defaults to 0.	0

Source code in src/dynamic_des/resources/container.py

def __init__(
    self, env: Environment, sim_id: str, container_id: str, init: float = 0
):
    """
    Initializes the DynamicContainer and binds its capacity to the registry.

    Args:
        env (Environment): The SimPy environment (must include Registry interface).
        sim_id (str): The parent simulation ID.
        container_id (str): The unique name of this container.
        init (float, optional): The initial amount of material in the container
            at simulation start. Defaults to 0.
    """
    super().__init__(env, sim_id, container_id, "containers")

    max_cap = float(self._max_cap_val.value)
    # Prevent initialization out-of-bounds
    initial_capacity = max(0.0, min(float(self._current_cap_val.value), max_cap))

    self._container = Container(env, init=init, capacity=initial_capacity)

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.

Functions

put(amount)

Request to put a specific amount of material into the container.

Parameters:

Name	Type	Description	Default
amount	float	The amount of material to add.	required

Returns:

Type	Description
Event	simpy.events.Event: A SimPy event that triggers when capacity is available.

Source code in src/dynamic_des/resources/container.py

def put(self, amount: float) -> Event:
    """
    Request to put a specific amount of material into the container.

    Args:
        amount (float): The amount of material to add.

    Returns:
        simpy.events.Event: A SimPy event that triggers when capacity is available.
    """
    return self._container.put(amount)

get(amount)

Request to get a specific amount of material from the container.

Parameters:

Name	Type	Description	Default
amount	float	The amount of material to withdraw.	required

Returns:

Type	Description
ContainerGet	simpy.resources.container.ContainerGet: A SimPy event that triggers when enough material is available.

Source code in src/dynamic_des/resources/container.py

def get(self, amount: float) -> ContainerGet:
    """
    Request to get a specific amount of material from the container.

    Args:
        amount (float): The amount of material to withdraw.

    Returns:
        simpy.resources.container.ContainerGet: A SimPy event that triggers
            when enough material is available.
    """
    return self._container.get(amount)

DynamicStore(env, sim_id, store_id, priority=False)

Bases: BaseDynamicResource

A wrapper for SimPy Store 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 simpy.PriorityStore instead of the standard FIFO simpy.Store. Defaults to False.	False

Source code in src/dynamic_des/resources/store.py

def __init__(
    self, env: Environment, sim_id: str, store_id: str, priority: bool = False
):
    """
    Initializes the DynamicStore and binds its capacity to the registry.

    Args:
        env (Environment): The SimPy environment (must include Registry interface).
        sim_id (str): The parent simulation ID.
        store_id (str): The unique name of this store.
        priority (bool): If True, uses `simpy.PriorityStore` instead of
            the standard FIFO `simpy.Store`. Defaults to False.
    """
    super().__init__(env, sim_id, store_id, "stores")

    max_cap = int(self._max_cap_val.value)
    # Prevent initialization out-of-bounds
    initial_capacity = max(0, min(int(self._current_cap_val.value), max_cap))

    # Dynamically select the underlying SimPy storage engine
    if priority:
        self._store = PriorityStore(env, capacity=initial_capacity)
    else:
        self._store = Store(env, capacity=initial_capacity)

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.

Functions

put(item)

Request to put a specific distinct item into the store.

Parameters:

Name	Type	Description	Default
item	Any	The Python object to place into the store.	required

Returns:

Type	Description
Event	simpy.events.Event: A SimPy event that triggers when a slot is available.

Source code in src/dynamic_des/resources/store.py

def put(self, item: Any) -> Event:
    """
    Request to put a specific distinct item into the store.

    Args:
        item (Any): The Python object to place into the store.

    Returns:
        simpy.events.Event: A SimPy event that triggers when a slot is available.
    """
    return self._store.put(item)

get()

Request to get the next available item from the store.

Returns:

Type	Description
StoreGet	simpy.resources.store.StoreGet: A SimPy event that yields the requested item object when available.

Source code in src/dynamic_des/resources/store.py

def get(self) -> StoreGet:
    """
    Request to get the next available item from the store.

    Returns:
        simpy.resources.store.StoreGet: A SimPy event that yields the
            requested item object when available.
    """
    return self._store.get()

Admin & 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/dynamic_des/connectors/admin/kafka.py

def __init__(self, bootstrap_servers: str, max_tasks: int = 100, **kwargs: Any):
    """
    Initialize the connector with broker settings and state limits.

    Args:
        bootstrap_servers: Kafka broker addresses.
        max_tasks: The maximum number of task records to keep per service
            in memory (rolling window).
        **kwargs: Additional arguments passed to Kafka clients (e.g., security settings).
    """
    self.bootstrap_servers = bootstrap_servers
    self.max_tasks = max_tasks
    self.kwargs = kwargs

    # Event State: sim_id -> service -> task_id -> {status: timestamp}
    self._state: DefaultDict[str, DefaultDict[str, Dict[str, Any]]] = defaultdict(
        lambda: defaultdict(dict)
    )
    # Telemetry State: path_id -> latest_value
    self._vitals: Dict[str, Any] = {}

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/dynamic_des/connectors/admin/kafka.py

def create_topics(self, topics_config: List[Dict[str, Any]]):
    """
    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.

    Args:
        topics_config: A list of topic configurations.
            Each dict should contain 'name', and optionally 'partitions'
            and 'replication'.
    """
    admin_client = KafkaAdminClient(
        bootstrap_servers=self.bootstrap_servers,
        client_id="sim_admin",
        **self.kwargs,
    )

    new_topics = [
        NewTopic(
            name=cfg["name"],
            num_partitions=cfg.get("partitions", 1),
            replication_factor=cfg.get("replication", 1),
        )
        for cfg in topics_config
    ]

    try:
        admin_client.create_topics(new_topics=new_topics, validate_only=False)
    except TopicAlreadyExistsError:
        pass
    finally:
        admin_client.close()

send_config(topic, path_id, 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.

Parameters:

Name	Type	Description	Default
topic	str	The Kafka topic the simulation is listening to for config.	required
path_id	str	The registry dot-notation path (e.g., 'Line_A.lathe.max_cap').	required
value	Any	The new value to be applied to the path.	required

Source code in src/dynamic_des/connectors/admin/kafka.py

async def send_config(self, topic: str, path_id: str, value: Any):
    """
    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.

    Args:
        topic: The Kafka topic the simulation is listening to for config.
        path_id: The registry dot-notation path (e.g., 'Line_A.lathe.max_cap').
        value: The new value to be applied to the path.
    """
    producer = AIOKafkaProducer(
        bootstrap_servers=self.bootstrap_servers, **self.kwargs
    )
    await producer.start()
    try:
        payload = json.dumps({"path_id": path_id, "value": value}).encode("utf-8")
        await producer.send_and_wait(topic, payload)
    finally:
        await producer.stop()

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/dynamic_des/connectors/admin/kafka.py

async def collect_data(self, topics: List[str], auto_offset_reset: str = "latest"):
    """
    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.

    Args:
        topics: List of topics to consume from (telemetry and events).
        auto_offset_reset: Where to start consuming if no offset is committed.
    """
    consumer = AIOKafkaConsumer(
        *topics,
        bootstrap_servers=self.bootstrap_servers,
        auto_offset_reset=auto_offset_reset,
        **self.kwargs,
    )
    await consumer.start()
    try:
        async for msg in consumer:
            data = json.loads(msg.value.decode("utf-8"))
            self._process_message(data)
    finally:
        await consumer.stop()

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}.

Source code in src/dynamic_des/connectors/admin/kafka.py

def get_state(self) -> Dict[str, Any]:
    """
    Returns the aggregated event state for task lifecycles.

    Returns:
        A nested dictionary: sim_id -> service -> task_id -> {status: timestamp}.
    """
    return self._state

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

Source code in src/dynamic_des/connectors/admin/kafka.py

def get_vitals(self) -> Dict[str, Any]:
    """
    Returns the latest system telemetry metrics.

    Returns:
        A dictionary of path_id to latest value (e.g., utilization, queue length).
    """
    return self._vitals

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/dynamic_des/connectors/ingress/base.py

async def run(self, ingress_queue: queue.Queue) -> None:
    """
    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.

    Args:
        ingress_queue: A thread-safe queue used to transmit updates
            to the main simulation environment.

    Raises:
        NotImplementedError: If the subclass does not override this method.
    """
    raise NotImplementedError("Subclasses must implement the run method.")

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_id' string and a 'value' of any serializable type.

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 JsonDeserializer.
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/dynamic_des/connectors/ingress/kafka.py

def __init__(
    self,
    topic: str,
    bootstrap_servers: str,
    topic_deserializers: Optional[Dict[str, MessageDeserializer]] = None,
    default_deserializer: Optional[MessageDeserializer] = None,
    **kwargs: Any,
):
    """
    Initializes the KafkaIngress with connection and subscription details.

    Args:
        topic: The Kafka topic to subscribe to.
        bootstrap_servers: Kafka broker address or list of addresses.
        topic_deserializers: Mapping of topics to their specific deserializers.
        default_deserializer: Fallback deserializer if the topic isn't explicitly mapped.
        **kwargs: Arbitrary keyword arguments passed to the underlying AIOKafkaConsumer.
    """
    self.topic = topic
    self.bootstrap_servers = bootstrap_servers

    self.topic_deserializers = topic_deserializers or {}
    self.default_deserializer = default_deserializer or JsonDeserializer()

    self.kwargs = kwargs

Functions

run(ingress_queue) async

Main execution loop that consumes Kafka messages, deserializes them, and populates the ingress queue.

Source code in src/dynamic_des/connectors/ingress/kafka.py

async def run(self, ingress_queue: queue.Queue) -> None:
    """
    Main execution loop that consumes Kafka messages, deserializes them,
    and populates the ingress queue.
    """
    backoff = 1.0
    while True:
        try:
            consumer = AIOKafkaConsumer(
                self.topic, bootstrap_servers=self.bootstrap_servers, **self.kwargs
            )
            await consumer.start()
            logger.info(f"Connected to Kafka Ingress topic: {self.topic}")
            backoff = 1.0

            try:
                async for msg in consumer:
                    try:
                        # Apply configured Deserializer strategy
                        deserializer = self.topic_deserializers.get(
                            msg.topic, self.default_deserializer
                        )

                        # Decode binary payload -> Python Dict
                        data = deserializer.deserialize(msg.topic, msg.value)

                        ingress_queue.put((data["path_id"], data["value"]))

                    except Exception as e:
                        logger.warning(
                            f"Malformed or undecodable Kafka message: {e}"
                        )
            finally:
                await consumer.stop()

        except asyncio.CancelledError:
            logger.info("Kafka Ingress shut down requested. Exiting loop.")
            break
        except Exception as e:
            logger.error(f"Kafka Ingress error: {e}. Retrying in {backoff}s")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60.0)

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_id, 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/dynamic_des/connectors/ingress/local.py

def __init__(self, schedule: List[Tuple[float, str, Any]]):
    """
    Initializes the LocalIngress and sorts the provided schedule.

    Args:
        schedule: A list of tuples representing scheduled updates.
            Format: [(delay_from_start, "registry_path", new_value), ...].
            The list is automatically sorted by the delay time.
    """
    # Sort schedule by delay to ensure chronological processing
    self.schedule = sorted(schedule, key=lambda x: x[0])

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_id, value) updates to the SimulationRegistry.	required

Source code in src/dynamic_des/connectors/ingress/local.py

async def run(self, ingress_queue: queue.Queue) -> None:
    """
    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.

    Args:
        ingress_queue: A thread-safe queue used to transmit the
            (path_id, value) updates to the SimulationRegistry.
    """
    last_time = 0.0
    for delay, path_id, value in self.schedule:
        # Wait for the relative time difference
        await asyncio.sleep(delay - last_time)
        ingress_queue.put((path_id, value))
        last_time = delay

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.

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), param_path (text), param_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/dynamic_des/connectors/ingress/postgres.py

def __init__(
    self,
    connection_dsn: str,
    table_name: str = "simulation_params",
    poll_interval: float = 2.0,
    **kwargs: Any,
):
    """
    Initializes the PostgresIngress.

    Args:
        connection_dsn: PostgreSQL connection string (DSN).
        table_name: Table to poll for parameter updates. Must have columns:
                    id (serial), param_path (text), param_value (text/json), is_applied (boolean).
        poll_interval: Seconds to wait between database polls.
        **kwargs: Additional connection pool arguments for asyncpg.
    """
    self.dsn = connection_dsn
    self.table_name = table_name
    self.poll_interval = poll_interval
    self.kwargs = kwargs
    self.pool: asyncpg.Pool | None = None

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/dynamic_des/connectors/ingress/postgres.py

async def run(self, ingress_queue: queue.Queue) -> None:
    """
    Main execution loop fetching dynamic parameter updates from PostgreSQL.

    Args:
        ingress_queue: A thread-safe queue used to transmit (path, value)
            tuples to the SimulationRegistry.
    """
    await self._init_pool()

    # Phase 1: Load the most recent applied state to resume seamlessly
    try:
        assert self.pool is not None
        async with self.pool.acquire() as conn:
            # Get the latest applied value for each distinct param_path
            rows = await conn.fetch(f"""
                SELECT DISTINCT ON (param_path) param_path, param_value
                FROM {self.table_name}
                WHERE is_applied = TRUE
                ORDER BY param_path, id DESC
            """)
            for row in rows:
                path = row["param_path"]
                raw_val = row["param_value"]
                try:
                    val = json.loads(raw_val)
                except json.JSONDecodeError:
                    val = raw_val
                ingress_queue.put((path, val))
                logger.info(f"Loaded historical parameter state: {path} = {val}")
    except Exception as e:
        logger.error(f"Error loading historical state from {self.table_name}: {e}")

    # Phase 2: Continuously poll for new unapplied updates
    while True:
        try:
            assert self.pool is not None
            async with self.pool.acquire() as conn:
                # Fetch all unapplied parameter updates
                rows = await conn.fetch(f"""
                    SELECT id, param_path, param_value
                    FROM {self.table_name}
                    WHERE is_applied = FALSE
                    ORDER BY id ASC
                """)

                if rows:
                    applied_ids = []
                    for row in rows:
                        row_id = row["id"]
                        path = row["param_path"]
                        raw_val = row["param_value"]

                        try:
                            # Attempt to parse JSON to maintain types (e.g. ints, dicts)
                            val = json.loads(raw_val)
                        except json.JSONDecodeError:
                            # Fallback to raw string if it's not valid JSON
                            val = raw_val

                        # Send to the simulation registry queue
                        ingress_queue.put((path, val))
                        applied_ids.append(row_id)
                        logger.info(f"Ingested parameter update: {path} = {val}")

                    # Mark the rows as applied so we don't process them again
                    if applied_ids:
                        await conn.execute(
                            f"""
                            UPDATE {self.table_name}
                            SET is_applied = TRUE
                            WHERE id = ANY($1::int[])
                        """,
                            applied_ids,
                        )

        except Exception as e:
            logger.error(
                f"Error polling PostgresIngress table {self.table_name}: {e}"
            )

        # Wait before polling again
        await asyncio.sleep(self.poll_interval)

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.

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 redis.from_url.

Initializes the RedisIngress with connection and channel settings.

Parameters:

Name	Type	Description	Default
url	str	Redis connection URL (e.g., redis://localhost: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 redis.from_url.	{}

Source code in src/dynamic_des/connectors/ingress/redis.py

def __init__(
    self,
    url: str,
    channel_name: str = "params",
    poll_interval: float = 0.5,
    **kwargs: Any,
):
    """
    Initializes the RedisIngress with connection and channel settings.

    Args:
        url: Redis connection URL (e.g., redis://localhost:6379/0).
        channel_name: Redis Pub/Sub channel to subscribe to.
        poll_interval: Sleep interval when no messages are found.
        **kwargs: Additional configuration dictionary passed to `redis.from_url`.
    """
    self.url = url
    self.channel_name = channel_name
    self.poll_interval = poll_interval
    self.kwargs = kwargs
    self.client: redis.Redis | None = None
    self.pubsub: redis.client.PubSub | None = None

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 param_path and param_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/dynamic_des/connectors/ingress/redis.py

async def run(self, update_queue: queue.Queue) -> None:
    """
    The main execution loop for subscribing to Redis and receiving updates.

    Continuously polls the Pub/Sub channel for incoming JSON messages containing
    `param_path` and `param_value`, placing successfully parsed updates onto the
    internal update queue.

    Args:
        update_queue: A thread-safe queue where incoming parameters are pushed
            so they can be applied within the SimPy context.
    """
    self.client = redis.from_url(self.url, **self.kwargs)
    self.pubsub = self.client.pubsub()
    await self.pubsub.subscribe(self.channel_name)
    logger.info(f"RedisIngress subscribed to channel '{self.channel_name}'")

    try:
        while True:
            message = await self.pubsub.get_message(
                ignore_subscribe_messages=True, timeout=self.poll_interval
            )
            if message and message["type"] == "message":
                try:
                    data = orjson.loads(message["data"])
                    param_path = data.get("param_path")
                    param_value = data.get("param_value")
                    if param_path is not None and param_value is not None:
                        update_queue.put((param_path, param_value))
                except Exception as e:
                    logger.error(f"Failed to parse Redis Ingress message: {e}")
            else:
                await asyncio.sleep(self.poll_interval)
    except asyncio.CancelledError:
        pass
    finally:
        if self.pubsub:
            await self.pubsub.unsubscribe(self.channel_name)
        if self.client:
            await self.client.close()

Deserializers

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/dynamic_des/connectors/ingress/kafka.py

def deserialize(self, topic: str, payload: bytes) -> Any:
    """
    Deserializes a JSON byte string.

    Args:
        topic (str): The Kafka topic from which the message was consumed (unused).
        payload (bytes): The UTF-8 encoded JSON byte string.

    Returns:
        Any: The parsed JSON data as a Python dictionary.
    """
    return json.loads(payload.decode("utf-8"))

ConfluentAvroDeserializer(registry_url, **registry_kwargs)

Lazy-loaded deserializer for Confluent Schema Registry.

This class converts Avro-encoded byte strings back into Python dictionaries. It automatically fetches the appropriate schema from the registry using the schema ID embedded within the message's binary payload, so no schema_str is required during initialization.

Initializes the Confluent Avro deserializer.

Parameters:

Name	Type	Description	Default
registry_url	str	The base URL of the Confluent Schema Registry.	required
**registry_kwargs	Any	Additional configuration arguments passed directly to the SchemaRegistryClient (e.g., {'basic.auth.user.info': 'user:pass'}).	{}

Raises:

Type	Description
ImportError	If the 'confluent-kafka' package is not installed.

Source code in src/dynamic_des/connectors/ingress/kafka.py

def __init__(self, registry_url: str, **registry_kwargs: Any):
    """
    Initializes the Confluent Avro deserializer.

    Args:
        registry_url (str): The base URL of the Confluent Schema Registry.
        **registry_kwargs: Additional configuration arguments passed directly
            to the `SchemaRegistryClient` (e.g., `{'basic.auth.user.info': 'user:pass'}`).

    Raises:
        ImportError: If the 'confluent-kafka' package is not installed.
    """
    try:
        from confluent_kafka.schema_registry import SchemaRegistryClient
        from confluent_kafka.schema_registry.avro import AvroDeserializer
        from confluent_kafka.serialization import MessageField, SerializationContext
    except ImportError:
        raise ImportError(
            "The 'confluent-kafka' package is required to use ConfluentAvroDeserializer. "
            "Install it using: pip install dynamic-des[confluent]"
        )

    conf = {"url": registry_url, **registry_kwargs}
    client = SchemaRegistryClient(conf)

    # No schema string required! It reads the Schema ID from the message bytes.
    self.avro_deserializer = AvroDeserializer(client)
    self.SerializationContext = SerializationContext
    self.MessageField = MessageField

Functions

deserialize(topic, payload)

Deserializes an Avro-encoded byte string.

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.

Source code in src/dynamic_des/connectors/ingress/kafka.py

def deserialize(self, topic: str, payload: bytes) -> Any:
    """
    Deserializes an Avro-encoded byte string.

    Args:
        topic (str): The Kafka topic from which the message was consumed.
        payload (bytes): The Avro-encoded binary payload containing the schema ID.

    Returns:
        Any: The deserialized data as a Python dictionary.
    """
    ctx = self.SerializationContext(topic, self.MessageField.VALUE)
    return self.avro_deserializer(payload, ctx)

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 boto3.client initialization (e.g., region_name, aws_access_key_id).	{}

Raises:

Type	Description
ImportError	If the 'aws-glue-schema-registry' or 'boto3' packages are not installed.

Source code in src/dynamic_des/connectors/ingress/kafka.py

def __init__(self, registry_name: str, **boto3_kwargs: Any):
    """
    Initializes the AWS Glue Avro deserializer.

    Args:
        registry_name (str): The name of the AWS Glue Schema Registry.
        **boto3_kwargs: Additional arguments passed directly to the `boto3.client`
            initialization (e.g., `region_name`, `aws_access_key_id`).

    Raises:
        ImportError: If the 'aws-glue-schema-registry' or 'boto3' packages are not installed.
    """
    try:
        import boto3
        from aws_schema_registry import SchemaRegistryClient
        from aws_schema_registry.adapter.kafka import KafkaDeserializer
    except ImportError:
        raise ImportError(
            "The 'aws-glue-schema-registry' and 'boto3' packages are required. "
            "Install them using: pip install dynamic-des[glue]"
        )

    if "region_name" not in boto3_kwargs:
        boto3_kwargs["region_name"] = "us-east-1"

    glue_client = boto3.client("glue", **boto3_kwargs)
    registry_client = SchemaRegistryClient(glue_client, registry_name=registry_name)

    self.deserializer = KafkaDeserializer(registry_client)

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/dynamic_des/connectors/ingress/kafka.py

def deserialize(self, topic: str, payload: bytes) -> Any:
    """
    Deserializes an Avro-encoded byte string using AWS Glue.

    Args:
        topic (str): The Kafka topic from which the message was consumed.
        payload (bytes): The Avro-encoded binary payload.

    Returns:
        Any: The raw data dictionary extracted from the AWS DataAndSchema object.
    """
    # AWS returns a DataAndSchema object, we just want the raw data dictionary
    result = self.deserializer.deserialize(topic, payload)
    return result.data

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/dynamic_des/connectors/egress/base.py

async def run(self, egress_queue: queue.Queue) -> None:
    """
    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.

    Args:
        egress_queue: A thread-safe queue containing batches of
            dictionaries to be exported.

    Raises:
        NotImplementedError: If the subclass does not override this method.
    """
    raise NotImplementedError("Subclasses must implement the run method.")

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

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 topic_serializers.
include_stream_type	bool	Whether the published message keeps the stream_type field.
producer_config	dict	Configuration dictionary passed to AIOKafkaProducer.

Examples:

Defining a custom topic router to split machine learning events from standard telemetry:

def custom_topic_router(data: dict) -> str:
    stream_type = data.get("stream_type")
    if stream_type == "telemetry":
        return "sim-telemetry"

    value = data.get("value", {})
    if isinstance(value, dict):
        event_type = value.get("event_type")
        if event_type == "prediction_request":
            return "mill-predictions"
        elif event_type == "ground_truth":
            return "mill-groundtruth"

    return "mill-lifecycle"

# Pass it to the egress connector
egress = KafkaEgress(
    bootstrap_servers="localhost:9092",
    topic_router=custom_topic_router
)

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 MessageSerializer implementations.	None
default_serializer	Optional[MessageSerializer]	The fallback serializer to use if a topic lacks a specific mapping. Defaults to JsonSerializer.	None
include_stream_type	bool	Keep stream_type in the published message. Defaults to False, which is the wire format every existing deployment already reads. Turn it on when the same consumer code has to work against both a Kafka topic and Parquet files, since the Parquet writer keeps the field. An Avro serializer only carries the field if the registered schema declares it: both ConfluentAvroSerializer and GlueAvroSerializer encode field by field from the schema, so an undeclared stream_type is dropped without an error rather than failing the send.	False
**kwargs	Any	Additional overrides for the AIOKafkaProducer configuration.	{}

Source code in src/dynamic_des/connectors/egress/kafka.py

def __init__(
    self,
    bootstrap_servers: str,
    telemetry_topic: str = "sim-telemetry",
    event_topic: str = "sim-events",
    topic_router: Optional[Callable[[dict], str]] = None,
    topic_serializers: Optional[Dict[str, MessageSerializer]] = None,
    default_serializer: Optional[MessageSerializer] = None,
    include_stream_type: bool = False,
    **kwargs: Any,
):
    """
    Initializes the KafkaEgress with topic and connection settings.

    Args:
        bootstrap_servers: Comma-separated list of Kafka brokers.
        telemetry_topic: Destination topic for telemetry stream.
        event_topic: Destination topic for event stream.
        topic_router: Optional callable to dynamically route payloads to specific topics.
        topic_serializers: Mapping of target topics to their specific `MessageSerializer` implementations.
        default_serializer: The fallback serializer to use if a topic lacks a specific mapping. Defaults to `JsonSerializer`.
        include_stream_type: Keep `stream_type` in the published message. Defaults to
            False, which is the wire format every existing deployment already reads.
            Turn it on when the same consumer code has to work against both a Kafka
            topic and Parquet files, since the Parquet writer keeps the field.
            An Avro serializer only carries the field if the registered schema
            declares it: both `ConfluentAvroSerializer` and `GlueAvroSerializer`
            encode field by field from the schema, so an undeclared `stream_type`
            is dropped without an error rather than failing the send.
        **kwargs: Additional overrides for the AIOKafkaProducer configuration.
    """
    self.telemetry_topic = telemetry_topic
    self.event_topic = event_topic
    self.topic_router = topic_router
    self.include_stream_type = include_stream_type

    # Initialize serializers in a backward-compatible way
    self.topic_serializers = topic_serializers or {}
    self.default_serializer = default_serializer or JsonSerializer()

    # High-performance defaults for 100k/sec
    self.producer_config = {
        "bootstrap_servers": bootstrap_servers,
        "linger_ms": 10,  # Batch messages for 10ms before sending
        "compression_type": "lz4",  # Fast compression for high volume
        "max_batch_size": 131072,  # 128KB batch size
        **kwargs,
    }

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/dynamic_des/connectors/egress/kafka.py

async def run(self, egress_queue: queue.Queue) -> None:
    """
    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.

    Args:
        egress_queue: A thread-safe queue containing lists of dictionaries
            or Pydantic models generated by the EgressMixIn.

    Note:
        The loop exits gracefully upon receiving an `asyncio.CancelledError`,
        ensuring the Kafka producer is stopped correctly.
    """
    backoff = 1.0
    max_backoff = 60.0

    while True:
        try:
            producer = AIOKafkaProducer(**self.producer_config)
            await producer.start()
            logger.info("Kafka Egress producer connected successfully.")
            backoff = 1.0  # Reset backoff on successful connection

            try:
                while True:
                    try:
                        # 'batch' is a list of dictionaries from the EgressMixIn
                        batch = egress_queue.get_nowait()

                        for data in batch:
                            if self.topic_router:
                                # Use externalized logic if provided
                                topic = self.topic_router(data)
                                stream = data.get("stream_type", "event")
                            else:
                                # Fallback to standard dynamic-des routing
                                stream = data["stream_type"]
                                topic = (
                                    self.telemetry_topic
                                    if stream == "telemetry"
                                    else self.event_topic
                                )

                            # Extract the Kafka Key
                            key = None
                            if stream == "telemetry":
                                key = (
                                    str(data["path_id"]).encode()
                                    if "path_id" in data
                                    else None
                                )
                            else:
                                key = (
                                    str(data["key"]).encode()
                                    if "key" in data
                                    else None
                                )

                            # Build the payload without touching the record itself.
                            # Every egress provider receives the same dictionary
                            # objects, so removing a key here would also remove it
                            # from what a Parquet or Redis sink writes.
                            if self.include_stream_type:
                                payload = data
                            else:
                                payload = {
                                    k: v
                                    for k, v in data.items()
                                    if k != "stream_type"
                                }

                            # Apply configured Serializer strategy
                            serializer = self.topic_serializers.get(
                                topic, self.default_serializer
                            )
                            payload_bytes = serializer.serialize(topic, payload)

                            await producer.send(topic, value=payload_bytes, key=key)

                    except queue.Empty:
                        # Yield to loop if queue is empty
                        await asyncio.sleep(0.001)
            finally:
                await producer.stop()

        except asyncio.CancelledError:
            logger.info("Kafka Egress shut down requested. Exiting loop.")
            break
        except Exception as e:
            logger.error(
                f"Kafka Egress connection failed: {e}. Retrying in {backoff} seconds..."
            )
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, max_backoff)

ConsoleEgress

Bases: BaseEgress

Local egress provider that outputs simulation data to the standard console.

This connector serves as a debugging and local development tool. It processes batched data from the simulation and prints formatted entries to stdout, distinguishing between telemetry metrics and discrete events via prefixes.

Functions

run(egress_queue) async

Continuously polls the egress queue and prints formatted results.

This method implements an asynchronous polling loop. It extracts batches from the internal queue, identifies the data stream type, and formats the output with specific prefixes: - [TEL]: Telemetry data (e.g., resource utilization, queue lengths). - [EVT]: Event data (e.g., task lifecycle transitions).

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/dynamic_des/connectors/egress/local.py

async def run(self, egress_queue: queue.Queue) -> None:
    """
    Continuously polls the egress queue and prints formatted results.

    This method implements an asynchronous polling loop. It extracts batches
    from the internal queue, identifies the data stream type, and formats the
    output with specific prefixes:
    - [TEL]: Telemetry data (e.g., resource utilization, queue lengths).
    - [EVT]: Event data (e.g., task lifecycle transitions).

    Args:
        egress_queue: A thread-safe queue containing lists of dictionaries
            (batches) generated by the EgressMixIn.

    Note:
        If the queue is empty, the method yields control to the asyncio
        event loop for a short duration to prevent CPU pinning.
    """
    while True:
        try:
            # Receive a batch (list) of messages
            batch = egress_queue.get_nowait()

            for data in batch:
                # Identify the stream type for the prefix
                stream = data.get("stream_type", "unknown")
                prefix = "[TEL]" if stream == "telemetry" else "[EVT]"

                # Log the remaining data (path_id/key, value, timestamp) without
                # altering the record. Every egress provider is handed the same
                # dictionary objects, so removing a key here would also remove it
                # from what another attached sink writes.
                rest = {k: v for k, v in data.items() if k != "stream_type"}
                logger.info(f"{prefix} {rest}")

        except queue.Empty:
            # Yield to the event loop
            await asyncio.sleep(0.1)

PostgresEgress(connection_dsn, table_name='simulation_data', **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.

Initializes the PostgresEgress with connection details.

Parameters:

Name	Type	Description	Default
connection_dsn	str	PostgreSQL connection string (DSN).	required
table_name	str	Target table for simulation records.	'simulation_data'
**kwargs	Any	Additional connection pool arguments for asyncpg.	{}

Source code in src/dynamic_des/connectors/egress/postgres.py

def __init__(
    self, connection_dsn: str, table_name: str = "simulation_data", **kwargs: Any
):
    """
    Initializes the PostgresEgress with connection details.

    Args:
        connection_dsn: PostgreSQL connection string (DSN).
        table_name: Target table for simulation records.
        **kwargs: Additional connection pool arguments for asyncpg.
    """
    self.dsn = connection_dsn
    self.table_name = table_name
    self.kwargs = kwargs
    self.pool: asyncpg.Pool | None = None
    self.valid_columns: set[str] = set()

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

Source code in src/dynamic_des/connectors/egress/postgres.py

async def run(self, egress_queue: queue.Queue) -> None:
    """
    Main execution loop for PostgreSQL data persistence.

    Args:
        egress_queue: A thread-safe queue containing batches of simulation data.
    """
    await self._init_pool()

    while True:
        try:
            # Receive a batch (list) of dictionaries
            batch = egress_queue.get_nowait()
            if not batch:
                continue

            # Support multi-table multiplexing using a __table__ key
            clean_batch = []
            for item in batch:
                # `item` is an EventPayload or TelemetryPayload dict.
                # The actual user data is in `item["value"]`.
                payload = item.get("value")

                if not isinstance(payload, dict):
                    # PostgresEgress requires dictionaries (or Pydantic models dumped to dicts)
                    continue

                # Merge the simulation metadata into the payload so it CAN be inserted if the user configured their schema to accept it
                payload_copy = payload.copy()
                payload_copy["sim_ts"] = item.get("sim_ts")
                payload_copy["timestamp"] = item.get("timestamp")

                if "key" in item:
                    payload_copy["event_id"] = item["key"]
                if "path_id" in item:
                    payload_copy["path_id"] = item["path_id"]

                # Determine target table, defaulting to self.table_name if not specified
                target = payload_copy.get("__table__", self.table_name)
                if target == self.table_name:
                    payload_copy.pop("__table__", None)
                    clean_batch.append(payload_copy)

            if not clean_batch:
                continue

            # Filter keys to only those that exist in the database table
            keys = [k for k in clean_batch[0].keys() if k in self.valid_columns]

            if not keys:
                logger.warning(
                    f"No matching columns for table {self.table_name}. Skipping."
                )
                continue

            columns = ", ".join(keys)
            placeholders = ", ".join(f"${i + 1}" for i in range(len(keys)))

            # Using ON CONFLICT DO NOTHING to ensure idempotency during simulation restarts
            query = f"INSERT INTO {self.table_name} ({columns}) VALUES ({placeholders}) ON CONFLICT DO NOTHING"

            # Extract the tuples of values matching the ordered keys
            values = [tuple(data.get(k) for k in keys) for data in clean_batch]

            # Execute the massive batch insert dynamically
            if self.valid_columns and values:
                assert self.pool is not None
                async with self.pool.acquire() as conn:
                    await conn.executemany(query, values)

            logger.debug(f"Inserted {len(values)} records into {self.table_name}")

        except queue.Empty:
            # Yield to the event loop if the queue is empty
            await asyncio.sleep(0.1)
        except Exception as e:
            logger.error(f"Error inserting batch into {self.table_name}: {e}")
            # Back-off on database failures to prevent rapid crash looping
            await asyncio.sleep(1)

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://localhost: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 redis.from_url.

Initializes the RedisEgress with topic and connection settings.

Parameters:

Name	Type	Description	Default
url	str	Redis connection URL (e.g., redis://localhost:6379/0).	required
stream_name	str	Target Redis Stream name.	'simulation_data'
**kwargs	Any	Additional arguments for redis.from_url.	{}

Source code in src/dynamic_des/connectors/egress/redis.py

def __init__(self, url: str, stream_name: str = "simulation_data", **kwargs: Any):
    """
    Initializes the RedisEgress with topic and connection settings.

    Args:
        url: Redis connection URL (e.g., redis://localhost:6379/0).
        stream_name: Target Redis Stream name.
        **kwargs: Additional arguments for `redis.from_url`.
    """
    self.url = url
    self.stream_name = stream_name
    self.kwargs = kwargs
    self.client: redis.Redis | None = None

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/dynamic_des/connectors/egress/redis.py

async def run(self, egress_queue: queue.Queue) -> None:
    """
    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.

    Args:
        egress_queue: A thread-safe queue containing lists of dictionaries
            or Pydantic models generated by the EgressMixIn.
    """
    await self._init_client()
    assert self.client is not None

    while True:
        try:
            batch = egress_queue.get_nowait()
            if not batch:
                continue

            pipe = self.client.pipeline()
            count = 0
            for item in batch:
                # Multi-stream multiplexing. `item` is an EventPayload or
                # TelemetryPayload dict, and the caller's own data sits under
                # `value`, so `__stream__` is looked for there first. Reading only
                # the top level found nothing, because publish_event never puts it
                # there, so every record went to the default stream and the
                # routing silently did nothing. PostgresEgress unwraps the same
                # way for `__table__`, which is why table routing always worked.
                value = item.get("value")
                nested = value if isinstance(value, dict) else {}
                target = nested.get(
                    "__stream__", item.get("__stream__", self.stream_name)
                )

                item_copy = item.copy()
                item_copy.pop("__stream__", None)
                if "__stream__" in nested:
                    item_copy["value"] = {
                        k: v for k, v in nested.items() if k != "__stream__"
                    }

                # Redis Streams expect a dict of string fields.
                # We serialize the entire item into a single 'payload' field.
                payload = orjson.dumps(item_copy).decode("utf-8")
                pipe.xadd(target, {"payload": payload})
                count += 1

            if count > 0:
                await pipe.execute()
            logger.debug(f"Inserted {count} records into Redis Streams")

        except queue.Empty:
            await asyncio.sleep(0.1)
        except Exception as e:
            logger.error(f"Error publishing batch to Redis: {e}")
            await asyncio.sleep(1)

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 utilizes PyArrow's Virtual File System (VFS) to seamlessly write data to local disks, AWS S3, Google Cloud Storage, or Azure Blob. It implements an enterprise-grade chunking strategy, generating a uniquely named file for every batch to prevent file locking and ensure crash resilience.

By default, all data is appended to the default_path. 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.

Attributes:

Name	Type	Description
default_path	Optional[str]	The fallback destination path if no router is provided.
filesystem	Optional[Any]	A PyArrow FileSystem instance. Defaults to local disk.
path_router	Optional[Callable]	Optional logic to dynamically route or drop payloads.

Examples:

Routing errors to a specific file and dropping meaningless metrics:

def custom_log_router(data: dict) -> str | None:
    # Drop lag metrics
    if data.get("path_id") == "system.simulation.lag_seconds":
        return None

    # Route errors to a separate JSONL file
    if data.get("status") == "error":
        return "data/error_logs.jsonl"

    return "data/standard_logs.jsonl"

egress = JsonlStorageEgress(path_router=custom_log_router)

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").	None
filesystem	Optional[Any]	An instantiated PyArrow FileSystem (e.g., fs.S3FileSystem()). If None, defaults to fs.LocalFileSystem().	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/dynamic_des/connectors/egress/storage.py

def __init__(
    self,
    default_path: Optional[str] = None,
    filesystem: Optional[Any] = None,
    path_router: Optional[Callable[[dict], Optional[str]]] = None,
):
    """
    Initializes the JsonlStorageEgress with routing and VFS settings.

    Args:
        default_path: The target file path prefix (e.g., "data/logs.jsonl").
        filesystem: An instantiated PyArrow FileSystem (e.g., `fs.S3FileSystem()`).
            If None, defaults to `fs.LocalFileSystem()`.
        path_router: A function taking a dict payload and returning a string path,
            or None to drop the record.
    """
    self.default_path = default_path
    self.filesystem = filesystem
    self.path_router = path_router

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/dynamic_des/connectors/egress/storage.py

async def run(self, egress_queue: queue.Queue) -> None:
    """
    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.

    Args:
        egress_queue: A thread-safe queue containing batches of dictionaries
            generated by the environment's EgressMixIn.

    Raises:
        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.
    """
    try:
        from pyarrow import fs
    except ImportError:
        raise ImportError("pyarrow is required. pip install dynamic-des[parquet]")

    self.filesystem = self.filesystem or fs.LocalFileSystem()
    batches_processed = 0

    try:
        while True:
            try:
                batch = egress_queue.get_nowait()
                self.active_tasks = getattr(self, "active_tasks", 0) + 1
                try:
                    await asyncio.to_thread(self._write_batch, batch)
                finally:
                    self.active_tasks -= 1
                # Log a progress heartbeat every 10 batches
                batches_processed += 1
                if batches_processed % 10 == 0:
                    q_size = egress_queue.qsize()
                    logger.info(
                        f"Parquet Writer: Processed {batches_processed} batches. "
                        f"(~{q_size} batches waiting in queue)"
                    )
            except queue.Empty:
                await asyncio.sleep(0.1)
    except asyncio.CancelledError:
        # This triggers during env.teardown()
        logger.info(
            f"JsonlStorageEgress shut down requested. "
            f"Successfully wrote {batches_processed} total chunks to storage."
        )

ParquetStorageEgress(default_path=None, filesystem=None, path_router=None)

Bases: BaseEgress

High-performance columnar Parquet writer for Data Lake ingestion patterns.

This connector buffers simulation records and converts them into heavily compressed Parquet tables using pyarrow. It natively supports writing to local disks or object stores (AWS S3, GCS, SeaweedFS) 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. Schemas are inferred from the first batch of data and strictly enforced on subsequent batches to prevent schema drift across chunks.

Attributes:

Name	Type	Description
default_path	Optional[str]	The fallback destination path.
filesystem	Optional[Any]	A PyArrow FileSystem instance. Defaults to local.
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:

from pyarrow import fs

def datalake_router(data: dict) -> str | None:
    if data.get("stream_type") == "telemetry":
        return "simulation_data/telemetry.parquet"
    return "simulation_data/events.parquet"

# Connect directly to SeaweedFS or AWS S3
s3 = fs.S3FileSystem(endpoint_override="localhost:8333", scheme="http")

egress = ParquetStorageEgress(
    filesystem=s3,
    path_router=datalake_router
)

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").	None
filesystem	Optional[Any]	An instantiated PyArrow FileSystem (e.g., fs.S3FileSystem()). If None, defaults to fs.LocalFileSystem().	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/dynamic_des/connectors/egress/storage.py

def __init__(
    self,
    default_path: Optional[str] = None,
    filesystem: Optional[Any] = None,
    path_router: Optional[Callable[[dict], Optional[str]]] = None,
):
    """
    Initializes the ParquetStorageEgress with routing and VFS settings.

    Args:
        default_path: The target file path prefix (e.g., "data/events.parquet").
        filesystem: An instantiated PyArrow FileSystem (e.g., `fs.S3FileSystem()`).
            If None, defaults to `fs.LocalFileSystem()`.
        path_router: A function taking a dict payload and returning a string path,
            or None to drop the record.
    """
    self.default_path = default_path
    self.filesystem = filesystem
    self.path_router = path_router
    self.schemas: Dict[str, Any] = {}

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/dynamic_des/connectors/egress/storage.py

async def run(self, egress_queue: queue.Queue) -> None:
    """
    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.

    Args:
        egress_queue: A thread-safe queue containing batches of dictionaries
            generated by the environment's EgressMixIn.

    Raises:
        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.
    """
    try:
        import pyarrow as pa
        import pyarrow.parquet as pq
        from pyarrow import fs
    except ImportError:
        raise ImportError("pyarrow is required. pip install dynamic-des[parquet]")

    self.filesystem = self.filesystem or fs.LocalFileSystem()
    batches_processed = 0

    try:
        while True:
            try:
                batch = egress_queue.get_nowait()
                self.active_tasks = getattr(self, "active_tasks", 0) + 1
                try:
                    await asyncio.to_thread(self._write_batch, batch, pa, pq)
                finally:
                    self.active_tasks -= 1
                # Log a progress heartbeat every 10 batches
                batches_processed += 1
                if batches_processed % 10 == 0:
                    q_size = egress_queue.qsize()
                    logger.info(
                        f"Parquet Writer: Processed {batches_processed} batches. "
                        f"(~{q_size} batches waiting in queue)"
                    )
            except queue.Empty:
                await asyncio.sleep(0.1)
    except asyncio.CancelledError:
        # This triggers during env.teardown()
        logger.info(
            f"ParquetStorageEgress shut down requested. "
            f"Successfully wrote {batches_processed} total chunks to storage."
        )

Serializers

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/dynamic_des/connectors/egress/kafka.py

def serialize(self, topic: str, data: Any) -> bytes:
    """
    Serializes the given data to a JSON byte string.

    Args:
        topic (str): The target Kafka topic (unused by this serializer but required by protocol).
        data (Any): The payload to serialize (dictionary or Pydantic model).

    Returns:
        bytes: The JSON-encoded byte string.
    """
    payload = extract_dict(data)
    return orjson.dumps(payload)

ConfluentAvroSerializer(registry_url, schema_str, **registry_kwargs)

Lazy-loaded serializer for Confluent Schema Registry using confluent-kafka.

This class converts simulation data into Avro-encoded byte strings, automatically fetching and validating against the schema from a Confluent-compatible Schema Registry.

Initializes the Confluent Avro serializer.

Parameters:

Name	Type	Description	Default
registry_url	str	The base URL of the Confluent Schema Registry.	required
schema_str	str	The Avro schema defined as a JSON string (typically generated from a dataclasses-avroschema AvroBaseModel via .avro_schema(); a plain Pydantic model does not provide it).	required
**registry_kwargs	Any	Additional configuration arguments passed directly to the SchemaRegistryClient (e.g., {'basic.auth.user.info': 'user:pass'}).	{}

Raises:

Type	Description
ImportError	If the 'confluent-kafka' package is not installed.

Source code in src/dynamic_des/connectors/egress/kafka.py

def __init__(self, registry_url: str, schema_str: str, **registry_kwargs: Any):
    """
    Initializes the Confluent Avro serializer.

    Args:
        registry_url (str): The base URL of the Confluent Schema Registry.
        schema_str (str): The Avro schema defined as a JSON string (typically
            generated from a dataclasses-avroschema `AvroBaseModel` via
            `.avro_schema()`; a plain Pydantic model does not provide it).
        **registry_kwargs: Additional configuration arguments passed directly
            to the `SchemaRegistryClient` (e.g., `{'basic.auth.user.info': 'user:pass'}`).

    Raises:
        ImportError: If the 'confluent-kafka' package is not installed.
    """
    try:
        from confluent_kafka.schema_registry import SchemaRegistryClient
        from confluent_kafka.schema_registry.avro import AvroSerializer
        from confluent_kafka.serialization import MessageField, SerializationContext
    except ImportError:
        raise ImportError(
            "The 'confluent-kafka' package is required to use ConfluentAvroSerializer. "
            "Install it using: pip install dynamic-des[confluent]"
        )

    # Merge the mandatory URL with any additional args (like basic.auth.user.info)
    conf = {"url": registry_url, **registry_kwargs}
    client = SchemaRegistryClient(conf)

    self.avro_serializer = AvroSerializer(client, schema_str)
    self.SerializationContext = SerializationContext
    self.MessageField = MessageField

Functions

serialize(topic, data)

Serializes the given data into an Avro byte string.

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.

Source code in src/dynamic_des/connectors/egress/kafka.py

def serialize(self, topic: str, data: Any) -> bytes:
    """
    Serializes the given data into an Avro byte string.

    Args:
        topic (str): The target Kafka topic, used for schema subject naming.
        data (Any): The payload to serialize (dictionary or Pydantic model).

    Returns:
        bytes: The Avro-encoded byte string.
    """
    payload = extract_dict(data)
    ctx = self.SerializationContext(topic, self.MessageField.VALUE)
    return self.avro_serializer(payload, ctx)

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 AvroBaseModel via .avro_schema(); a plain Pydantic model does not provide it).	required
**boto3_kwargs	Any	Additional arguments passed directly to the boto3.client initialization (e.g., region_name, aws_access_key_id).	{}

Raises:

Type	Description
ImportError	If the 'aws-glue-schema-registry' or 'boto3' packages are not installed.

Source code in src/dynamic_des/connectors/egress/kafka.py

def __init__(self, registry_name: str, schema_str: str, **boto3_kwargs: Any):
    """
    Initializes the AWS Glue Avro serializer.

    Args:
        registry_name (str): The name of the AWS Glue Schema Registry.
        schema_str (str): The Avro schema defined as a JSON string (typically
            generated from a dataclasses-avroschema `AvroBaseModel` via
            `.avro_schema()`; a plain Pydantic model does not provide it).
        **boto3_kwargs: Additional arguments passed directly to the `boto3.client`
            initialization (e.g., `region_name`, `aws_access_key_id`).

    Raises:
        ImportError: If the 'aws-glue-schema-registry' or 'boto3' packages are not installed.
    """
    try:
        import boto3
        from aws_schema_registry import SchemaRegistryClient
        from aws_schema_registry.adapter.kafka import KafkaSerializer
        from aws_schema_registry.avro import AvroSchema
    except ImportError:
        raise ImportError(
            "The 'aws-glue-schema-registry' and 'boto3' packages are required. "
            "Install them using: pip install dynamic-des[glue]"
        )

    # Provide a default region if the user didn't explicitly pass one
    if "region_name" not in boto3_kwargs:
        boto3_kwargs["region_name"] = "us-east-1"

    # Pass the kwargs directly into boto3
    glue_client = boto3.client("glue", **boto3_kwargs)

    registry_client = SchemaRegistryClient(glue_client, registry_name=registry_name)
    schema = AvroSchema(schema_str)
    self.serializer = KafkaSerializer(registry_client, schema)

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/dynamic_des/connectors/egress/kafka.py

def serialize(self, topic: str, data: Any) -> bytes:
    """
    Serializes the given data into an Avro byte string using AWS Glue.

    Args:
        topic (str): The target Kafka topic.
        data (Any): The payload to serialize (dictionary or Pydantic model).

    Returns:
        bytes: The Avro-encoded byte string.
    """
    payload = extract_dict(data)
    return self.serializer.serialize(topic, payload)