now the simplified metered ramp is able to use two different flow eq: limited and unlimited

Author: FilippoAiraldi

src/sym_metanet/blocks/origins.py

@@ -225,19 +225,28 @@ class SimplifiedMeteredOnRamp(MeteredOnRamp[VarType]):
     _actions = {"q"}
 
     def __init__(
-        self, capacity: Union[VarType, float], name: Optional[str] = None
+        self,
+        capacity: Union[VarType, float],
+        flow_eq_type: Literal["limited", "unlimited"] = "limited",
+        name: Optional[str] = None,
     ) -> None:
         """Instantiates a simplified on-ramp with the given capacity.
 
         Parameters
         ----------
         capacity : float or variable
             Capacity of the on-ramp, i.e., `C`.
+        flow_eq_type : 'limited' or 'unlimited', optional
+            Type of flow equation for the ramp. See
+            `engine.origins.get_simplifiedramp_flow` for more details.
         name : str, optional
             Name of the on-ramp, by default None.
         """
-        super().__init__(capacity=capacity, name=name)
-        del self.flow_eq_type
+        super().__init__(
+            capacity=capacity,
+            flow_eq_type=flow_eq_type,  # type: ignore[arg-type]
+            name=name,
+        )
 
     def init_vars(
         self,
@@ -306,4 +315,5 @@ def get_flow(  # type: ignore[override]
             rho_crit=link_down.rho_crit,
             rho_first=link_down.states["rho"][0],
             T=T,
+            type=self.flow_eq_type,  # type: ignore[arg-type]
         )

src/sym_metanet/engines/casadi.py

@@ -131,14 +131,17 @@ def get_ramp_flow(
     @staticmethod
     def get_simplifiedramp_flow(
         qdes: VarType,
-        d: VarType,
-        w: VarType,
-        C: VarType,
-        rho_max: VarType,
-        rho_first: VarType,
-        rho_crit: VarType,
-        T: VarType,
+        d: VarType = None,
+        w: VarType = None,
+        C: VarType = None,
+        rho_max: VarType = None,
+        rho_first: VarType = None,
+        rho_crit: VarType = None,
+        T: VarType = None,
+        type: Literal["limited", "unlimited"] = "limited",
     ) -> VarType:
+        if type == "unlimited":
+            return qdes
         term2 = d + w / T
         term3 = C * cs.fmin(1, (rho_max - rho_first) / (rho_max - rho_crit))
         return cs.fmin(qdes, cs.fmin(term2, term3))

src/sym_metanet/engines/core.py

@@ -311,37 +311,45 @@ def get_ramp_flow(
     @abstractmethod
     def get_simplifiedramp_flow(
         qdes,
-        d,
-        w,
-        C,
-        rho_max,
-        rho_first,
-        rho_crit,
-        T,
+        d=None,
+        w=None,
+        C=None,
+        rho_max=None,
+        rho_first=None,
+        rho_crit=None,
+        T=None,
+        type: Literal["limited", "unlimited"] = "limited",
     ):
         """Computes the flows of a simplified ramp origin.
 
         Parameters
         ----------
         qdes
             Desired flow at the ramp.
-        d
+        d : optional
             Demand at the ramp.
-        w
+        w : optional
             Queue of the ramp.
-        C
+        C : optional
             Capacity of the ramp.
-        rho_first
+        rho_first : optional
             Density of the first segment of the link the ramp is attached to. To be
             provided only if `type=limited`.
-        rho_max
+        rho_max : optional
             Maximum density of the link the ramp is attached to. To be provided only if
             `type=limited`.
-        rho_crit
+        rho_crit : optional
             Critical density of the link the ramp is attached to. To be provided only if
             `type=limited`.
-        T
+        T : optional
             Sampling time.
+        type : "limited" or "unlimited", optional
+            Type of equation to compute the flow for a simplified metered ramp. By
+            default, "limited" is selected, which means that the desired flow is limited
+            according to the capacity, current demands, queues and densities of the ramp
+            and its attached segment. This equation needs that all optional arguments
+            are provided. Otherwise, "unlimited" makes sure that the  desired flow is
+            the actual flow of the ramp, which can cause negative queues though.
 
         Returns
         -------

src/sym_metanet/engines/numpy.py

@@ -123,16 +123,21 @@ def get_ramp_flow(
     @staticmethod
     def get_simplifiedramp_flow(
         qdes: np.ndarray,
-        d: np.ndarray,
-        w: np.ndarray,
-        C: np.ndarray,
-        rho_max: np.ndarray,
-        rho_first: np.ndarray,
-        rho_crit: np.ndarray,
-        T: np.ndarray,
+        d: Optional[np.ndarray] = None,
+        w: Optional[np.ndarray] = None,
+        C: Optional[np.ndarray] = None,
+        rho_max: Optional[np.ndarray] = None,
+        rho_first: Optional[np.ndarray] = None,
+        rho_crit: Optional[np.ndarray] = None,
+        T: Optional[np.ndarray] = None,
+        type: Literal["limited", "unlimited"] = "limited",
     ) -> np.ndarray:
-        term2 = d + w / T
-        term3 = C * np.minimum(1, (rho_max - rho_first) / (rho_max - rho_crit))
+        if type == "unlimited":
+            return qdes
+        term2 = d + w / T  # type: ignore[operator]
+        term3 = C * np.minimum(
+            1, (rho_max - rho_first) / (rho_max - rho_crit)  # type: ignore[operator]
+        )
         return np.minimum(qdes, np.minimum(term2, term3))