defined edge runtime and more central logic for sampler

Author: GioeleB00

src/app/api/simulation.py

@@ -1,18 +1,4 @@
 """"Api to simulate the process"""
 
-import numpy as np
-from fastapi import APIRouter
-
-from app.core.simulation.simulation_run import run_simulation
-from app.schemas.full_simulation_input import SimulationPayload
-from app.schemas.simulation_output import SimulationOutput
-
-router = APIRouter()
-
-@router.post("/simulation")
-async def event_loop_simulation(input_data: SimulationPayload) -> SimulationOutput:
-    """Run the simulation and return aggregate KPIs."""
-    rng = np.random.default_rng()
-    return run_simulation(input_data, rng=rng)
 
 

src/app/config/constants.py

@@ -49,6 +49,7 @@ class Distribution(StrEnum):
     NORMAL = "normal"
     LOG_NORMAL = "log_normal"
     EXPONENTIAL = "exponential"
+    UNIFORM = "uniform"
 
 # ======================================================================
 # CONSTANTS FOR ENDPOINT STEP DEFINITION (REQUEST-HANDLER)
@@ -123,7 +124,7 @@ class EndpointStepRAM(StrEnum):
     RAM = "ram"
 
 
-class Metrics(StrEnum):
+class StepOperation(StrEnum):
     """
     Keys used inside the ``metrics`` dictionary of a *step*.
 
@@ -153,6 +154,15 @@ class ServerResourcesDefaults:
     MINIMUM_RAM_MB = 256
     DB_CONNECTION_POOL = None
 
+# ======================================================================
+# CONSTANTS FOR NETWORK PARAMETERS
+# ======================================================================
+
+class NetworkParameters:
+  """parameters for the network"""
+
+  DROPOUT_RATE = 0.01
+
 # ======================================================================
 # CONSTANTS FOR THE MACRO-TOPOLOGY GRAPH
 # ======================================================================
@@ -164,7 +174,7 @@ class SystemNodes(StrEnum):
     Each member represents a *macro-component* that may have its own SimPy
     resources (CPU cores, DB pool, etc.).
     """
-    
+
     GENERATOR     = "generator"
     SERVER        = "server"
     CLIENT        = "client"

src/app/config/rqs_state.py

@@ -5,6 +5,7 @@
 """
 
 from __future__ import annotations
+
 from dataclasses import dataclass, field
 
 
@@ -17,27 +18,32 @@ class RequestState:
         id: Unique identifier of the request.
         t0: Timestamp (simulated env.now) when the request was generated.
         history: List of hop records, each noting a node/edge visit.
+        finish_time: Timestamp when the requests is satisfied
+
     """
+
     id: int                                # Unique request identifier
     initial_time: float                    # Generation timestamp (env.now)
     finish_time: float | None = None       # a requests might be dropped
     history: list[str] = field(default_factory=list)  # Trace of hops
 
-    def record_hop(self, node_name: str) -> None:
+    def record_hop(self, node_name: str, now: float) -> None:
         """
         Append a record of visiting a node or edge.
 
         Args:
             node_name: Name of the node or edge being recorded.
+            now: register the time of the operation
+
         """
         # Record hop as "NodeName@Timestamp"
-        self.history.append(f"{node_name}@{self.t0:.3f}")
-    
+        self.history.append(f"{node_name}@{now:.3f}")
+
     @property
     def latency(self) -> float | None:
         """
         Return the total time in the system (finish_time - initial_time),
-        or None if the request hasn’t completed yet.
+        or None if the request hasn't completed yet.
         """
         if self.finish_time is None:
             return None

src/app/core/event_samplers/common_helpers.py

@@ -3,31 +3,74 @@
 
 import numpy as np
 
+from app.config.constants import Distribution
+from app.schemas.random_variables_config import RVConfig
 
-def uniform_variable_generator(rng: np.random.Generator | None = None) -> float:
+
+def uniform_variable_generator(rng: np.random.Generator) -> float:
     """Return U~Uniform(0, 1)."""
     rng = rng or np.random.default_rng()
-    return float(rng.random())
-
+    return rng.random()
 
 def poisson_variable_generator(
     mean: float,
-    rng: np.random.Generator | None = None,
-) -> int:
-    """Return a Poisson-distributed integer with expectation *mean*."""
-    rng = rng or np.random.default_rng()
-    return int(rng.poisson(mean))
-
+    rng: np.random.Generator,
+) -> float:
+    """Return a Poisson-distributed floateger with expectation *mean*."""
+    return rng.poisson(mean)
 
 def truncated_gaussian_generator(
     mean: float,
     variance: float,
     rng: np.random.Generator,
-) -> int:
+) -> float:
     """
     Generate a Normal-distributed variable
     with mean and variance
     """
-    rng = rng or np.random.default_rng()
     value = rng.normal(mean, variance)
-    return max(0, int(value))
+    return max(0.0, value)
+
+def lognormal_variable_generator(
+    mean: float,
+    variance: float,
+    rng: np.random.Generator,
+) -> float:
+    """Return a Poisson-distributed floateger with expectation *mean*."""
+    return rng.lognormal(mean, variance)
+
+def exponential_variable_generator(
+    mean: float,
+    rng: np.random.Generator,
+) -> float:
+    """Return a Poisson-distributed floateger with expectation *mean*."""
+    return float(rng.exponential(mean))
+
+def general_sampler(random_variable: RVConfig, rng: np.random.Generator) -> float:
+    """Sample a number according to the distribution described in `random_variable`."""
+    dist = random_variable.distribution
+    mean = random_variable.mean
+
+    match dist:
+        case Distribution.UNIFORM:
+
+            assert random_variable.variance is None
+            return uniform_variable_generator(rng)
+
+        case _:
+
+            variance = random_variable.variance
+            assert variance is not None
+
+            match dist:
+                case Distribution.NORMAL:
+                    return truncated_gaussian_generator(mean, variance, rng)
+                case Distribution.LOG_NORMAL:
+                    return lognormal_variable_generator(mean, variance, rng)
+                case Distribution.POISSON:
+                    return float(poisson_variable_generator(mean, rng))
+                case Distribution.EXPONENTIAL:
+                    return exponential_variable_generator(mean, rng)
+                case _:
+                    msg = f"Unsupported distribution: {dist}"
+                    raise ValueError(msg)

src/app/core/event_samplers/gaussian_poisson.py

@@ -39,7 +39,6 @@ def gaussian_poisson_sampling(
          Δt ~ Exponential(Λ)   using inverse-CDF.
     4. Stop once the virtual clock exceeds *total_simulation_time*.
     """
-
     simulation_time = sim_settings.total_simulation_time
     user_sampling_window = input_data.user_sampling_window
 

src/app/core/event_samplers/poisson_poisson.py

@@ -36,7 +36,6 @@ def poisson_poisson_sampling(
          Δt ~ Exponential(Λ)   using inverse-CDF.
     4. Stop once the virtual clock exceeds *total_simulation_time*.
     """
-    
     simulation_time = sim_settings.total_simulation_time
     user_sampling_window = input_data.user_sampling_window
 

src/app/core/helpers/requests_generator.py

@@ -1,6 +1,10 @@
 """
-Continuous-time event sampling for the Poisson-Poisson
-and Gaussian-Poisson workload model.
+SimPy process that generates user requests at stochastic intervals.
+
+This node samples inter-arrival times according to the configured
+distribution (Gaussian-Poisson or Poisson-Poisson), constructs a
+RequestState for each new request, records its origin hop, and
+immediately pushes it into the next pipeline stage via an EdgeRuntime.
 """
 
 from __future__ import annotations
@@ -24,7 +28,7 @@ def requests_generator(
     input_data: RqsGeneratorInput,
     sim_settings: SimulationSettings,
     *,
-    rng: np.random.Generator | None = None,
+    rng: np.random.Generator,
 ) -> Generator[float, None, None]:
     """
     Return an iterator of inter-arrival gaps (seconds) according to the model

src/app/core/runtime/edge.py

@@ -0,0 +1,66 @@
+"""
+Unidirectional link that simulates message transmission between nodes.
+Encapsulates network behavior—latency sampling (LogNormal, Exponential, etc.),
+drop probability, and optional connection-pool contention—by exposing a
+`send(msg)` method. Each `send` call schedules a SimPy subprocess that
+waits the sampled delay (and any resource wait) before delivering the
+message to the target node's inbox.
+"""
+
+from collections.abc import Generator
+
+import numpy as np
+import simpy
+
+from app.config.constants import NetworkParameters
+from app.config.rqs_state import RequestState
+from app.core.event_samplers.common_helpers import general_sampler
+from app.schemas.system_topology_schema.full_system_topology_schema import Edge
+
+
+class EdgeRuntime:
+    """definining the logic to handle the edges during the simulation"""
+
+    def __init__(
+        self,
+        *,
+        env: simpy.Environment,
+        edge_config: Edge,
+        rng: np.random.Generator | None = None,
+        target_box: simpy.Store,
+        ) -> None:
+        """Definition of the instance attributes"""
+        self.env = env
+        self.edge_config = edge_config
+        self.target_box = target_box
+        self.rng = rng or np.random.default_rng()
+
+    def _deliver(self, state: RequestState) -> Generator[simpy.Event, None, None]:
+        """Function to deliver the state to the next node"""
+        # assign a probability that the network will drop the request
+        distribution = self.edge_config.latency.distribution
+
+        uniform_variable = self.rng.uniform()
+        if uniform_variable < NetworkParameters.DROPOUT_RATE:
+            state.finish_time = self.env.now
+            state.record_hop(f"{self.edge_config.id}-dropped")
+            return
+
+        transit_time = general_sampler(distribution, self.rng)
+        yield self.env.timeout(transit_time)
+        state.record_hop(self.edge_config.id, self.env.now)
+        self.target_box.put(state)
+
+
+    def transport(self, state: RequestState) -> simpy.Process:
+        """
+        Called by the upstream node. Immediately spins off a SimPy process
+        that will handle drop + delay + delivery of `state`.
+        """
+        return self.env.process(self._deliver(state))
+
+
+
+
+
+