Biological Components API

pyadm1.components.biological.digester.Digester

Bases: Component

ADM1da digester component (41-state).

Encapsulates a digester's geometry, attached gas storage, and the underlying :class:pyadm1.core.adm1.ADM1 instance. The same component class is used for hydrolysis pre-tanks, main fermenters and post-digesters — the operating temperature and HRT determine the dominant biochemistry.

Parameters

component_id : str Unique identifier for this component. feedstock : Feedstock or None Feedstock used to derive the influent. May be None if you intend to drive the digester exclusively via set_influent_dataframe on the underlying :class:ADM1 instance. V_liq, V_gas, T_ad : float Reactor liquid volume [m³], gas headspace [m³], temperature [K]. When dynamic_volume=True, V_liq is interpreted as the geometric maximum (the overflow-weir setpoint) and the actual sludge volume evolves with the mass balance. name : str, optional dynamic_volume : bool, default False Enable a dynamic sludge-volume balance dV/dt = Q_in − Q_out − q_S,loss, with Q_out from an overflow weir at V_liq. When False, sludge volume stays constant and Q_out = Q_in − q_S,loss enforces volume conservation. initial_fill_fraction : float, default 1.0 Starting sludge fill as a fraction of the geometric maximum. Only used when dynamic_volume=True. Set below 1.0 to simulate a partially-filled startup transient. outflow_time_constant : float, default 1.0 Overflow-weir time constant τ_out [d]: Q_out = max(0, (V − V_max) / τ_out). Only used when dynamic_volume=True. Steady- state sludge volume sits slightly above V_max at V_max + (Q_in − q_S,loss)·τ_out.

Source code in pyadm1/components/biological/digester.py

class Digester(Component):
    """
    ADM1da digester component (41-state).

    Encapsulates a digester's geometry, attached gas storage, and the
    underlying :class:`pyadm1.core.adm1.ADM1` instance.  The same component
    class is used for hydrolysis pre-tanks, main fermenters and post-digesters
    — the operating temperature and HRT determine the dominant biochemistry.

    Parameters
    ----------
    component_id : str
        Unique identifier for this component.
    feedstock : Feedstock or None
        Feedstock used to derive the influent.  May be ``None`` if you intend
        to drive the digester exclusively via ``set_influent_dataframe`` on
        the underlying :class:`ADM1` instance.
    V_liq, V_gas, T_ad : float
        Reactor liquid volume [m³], gas headspace [m³], temperature [K]. When
        ``dynamic_volume=True``, ``V_liq`` is interpreted as the geometric
        maximum (the overflow-weir setpoint) and the actual sludge volume
        evolves with the mass balance.
    name : str, optional
    dynamic_volume : bool, default False
        Enable a dynamic sludge-volume balance
        ``dV/dt = Q_in − Q_out − q_S,loss``, with ``Q_out`` from an overflow
        weir at ``V_liq``. When False, sludge volume stays constant and
        ``Q_out = Q_in − q_S,loss`` enforces volume conservation.
    initial_fill_fraction : float, default 1.0
        Starting sludge fill as a fraction of the geometric maximum. Only
        used when ``dynamic_volume=True``. Set below 1.0 to simulate a
        partially-filled startup transient.
    outflow_time_constant : float, default 1.0
        Overflow-weir time constant ``τ_out`` [d]: ``Q_out = max(0, (V −
        V_max) / τ_out)``. Only used when ``dynamic_volume=True``. Steady-
        state sludge volume sits slightly above ``V_max`` at
        ``V_max + (Q_in − q_S,loss)·τ_out``.
    """

    def __init__(
        self,
        component_id: str,
        feedstock,
        V_liq: float = 1977.0,
        V_gas: float = 304.0,
        T_ad: float = 308.15,
        name: Optional[str] = None,
        dynamic_volume: bool = False,
        initial_fill_fraction: float = 1.0,
        outflow_time_constant: float = 1.0,
    ):
        super().__init__(component_id, ComponentType.DIGESTER, name)

        self.feedstock = feedstock
        self.V_liq = V_liq
        self.V_gas = V_gas
        self.T_ad = T_ad

        # Dynamic sludge volume with overflow-weir effluent.
        self._dynamic_volume = bool(dynamic_volume)
        self._V_liq_max = float(V_liq)
        self._initial_fill_fraction = float(initial_fill_fraction)
        self._tau_out = float(outflow_time_constant)

        self.gas_storage: GasStorage = GasStorage(
            component_id=f"{self.component_id}_storage",
            storage_type="membrane",
            capacity_m3=max(50.0, float(self.V_gas)),
            p_min_bar=0.95,
            p_max_bar=1.05,
            initial_fill_fraction=0.1,
            name=f"{self.name} Gas Storage",
        )

        self.adm1_state: List[float] = []
        self.Q_substrates: List[float] = [0.0] * 10

        self.adm1 = ADM1(feedstock=feedstock, V_liq=V_liq, V_gas=V_gas, T_ad=T_ad)

    # ------------------------------------------------------------------
    # Component lifecycle
    # ------------------------------------------------------------------

    def initialize(self, initial_state: Optional[Dict[str, Any]] = None) -> None:
        """
        Initialize the digester state and attached gas storage.

        Accepted keys in *initial_state*:

        * ``adm1_state`` : list of 41 floats (full ADM1 state vector).
          When omitted and ``Q_substrates`` is supplied with a
          :class:`Feedstock`, a pre-inoculated steady-state vector at pH 7
          is computed automatically from the blended influent.
        * ``Q_substrates`` : list of substrate feed rates [m³/d].  When
          supplied with a :class:`Feedstock`, the underlying ADM1 solver's
          influent DataFrame and density are wired automatically.
        * ``gas_storage`` : dict forwarded to the attached gas storage.
        """
        if initial_state is None:
            initial_state = {}

        # --- Substrate feed: auto-wire the influent DataFrame/density ---
        Q_substrates = initial_state.get("Q_substrates")
        if Q_substrates is not None:
            self.Q_substrates = list(Q_substrates)

            from ...substrates.feedstock import Feedstock as _Feedstock

            if isinstance(self.feedstock, _Feedstock):
                self.adm1.set_influent_dataframe(self.feedstock.get_influent_dataframe(Q=self.Q_substrates))
                self.adm1.set_influent_density(self.feedstock.blended_density(self.Q_substrates))

        # --- Biological state vector ---
        if "adm1_state" in initial_state and initial_state["adm1_state"] is not None:
            self.adm1_state = list(initial_state["adm1_state"])
        elif Q_substrates is not None:
            from ...substrates.feedstock import Feedstock as _Feedstock

            if isinstance(self.feedstock, _Feedstock):
                self.adm1_state = self._build_pre_inoculated_state(self.Q_substrates)
            else:
                self.adm1_state = [0.01] * _ADM1_STATE_SIZE
        else:
            self.adm1_state = [0.01] * _ADM1_STATE_SIZE

        # When dynamic volume is enabled, initialise V_liq below the
        # geometric maximum so the digester fills up over the first transient.
        if self._dynamic_volume:
            V_liq_init = max(1.0e-6, self._V_liq_max * self._initial_fill_fraction)
            self.V_liq = V_liq_init
            self.adm1.V_liq = V_liq_init

        self.state = {
            "adm1_state": self.adm1_state,
            "Q_substrates": self.Q_substrates,
            "Q_gas": 0.0,
            "Q_ch4": 0.0,
            "Q_co2": 0.0,
            "pH": 7.0,
            "VFA": 0.0,
            "TAC": 0.0,
            # Filtered hydraulic retention time tracked via the first-order
            # lag dHRT/dt + HRT*(Q_in/V_S) = 1, so the reported value rises
            # from start-up toward V_S/Q_in with time constant HRT_ss itself.
            "HRT": 0.0,
            # Current sludge volume [m³]. Equals self.V_liq; kept in
            # self.state so it round-trips through to_dict / from_dict.
            "V_liq": self.V_liq,
        }

        gs_state = initial_state.get("gas_storage")
        try:
            self.gas_storage.initialize(gs_state)
        except Exception:
            self.gas_storage.initialize()

        self._initialized = True

    def _build_pre_inoculated_state(self, Q_substrates) -> list:
        """
        Build a pre-inoculated initial state from the feedstock blend.

        Particulate pools at retention-factor steady state; dissolved species
        for a healthy digester at pH 7; biomass concentrations above the
        bistability washout threshold so methanogenesis is active from t = 0.
        """
        fs = self.feedstock
        conc = fs.blended_concentrations(Q_substrates)
        Q_total = float(np.sum(fs.actual_Q(Q_substrates))) if hasattr(fs, "actual_Q") else float(np.sum(Q_substrates))
        V_liq = float(self.V_liq)
        T_ad = float(self.T_ad)

        # --- Acid-base constants (temperature-corrected to T_ad) ---
        R_gas = 0.08314
        T_base = 298.15
        K_a_va = 10.0**-4.86
        K_a_bu = 10.0**-4.82
        K_a_pro = 10.0**-4.88
        K_a_ac = 10.0**-4.76
        K_a_co2 = 10.0**-6.35
        K_a_IN = 10.0**-9.25
        K_w = 1.0e-14 * np.exp((55900.0 / (100.0 * R_gas)) * (1.0 / T_base - 1.0 / T_ad))

        # --- Target pH 7 and typical healthy-digester dissolved concentrations ---
        S_H_0 = 10.0**-7.0
        S_ac_0, S_pro_0, S_bu_0, S_va_0 = 0.10, 0.02, 0.01, 0.01
        S_co2_0 = 0.18
        S_nh4_0 = conc.get("S_nh4", 0.0) * 1.5

        S_ac_ion_0 = K_a_ac / (K_a_ac + S_H_0) * S_ac_0
        S_pro_ion_0 = K_a_pro / (K_a_pro + S_H_0) * S_pro_0
        S_bu_ion_0 = K_a_bu / (K_a_bu + S_H_0) * S_bu_0
        S_va_ion_0 = K_a_va / (K_a_va + S_H_0) * S_va_0
        S_nh3_0 = K_a_IN / (K_a_IN + S_H_0) * S_nh4_0
        S_hco3_0 = K_a_co2 * S_co2_0 / (S_H_0 + K_a_co2)

        vfa_kmol_0 = S_ac_ion_0 / 64.0 + S_pro_ion_0 / 112.0 + S_bu_ion_0 / 160.0 + S_va_ion_0 / 208.0

        S_anion_0 = conc.get("S_anion", 0.0)
        S_cation_0 = S_anion_0 + S_hco3_0 + vfa_kmol_0 + K_w / S_H_0 - S_nh4_0 + S_nh3_0 - S_H_0

        # --- Particulate pools at retention-factor steady state ---
        D = Q_total / V_liq if V_liq > 0.0 else 0.0
        dT = T_ad - 308.15
        k_dis_PF = 0.4 * (1.035**dT)
        k_dis_PS = 0.04 * (1.035**dT)
        k_hyd = 4.0 * (1.07**dT)

        ret_PF = D / (D + k_dis_PF) if (D + k_dis_PF) > 0.0 else 0.0
        ret_PS = D / (D + k_dis_PS) if (D + k_dis_PS) > 0.0 else 0.0

        X_PF_ch_ss = conc.get("X_PF_ch", 0.0) * ret_PF
        X_PF_pr_ss = conc.get("X_PF_pr", 0.0) * ret_PF
        X_PF_li_ss = conc.get("X_PF_li", 0.0) * ret_PF
        X_PS_ch_ss = conc.get("X_PS_ch", 0.0) * ret_PS
        X_PS_pr_ss = conc.get("X_PS_pr", 0.0) * ret_PS
        X_PS_li_ss = conc.get("X_PS_li", 0.0) * ret_PS

        denom_hyd = D + k_hyd
        X_S_ch_0 = (k_dis_PF * X_PF_ch_ss + k_dis_PS * X_PS_ch_ss) / denom_hyd
        X_S_pr_0 = (k_dis_PF * X_PF_pr_ss + k_dis_PS * X_PS_pr_ss) / denom_hyd
        X_S_li_0 = (k_dis_PF * X_PF_li_ss + k_dis_PS * X_PS_li_ss) / denom_hyd

        X_I_0 = conc.get("X_I", 0.0)

        p_h2_0, p_ch4_0, p_co2_0 = 1.02e-5, 0.65, 0.33
        p_tot_0 = p_h2_0 + p_ch4_0 + p_co2_0

        return [
            0.01,
            0.001,
            0.05,
            S_va_0,
            S_bu_0,
            S_pro_0,
            S_ac_0,
            1.0e-7,
            1.0e-4,
            S_co2_0,
            S_nh4_0,
            0.0,
            X_PS_ch_ss,
            X_PS_pr_ss,
            X_PS_li_ss,
            X_PF_ch_ss,
            X_PF_pr_ss,
            X_PF_li_ss,
            X_S_ch_0,
            X_S_pr_0,
            X_S_li_0,
            X_I_0,
            0.50,
            0.50,
            0.30,
            0.40,
            0.30,
            1.20,
            0.30,
            S_cation_0,
            S_anion_0,
            S_va_ion_0,
            S_bu_ion_0,
            S_pro_ion_0,
            S_ac_ion_0,
            S_hco3_0,
            S_nh3_0,
            p_h2_0,
            p_ch4_0,
            p_co2_0,
            p_tot_0,
        ]

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def _has_valid_state_input(self) -> bool:
        """Return True if the ADM1 model has both a state input and a non-empty Q array."""
        state_input = getattr(self.adm1, "_state_input", None)
        q_array = getattr(self.adm1, "_Q", None)
        return state_input is not None and q_array is not None and len(q_array) > 0

    def _compute_indicators(self) -> Dict[str, float]:
        """Compute pH, VFA, and TAC from the current state (Schlattmann 2011)."""
        st = self.adm1_state
        inhib = ADMParams.get_inhibition_params(self.adm1._R, self.adm1._T_base, self.adm1._T_ad)
        try:
            S_H = ADM1._calc_ph(
                st[_IDX_S_NH4],
                st[_IDX_S_NH3],
                st[_IDX_S_HCO3],
                st[_IDX_S_AC_ION],
                st[_IDX_S_PRO_ION],
                st[_IDX_S_BU_ION],
                st[_IDX_S_VA_ION],
                st[_IDX_S_CATION],
                st[_IDX_S_ANION],
                inhib["K_w"],
            )
            pH = -np.log10(max(float(S_H), 1.0e-14))
        except Exception:
            pH = 7.0

        # VFA [kg HAc-eq/m³] (Schlattmann 2011)
        M_HAc = 60.0
        COD_AC, COD_PRO, COD_BU, COD_VA = 64.0, 112.0, 160.0, 208.0
        vfa = M_HAc * (
            float(st[_IDX_S_AC]) / COD_AC
            + float(st[_IDX_S_PRO]) / COD_PRO
            + float(st[_IDX_S_BU]) / COD_BU
            + float(st[_IDX_S_VA]) / COD_VA
        )

        # TAC [kg CaCO3/m³] (titration endpoint pH 5; Schlattmann 2011)
        H_pH5 = 1.0e-5
        a_nh4 = inhib["K_a_IN"] / (H_pH5 + inhib["K_a_IN"])
        a_co2 = inhib["K_a_co2"] / (H_pH5 + inhib["K_a_co2"])
        a_ac = inhib["K_a_ac"] / (H_pH5 + inhib["K_a_ac"])
        a_pro = inhib["K_a_pro"] / (H_pH5 + inhib["K_a_pro"])
        a_bu = inhib["K_a_bu"] / (H_pH5 + inhib["K_a_bu"])
        a_va = inhib["K_a_va"] / (H_pH5 + inhib["K_a_va"])

        total_N = float(st[_IDX_S_NH4]) + float(st[_IDX_S_NH3])
        total_IC = float(st[_IDX_S_CO2])
        total_ac_mol = float(st[_IDX_S_AC]) / COD_AC
        total_pro_mol = float(st[_IDX_S_PRO]) / COD_PRO
        total_bu_mol = float(st[_IDX_S_BU]) / COD_BU
        total_va_mol = float(st[_IDX_S_VA]) / COD_VA
        ion_ac_mol = float(st[_IDX_S_AC_ION]) / COD_AC
        ion_pro_mol = float(st[_IDX_S_PRO_ION]) / COD_PRO
        ion_bu_mol = float(st[_IDX_S_BU_ION]) / COD_BU
        ion_va_mol = float(st[_IDX_S_VA_ION]) / COD_VA

        tac_mol = (
            (float(st[_IDX_S_NH3]) - a_nh4 * total_N)
            + (float(st[_IDX_S_HCO3]) - a_co2 * total_IC)
            + (ion_ac_mol - a_ac * total_ac_mol)
            + (ion_pro_mol - a_pro * total_pro_mol)
            + (ion_bu_mol - a_bu * total_bu_mol)
            + (ion_va_mol - a_va * total_va_mol)
            + float(st[_IDX_S_ANION])
            - float(st[_IDX_S_CATION])
        )
        tac = 50.0 * tac_mol
        return {"pH": pH, "VFA": vfa, "TAC": tac}

    # ------------------------------------------------------------------
    # Step
    # ------------------------------------------------------------------

    def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """
        Integrate the ADM1 ODE by *dt* days and return outputs.

        Inputs may include:
          - ``Q_substrates`` : fresh substrate feed rates [m³/d]
          - ``Q_in``         : effluent flow from an upstream digester [m³/d]
          - ``state_in``     : 41-element state from upstream digester

        Returns a dict with ``Q_out``, ``state_out``, ``Q_gas``, ``Q_ch4``,
        ``Q_co2``, ``pH``, ``VFA``, ``TAC``, ``gas_storage``.
        """
        if "Q_substrates" in inputs:
            self.Q_substrates = inputs["Q_substrates"]

        if "state_in" in inputs and "Q_in" in inputs:
            Q_in = float(inputs["Q_in"])
            state_in = inputs["state_in"]

            self.adm1.create_influent(self.Q_substrates, int(t / dt))
            Q_sub = float(np.sum(self.adm1._Q)) if self.adm1._Q is not None else 0.0
            Q_total = Q_in + Q_sub

            if Q_total > 0 and len(state_in) >= _N_LIQUID_STATE:
                for idx in range(_N_LIQUID_STATE):
                    c_sub = float(self.adm1._state_input[idx])
                    c_in = float(state_in[idx])
                    self.adm1._state_input[idx] = (c_sub * Q_sub + c_in * Q_in) / Q_total
                # The ODE reads the total volumetric flow from ``_Q`` (sum), so
                # collapse the per-substrate breakdown into a single mixed flow.
                self.adm1._Q = [Q_total]
        else:
            self.adm1.create_influent(self.Q_substrates, int(t / dt))

        # When dynamic volume is enabled, drive Q_out from an overflow weir
        # at V_max instead of letting the ODE enforce volume conservation.
        # V_liq is treated as constant within one ODE step and advanced
        # explicitly afterwards using the cached q_S_loss.
        if self._dynamic_volume:
            V_prev = float(self.adm1.V_liq)
            Q_out_weir = max(0.0, (V_prev - self._V_liq_max) / self._tau_out)
            self.adm1._Q_out_override = Q_out_weir
        else:
            Q_out_weir = None
            self.adm1._Q_out_override = None

        result = solve_ivp(
            fun=self.adm1.ADM_ODE,
            t_span=(t, t + dt),
            y0=self.adm1_state,
            method="BDF",
            rtol=1.0e-6,
            atol=1.0e-8,
            max_step=max(0.1 * dt, 1.0e-3),
        )
        if not result.success:
            raise RuntimeError(f"ADM1 integration failed in '{self.component_id}': {result.message}")
        self.adm1_state = list(result.y[:, -1])

        # Advance the sludge-volume mass balance now that the ODE has
        # settled and q_S_loss has been cached on the model.
        if self._dynamic_volume:
            Q_in_for_V = float(np.sum(self.adm1._Q)) if self.adm1._Q is not None else 0.0
            q_loss = float(self.adm1._q_S_loss_last)
            V_prev = float(self.adm1.V_liq)
            V_new = max(1.0e-6, V_prev + (Q_in_for_V - Q_out_weir - q_loss) * dt)
            self.adm1.V_liq = V_new
            self.V_liq = V_new
            self.state["V_liq"] = V_new

        q_gas, q_ch4, q_co2, _, _ = self.adm1.calc_gas(
            self.adm1_state[_IDX_P_H2],
            self.adm1_state[_IDX_P_CH4],
            self.adm1_state[_IDX_P_CO2],
            self.adm1_state[_IDX_P_TOTAL],
        )

        indicators = self._compute_indicators()
        self.state["adm1_state"] = self.adm1_state
        self.state["Q_gas"] = q_gas
        self.state["Q_ch4"] = q_ch4
        self.state["Q_co2"] = q_co2
        self.state.update(indicators)

        if self._dynamic_volume:
            # Effluent leaving the reactor over this step (overflow weir).
            Q_out = float(Q_out_weir)
        else:
            # Legacy: Q_out tracks Q_in (volume-conservation enforced by the ODE).
            Q_out = float(np.sum(self.adm1._Q)) if self._has_valid_state_input() else float(np.sum(self.Q_substrates))

        # First-order lag for the reported HRT: dHRT/dt + HRT*(Q_in/V_S) = 1.
        # Exact discrete update over dt with piecewise-constant a = Q_in/V_S.
        Q_in_total = float(np.sum(self.adm1._Q)) if self.adm1._Q is not None else 0.0
        hrt_prev = float(self.state.get("HRT", 0.0))
        if Q_in_total > 0.0 and self.V_liq > 0.0:
            a = Q_in_total / self.V_liq
            decay = float(np.exp(-a * dt))
            hrt_new = hrt_prev * decay + (1.0 / a) * (1.0 - decay)
        else:
            hrt_new = hrt_prev + dt
        self.state["HRT"] = hrt_new

        gs_outputs = self.gas_storage.step(
            t=t,
            dt=dt,
            inputs={
                "Q_gas_in_m3_per_day": float(q_gas),
                "Q_gas_out_m3_per_day": 0.0,
                "vent_to_flare": True,
            },
        )
        self.gas_storage.outputs_data["Q_gas_in_m3_per_day"] = float(q_gas)

        self.outputs_data = {
            "Q_out": Q_out,
            "state_out": self.adm1_state,
            "Q_gas": float(q_gas),
            "Q_ch4": float(q_ch4),
            "Q_co2": float(q_co2),
            "pH": indicators["pH"],
            "VFA": indicators["VFA"],
            "TAC": indicators["TAC"],
            "HRT": float(hrt_new),
            "V_liq": float(self.V_liq),
            "Q_gas_to_storage_m3_per_day": float(q_gas),
            "gas_storage": {
                "component_id": self.gas_storage.component_id,
                "stored_volume_m3": gs_outputs["stored_volume_m3"],
                "pressure_bar": gs_outputs["pressure_bar"],
                "vented_volume_m3": gs_outputs["vented_volume_m3"],
                "Q_gas_supplied_m3_per_day": gs_outputs["Q_gas_supplied_m3_per_day"],
            },
        }
        return self.outputs_data

    # ------------------------------------------------------------------
    # Calibration parameter API (delegates to the underlying ADM1 instance)
    # ------------------------------------------------------------------

    def apply_calibration_parameters(self, parameters: dict) -> None:
        """Apply calibration overrides to the underlying ADM1 model."""
        self.adm1.set_calibration_parameters(parameters)

    def get_calibration_parameters(self) -> dict:
        """Return currently applied calibration parameters."""
        return self.adm1.get_calibration_parameters()

    def clear_calibration_parameters(self) -> None:
        """Clear all calibration overrides."""
        self.adm1.clear_calibration_parameters()

    # ------------------------------------------------------------------
    # Serialization
    # ------------------------------------------------------------------

    def to_dict(self) -> Dict[str, Any]:
        """Serialize the digester to a configuration dictionary."""
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            # Persist the geometric maximum so a round-trip rebuild gets the
            # same tank size. The current dynamic V lives in state["V_liq"].
            "V_liq": self._V_liq_max,
            "V_gas": self.V_gas,
            "T_ad": self.T_ad,
            "dynamic_volume": self._dynamic_volume,
            "initial_fill_fraction": self._initial_fill_fraction,
            "outflow_time_constant": self._tau_out,
            "inputs": self.inputs,
            "outputs": self.outputs,
            "state": self.state,
        }

    @classmethod
    def from_dict(cls, config: Dict[str, Any], feedstock=None) -> "Digester":
        """Reconstruct a Digester from a configuration dictionary."""
        digester = cls(
            component_id=config["component_id"],
            feedstock=feedstock,
            V_liq=config.get("V_liq", 1977.0),
            V_gas=config.get("V_gas", 304.0),
            T_ad=config.get("T_ad", 308.15),
            name=config.get("name"),
            dynamic_volume=config.get("dynamic_volume", False),
            initial_fill_fraction=config.get("initial_fill_fraction", 1.0),
            outflow_time_constant=config.get("outflow_time_constant", 1.0),
        )
        digester.inputs = config.get("inputs", [])
        digester.outputs = config.get("outputs", [])
        if "state" in config:
            digester.initialize(config["state"])
            # Restore the live sludge volume after initialize() applied the
            # nominal initial_fill_fraction.
            if digester._dynamic_volume and "V_liq" in config["state"]:
                V_curr = float(config["state"]["V_liq"])
                digester.V_liq = V_curr
                digester.adm1.V_liq = V_curr
                digester.state["V_liq"] = V_curr
        return digester

Functions

apply_calibration_parameters(parameters)

Apply calibration overrides to the underlying ADM1 model.

Source code in pyadm1/components/biological/digester.py

def apply_calibration_parameters(self, parameters: dict) -> None:
    """Apply calibration overrides to the underlying ADM1 model."""
    self.adm1.set_calibration_parameters(parameters)

clear_calibration_parameters()

Clear all calibration overrides.

Source code in pyadm1/components/biological/digester.py

def clear_calibration_parameters(self) -> None:
    """Clear all calibration overrides."""
    self.adm1.clear_calibration_parameters()

from_dict(config, feedstock=None) classmethod

Reconstruct a Digester from a configuration dictionary.

Source code in pyadm1/components/biological/digester.py

@classmethod
def from_dict(cls, config: Dict[str, Any], feedstock=None) -> "Digester":
    """Reconstruct a Digester from a configuration dictionary."""
    digester = cls(
        component_id=config["component_id"],
        feedstock=feedstock,
        V_liq=config.get("V_liq", 1977.0),
        V_gas=config.get("V_gas", 304.0),
        T_ad=config.get("T_ad", 308.15),
        name=config.get("name"),
        dynamic_volume=config.get("dynamic_volume", False),
        initial_fill_fraction=config.get("initial_fill_fraction", 1.0),
        outflow_time_constant=config.get("outflow_time_constant", 1.0),
    )
    digester.inputs = config.get("inputs", [])
    digester.outputs = config.get("outputs", [])
    if "state" in config:
        digester.initialize(config["state"])
        # Restore the live sludge volume after initialize() applied the
        # nominal initial_fill_fraction.
        if digester._dynamic_volume and "V_liq" in config["state"]:
            V_curr = float(config["state"]["V_liq"])
            digester.V_liq = V_curr
            digester.adm1.V_liq = V_curr
            digester.state["V_liq"] = V_curr
    return digester

get_calibration_parameters()

Return currently applied calibration parameters.

Source code in pyadm1/components/biological/digester.py

def get_calibration_parameters(self) -> dict:
    """Return currently applied calibration parameters."""
    return self.adm1.get_calibration_parameters()

initialize(initial_state=None)

Initialize the digester state and attached gas storage.

Accepted keys in initial_state:

• adm1_state : list of 41 floats (full ADM1 state vector). When omitted and Q_substrates is supplied with a :class:Feedstock, a pre-inoculated steady-state vector at pH 7 is computed automatically from the blended influent.
• Q_substrates : list of substrate feed rates [m³/d]. When supplied with a :class:Feedstock, the underlying ADM1 solver's influent DataFrame and density are wired automatically.
• gas_storage : dict forwarded to the attached gas storage.

Source code in pyadm1/components/biological/digester.py

def initialize(self, initial_state: Optional[Dict[str, Any]] = None) -> None:
    """
    Initialize the digester state and attached gas storage.

    Accepted keys in *initial_state*:

    * ``adm1_state`` : list of 41 floats (full ADM1 state vector).
      When omitted and ``Q_substrates`` is supplied with a
      :class:`Feedstock`, a pre-inoculated steady-state vector at pH 7
      is computed automatically from the blended influent.
    * ``Q_substrates`` : list of substrate feed rates [m³/d].  When
      supplied with a :class:`Feedstock`, the underlying ADM1 solver's
      influent DataFrame and density are wired automatically.
    * ``gas_storage`` : dict forwarded to the attached gas storage.
    """
    if initial_state is None:
        initial_state = {}

    # --- Substrate feed: auto-wire the influent DataFrame/density ---
    Q_substrates = initial_state.get("Q_substrates")
    if Q_substrates is not None:
        self.Q_substrates = list(Q_substrates)

        from ...substrates.feedstock import Feedstock as _Feedstock

        if isinstance(self.feedstock, _Feedstock):
            self.adm1.set_influent_dataframe(self.feedstock.get_influent_dataframe(Q=self.Q_substrates))
            self.adm1.set_influent_density(self.feedstock.blended_density(self.Q_substrates))

    # --- Biological state vector ---
    if "adm1_state" in initial_state and initial_state["adm1_state"] is not None:
        self.adm1_state = list(initial_state["adm1_state"])
    elif Q_substrates is not None:
        from ...substrates.feedstock import Feedstock as _Feedstock

        if isinstance(self.feedstock, _Feedstock):
            self.adm1_state = self._build_pre_inoculated_state(self.Q_substrates)
        else:
            self.adm1_state = [0.01] * _ADM1_STATE_SIZE
    else:
        self.adm1_state = [0.01] * _ADM1_STATE_SIZE

    # When dynamic volume is enabled, initialise V_liq below the
    # geometric maximum so the digester fills up over the first transient.
    if self._dynamic_volume:
        V_liq_init = max(1.0e-6, self._V_liq_max * self._initial_fill_fraction)
        self.V_liq = V_liq_init
        self.adm1.V_liq = V_liq_init

    self.state = {
        "adm1_state": self.adm1_state,
        "Q_substrates": self.Q_substrates,
        "Q_gas": 0.0,
        "Q_ch4": 0.0,
        "Q_co2": 0.0,
        "pH": 7.0,
        "VFA": 0.0,
        "TAC": 0.0,
        # Filtered hydraulic retention time tracked via the first-order
        # lag dHRT/dt + HRT*(Q_in/V_S) = 1, so the reported value rises
        # from start-up toward V_S/Q_in with time constant HRT_ss itself.
        "HRT": 0.0,
        # Current sludge volume [m³]. Equals self.V_liq; kept in
        # self.state so it round-trips through to_dict / from_dict.
        "V_liq": self.V_liq,
    }

    gs_state = initial_state.get("gas_storage")
    try:
        self.gas_storage.initialize(gs_state)
    except Exception:
        self.gas_storage.initialize()

    self._initialized = True

step(t, dt, inputs)

Integrate the ADM1 ODE by dt days and return outputs.

Inputs may include

• Q_substrates : fresh substrate feed rates [m³/d]
• Q_in : effluent flow from an upstream digester [m³/d]
• state_in : 41-element state from upstream digester

Returns a dict with Q_out, state_out, Q_gas, Q_ch4, Q_co2, pH, VFA, TAC, gas_storage.

Source code in pyadm1/components/biological/digester.py

def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
    """
    Integrate the ADM1 ODE by *dt* days and return outputs.

    Inputs may include:
      - ``Q_substrates`` : fresh substrate feed rates [m³/d]
      - ``Q_in``         : effluent flow from an upstream digester [m³/d]
      - ``state_in``     : 41-element state from upstream digester

    Returns a dict with ``Q_out``, ``state_out``, ``Q_gas``, ``Q_ch4``,
    ``Q_co2``, ``pH``, ``VFA``, ``TAC``, ``gas_storage``.
    """
    if "Q_substrates" in inputs:
        self.Q_substrates = inputs["Q_substrates"]

    if "state_in" in inputs and "Q_in" in inputs:
        Q_in = float(inputs["Q_in"])
        state_in = inputs["state_in"]

        self.adm1.create_influent(self.Q_substrates, int(t / dt))
        Q_sub = float(np.sum(self.adm1._Q)) if self.adm1._Q is not None else 0.0
        Q_total = Q_in + Q_sub

        if Q_total > 0 and len(state_in) >= _N_LIQUID_STATE:
            for idx in range(_N_LIQUID_STATE):
                c_sub = float(self.adm1._state_input[idx])
                c_in = float(state_in[idx])
                self.adm1._state_input[idx] = (c_sub * Q_sub + c_in * Q_in) / Q_total
            # The ODE reads the total volumetric flow from ``_Q`` (sum), so
            # collapse the per-substrate breakdown into a single mixed flow.
            self.adm1._Q = [Q_total]
    else:
        self.adm1.create_influent(self.Q_substrates, int(t / dt))

    # When dynamic volume is enabled, drive Q_out from an overflow weir
    # at V_max instead of letting the ODE enforce volume conservation.
    # V_liq is treated as constant within one ODE step and advanced
    # explicitly afterwards using the cached q_S_loss.
    if self._dynamic_volume:
        V_prev = float(self.adm1.V_liq)
        Q_out_weir = max(0.0, (V_prev - self._V_liq_max) / self._tau_out)
        self.adm1._Q_out_override = Q_out_weir
    else:
        Q_out_weir = None
        self.adm1._Q_out_override = None

    result = solve_ivp(
        fun=self.adm1.ADM_ODE,
        t_span=(t, t + dt),
        y0=self.adm1_state,
        method="BDF",
        rtol=1.0e-6,
        atol=1.0e-8,
        max_step=max(0.1 * dt, 1.0e-3),
    )
    if not result.success:
        raise RuntimeError(f"ADM1 integration failed in '{self.component_id}': {result.message}")
    self.adm1_state = list(result.y[:, -1])

    # Advance the sludge-volume mass balance now that the ODE has
    # settled and q_S_loss has been cached on the model.
    if self._dynamic_volume:
        Q_in_for_V = float(np.sum(self.adm1._Q)) if self.adm1._Q is not None else 0.0
        q_loss = float(self.adm1._q_S_loss_last)
        V_prev = float(self.adm1.V_liq)
        V_new = max(1.0e-6, V_prev + (Q_in_for_V - Q_out_weir - q_loss) * dt)
        self.adm1.V_liq = V_new
        self.V_liq = V_new
        self.state["V_liq"] = V_new

    q_gas, q_ch4, q_co2, _, _ = self.adm1.calc_gas(
        self.adm1_state[_IDX_P_H2],
        self.adm1_state[_IDX_P_CH4],
        self.adm1_state[_IDX_P_CO2],
        self.adm1_state[_IDX_P_TOTAL],
    )

    indicators = self._compute_indicators()
    self.state["adm1_state"] = self.adm1_state
    self.state["Q_gas"] = q_gas
    self.state["Q_ch4"] = q_ch4
    self.state["Q_co2"] = q_co2
    self.state.update(indicators)

    if self._dynamic_volume:
        # Effluent leaving the reactor over this step (overflow weir).
        Q_out = float(Q_out_weir)
    else:
        # Legacy: Q_out tracks Q_in (volume-conservation enforced by the ODE).
        Q_out = float(np.sum(self.adm1._Q)) if self._has_valid_state_input() else float(np.sum(self.Q_substrates))

    # First-order lag for the reported HRT: dHRT/dt + HRT*(Q_in/V_S) = 1.
    # Exact discrete update over dt with piecewise-constant a = Q_in/V_S.
    Q_in_total = float(np.sum(self.adm1._Q)) if self.adm1._Q is not None else 0.0
    hrt_prev = float(self.state.get("HRT", 0.0))
    if Q_in_total > 0.0 and self.V_liq > 0.0:
        a = Q_in_total / self.V_liq
        decay = float(np.exp(-a * dt))
        hrt_new = hrt_prev * decay + (1.0 / a) * (1.0 - decay)
    else:
        hrt_new = hrt_prev + dt
    self.state["HRT"] = hrt_new

    gs_outputs = self.gas_storage.step(
        t=t,
        dt=dt,
        inputs={
            "Q_gas_in_m3_per_day": float(q_gas),
            "Q_gas_out_m3_per_day": 0.0,
            "vent_to_flare": True,
        },
    )
    self.gas_storage.outputs_data["Q_gas_in_m3_per_day"] = float(q_gas)

    self.outputs_data = {
        "Q_out": Q_out,
        "state_out": self.adm1_state,
        "Q_gas": float(q_gas),
        "Q_ch4": float(q_ch4),
        "Q_co2": float(q_co2),
        "pH": indicators["pH"],
        "VFA": indicators["VFA"],
        "TAC": indicators["TAC"],
        "HRT": float(hrt_new),
        "V_liq": float(self.V_liq),
        "Q_gas_to_storage_m3_per_day": float(q_gas),
        "gas_storage": {
            "component_id": self.gas_storage.component_id,
            "stored_volume_m3": gs_outputs["stored_volume_m3"],
            "pressure_bar": gs_outputs["pressure_bar"],
            "vented_volume_m3": gs_outputs["vented_volume_m3"],
            "Q_gas_supplied_m3_per_day": gs_outputs["Q_gas_supplied_m3_per_day"],
        },
    }
    return self.outputs_data

to_dict()

Serialize the digester to a configuration dictionary.

Source code in pyadm1/components/biological/digester.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize the digester to a configuration dictionary."""
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        # Persist the geometric maximum so a round-trip rebuild gets the
        # same tank size. The current dynamic V lives in state["V_liq"].
        "V_liq": self._V_liq_max,
        "V_gas": self.V_gas,
        "T_ad": self.T_ad,
        "dynamic_volume": self._dynamic_volume,
        "initial_fill_fraction": self._initial_fill_fraction,
        "outflow_time_constant": self._tau_out,
        "inputs": self.inputs,
        "outputs": self.outputs,
        "state": self.state,
    }

pyadm1.components.biological.separator.Separator

Bases: Component

Solid-liquid separator for digestate processing.

Splits the effluent flow from a digester into a solid fraction (press cake) and a liquid fraction (separated effluent) using a mass-balance model. Nutrient (N, P) partitioning between fractions is included.

Attributes:

Name	Type	Description
separator_type		Mechanical type (SeparatorType enum).
separation_efficiency		Fraction of total solids going to solid phase (0-1).
ts_solid_target		Target TS concentration in solid fraction [kg/m3].
n_to_solid		Fraction of input nitrogen transferred to solid (0-1).
p_to_solid		Fraction of input phosphorus transferred to solid (0-1).
specific_energy		Power demand per tonne of fresh-mass input [kWh/t FM].
fluid_density		Digestate density [kg/m3].

Source code in pyadm1/components/biological/separator.py

class Separator(Component):
    """
    Solid-liquid separator for digestate processing.

    Splits the effluent flow from a digester into a solid fraction (press
    cake) and a liquid fraction (separated effluent) using a mass-balance
    model.  Nutrient (N, P) partitioning between fractions is included.

    Attributes:
        separator_type:       Mechanical type (SeparatorType enum).
        separation_efficiency: Fraction of total solids going to solid phase (0-1).
        ts_solid_target:      Target TS concentration in solid fraction [kg/m3].
        n_to_solid:           Fraction of input nitrogen transferred to solid (0-1).
        p_to_solid:           Fraction of input phosphorus transferred to solid (0-1).
        specific_energy:      Power demand per tonne of fresh-mass input [kWh/t FM].
        fluid_density:        Digestate density [kg/m3].
    """

    def __init__(
        self,
        component_id: str,
        separator_type: str = "screw_press",
        separation_efficiency: Optional[float] = None,
        ts_solid_target: Optional[float] = None,
        n_to_solid: Optional[float] = None,
        p_to_solid: Optional[float] = None,
        specific_energy: Optional[float] = None,
        fluid_density: float = _DIGESTATE_DENSITY,
        name: Optional[str] = None,
    ):
        """
        Initialize separator component.

        Args:
            component_id:         Unique identifier.
            separator_type:       One of "screw_press", "decanter",
                                  "belt_press", "vibrating_screen".
            separation_efficiency: Override default solid-capture efficiency (0-1).
            ts_solid_target:      Override default solid-fraction TS [kg/m3].
            n_to_solid:           Override default N fraction to solid (0-1).
            p_to_solid:           Override default P fraction to solid (0-1).
            specific_energy:      Override default power demand [kWh/t FM].
            fluid_density:        Digestate density [kg/m3]. Default 1020.
            name:                 Human-readable display name.
        """
        super().__init__(component_id, ComponentType.SEPARATOR, name)

        self.separator_type = SeparatorType(separator_type.lower())
        defaults = _SEPARATOR_DEFAULTS[self.separator_type.value]

        self.separation_efficiency = (
            separation_efficiency if separation_efficiency is not None else defaults["separation_efficiency"]
        )
        self.ts_solid_target = ts_solid_target if ts_solid_target is not None else defaults["ts_solid_target"]
        self.n_to_solid = n_to_solid if n_to_solid is not None else defaults["n_to_solid"]
        self.p_to_solid = p_to_solid if p_to_solid is not None else defaults["p_to_solid"]
        self.specific_energy = specific_energy if specific_energy is not None else defaults["specific_energy"]
        self.fluid_density = float(fluid_density)

        # Cumulative tracking
        self.total_solid_mass = 0.0  # kg
        self.total_liquid_vol = 0.0  # m3
        self.energy_consumed = 0.0  # kWh

        self.initialize()

    # ------------------------------------------------------------------
    # Component interface
    # ------------------------------------------------------------------

    def initialize(self, initial_state: Optional[Dict[str, Any]] = None) -> None:
        """
        Initialize separator state.

        Args:
            initial_state: Optional dict with keys:
                - 'total_solid_mass': cumulative solid output [kg]
                - 'total_liquid_vol': cumulative liquid output [m3]
                - 'energy_consumed':  cumulative energy use [kWh]
        """
        if initial_state:
            self.total_solid_mass = float(initial_state.get("total_solid_mass", 0.0))
            self.total_liquid_vol = float(initial_state.get("total_liquid_vol", 0.0))
            self.energy_consumed = float(initial_state.get("energy_consumed", 0.0))

        self.state = {
            "total_solid_mass": self.total_solid_mass,
            "total_liquid_vol": self.total_liquid_vol,
            "energy_consumed": self.energy_consumed,
        }

        self.outputs_data = {
            "Q_liquid": 0.0,
            "Q_solid": 0.0,
            "TS_liquid": 0.0,
            "TS_solid": self.ts_solid_target,
            "VS_liquid": 0.0,
            "VS_solid": 0.0,
            "TAN_liquid": 0.0,
            "TAN_solid": 0.0,
            "TP_liquid": 0.0,
            "TP_solid": 0.0,
            "P_consumed": 0.0,
            "separation_efficiency": self.separation_efficiency,
        }

        self._initialized = True

    def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """
        Perform one simulation time step.

        Args:
            t:   Current simulation time [days].
            dt:  Time step [days].
            inputs: Dict with keys:
                - 'Q_in'    [m3/d]    Influent flow (required, or from Q_out
                                       of connected Digester).
                - 'Q_out'   [m3/d]    Alias for Q_in (Digester output key).
                - 'TS_in'   [kg/m3]   Total solids concentration in influent.
                                       If omitted, estimated from ADM1 state.
                - 'VS_in'   [kg/m3]   Volatile solids (optional, default 0.85*TS).
                - 'TAN_in'  [kg/m3]   Total ammonium nitrogen (optional).
                - 'TP_in'   [kg/m3]   Total phosphorus (optional).
                - 'state_out' [list]  ADM1 state vector from upstream Digester.
                                       Used to estimate TS_in if not given.

        Returns:
            Dict with keys:
                - 'Q_liquid'   [m3/d]  Liquid fraction flow rate.
                - 'Q_solid'    [m3/d]  Solid fraction flow rate.
                - 'TS_liquid'  [kg/m3] TS in liquid fraction.
                - 'TS_solid'   [kg/m3] TS in solid fraction (= ts_solid_target).
                - 'VS_liquid'  [kg/m3] VS in liquid fraction.
                - 'VS_solid'   [kg/m3] VS in solid fraction.
                - 'TAN_liquid' [kg/m3] TAN in liquid fraction.
                - 'TAN_solid'  [kg/m3] TAN in solid fraction.
                - 'TP_liquid'  [kg/m3] Total P in liquid fraction.
                - 'TP_solid'   [kg/m3] Total P in solid fraction.
                - 'P_consumed' [kW]    Electrical power draw.
                - 'separation_efficiency' [-] Active efficiency value.
        """
        # --- resolve influent flow -----------------------------------------
        Q_in = float(inputs.get("Q_in", inputs.get("Q_out", 0.0)))

        if Q_in <= 0.0:
            return self.outputs_data  # no flow, no separation

        # --- resolve total solids ------------------------------------------
        TS_in = float(inputs.get("TS_in", 0.0))

        if TS_in <= 0.0:
            # Estimate from ADM1 state vector if available
            adm1_state = inputs.get("state_out")
            if adm1_state is not None:
                TS_in = self._estimate_ts_from_adm1(adm1_state)
            else:
                # Fallback: typical mesophilic digestate TS ~40 kg/m3 (4%)
                TS_in = 40.0

        VS_in = float(inputs.get("VS_in", TS_in * 0.75))  # ~75% VS/TS
        TAN_in = float(inputs.get("TAN_in", 0.0))
        TP_in = float(inputs.get("TP_in", 0.0))

        # --- mass balance --------------------------------------------------
        m_TS_total = Q_in * TS_in  # kg/d total solids in
        m_VS_total = Q_in * VS_in  # kg/d VS in

        # VS separation efficiency slightly higher than TS (VS is more
        # particulate than mineral ash fraction); use 1.1x factor, capped at 0.95
        vs_sep_eff = min(self.separation_efficiency * 1.1, 0.95)

        m_TS_solid = m_TS_total * self.separation_efficiency  # kg/d to solid
        m_VS_solid = m_VS_total * vs_sep_eff  # kg/d VS to solid

        m_TS_liquid = m_TS_total - m_TS_solid  # kg/d to liquid
        m_VS_liquid = m_VS_total - m_VS_solid  # kg/d VS to liquid

        # Volume of solid fraction: m_TS_solid = Q_solid * ts_solid_target
        Q_solid = m_TS_solid / max(self.ts_solid_target, 1.0)  # m3/d
        Q_liquid = max(Q_in - Q_solid, 0.0)  # m3/d

        # Concentrations in each fraction
        TS_liquid = m_TS_liquid / max(Q_liquid, 1e-9)  # kg/m3
        VS_liquid = m_VS_liquid / max(Q_liquid, 1e-9)  # kg/m3
        VS_solid = m_VS_solid / max(Q_solid, 1e-9)  # kg/m3

        # Nutrient partitioning
        TAN_solid_total = TAN_in * Q_in * self.n_to_solid  # kg/d
        TAN_liquid_total = TAN_in * Q_in * (1.0 - self.n_to_solid)

        TP_solid_total = TP_in * Q_in * self.p_to_solid  # kg/d
        TP_liquid_total = TP_in * Q_in * (1.0 - self.p_to_solid)

        TAN_solid_conc = TAN_solid_total / max(Q_solid, 1e-9)  # kg/m3
        TAN_liquid_conc = TAN_liquid_total / max(Q_liquid, 1e-9)  # kg/m3
        TP_solid_conc = TP_solid_total / max(Q_solid, 1e-9)  # kg/m3
        TP_liquid_conc = TP_liquid_total / max(Q_liquid, 1e-9)  # kg/m3

        # --- power consumption --------------------------------------------
        # P [kW] = specific_energy [kWh/t] * FM_rate [t/d] / 24 [h/d]
        FM_rate_t_per_day = Q_in * self.fluid_density / 1000.0  # t FM/d
        P_consumed = self.specific_energy * FM_rate_t_per_day / 24.0  # kW

        # --- cumulative accounting ----------------------------------------
        dt_hours = dt * 24.0
        self.total_solid_mass += m_TS_solid * dt  # kg
        self.total_liquid_vol += Q_liquid * dt  # m3
        self.energy_consumed += P_consumed * dt_hours  # kWh

        # --- update state and outputs ------------------------------------
        self.state.update(
            {
                "total_solid_mass": self.total_solid_mass,
                "total_liquid_vol": self.total_liquid_vol,
                "energy_consumed": self.energy_consumed,
            }
        )

        self.outputs_data = {
            "Q_liquid": float(Q_liquid),
            "Q_solid": float(Q_solid),
            "TS_liquid": float(TS_liquid),
            "TS_solid": float(self.ts_solid_target),
            "VS_liquid": float(VS_liquid),
            "VS_solid": float(VS_solid),
            "TAN_liquid": float(TAN_liquid_conc),
            "TAN_solid": float(TAN_solid_conc),
            "TP_liquid": float(TP_liquid_conc),
            "TP_solid": float(TP_solid_conc),
            "P_consumed": float(P_consumed),
            "separation_efficiency": float(self.separation_efficiency),
            # Convenience summary
            "solid_fraction_ts_pct": float(self.ts_solid_target / (self.fluid_density * 10.0)),  # % TS
            "recovery_solid_pct": float(self.separation_efficiency * 100.0),
        }

        return self.outputs_data

    # ------------------------------------------------------------------
    # Serialization
    # ------------------------------------------------------------------

    def to_dict(self) -> Dict[str, Any]:
        """Serialize separator configuration to dictionary."""
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "separator_type": self.separator_type.value,
            "separation_efficiency": self.separation_efficiency,
            "ts_solid_target": self.ts_solid_target,
            "n_to_solid": self.n_to_solid,
            "p_to_solid": self.p_to_solid,
            "specific_energy": self.specific_energy,
            "fluid_density": self.fluid_density,
            "inputs": self.inputs,
            "outputs": self.outputs,
            "state": self.state,
        }

    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> "Separator":
        """
        Create separator from dictionary.

        Args:
            config: Configuration dictionary (from to_dict()).

        Returns:
            Initialized Separator instance.
        """
        sep = cls(
            component_id=config["component_id"],
            separator_type=config.get("separator_type", "screw_press"),
            separation_efficiency=config.get("separation_efficiency"),
            ts_solid_target=config.get("ts_solid_target"),
            n_to_solid=config.get("n_to_solid"),
            p_to_solid=config.get("p_to_solid"),
            specific_energy=config.get("specific_energy"),
            fluid_density=config.get("fluid_density", _DIGESTATE_DENSITY),
            name=config.get("name"),
        )

        sep.inputs = config.get("inputs", [])
        sep.outputs = config.get("outputs", [])

        if "state" in config:
            sep.initialize(config["state"])

        return sep

    # ------------------------------------------------------------------
    # Private helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _estimate_ts_from_adm1(state: Any) -> float:
        """
        Estimate digestate TS from ADM1 state vector.

        Sums particulate COD components (indices 12-24, units kg COD/m3)
        and converts to kg TS/m3 using standard stoichiometric factors.

        Args:
            state: Sequence of 37 ADM1 state values [kg COD/m3 or kmol/m3].

        Returns:
            Estimated TS concentration [kg/m3].
        """
        try:
            particulate_cod = sum(float(state[i]) for i in _ADM1_PARTICULATE_INDICES if i < len(state))
            # Convert kg COD/m3 -> kg TS/m3
            return max(particulate_cod * _COD_TO_TS, 0.0)
        except (TypeError, IndexError):
            return 40.0  # fallback: typical digestate 4% TS

Functions

__init__(component_id, separator_type='screw_press', separation_efficiency=None, ts_solid_target=None, n_to_solid=None, p_to_solid=None, specific_energy=None, fluid_density=_DIGESTATE_DENSITY, name=None)

Initialize separator component.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier.	required
separator_type	str	One of "screw_press", "decanter", "belt_press", "vibrating_screen".	'screw_press'
separation_efficiency	Optional[float]	Override default solid-capture efficiency (0-1).	None
ts_solid_target	Optional[float]	Override default solid-fraction TS [kg/m3].	None
n_to_solid	Optional[float]	Override default N fraction to solid (0-1).	None
p_to_solid	Optional[float]	Override default P fraction to solid (0-1).	None
specific_energy	Optional[float]	Override default power demand [kWh/t FM].	None
fluid_density	float	Digestate density [kg/m3]. Default 1020.	_DIGESTATE_DENSITY
name	Optional[str]	Human-readable display name.	None

Source code in pyadm1/components/biological/separator.py

def __init__(
    self,
    component_id: str,
    separator_type: str = "screw_press",
    separation_efficiency: Optional[float] = None,
    ts_solid_target: Optional[float] = None,
    n_to_solid: Optional[float] = None,
    p_to_solid: Optional[float] = None,
    specific_energy: Optional[float] = None,
    fluid_density: float = _DIGESTATE_DENSITY,
    name: Optional[str] = None,
):
    """
    Initialize separator component.

    Args:
        component_id:         Unique identifier.
        separator_type:       One of "screw_press", "decanter",
                              "belt_press", "vibrating_screen".
        separation_efficiency: Override default solid-capture efficiency (0-1).
        ts_solid_target:      Override default solid-fraction TS [kg/m3].
        n_to_solid:           Override default N fraction to solid (0-1).
        p_to_solid:           Override default P fraction to solid (0-1).
        specific_energy:      Override default power demand [kWh/t FM].
        fluid_density:        Digestate density [kg/m3]. Default 1020.
        name:                 Human-readable display name.
    """
    super().__init__(component_id, ComponentType.SEPARATOR, name)

    self.separator_type = SeparatorType(separator_type.lower())
    defaults = _SEPARATOR_DEFAULTS[self.separator_type.value]

    self.separation_efficiency = (
        separation_efficiency if separation_efficiency is not None else defaults["separation_efficiency"]
    )
    self.ts_solid_target = ts_solid_target if ts_solid_target is not None else defaults["ts_solid_target"]
    self.n_to_solid = n_to_solid if n_to_solid is not None else defaults["n_to_solid"]
    self.p_to_solid = p_to_solid if p_to_solid is not None else defaults["p_to_solid"]
    self.specific_energy = specific_energy if specific_energy is not None else defaults["specific_energy"]
    self.fluid_density = float(fluid_density)

    # Cumulative tracking
    self.total_solid_mass = 0.0  # kg
    self.total_liquid_vol = 0.0  # m3
    self.energy_consumed = 0.0  # kWh

    self.initialize()

from_dict(config) classmethod

Create separator from dictionary.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Configuration dictionary (from to_dict()).	required

Returns:

Type	Description
Separator	Initialized Separator instance.

Source code in pyadm1/components/biological/separator.py

@classmethod
def from_dict(cls, config: Dict[str, Any]) -> "Separator":
    """
    Create separator from dictionary.

    Args:
        config: Configuration dictionary (from to_dict()).

    Returns:
        Initialized Separator instance.
    """
    sep = cls(
        component_id=config["component_id"],
        separator_type=config.get("separator_type", "screw_press"),
        separation_efficiency=config.get("separation_efficiency"),
        ts_solid_target=config.get("ts_solid_target"),
        n_to_solid=config.get("n_to_solid"),
        p_to_solid=config.get("p_to_solid"),
        specific_energy=config.get("specific_energy"),
        fluid_density=config.get("fluid_density", _DIGESTATE_DENSITY),
        name=config.get("name"),
    )

    sep.inputs = config.get("inputs", [])
    sep.outputs = config.get("outputs", [])

    if "state" in config:
        sep.initialize(config["state"])

    return sep

initialize(initial_state=None)

Initialize separator state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Optional dict with keys: - 'total_solid_mass': cumulative solid output [kg] - 'total_liquid_vol': cumulative liquid output [m3] - 'energy_consumed': cumulative energy use [kWh]	None

Source code in pyadm1/components/biological/separator.py

def initialize(self, initial_state: Optional[Dict[str, Any]] = None) -> None:
    """
    Initialize separator state.

    Args:
        initial_state: Optional dict with keys:
            - 'total_solid_mass': cumulative solid output [kg]
            - 'total_liquid_vol': cumulative liquid output [m3]
            - 'energy_consumed':  cumulative energy use [kWh]
    """
    if initial_state:
        self.total_solid_mass = float(initial_state.get("total_solid_mass", 0.0))
        self.total_liquid_vol = float(initial_state.get("total_liquid_vol", 0.0))
        self.energy_consumed = float(initial_state.get("energy_consumed", 0.0))

    self.state = {
        "total_solid_mass": self.total_solid_mass,
        "total_liquid_vol": self.total_liquid_vol,
        "energy_consumed": self.energy_consumed,
    }

    self.outputs_data = {
        "Q_liquid": 0.0,
        "Q_solid": 0.0,
        "TS_liquid": 0.0,
        "TS_solid": self.ts_solid_target,
        "VS_liquid": 0.0,
        "VS_solid": 0.0,
        "TAN_liquid": 0.0,
        "TAN_solid": 0.0,
        "TP_liquid": 0.0,
        "TP_solid": 0.0,
        "P_consumed": 0.0,
        "separation_efficiency": self.separation_efficiency,
    }

    self._initialized = True

step(t, dt, inputs)

Perform one simulation time step.

Parameters:

Name	Type	Description	Default
t	float	Current simulation time [days].	required
dt	float	Time step [days].	required
inputs	Dict[str, Any]	Dict with keys: - 'Q_in' [m3/d] Influent flow (required, or from Q_out of connected Digester). - 'Q_out' [m3/d] Alias for Q_in (Digester output key). - 'TS_in' [kg/m3] Total solids concentration in influent. If omitted, estimated from ADM1 state. - 'VS_in' [kg/m3] Volatile solids (optional, default 0.85*TS). - 'TAN_in' [kg/m3] Total ammonium nitrogen (optional). - 'TP_in' [kg/m3] Total phosphorus (optional). - 'state_out' [list] ADM1 state vector from upstream Digester. Used to estimate TS_in if not given.	required

Returns:

Type

Description

Dict[str, Any]

Dict with keys: - 'Q_liquid' [m3/d] Liquid fraction flow rate. - 'Q_solid' [m3/d] Solid fraction flow rate. - 'TS_liquid' [kg/m3] TS in liquid fraction. - 'TS_solid' [kg/m3] TS in solid fraction (= ts_solid_target). - 'VS_liquid' [kg/m3] VS in liquid fraction. - 'VS_solid' [kg/m3] VS in solid fraction. - 'TAN_liquid' [kg/m3] TAN in liquid fraction. - 'TAN_solid' [kg/m3] TAN in solid fraction. - 'TP_liquid' [kg/m3] Total P in liquid fraction. - 'TP_solid' [kg/m3] Total P in solid fraction. - 'P_consumed' [kW] Electrical power draw. - 'separation_efficiency' [-] Active efficiency value.

Source code in pyadm1/components/biological/separator.py

def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
    """
    Perform one simulation time step.

    Args:
        t:   Current simulation time [days].
        dt:  Time step [days].
        inputs: Dict with keys:
            - 'Q_in'    [m3/d]    Influent flow (required, or from Q_out
                                   of connected Digester).
            - 'Q_out'   [m3/d]    Alias for Q_in (Digester output key).
            - 'TS_in'   [kg/m3]   Total solids concentration in influent.
                                   If omitted, estimated from ADM1 state.
            - 'VS_in'   [kg/m3]   Volatile solids (optional, default 0.85*TS).
            - 'TAN_in'  [kg/m3]   Total ammonium nitrogen (optional).
            - 'TP_in'   [kg/m3]   Total phosphorus (optional).
            - 'state_out' [list]  ADM1 state vector from upstream Digester.
                                   Used to estimate TS_in if not given.

    Returns:
        Dict with keys:
            - 'Q_liquid'   [m3/d]  Liquid fraction flow rate.
            - 'Q_solid'    [m3/d]  Solid fraction flow rate.
            - 'TS_liquid'  [kg/m3] TS in liquid fraction.
            - 'TS_solid'   [kg/m3] TS in solid fraction (= ts_solid_target).
            - 'VS_liquid'  [kg/m3] VS in liquid fraction.
            - 'VS_solid'   [kg/m3] VS in solid fraction.
            - 'TAN_liquid' [kg/m3] TAN in liquid fraction.
            - 'TAN_solid'  [kg/m3] TAN in solid fraction.
            - 'TP_liquid'  [kg/m3] Total P in liquid fraction.
            - 'TP_solid'   [kg/m3] Total P in solid fraction.
            - 'P_consumed' [kW]    Electrical power draw.
            - 'separation_efficiency' [-] Active efficiency value.
    """
    # --- resolve influent flow -----------------------------------------
    Q_in = float(inputs.get("Q_in", inputs.get("Q_out", 0.0)))

    if Q_in <= 0.0:
        return self.outputs_data  # no flow, no separation

    # --- resolve total solids ------------------------------------------
    TS_in = float(inputs.get("TS_in", 0.0))

    if TS_in <= 0.0:
        # Estimate from ADM1 state vector if available
        adm1_state = inputs.get("state_out")
        if adm1_state is not None:
            TS_in = self._estimate_ts_from_adm1(adm1_state)
        else:
            # Fallback: typical mesophilic digestate TS ~40 kg/m3 (4%)
            TS_in = 40.0

    VS_in = float(inputs.get("VS_in", TS_in * 0.75))  # ~75% VS/TS
    TAN_in = float(inputs.get("TAN_in", 0.0))
    TP_in = float(inputs.get("TP_in", 0.0))

    # --- mass balance --------------------------------------------------
    m_TS_total = Q_in * TS_in  # kg/d total solids in
    m_VS_total = Q_in * VS_in  # kg/d VS in

    # VS separation efficiency slightly higher than TS (VS is more
    # particulate than mineral ash fraction); use 1.1x factor, capped at 0.95
    vs_sep_eff = min(self.separation_efficiency * 1.1, 0.95)

    m_TS_solid = m_TS_total * self.separation_efficiency  # kg/d to solid
    m_VS_solid = m_VS_total * vs_sep_eff  # kg/d VS to solid

    m_TS_liquid = m_TS_total - m_TS_solid  # kg/d to liquid
    m_VS_liquid = m_VS_total - m_VS_solid  # kg/d VS to liquid

    # Volume of solid fraction: m_TS_solid = Q_solid * ts_solid_target
    Q_solid = m_TS_solid / max(self.ts_solid_target, 1.0)  # m3/d
    Q_liquid = max(Q_in - Q_solid, 0.0)  # m3/d

    # Concentrations in each fraction
    TS_liquid = m_TS_liquid / max(Q_liquid, 1e-9)  # kg/m3
    VS_liquid = m_VS_liquid / max(Q_liquid, 1e-9)  # kg/m3
    VS_solid = m_VS_solid / max(Q_solid, 1e-9)  # kg/m3

    # Nutrient partitioning
    TAN_solid_total = TAN_in * Q_in * self.n_to_solid  # kg/d
    TAN_liquid_total = TAN_in * Q_in * (1.0 - self.n_to_solid)

    TP_solid_total = TP_in * Q_in * self.p_to_solid  # kg/d
    TP_liquid_total = TP_in * Q_in * (1.0 - self.p_to_solid)

    TAN_solid_conc = TAN_solid_total / max(Q_solid, 1e-9)  # kg/m3
    TAN_liquid_conc = TAN_liquid_total / max(Q_liquid, 1e-9)  # kg/m3
    TP_solid_conc = TP_solid_total / max(Q_solid, 1e-9)  # kg/m3
    TP_liquid_conc = TP_liquid_total / max(Q_liquid, 1e-9)  # kg/m3

    # --- power consumption --------------------------------------------
    # P [kW] = specific_energy [kWh/t] * FM_rate [t/d] / 24 [h/d]
    FM_rate_t_per_day = Q_in * self.fluid_density / 1000.0  # t FM/d
    P_consumed = self.specific_energy * FM_rate_t_per_day / 24.0  # kW

    # --- cumulative accounting ----------------------------------------
    dt_hours = dt * 24.0
    self.total_solid_mass += m_TS_solid * dt  # kg
    self.total_liquid_vol += Q_liquid * dt  # m3
    self.energy_consumed += P_consumed * dt_hours  # kWh

    # --- update state and outputs ------------------------------------
    self.state.update(
        {
            "total_solid_mass": self.total_solid_mass,
            "total_liquid_vol": self.total_liquid_vol,
            "energy_consumed": self.energy_consumed,
        }
    )

    self.outputs_data = {
        "Q_liquid": float(Q_liquid),
        "Q_solid": float(Q_solid),
        "TS_liquid": float(TS_liquid),
        "TS_solid": float(self.ts_solid_target),
        "VS_liquid": float(VS_liquid),
        "VS_solid": float(VS_solid),
        "TAN_liquid": float(TAN_liquid_conc),
        "TAN_solid": float(TAN_solid_conc),
        "TP_liquid": float(TP_liquid_conc),
        "TP_solid": float(TP_solid_conc),
        "P_consumed": float(P_consumed),
        "separation_efficiency": float(self.separation_efficiency),
        # Convenience summary
        "solid_fraction_ts_pct": float(self.ts_solid_target / (self.fluid_density * 10.0)),  # % TS
        "recovery_solid_pct": float(self.separation_efficiency * 100.0),
    }

    return self.outputs_data

to_dict()

Serialize separator configuration to dictionary.

Source code in pyadm1/components/biological/separator.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize separator configuration to dictionary."""
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "separator_type": self.separator_type.value,
        "separation_efficiency": self.separation_efficiency,
        "ts_solid_target": self.ts_solid_target,
        "n_to_solid": self.n_to_solid,
        "p_to_solid": self.p_to_solid,
        "specific_energy": self.specific_energy,
        "fluid_density": self.fluid_density,
        "inputs": self.inputs,
        "outputs": self.outputs,
        "state": self.state,
    }