API Reference

Simulation Context

SimulationContext

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

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

class SimulationContext:
    """
    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:
        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).
        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.
    """

    def __init__(
        self,
        sim_id: str,
        factor: float = 1.0,
        random_seed: Optional[int] = None,
        logical_start_time: 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).
        """
        self.sim_id = sim_id
        self.factor = factor
        self.random_seed = random_seed
        self.logical_start_time = logical_start_time

        # Builder State (Pre-Compilation)
        self._ingress_providers: List[Any] = []
        self._egress_providers: List[Any] = []
        self._batch_size = 500
        self._flush_interval = 1.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]]] = []

    # ==========================================
    # BUILDER METHODS (Fluent API)
    # ==========================================

    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

    def add_egress(self, provider: Any) -> "SimulationContext":
        """
        Registers an asynchronous egress provider for data exfiltration.

        Args:
            provider: An initialized connector subclassing `BaseEgress`
                (e.g., `KafkaEgress` or `ParquetStorageEgress`).

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

    def with_batching(
        self, batch_size: int, flush_interval: float
    ) -> "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.

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

    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

    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

    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

    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

    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

    # ==========================================
    # DECORATORS (Execution Logic)
    # ==========================================

    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

    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

    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

    # ==========================================
    # RUNTIME HELPERS (For Raw Generators)
    # ==========================================

    @property
    def env(self) -> DynamicRealtimeEnvironment:
        """
        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:
            RuntimeError: If accessed during the Builder Phase (before `.run()`).
        """
        if self._env is None:
            raise RuntimeError("Environment is not attached until context is run.")
        return self._env

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

    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]

    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)

    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)

    # ==========================================
    # ORCHESTRATION (The Compilation Phase)
    # ==========================================

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

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

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

__init__(sim_id, factor=1.0, random_seed=None, logical_start_time=None)

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

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,
):
    """
    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).
    """
    self.sim_id = sim_id
    self.factor = factor
    self.random_seed = random_seed
    self.logical_start_time = logical_start_time

    # Builder State (Pre-Compilation)
    self._ingress_providers: List[Any] = []
    self._egress_providers: List[Any] = []
    self._batch_size = 500
    self._flush_interval = 1.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]]] = []

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)

Registers an asynchronous egress provider for data exfiltration.

Parameters:

Name	Type	Description	Default
provider	Any	An initialized connector subclassing BaseEgress (e.g., KafkaEgress or ParquetStorageEgress).	required

Returns:

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

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

def add_egress(self, provider: Any) -> "SimulationContext":
    """
    Registers an asynchronous egress provider for data exfiltration.

    Args:
        provider: An initialized connector subclassing `BaseEgress`
            (e.g., `KafkaEgress` or `ParquetStorageEgress`).

    Returns:
        SimulationContext: The current instance for method chaining.
    """
    self._egress_providers.append(provider)
    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
    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,
        )

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

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

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

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

Environment

DynamicRealtimeEnvironment

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.

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

class DynamicRealtimeEnvironment(
    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:
        start_datetime (datetime): The real-world clock time when the simulation started.
    """

    def __init__(
        self,
        initial_time=0,
        factor=1.0,
        strict=False,
        logical_start_time: 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).
        """
        # 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.setup_registry()

    def teardown(self):
        """
        Gracefully terminates the environment.

        Ensures that any remaining data in event buffers is flushed to the
        egress connectors, and that background asyncio threads for both
        ingress and egress are cleanly stopped. Should be called in a `finally` block.
        """
        logger.info("Environment teardown initiated.")
        if hasattr(self, "teardown_egress"):
            self.teardown_egress()
        if hasattr(self, "teardown_ingress"):
            self.teardown_ingress()
        logger.info("Environment teardown complete.")

Registry & Parameters

SimulationRegistry

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

class SimulationRegistry:
    """
    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:
        env (Environment): The active SimPy environment.
    """

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

    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]

    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]

    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}")

    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)

    def _register_dist(self, path_prefix: str, dist_config: Any):
        """Internal: Flattens a DistributionConfig."""
        if dist_config.dist == "exponential":
            self._values[f"{path_prefix}.rate"] = DynamicValue(
                self.env, f"{path_prefix}.rate", dist_config.rate
            )
        else:
            self._values[f"{path_prefix}.mean"] = DynamicValue(
                self.env, f"{path_prefix}.mean", dist_config.mean
            )
            self._values[f"{path_prefix}.std"] = DynamicValue(
                self.env, f"{path_prefix}.std", dist_config.std
            )

    def _register_cap(self, path_prefix: str, cap_config: Any):
        """Internal: Flattens a CapacityConfig."""
        self._values[f"{path_prefix}.current_cap"] = DynamicValue(
            self.env, f"{path_prefix}.current_cap", cap_config.current_cap
        )
        self._values[f"{path_prefix}.max_cap"] = DynamicValue(
            self.env, f"{path_prefix}.max_cap", cap_config.max_cap
        )

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

Source code in src/dynamic_des/models/params.py

@dataclass
class SimParameter:
    """
    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:
        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.
    """

    sim_id: str
    arrival: Dict[str, DistributionConfig] = field(default_factory=dict)
    service: Dict[str, DistributionConfig] = field(default_factory=dict)
    # Standardized categories
    resources: Dict[str, CapacityConfig] = field(default_factory=dict)
    containers: Dict[str, CapacityConfig] = field(default_factory=dict)
    stores: Dict[str, CapacityConfig] = field(default_factory=dict)
    # Custom / User-Defined states
    variables: Dict[str, Any] = field(default_factory=dict)

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

Source code in src/dynamic_des/models/params.py

@dataclass
class CapacityConfig:
    """
    Configuration for the capacity of a simulated resource, container, or store.

    Attributes:
        current_cap (Union[int, float]): The currently active capacity.
        max_cap (Union[int, float]): The absolute physical maximum capacity.
    """

    current_cap: Union[int, float]
    max_cap: Union[int, float]

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

Source code in src/dynamic_des/models/params.py

@dataclass
class DistributionConfig:
    """
    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:
        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.
    """

    dist: Literal["exponential", "normal", "lognormal"]
    rate: Optional[float] = None
    mean: Optional[float] = None
    std: Optional[float] = None

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.

Source code in src/dynamic_des/models/schemas.py

class TelemetryPayload(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:
        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.
    """

    stream_type: Literal["telemetry"] = "telemetry"
    path_id: str = Field(
        ...,
        description="The dot-notation path of the metric (e.g., 'Line_A.lathe.utilization').",
    )
    value: Any = Field(..., description="The scalar value of the metric.")

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.

Source code in src/dynamic_des/models/schemas.py

class EventPayload(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:
        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.
    """

    stream_type: Literal["event"] = "event"
    key: str = Field(
        ...,
        description="A unique identifier/partition key for the event (e.g., 'task-001').",
    )
    value: Any = Field(
        ..., description="A dictionary or Pydantic model containing the event 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.

Source code in src/dynamic_des/models/schemas.py

class BaseStreamPayload(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:
        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.
    """

    stream_type: Literal["telemetry", "event"]
    sim_ts: float = Field(..., description="The simulation clock time (in seconds).")
    timestamp: str = Field(..., description="The real-world ISO 8601 timestamp.")

Utilities

Sampler

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.

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

class Sampler:
    """
    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:
        rng (np.random.Generator): The NumPy random number generator instance.
    """

    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

    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

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

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

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

class DynamicResource(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:
        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').
    """

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

    @property
    def capacity(self) -> int:
        """
        The current active capacity of the resource.

        This value reflects the live state from the Registry and may change
        during the simulation.
        """
        return self._capacity

    @property
    def in_use(self) -> int:
        """Currently occupied slots (Total Capacity - Idle Tokens)."""
        return self._capacity - self.pool.level

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

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

    def _dispatcher(self):
        """Bridges PriorityStore and Container without pre-fetching tokens."""
        while True:
            # If no one is waiting, sleep until a request is made
            if not self.queue.items:
                yield self._request_event
                self._request_event = self.env.event()

            # Wait for a physical token to be available
            yield self.pool.get(1)

            # Pull the highest-priority request currently in the queue
            ticket = yield self.queue.get()

            # If the user cancelled their request (e.g., timed out while waiting),
            # return the token to the pool and move to the next person.
            if getattr(ticket.item, "cancelled", False):
                self.pool.put(1)
                continue

            # Signal the user process
            if not ticket.item.triggered:
                ticket.item.succeed()

    def _handle_capacity_change(self, new_target: int):
        # Safely bind the new target to physical limits [0, max_cap]
        new_target = max(0, min(new_target, self.pool.capacity))

        diff = new_target - self._capacity
        if diff > 0:
            self._capacity += diff
            self.env.process(self._grow_pool(diff))
        elif diff < 0:
            amount = abs(diff)
            self._capacity -= amount
            self.env.process(self._shrink_pool(amount))

    def _grow_pool(self, amount: int):
        yield self.pool.put(amount)

    def _shrink_pool(self, amount: int):
        yield self.pool.get(amount)

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

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.

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

class DynamicContainer(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:
        env (Environment): The active SimPy simulation environment.
        sim_id (str): The parent simulation ID prefix.
        obj_id (str): The specific container name/identifier.
    """

    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)

    @property
    def capacity(self) -> float:
        """
        float: The current active maximum capacity of the container.
        This value reflects the live state from the Registry.
        """
        return self._container.capacity

    @property
    def level(self) -> float:
        """float: The current amount of bulk material stored inside the container."""
        return self._container.level

    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)

    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)

    def _handle_capacity_change(self, new_target: float):
        """
        Internal callback triggered when the Registry capacity value changes.
        Updates the physical SimPy container and processes pending events.

        Args:
            new_target (float): The new capacity limit dictated by the control plane.
        """
        # Safely bind the new target to absolute physical limits [0, max_cap]
        max_cap = float(self._max_cap_val.value)
        new_target = max(0.0, min(float(new_target), max_cap))

        self._container._capacity = new_target

        # If capacity grew, pending put requests might now have enough room to succeed.
        # We manually trigger SimPy's internal put processor to wake them up.
        if self._container.put_queue:
            self._container._trigger_put(None)

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

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.

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

class DynamicStore(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:
        env (Environment): The active SimPy simulation environment.
        sim_id (str): The parent simulation ID prefix.
        obj_id (str): The specific store name/identifier.
    """

    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)

    @property
    def capacity(self) -> int:
        """
        int: The current active maximum slot capacity of the store.
        This value reflects the live state from the Registry.
        """
        return self._store.capacity

    @property
    def items(self) -> list:
        """
        list: The actual list of distinct item objects currently residing
        in the store.
        """
        return self._store.items

    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)

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

    def _handle_capacity_change(self, new_target: int):
        """
        Internal callback triggered when the Registry capacity value changes.
        Updates the physical SimPy store and processes pending events.

        Args:
            new_target (int): The new slot capacity limit dictated by the control plane.
        """
        # Safely bind the new target to absolute physical limits [0, max_cap]
        max_cap = int(self._max_cap_val.value)
        new_target = max(0, min(int(new_target), max_cap))

        self._store._capacity = new_target

        # If capacity grew, pending put requests might now have open slots to succeed.
        # This exact same trigger works for both Store and PriorityStore.
        if self._store.put_queue:
            self._store._trigger_put(None)

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

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.

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

class KafkaAdminConnector:
    """
    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:
        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.
    """

    def __init__(self, bootstrap_servers: str, max_tasks: int = 100, **kwargs):
        """
        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] = {}

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

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

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

    def _process_message(self, data: Dict[str, Any]):
        """
        Routes incoming messages based on their JSON structure.

        Distinguishes between 'telemetry' (vitals like utilization) and
        'events' (task lifecycle steps like 'started' or 'finished').

        Args:
            data: The decoded JSON payload from Kafka.
        """

        # 1. Telemetry: 'path_id' is at the root level
        if "path_id" in data and not isinstance(data.get("value"), dict):
            self._vitals[data["path_id"]] = data["value"]

        # 2. Events: 'key' is at the root, 'value' is a dictionary
        elif "key" in data and isinstance(data.get("value"), dict):
            task_id = data["key"]
            payload = data["value"]
            status = payload.get("status")

            # Using 'timestamp' based on your actual JSON payload
            ts = data.get("timestamp")

            path_id = payload.get("path_id", "unknown.unknown.unknown")
            parts = path_id.split(".")
            sim_id = parts[0]
            service = parts[2] if len(parts) > 2 else "default"

            service_data = self._state[sim_id][service]

            # Prune oldest if at max capacity
            if task_id not in service_data:
                if len(service_data) >= self.max_tasks:
                    oldest_key = next(iter(service_data))
                    service_data.pop(oldest_key)

            # Update task status timestamp
            if task_id not in service_data:
                service_data[task_id] = {}

            service_data[task_id][status] = ts

    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

    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

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.

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

class 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.
    """

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

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

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.

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

class KafkaIngress(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:
        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.
    """

    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

    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)

Functions

__init__(topic, bootstrap_servers, topic_deserializers=None, default_deserializer=None, **kwargs)

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

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

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

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

class LocalIngress(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:
        schedule (List[Tuple[float, str, Any]]): A chronologically sorted
            list of updates, where each tuple contains (delay_seconds,
            path_id, value).
    """

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

    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

Functions

__init__(schedule)

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

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

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.

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

class PostgresIngress(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.
    """

    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

    async def _init_pool(self) -> None:
        """Initializes the asyncpg connection pool if not already created."""
        if self.pool is None:
            self.pool = await asyncpg.create_pool(self.dsn, **self.kwargs)

            # Ensure the table exists
            assert self.pool is not None
            async with self.pool.acquire() as conn:
                await conn.execute(f"""
                    CREATE TABLE IF NOT EXISTS {self.table_name} (
                        id SERIAL PRIMARY KEY,
                        param_path VARCHAR(255) NOT NULL,
                        param_value TEXT NOT NULL,
                        is_applied BOOLEAN DEFAULT FALSE,
                        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                    );
                """)

    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)

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

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.

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

class RedisIngress(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:
        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.Redis | None): The underlying redis-py asynchronous client instance.
        pubsub (redis.client.PubSub | None): The active Pub/Sub subscription instance.
        **kwargs: Additional configuration dictionary passed to `redis.from_url`.
    """

    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

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

Functions

__init__(url, channel_name='params', poll_interval=0.5, **kwargs)

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

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.

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

class JsonDeserializer:
    """
    Default fallback deserializer providing backward compatibility via standard JSON.

    This deserializer assumes the incoming payload is a UTF-8 encoded JSON string.
    """

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

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

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.

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

class ConfluentAvroDeserializer:
    """
    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.
    """

    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

    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)

Functions

__init__(registry_url, **registry_kwargs)

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

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

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.

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

class GlueAvroDeserializer:
    """
    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.
    """

    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)

    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

Functions

__init__(registry_name, **boto3_kwargs)

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)

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.

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

class 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.
    """

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

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

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

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

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

class KafkaEgress(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).

    Attributes:
        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`.
        producer_config (dict): Configuration dictionary passed to AIOKafkaProducer.

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

        ```python
        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
        )
        ```
    """

    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,
        **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`.
            **kwargs: Additional overrides for the AIOKafkaProducer configuration.
        """
        self.telemetry_topic = telemetry_topic
        self.event_topic = event_topic
        self.topic_router = topic_router

        # 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,
        }

    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.pop("stream_type", "event")
                                else:
                                    # Fallback to standard dynamic-des routing
                                    stream = data.pop("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
                                    )

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

                                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)

Functions

__init__(bootstrap_servers, telemetry_topic='sim-telemetry', event_topic='sim-events', topic_router=None, topic_serializers=None, default_serializer=None, **kwargs)

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
**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,
    **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`.
        **kwargs: Additional overrides for the AIOKafkaProducer configuration.
    """
    self.telemetry_topic = telemetry_topic
    self.event_topic = event_topic
    self.topic_router = topic_router

    # 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,
    }

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.pop("stream_type", "event")
                            else:
                                # Fallback to standard dynamic-des routing
                                stream = data.pop("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
                                )

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

                            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.

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

class ConsoleEgress(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.
    """

    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.pop("stream_type", "unknown")
                    prefix = "[TEL]" if stream == "telemetry" else "[EVT]"

                    # Log the remaining data (path_id/key, value, timestamp)
                    logger.info(f"{prefix} {data}")

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

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.pop("stream_type", "unknown")
                prefix = "[TEL]" if stream == "telemetry" else "[EVT]"

                # Log the remaining data (path_id/key, value, timestamp)
                logger.info(f"{prefix} {data}")

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

PostgresEgress

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.

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

class PostgresEgress(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.
    """

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

    async def _init_pool(self) -> None:
        if self.pool is None:
            self.pool = await asyncpg.create_pool(dsn=self.dsn, **self.kwargs)
            logger.info(f"PostgresEgress connected to {self.table_name}")

            # Cache valid columns to safely ignore extra injected fields (like sim_ts)
            self.valid_columns = set()
            assert self.pool is not None
            async with self.pool.acquire() as conn:
                rows = await conn.fetch(
                    "SELECT column_name FROM information_schema.columns WHERE table_name = $1",
                    self.table_name,
                )
                self.valid_columns = {row["column_name"] for row in rows}
                if not self.valid_columns:
                    logger.warning(
                        f"Table '{self.table_name}' does not exist or has no columns!"
                    )

    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)

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

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.

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

class RedisEgress(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:
        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.Redis | None): The underlying redis-py asynchronous client instance.
        **kwargs: Additional configuration dictionary passed to `redis.from_url`.
    """

    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

    async def _init_client(self) -> None:
        if self.client is None:
            self.client = redis.from_url(self.url, **self.kwargs)
            logger.info(f"RedisEgress connected to stream '{self.stream_name}'")

    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:
                    # Support multi-stream multiplexing
                    target = item.get("__stream__", self.stream_name)
                    item_copy = item.copy()
                    item_copy.pop("__stream__", None)

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

Functions

__init__(url, stream_name='simulation_data', **kwargs)

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

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:
                # Support multi-stream multiplexing
                target = item.get("__stream__", self.stream_name)
                item_copy = item.copy()
                item_copy.pop("__stream__", None)

                # 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

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)

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

class JsonlStorageEgress(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:
        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:

        ```python
        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)
        ```
    """

    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

    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()
                    await asyncio.to_thread(self._write_batch, batch)
                    # 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."
            )

    def _write_batch(self, batch: list):
        """
        Groups a batch of records by destination path and writes them to chunked files.

        Args:
            batch (list): A list of dictionaries or Pydantic models to be written.
        """
        if self.filesystem is None:
            logger.error("Filesystem not initialized. Batch dropped.")
            return

        grouped_batches: dict[str, list[dict]] = {}

        for data in batch:
            target_path = (
                self.path_router(data) if self.path_router else self.default_path
            )
            if not target_path or not self.filesystem:
                continue

            if target_path not in grouped_batches:
                grouped_batches[target_path] = []
            grouped_batches[target_path].append(extract_dict(data))

        for target_path, records in grouped_batches.items():
            chunk_path = _generate_chunk_filename(target_path)
            with self.filesystem.open_output_stream(chunk_path) as stream:
                for payload in records:
                    stream.write(
                        orjson.dumps(payload, option=orjson.OPT_APPEND_NEWLINE)
                    )

Functions

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

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

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()
                await asyncio.to_thread(self._write_batch, batch)
                # 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

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
)

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

class ParquetStorageEgress(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:
        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:

        ```python
        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
        )
        ```
    """

    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] = {}

    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()
                    await asyncio.to_thread(self._write_batch, batch, pa, pq)
                    # 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."
            )

    def _write_batch(self, batch: list, pa: Any, pq: Any):
        """
        Groups a batch of records, enforces strict schema typing, and writes Parquet chunks.

        This method dynamically infers the PyArrow schema from the first chunk of data
        destined for a specific path. It caches this schema and safely casts all
        subsequent chunks to match, preventing pipeline-breaking schema drift
        (e.g., if a float unexpectedly arrives as an integer in a later batch).

        Args:
            batch (list): A list of dictionaries or Pydantic models to be written.
            pa (Any): Injected reference to the `pyarrow` module.
            pq (Any): Injected reference to the `pyarrow.parquet` module.
        """
        grouped_batches: dict[str, list[dict]] = {}

        for data in batch:
            target_path = (
                self.path_router(data) if self.path_router else self.default_path
            )
            if not target_path:
                continue

            if target_path not in grouped_batches:
                grouped_batches[target_path] = []
            grouped_batches[target_path].append(extract_dict(data))

        for target_path, records in grouped_batches.items():
            if not self.filesystem:
                continue

            table = pa.Table.from_pylist(records)

            # Infer schema on first batch and cache it
            if target_path not in self.schemas:
                self.schemas[target_path] = table.schema

            # Cast current batch to match the canonical schema
            expected_schema = self.schemas[target_path]
            if table.schema != expected_schema:
                table = table.cast(expected_schema)

            chunk_path = _generate_chunk_filename(target_path)

            pq.write_table(table, chunk_path, filesystem=self.filesystem)

Functions

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

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] = {}

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()
                await asyncio.to_thread(self._write_batch, batch, pa, pq)
                # 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.

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

class JsonSerializer:
    """
    Default fallback serializer providing backward compatibility via orjson.

    This serializer converts dictionaries or Pydantic models into standard
    JSON byte strings.
    """

    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)

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

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.

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

class ConfluentAvroSerializer:
    """
    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.
    """

    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 directly from a Pydantic model via `.avro_schema()`).
            **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

    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)

Functions

__init__(registry_url, schema_str, **registry_kwargs)

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 directly from a Pydantic model via .avro_schema()).	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 directly from a Pydantic model via `.avro_schema()`).
        **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

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

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.

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

class GlueAvroSerializer:
    """
    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.
    """

    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 directly from a Pydantic model via `.avro_schema()`).
            **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)

    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)

Functions

__init__(registry_name, schema_str, **boto3_kwargs)

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 directly from a Pydantic model via .avro_schema()).	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 directly from a Pydantic model via `.avro_schema()`).
        **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)

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)