API Reference

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):
        """
        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.
        """
        self.start_datetime = 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 is intended to act as a bridge for scenarios where simulation parameters are managed within a relational database. It would typically poll a configuration table for changes or listen to database notifications (LISTEN/NOTIFY) to trigger real-time updates in the simulation registry.

Note

This class is currently a placeholder and is planned for a future release.

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 is intended to act as a bridge for scenarios where simulation
    parameters are managed within a relational database. It would typically
    poll a configuration table for changes or listen to database notifications
    (LISTEN/NOTIFY) to trigger real-time updates in the simulation registry.

    Note:
        This class is currently a placeholder and is planned for a future release.
    """

    def __init__(self, *args, **kwargs):
        """
        Initializes the PostgresIngress placeholder.

        Raises:
            NotImplementedError: Always raised as the connector is not yet implemented.
        """
        raise NotImplementedError("PostgresIngress is planned for a future release.")

    async def run(self, ingress_queue: queue.Queue) -> None:
        """
        Placeholder for the database polling execution loop.

        Future implementations will likely utilize an async driver (e.g., asyncpg)
        to fetch rows and convert them into (path, value) tuples for the
        ingress queue.

        Args:
            ingress_queue: A thread-safe queue used to transmit updates
                to the SimulationRegistry.
        """
        pass

Functions

__init__(*args, **kwargs)

Initializes the PostgresIngress placeholder.

Raises:

Type	Description
NotImplementedError	Always raised as the connector is not yet implemented.

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

def __init__(self, *args, **kwargs):
    """
    Initializes the PostgresIngress placeholder.

    Raises:
        NotImplementedError: Always raised as the connector is not yet implemented.
    """
    raise NotImplementedError("PostgresIngress is planned for a future release.")

run(ingress_queue) async

Placeholder for the database polling execution loop.

Future implementations will likely utilize an async driver (e.g., asyncpg) to fetch rows and convert them into (path, value) tuples for the ingress queue.

Parameters:

Name	Type	Description	Default
ingress_queue	Queue	A thread-safe queue used to transmit updates to the SimulationRegistry.	required

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

async def run(self, ingress_queue: queue.Queue) -> None:
    """
    Placeholder for the database polling execution loop.

    Future implementations will likely utilize an async driver (e.g., asyncpg)
    to fetch rows and convert them into (path, value) tuples for the
    ingress queue.

    Args:
        ingress_queue: A thread-safe queue used to transmit updates
            to the SimulationRegistry.
    """
    pass

RedisIngress

Bases: BaseIngress

Asynchronous ingress provider for consuming updates from Redis.

This connector is designed to facilitate high-speed, low-latency parameter updates by leveraging Redis as a middleware layer. Potential implementations include subscribing to Pub/Sub channels for real-time broadcast signals or monitoring Redis Streams for sequential state updates.

Note

This class is currently a placeholder and is planned for a future release.

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

class RedisIngress(BaseIngress):
    """
    Asynchronous ingress provider for consuming updates from Redis.

    This connector is designed to facilitate high-speed, low-latency parameter
    updates by leveraging Redis as a middleware layer. Potential implementations
    include subscribing to Pub/Sub channels for real-time broadcast signals
    or monitoring Redis Streams for sequential state updates.

    Note:
        This class is currently a placeholder and is planned for a future release.
    """

    def __init__(self, *args, **kwargs):
        """
        Initializes the RedisIngress placeholder.

        Raises:
            NotImplementedError: Always raised as the connector is not yet implemented.
        """
        raise NotImplementedError("RedisIngress is planned for a future release.")

    async def run(self, ingress_queue: queue.Queue) -> None:
        """
        Placeholder for the Redis consumption execution loop.

        Future implementations will likely utilize an async Redis client
        to listen for messages and transform them into (path, value)
        tuples for the internal ingress queue.

        Args:
            ingress_queue: A thread-safe queue used to transmit updates
                to the SimulationRegistry.
        """
        pass

Functions

__init__(*args, **kwargs)

Initializes the RedisIngress placeholder.

Raises:

Type	Description
NotImplementedError	Always raised as the connector is not yet implemented.

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

def __init__(self, *args, **kwargs):
    """
    Initializes the RedisIngress placeholder.

    Raises:
        NotImplementedError: Always raised as the connector is not yet implemented.
    """
    raise NotImplementedError("RedisIngress is planned for a future release.")

run(ingress_queue) async

Placeholder for the Redis consumption execution loop.

Future implementations will likely utilize an async Redis client to listen for messages and transform them into (path, value) tuples for the internal ingress queue.

Parameters:

Name	Type	Description	Default
ingress_queue	Queue	A thread-safe queue used to transmit updates to the SimulationRegistry.	required

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

async def run(self, ingress_queue: queue.Queue) -> None:
    """
    Placeholder for the Redis consumption execution loop.

    Future implementations will likely utilize an async Redis client
    to listen for messages and transform them into (path, value)
    tuples for the internal ingress queue.

    Args:
        ingress_queue: A thread-safe queue used to transmit updates
            to the SimulationRegistry.
    """
    pass

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 is intended to handle long-term storage of simulation results, allowing for historical analysis and complex SQL queries on telemetry and event data.

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

class PostgresEgress(BaseEgress):
    """
    Asynchronous egress provider for persisting simulation data to PostgreSQL.

    This connector is intended to handle long-term storage of simulation results,
    allowing for historical analysis and complex SQL queries on telemetry
    and event data.
    """

    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.
        """
        self.dsn = connection_dsn
        self.table_name = table_name
        self.kwargs = kwargs

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

        Should implement a resilient connection using an async driver (like asyncpg),
        performing bulk inserts of batched data from the egress queue.

        Args:
            egress_queue: A thread-safe queue containing batches of simulation data.

        Raises:
            NotImplementedError: This connector is currently a placeholder.
        """
        logger.warning("PostgresEgress is not yet implemented.")
        raise NotImplementedError("PostgresEgress.run() is not implemented.")

Functions

__init__(connection_dsn, table_name='simulation_data', **kwargs)

Initializes the PostgresEgress with connection details.

Parameters:

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

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

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

    Args:
        connection_dsn: PostgreSQL connection string (DSN).
        table_name: Target table for simulation records.
        **kwargs: Additional connection pool arguments.
    """
    self.dsn = connection_dsn
    self.table_name = table_name
    self.kwargs = kwargs

run(egress_queue) async

Main execution loop for PostgreSQL data persistence.

Should implement a resilient connection using an async driver (like asyncpg), performing bulk inserts of batched data from the egress queue.

Parameters:

Name	Type	Description	Default
egress_queue	Queue	A thread-safe queue containing batches of simulation data.	required

Raises:

Type	Description
NotImplementedError	This connector is currently a placeholder.

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.

    Should implement a resilient connection using an async driver (like asyncpg),
    performing bulk inserts of batched data from the egress queue.

    Args:
        egress_queue: A thread-safe queue containing batches of simulation data.

    Raises:
        NotImplementedError: This connector is currently a placeholder.
    """
    logger.warning("PostgresEgress is not yet implemented.")
    raise NotImplementedError("PostgresEgress.run() is not implemented.")

RedisEgress

Bases: BaseEgress

Asynchronous egress provider for streaming simulation data to Redis.

Designed for real-time dashboarding or inter-process communication, publishing simulation updates to Redis Pub/Sub channels or Streams.

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

class RedisEgress(BaseEgress):
    """
    Asynchronous egress provider for streaming simulation data to Redis.

    Designed for real-time dashboarding or inter-process communication,
    publishing simulation updates to Redis Pub/Sub channels or Streams.
    """

    def __init__(
        self,
        host: str = "localhost",
        port: int = 6379,
        channel: str = "sim_stream",
        **kwargs: Any,
    ):
        """
        Initializes the RedisEgress with connection and routing details.

        Args:
            host: Redis server hostname.
            port: Redis server port.
            channel: The target Pub/Sub channel or Stream key.
            **kwargs: Additional redis-py configuration.
        """
        self.host = host
        self.port = port
        self.channel = channel
        self.kwargs = kwargs

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

        Should implement an async client (like redis-py's async mode) to
        publish or add entries to Redis based on the simulation egress queue.

        Args:
            egress_queue: A thread-safe queue containing batches of simulation data.

        Raises:
            NotImplementedError: This connector is currently a placeholder.
        """
        logger.warning("RedisEgress is not yet implemented.")
        raise NotImplementedError("RedisEgress.run() is not implemented.")

Functions

__init__(host='localhost', port=6379, channel='sim_stream', **kwargs)

Initializes the RedisEgress with connection and routing details.

Parameters:

Name	Type	Description	Default
host	str	Redis server hostname.	'localhost'
port	int	Redis server port.	6379
channel	str	The target Pub/Sub channel or Stream key.	'sim_stream'
**kwargs	Any	Additional redis-py configuration.	{}

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

def __init__(
    self,
    host: str = "localhost",
    port: int = 6379,
    channel: str = "sim_stream",
    **kwargs: Any,
):
    """
    Initializes the RedisEgress with connection and routing details.

    Args:
        host: Redis server hostname.
        port: Redis server port.
        channel: The target Pub/Sub channel or Stream key.
        **kwargs: Additional redis-py configuration.
    """
    self.host = host
    self.port = port
    self.channel = channel
    self.kwargs = kwargs

run(egress_queue) async

Main execution loop for Redis data streaming.

Should implement an async client (like redis-py's async mode) to publish or add entries to Redis based on the simulation egress queue.

Parameters:

Name	Type	Description	Default
egress_queue	Queue	A thread-safe queue containing batches of simulation data.	required

Raises:

Type	Description
NotImplementedError	This connector is currently a placeholder.

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

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

    Should implement an async client (like redis-py's async mode) to
    publish or add entries to Redis based on the simulation egress queue.

    Args:
        egress_queue: A thread-safe queue containing batches of simulation data.

    Raises:
        NotImplementedError: This connector is currently a placeholder.
    """
    logger.warning("RedisEgress is not yet implemented.")
    raise NotImplementedError("RedisEgress.run() is not implemented.")

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)