Energy Components API

pyadm1.components.energy.chp.CHP

Bases: Component

Combined Heat and Power unit.

Converts biogas to electricity and heat with configurable efficiency.

Attributes:

Name	Type	Description
P_el_nom	float	Nominal electrical power in kW.
eta_el	float	Electrical efficiency (0-1).
eta_th	float	Thermal efficiency (0-1).
load_factor	float	Current operating point (0-1).

Example

chp = CHP("chp1", P_el_nom=500, eta_el=0.40, eta_th=0.45) chp.initialize() result = chp.step(t=0, dt=1/24, inputs={"Q_ch4": 1000})

Source code in pyadm1/components/energy/chp.py

class CHP(Component):
    """
    Combined Heat and Power unit.

    Converts biogas to electricity and heat with configurable efficiency.

    Attributes:
        P_el_nom (float): Nominal electrical power in kW.
        eta_el (float): Electrical efficiency (0-1).
        eta_th (float): Thermal efficiency (0-1).
        load_factor (float): Current operating point (0-1).

    Example:
        >>> chp = CHP("chp1", P_el_nom=500, eta_el=0.40, eta_th=0.45)
        >>> chp.initialize()
        >>> result = chp.step(t=0, dt=1/24, inputs={"Q_ch4": 1000})
    """

    def __init__(
        self,
        component_id: str,
        P_el_nom: float = 500.0,
        eta_el: float = 0.40,
        eta_th: float = 0.45,
        name: Optional[str] = None,
    ):
        """
        Initialize CHP unit.

        Args:
            component_id (str): Unique identifier.
            P_el_nom (float): Nominal electrical power in kW. Defaults to 500.0.
            eta_el (float): Electrical efficiency (0-1). Defaults to 0.40.
            eta_th (float): Thermal efficiency (0-1). Defaults to 0.45.
            name (Optional[str]): Human-readable name. Defaults to component_id.
        """
        super().__init__(component_id, ComponentType.CHP, name)

        self.P_el_nom = P_el_nom
        self.eta_el = eta_el
        self.eta_th = eta_th

        # Operating point (0-1)
        self.load_factor = 0.0

        # Auto-initialize with default state
        self.initialize()

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

        Args:
            initial_state (Optional[Dict[str, Any]]): Initial state with keys:
                - 'load_factor': Initial load factor (0-1)
                If None, uses default initialization.
        """
        if initial_state is None:
            initial_state = {}

        self.load_factor = initial_state.get("load_factor", 0.0)

        self.state = {
            "load_factor": self.load_factor,
            "P_el": 0.0,
            "P_th": 0.0,
            "Q_gas_consumed": 0.0,
            "operating_hours": 0.0,
        }

        # Ensure outputs_data is also initialized
        self.outputs_data = {
            "P_el": 0.0,
            "P_th": 0.0,
            "Q_gas_consumed": 0.0,
            "Q_ch4_remaining": 0.0,
        }

        self._initialized = True

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

        Args:
            t (float): Current time in days.
            dt (float): Time step in days.
            inputs (Dict[str, Any]): Input data with keys:
                - 'Q_ch4': Methane flow rate [m³/d] (direct input)
                - 'Q_gas_supplied_m3_per_day': Available biogas from storage [m³/d]
                - 'load_setpoint': Desired load factor [0-1] (optional)

        Returns:
            Dict[str, Any]: Output data with keys:
                - 'P_el': Electrical power [kW]
                - 'P_th': Thermal power [kW]
                - 'Q_gas_consumed': Biogas consumption [m³/d]
                - 'Q_gas_out_m3_per_day': Gas demand for connected storages [m³/d]
                - 'Q_ch4_remaining': Remaining methane [m³/d]
        """
        # Determine gas source priority:
        # 1. Direct gas supply from storage (preferred, plant_builder managed)
        # 2. Direct CH4 from digester (fallback for simple configurations)

        Q_gas_from_storage = inputs.get("Q_gas_supplied_m3_per_day", 0.0)
        Q_ch4_direct = inputs.get("Q_ch4", 0.0)

        # Assume 60% methane content in biogas from storage
        CH4_content = 0.60

        if Q_gas_from_storage > 0:
            # Use gas from storage (normal operation)
            Q_ch4_available = Q_gas_from_storage * CH4_content
        else:
            # Fallback: use direct CH4 from digester
            Q_ch4_available = Q_ch4_direct

        load_setpoint = inputs.get("load_setpoint", 1.0)

        # Methane energy content: ~10 kWh/m³
        E_ch4 = 10.0  # kWh/m³

        # Available power from methane
        P_available = Q_ch4_available / 24.0 * E_ch4  # kW (convert m³/d to m³/h)

        # Determine actual load
        P_el_max = min(self.P_el_nom, P_available / self.eta_el)
        self.load_factor = min(load_setpoint, P_el_max / self.P_el_nom) if self.P_el_nom > 0 else 0.0

        # Calculate outputs
        P_el = self.load_factor * self.P_el_nom
        P_th = P_el * self.eta_th / self.eta_el if self.eta_el > 0 else 0.0
        Q_ch4_consumed = (P_el / self.eta_el) * 24.0 / E_ch4 if self.eta_el > 0 else 0.0  # m³/d
        Q_gas_consumed = Q_ch4_consumed / CH4_content if CH4_content > 0 else 0.0  # m³/d biogas

        # Update state
        self.state["load_factor"] = self.load_factor
        self.state["P_el"] = P_el
        self.state["P_th"] = P_th
        self.state["Q_gas_consumed"] = Q_gas_consumed
        self.state["operating_hours"] += dt * 24.0

        self.outputs_data = {
            "P_el": P_el,
            "P_th": P_th,
            "Q_gas_consumed": Q_gas_consumed,
            "Q_gas_out_m3_per_day": Q_gas_consumed,  # Demand signal for storages
            "Q_ch4_remaining": Q_ch4_available - Q_ch4_consumed,
        }

        return self.outputs_data

    def to_dict(self) -> Dict[str, Any]:
        """
        Serialize to dictionary.

        Returns:
            Dict[str, Any]: Component configuration as dictionary.
        """
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "P_el_nom": self.P_el_nom,
            "eta_el": self.eta_el,
            "eta_th": self.eta_th,
            "inputs": self.inputs,
            "outputs": self.outputs,
            "state": self.state,
        }

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

        Args:
            config (Dict[str, Any]): Component configuration.

        Returns:
            CHP: Initialized CHP component.
        """
        chp = cls(
            component_id=config["component_id"],
            P_el_nom=config.get("P_el_nom", 500.0),
            eta_el=config.get("eta_el", 0.40),
            eta_th=config.get("eta_th", 0.45),
            name=config.get("name"),
        )

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

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

        return chp

Functions

__init__(component_id, P_el_nom=500.0, eta_el=0.4, eta_th=0.45, name=None)

Initialize CHP unit.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier.	required
P_el_nom	float	Nominal electrical power in kW. Defaults to 500.0.	500.0
eta_el	float	Electrical efficiency (0-1). Defaults to 0.40.	0.4
eta_th	float	Thermal efficiency (0-1). Defaults to 0.45.	0.45
name	Optional[str]	Human-readable name. Defaults to component_id.	None

Source code in pyadm1/components/energy/chp.py

def __init__(
    self,
    component_id: str,
    P_el_nom: float = 500.0,
    eta_el: float = 0.40,
    eta_th: float = 0.45,
    name: Optional[str] = None,
):
    """
    Initialize CHP unit.

    Args:
        component_id (str): Unique identifier.
        P_el_nom (float): Nominal electrical power in kW. Defaults to 500.0.
        eta_el (float): Electrical efficiency (0-1). Defaults to 0.40.
        eta_th (float): Thermal efficiency (0-1). Defaults to 0.45.
        name (Optional[str]): Human-readable name. Defaults to component_id.
    """
    super().__init__(component_id, ComponentType.CHP, name)

    self.P_el_nom = P_el_nom
    self.eta_el = eta_el
    self.eta_th = eta_th

    # Operating point (0-1)
    self.load_factor = 0.0

    # Auto-initialize with default state
    self.initialize()

from_dict(config) classmethod

Create from dictionary.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Component configuration.	required

Returns:

Name	Type	Description
CHP	CHP	Initialized CHP component.

Source code in pyadm1/components/energy/chp.py

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

    Args:
        config (Dict[str, Any]): Component configuration.

    Returns:
        CHP: Initialized CHP component.
    """
    chp = cls(
        component_id=config["component_id"],
        P_el_nom=config.get("P_el_nom", 500.0),
        eta_el=config.get("eta_el", 0.40),
        eta_th=config.get("eta_th", 0.45),
        name=config.get("name"),
    )

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

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

    return chp

initialize(initial_state=None)

Initialize CHP state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Initial state with keys: - 'load_factor': Initial load factor (0-1) If None, uses default initialization.	None

Source code in pyadm1/components/energy/chp.py

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

    Args:
        initial_state (Optional[Dict[str, Any]]): Initial state with keys:
            - 'load_factor': Initial load factor (0-1)
            If None, uses default initialization.
    """
    if initial_state is None:
        initial_state = {}

    self.load_factor = initial_state.get("load_factor", 0.0)

    self.state = {
        "load_factor": self.load_factor,
        "P_el": 0.0,
        "P_th": 0.0,
        "Q_gas_consumed": 0.0,
        "operating_hours": 0.0,
    }

    # Ensure outputs_data is also initialized
    self.outputs_data = {
        "P_el": 0.0,
        "P_th": 0.0,
        "Q_gas_consumed": 0.0,
        "Q_ch4_remaining": 0.0,
    }

    self._initialized = True

step(t, dt, inputs)

Perform one simulation time step.

Parameters:

Name	Type	Description	Default
t	float	Current time in days.	required
dt	float	Time step in days.	required
inputs	Dict[str, Any]	Input data with keys: - 'Q_ch4': Methane flow rate [m³/d] (direct input) - 'Q_gas_supplied_m3_per_day': Available biogas from storage [m³/d] - 'load_setpoint': Desired load factor [0-1] (optional)	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Output data with keys: - 'P_el': Electrical power [kW] - 'P_th': Thermal power [kW] - 'Q_gas_consumed': Biogas consumption [m³/d] - 'Q_gas_out_m3_per_day': Gas demand for connected storages [m³/d] - 'Q_ch4_remaining': Remaining methane [m³/d]

Source code in pyadm1/components/energy/chp.py

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

    Args:
        t (float): Current time in days.
        dt (float): Time step in days.
        inputs (Dict[str, Any]): Input data with keys:
            - 'Q_ch4': Methane flow rate [m³/d] (direct input)
            - 'Q_gas_supplied_m3_per_day': Available biogas from storage [m³/d]
            - 'load_setpoint': Desired load factor [0-1] (optional)

    Returns:
        Dict[str, Any]: Output data with keys:
            - 'P_el': Electrical power [kW]
            - 'P_th': Thermal power [kW]
            - 'Q_gas_consumed': Biogas consumption [m³/d]
            - 'Q_gas_out_m3_per_day': Gas demand for connected storages [m³/d]
            - 'Q_ch4_remaining': Remaining methane [m³/d]
    """
    # Determine gas source priority:
    # 1. Direct gas supply from storage (preferred, plant_builder managed)
    # 2. Direct CH4 from digester (fallback for simple configurations)

    Q_gas_from_storage = inputs.get("Q_gas_supplied_m3_per_day", 0.0)
    Q_ch4_direct = inputs.get("Q_ch4", 0.0)

    # Assume 60% methane content in biogas from storage
    CH4_content = 0.60

    if Q_gas_from_storage > 0:
        # Use gas from storage (normal operation)
        Q_ch4_available = Q_gas_from_storage * CH4_content
    else:
        # Fallback: use direct CH4 from digester
        Q_ch4_available = Q_ch4_direct

    load_setpoint = inputs.get("load_setpoint", 1.0)

    # Methane energy content: ~10 kWh/m³
    E_ch4 = 10.0  # kWh/m³

    # Available power from methane
    P_available = Q_ch4_available / 24.0 * E_ch4  # kW (convert m³/d to m³/h)

    # Determine actual load
    P_el_max = min(self.P_el_nom, P_available / self.eta_el)
    self.load_factor = min(load_setpoint, P_el_max / self.P_el_nom) if self.P_el_nom > 0 else 0.0

    # Calculate outputs
    P_el = self.load_factor * self.P_el_nom
    P_th = P_el * self.eta_th / self.eta_el if self.eta_el > 0 else 0.0
    Q_ch4_consumed = (P_el / self.eta_el) * 24.0 / E_ch4 if self.eta_el > 0 else 0.0  # m³/d
    Q_gas_consumed = Q_ch4_consumed / CH4_content if CH4_content > 0 else 0.0  # m³/d biogas

    # Update state
    self.state["load_factor"] = self.load_factor
    self.state["P_el"] = P_el
    self.state["P_th"] = P_th
    self.state["Q_gas_consumed"] = Q_gas_consumed
    self.state["operating_hours"] += dt * 24.0

    self.outputs_data = {
        "P_el": P_el,
        "P_th": P_th,
        "Q_gas_consumed": Q_gas_consumed,
        "Q_gas_out_m3_per_day": Q_gas_consumed,  # Demand signal for storages
        "Q_ch4_remaining": Q_ch4_available - Q_ch4_consumed,
    }

    return self.outputs_data

to_dict()

Serialize to dictionary.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Component configuration as dictionary.

Source code in pyadm1/components/energy/chp.py

def to_dict(self) -> Dict[str, Any]:
    """
    Serialize to dictionary.

    Returns:
        Dict[str, Any]: Component configuration as dictionary.
    """
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "P_el_nom": self.P_el_nom,
        "eta_el": self.eta_el,
        "eta_th": self.eta_th,
        "inputs": self.inputs,
        "outputs": self.outputs,
        "state": self.state,
    }

pyadm1.components.energy.heating.HeatingSystem

Bases: Component

Heating system for maintaining digester temperature.

Calculates heat demand and energy consumption based on temperature difference and available waste heat from CHP.

Attributes:

Name	Type	Description
target_temperature		Target digester temperature in K.
heat_loss_coefficient		Heat loss coefficient in kW/K.

Example

heating = HeatingSystem("heat1", target_temperature=308.15, heat_loss_coefficient=0.5) heating.initialize() result = heating.step(t=0, dt=1/24, inputs={"T_digester": 308.15, "P_th_available": 200})

Source code in pyadm1/components/energy/heating.py

class HeatingSystem(Component):
    """
    Heating system for maintaining digester temperature.

    Calculates heat demand and energy consumption based on temperature
    difference and available waste heat from CHP.

    Attributes:
        target_temperature: Target digester temperature in K.
        heat_loss_coefficient: Heat loss coefficient in kW/K.

    Example:
        >>> heating = HeatingSystem("heat1", target_temperature=308.15, heat_loss_coefficient=0.5)
        >>> heating.initialize()
        >>> result = heating.step(t=0, dt=1/24, inputs={"T_digester": 308.15, "P_th_available": 200})
    """

    def __init__(
        self,
        component_id: str,
        target_temperature: float = 308.15,
        heat_loss_coefficient: float = 0.5,
        name: Optional[str] = None,
    ):
        """
        Initialize heating system.

        Args:
            component_id: Unique identifier.
            target_temperature: Target digester temperature in K. Defaults to 308.15 (35°C).
            heat_loss_coefficient: Heat loss coefficient in kW/K. Defaults to 0.5.
            name: Human-readable name. Defaults to component_id.
        """
        super().__init__(component_id, ComponentType.HEATING, name)

        self.target_temperature = target_temperature
        self.heat_loss_coefficient = heat_loss_coefficient

        # Auto-initialize with default state
        self.initialize()

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

        Args:
            initial_state: Initial state (not used currently).
        """
        self.state = {
            "Q_heat_demand": 0.0,
            "Q_heat_supplied": 0.0,
            "energy_consumed": 0.0,
        }

        # Ensure outputs_data is also initialized
        self.outputs_data = {
            "Q_heat_supplied": 0.0,
            "P_th_used": 0.0,
            "P_aux_heat": 0.0,
        }

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

        Args:
            t: Current time in days.
            dt: Time step in days.
            inputs: Input data with keys:
                - 'T_digester': Current digester temperature [K]
                - 'T_ambient': Ambient temperature [K]
                - 'V_liq': Liquid volume [m³]
                - 'P_th_available': Available thermal power from CHP [kW]

        Returns:
            Output data with keys:
                - 'Q_heat_supplied': Heat supplied [kW]
                - 'P_th_used': Thermal power used from CHP [kW]
                - 'P_aux_heat': Auxiliary heating needed [kW]
        """
        T_digester = inputs.get("T_digester", self.target_temperature)
        T_ambient = inputs.get("T_ambient", 288.15)  # 15°C default
        P_th_available = inputs.get("P_th_available", 0.0)
        q_substrates = inputs.get("Q_substrates")

        # Heat loss to environment
        T_diff = T_digester - T_ambient
        Q_loss = self.heat_loss_coefficient * T_diff
        Q_process = _calc_process_heat_kw(q_substrates, self.target_temperature)

        # Heat demand to maintain temperature
        Q_demand = Q_loss + Q_process

        # Use CHP heat first
        P_th_used = min(P_th_available, Q_demand)

        # Additional heating needed
        P_aux_heat = max(0.0, Q_demand - P_th_used)

        Q_heat_supplied = P_th_used + P_aux_heat

        # Update state
        self.state["Q_heat_demand"] = Q_demand
        self.state["Q_heat_supplied"] = Q_heat_supplied
        self.state["energy_consumed"] += P_aux_heat * dt * 24.0  # kWh

        self.outputs_data = {
            "Q_heat_supplied": Q_heat_supplied,
            "P_th_used": P_th_used,
            "P_aux_heat": P_aux_heat,
        }

        return self.outputs_data

    def to_dict(self) -> Dict[str, Any]:
        """
        Serialize to dictionary.

        Returns:
            Component configuration as dictionary.
        """
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "target_temperature": self.target_temperature,
            "heat_loss_coefficient": self.heat_loss_coefficient,
            "inputs": self.inputs,
            "outputs": self.outputs,
            "state": self.state,
        }

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

        Args:
            config: Component configuration.

        Returns:
            Initialized heating system component.
        """
        heating = cls(
            component_id=config["component_id"],
            target_temperature=config.get("target_temperature", 308.15),
            heat_loss_coefficient=config.get("heat_loss_coefficient", 0.5),
            name=config.get("name"),
        )

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

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

        return heating

Functions

__init__(component_id, target_temperature=308.15, heat_loss_coefficient=0.5, name=None)

Initialize heating system.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier.	required
target_temperature	float	Target digester temperature in K. Defaults to 308.15 (35°C).	308.15
heat_loss_coefficient	float	Heat loss coefficient in kW/K. Defaults to 0.5.	0.5
name	Optional[str]	Human-readable name. Defaults to component_id.	None

Source code in pyadm1/components/energy/heating.py

def __init__(
    self,
    component_id: str,
    target_temperature: float = 308.15,
    heat_loss_coefficient: float = 0.5,
    name: Optional[str] = None,
):
    """
    Initialize heating system.

    Args:
        component_id: Unique identifier.
        target_temperature: Target digester temperature in K. Defaults to 308.15 (35°C).
        heat_loss_coefficient: Heat loss coefficient in kW/K. Defaults to 0.5.
        name: Human-readable name. Defaults to component_id.
    """
    super().__init__(component_id, ComponentType.HEATING, name)

    self.target_temperature = target_temperature
    self.heat_loss_coefficient = heat_loss_coefficient

    # Auto-initialize with default state
    self.initialize()

from_dict(config) classmethod

Create from dictionary.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Component configuration.	required

Returns:

Type	Description
HeatingSystem	Initialized heating system component.

Source code in pyadm1/components/energy/heating.py

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

    Args:
        config: Component configuration.

    Returns:
        Initialized heating system component.
    """
    heating = cls(
        component_id=config["component_id"],
        target_temperature=config.get("target_temperature", 308.15),
        heat_loss_coefficient=config.get("heat_loss_coefficient", 0.5),
        name=config.get("name"),
    )

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

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

    return heating

initialize(initial_state=None)

Initialize heating system state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Initial state (not used currently).	None

Source code in pyadm1/components/energy/heating.py

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

    Args:
        initial_state: Initial state (not used currently).
    """
    self.state = {
        "Q_heat_demand": 0.0,
        "Q_heat_supplied": 0.0,
        "energy_consumed": 0.0,
    }

    # Ensure outputs_data is also initialized
    self.outputs_data = {
        "Q_heat_supplied": 0.0,
        "P_th_used": 0.0,
        "P_aux_heat": 0.0,
    }

step(t, dt, inputs)

Perform one simulation time step.

Parameters:

Name	Type	Description	Default
t	float	Current time in days.	required
dt	float	Time step in days.	required
inputs	Dict[str, Any]	Input data with keys: - 'T_digester': Current digester temperature [K] - 'T_ambient': Ambient temperature [K] - 'V_liq': Liquid volume [m³] - 'P_th_available': Available thermal power from CHP [kW]	required

Returns:

Type	Description
Dict[str, Any]	Output data with keys: - 'Q_heat_supplied': Heat supplied [kW] - 'P_th_used': Thermal power used from CHP [kW] - 'P_aux_heat': Auxiliary heating needed [kW]

Source code in pyadm1/components/energy/heating.py

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

    Args:
        t: Current time in days.
        dt: Time step in days.
        inputs: Input data with keys:
            - 'T_digester': Current digester temperature [K]
            - 'T_ambient': Ambient temperature [K]
            - 'V_liq': Liquid volume [m³]
            - 'P_th_available': Available thermal power from CHP [kW]

    Returns:
        Output data with keys:
            - 'Q_heat_supplied': Heat supplied [kW]
            - 'P_th_used': Thermal power used from CHP [kW]
            - 'P_aux_heat': Auxiliary heating needed [kW]
    """
    T_digester = inputs.get("T_digester", self.target_temperature)
    T_ambient = inputs.get("T_ambient", 288.15)  # 15°C default
    P_th_available = inputs.get("P_th_available", 0.0)
    q_substrates = inputs.get("Q_substrates")

    # Heat loss to environment
    T_diff = T_digester - T_ambient
    Q_loss = self.heat_loss_coefficient * T_diff
    Q_process = _calc_process_heat_kw(q_substrates, self.target_temperature)

    # Heat demand to maintain temperature
    Q_demand = Q_loss + Q_process

    # Use CHP heat first
    P_th_used = min(P_th_available, Q_demand)

    # Additional heating needed
    P_aux_heat = max(0.0, Q_demand - P_th_used)

    Q_heat_supplied = P_th_used + P_aux_heat

    # Update state
    self.state["Q_heat_demand"] = Q_demand
    self.state["Q_heat_supplied"] = Q_heat_supplied
    self.state["energy_consumed"] += P_aux_heat * dt * 24.0  # kWh

    self.outputs_data = {
        "Q_heat_supplied": Q_heat_supplied,
        "P_th_used": P_th_used,
        "P_aux_heat": P_aux_heat,
    }

    return self.outputs_data

to_dict()

Serialize to dictionary.

Returns:

Type	Description
Dict[str, Any]	Component configuration as dictionary.

Source code in pyadm1/components/energy/heating.py

def to_dict(self) -> Dict[str, Any]:
    """
    Serialize to dictionary.

    Returns:
        Component configuration as dictionary.
    """
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "target_temperature": self.target_temperature,
        "heat_loss_coefficient": self.heat_loss_coefficient,
        "inputs": self.inputs,
        "outputs": self.outputs,
        "state": self.state,
    }

pyadm1.components.energy.gas_storage.GasStorage

Bases: Component

Gas storage component.

Parameters:

Name	Type	Description	Default
component_id	str	unique id	required
storage_type	str	'membrane' | 'dome' | 'compressed'	'membrane'
capacity_m3	float	usable gas volume at STP (m^3)	1000.0
p_min_bar	float	minimum operating pressure (bar)	0.95
p_max_bar	float	maximum safe pressure (bar)	1.05
name	Optional[str]	optional human-readable name	None

Source code in pyadm1/components/energy/gas_storage.py

class GasStorage(Component):
    """
    Gas storage component.

    Arguments:
        component_id: unique id
        storage_type: 'membrane' | 'dome' | 'compressed'
        capacity_m3: usable gas volume at STP (m^3)
        p_min_bar: minimum operating pressure (bar)
        p_max_bar: maximum safe pressure (bar)
        name: optional human-readable name
    """

    def __init__(
        self,
        component_id: str,
        storage_type: str = "membrane",
        capacity_m3: float = 1000.0,
        p_min_bar: float = 0.95,
        p_max_bar: float = 1.05,
        initial_fill_fraction: float = 0.1,
        name: Optional[str] = None,
    ):
        """

        Args:
            component_id: unique id
            storage_type: 'membrane' | 'dome' | 'compressed'
            capacity_m3: usable gas volume at STP (m^3)
            p_min_bar: minimum operating pressure (bar)
            p_max_bar: maximum safe pressure (bar)
            initial_fill_fraction: initial stored fraction of capacity (0-1)
            name: optional human-readable name
        """
        super().__init__(component_id, ComponentType.STORAGE, name)

        # Config
        self.storage_type = storage_type.lower()
        if self.storage_type not in ("membrane", "dome", "compressed"):
            raise ValueError("storage_type must be 'membrane', 'dome' or 'compressed'")

        self.capacity_m3 = float(capacity_m3)
        self.p_atm_bar = 1.01325
        self.p_min_bar = float(p_min_bar)
        self.p_max_bar = float(p_max_bar)

        # State variables
        self.stored_volume_m3 = float(self.capacity_m3) * float(initial_fill_fraction)
        # cumulative vented (for logging)
        self._cum_vented_m3 = 0.0

        # control setpoint (pressure in bar) - None means no active setpoint
        self.pressure_setpoint_bar: Optional[float] = None

        # initialize default state
        self.initialize()

    def initialize(self, initial_state: Optional[Dict[str, Any]] = None) -> None:
        """Initialize storage state; initial_state may contain stored_volume_m3, pressure_setpoint_bar."""
        if initial_state is not None:
            if "stored_volume_m3" in initial_state:
                self.stored_volume_m3 = float(initial_state["stored_volume_m3"])
            if "pressure_setpoint_bar" in initial_state:
                self.pressure_setpoint_bar = initial_state["pressure_setpoint_bar"]

        # clamp stored volume
        self.stored_volume_m3 = max(0.0, min(self.stored_volume_m3, self.capacity_m3))

        self._cum_vented_m3 = 0.0

        # outputs_data template
        self.outputs_data = {
            "stored_volume_m3": self.stored_volume_m3,
            "pressure_bar": self._estimate_pressure_bar(),
            "utilization": self.stored_volume_m3 / max(1e-9, self.capacity_m3),
            "vented_volume_m3": 0.0,
            "Q_gas_supplied_m3_per_day": 0.0,
        }

        self._initialized = True

    # ------------------------------
    # Internal helpers
    # ------------------------------
    def _estimate_pressure_bar(self) -> float:
        """
        Estimate pressure [bar] from stored volume and storage type.

        Models:
        - low-pressure (membrane/dome): small overpressure scales with stored fraction
            p = p_atm + frac * (p_max_bar - p_atm)
        - compressed: pressure scales from p_min to p_max with stored fraction
            p = p_min + frac^alpha * (p_max - p_min)  (alpha > 1 to reflect nonlinear increase)
        """
        frac = 0.0
        if self.capacity_m3 > 0:
            frac = max(0.0, min(self.stored_volume_m3 / self.capacity_m3, 1.0))

        if self.storage_type in ("membrane", "dome"):
            # low-pressure head/membrane; small overpressure
            p = self.p_atm_bar + frac * (self.p_max_bar - self.p_atm_bar)
        else:  # compressed
            # use nonlinear exponent to mimic gas compression behaviour (very simplified)
            alpha = 2.0
            p = self.p_min_bar + (frac**alpha) * (self.p_max_bar - self.p_min_bar)

        return float(p)

    # ------------------------------
    # Main simulation step
    # ------------------------------
    def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """
        One simulation step.

        Args:
            t:
            dt:
            inputs: Inputs dictionary may contain:
                - 'Q_gas_in_m3_per_day'   : gas inflow from digesters/other sources (m^3/day)
                - 'Q_gas_out_m3_per_day'  : requested gas outflow (demand) (m^3/day)
                - 'set_pressure'          : desired pressure setpoint (bar)  (optional)
                - 'vent_to_flare'         : bool, if True allow venting to flare when overpressure (default True)

        Returns:
            object: Returns outputs_data with keys:
                - 'stored_volume_m3'
                - 'pressure_bar'
                - 'utilization' (0-1)
                - 'vented_volume_m3' (this timestep)
                - 'Q_gas_supplied_m3_per_day' (actual supply that was delivered)
        """
        # get flows (units: m^3/day) -> convert to volume for this timestep: m^3 = Q * dt (dt in days)
        Q_in = float(inputs.get("Q_gas_in_m3_per_day", 0.0))
        Q_out_req = float(inputs.get("Q_gas_out_m3_per_day", 0.0))
        allow_vent = bool(inputs.get("vent_to_flare", True))

        print(f"gas storage Q_in {Q_in}, Q_out: {Q_out_req}")

        # pressure setpoint handling (bar) - if provided, store it for internal control
        if "set_pressure" in inputs:
            sp = inputs["set_pressure"]
            try:
                self.pressure_setpoint_bar = None if sp is None else float(sp)
            except Exception as e:
                # ignore invalid setpoint
                print(e)

        # convert to volumes for this dt
        vol_in = Q_in * dt
        vol_out_req = Q_out_req * dt

        vented_this_step = 0.0
        # supplied_this_step = 0.0

        # Inflow: attempt to store incoming gas
        free_capacity_m3 = self.capacity_m3 - self.stored_volume_m3
        if vol_in <= free_capacity_m3 + 1e-12:
            # all incoming stored
            self.stored_volume_m3 += vol_in
            overflow = 0.0
        else:
            # store what fits, overflow is vented (or forwarded externally)
            stored = max(0.0, free_capacity_m3)
            overflow = vol_in - stored
            self.stored_volume_m3 += stored
            if allow_vent:
                vented_this_step += overflow
                self._cum_vented_m3 += overflow
            else:
                # if venting not allowed, assume overflow is rejected (lost) (count as vented for safety)
                vented_this_step += overflow
                self._cum_vented_m3 += overflow

        # Supply (outflow): cannot supply more than stored
        # If pressure below p_min, limit supply (simulate pressure control)
        current_pressure = self._estimate_pressure_bar()

        # If a pressure setpoint is requested and higher than current,
        # we might choose to deny outflow to increase pressure
        restrict_factor = 1.0
        if self.pressure_setpoint_bar is not None:
            # If setpoint is higher than current pressure, prioritize charging (i.e., restrict outflow)
            if self.pressure_setpoint_bar > current_pressure:
                # reduce allowed outflow proportionally
                # factor between 0..1 -> 0 means block outflow, 1 means full
                gap = self.pressure_setpoint_bar - current_pressure
                span = max(1e-6, self.p_max_bar - self.p_atm_bar)
                restrict_factor = max(0.0, 1.0 - gap / span)

        allowed_out_vol = self.stored_volume_m3 * restrict_factor  # simple limit
        desired_out_vol = min(vol_out_req, allowed_out_vol)

        # Also ensure we don't drop below zero or below some minimum reserve needed to maintain p_min:
        # approximate required volume to keep p >= p_min: invert pressure estimate crudely by fraction
        if self.storage_type in ("membrane", "dome"):
            # frac required to maintain p_min: (p_min - p_atm)/(p_max - p_atm)
            if (self.p_max_bar - self.p_atm_bar) > 1e-9:
                frac_needed = max(
                    0.0,
                    (self.p_min_bar - self.p_atm_bar) / (self.p_max_bar - self.p_atm_bar),
                )
            else:
                frac_needed = 0.0
        else:  # compressed
            # invert nonlinear mapping p = p_min + frac^alpha*(p_max-p_min)
            if (self.p_max_bar - self.p_min_bar) > 1e-9:
                alpha = 2.0
                frac_needed = max(
                    0.0,
                    ((self.p_min_bar - self.p_min_bar) / (self.p_max_bar - self.p_min_bar)) ** (1.0 / alpha),
                )
                # above formula trivial -> 0, but keep general structure; use small reserve fraction
                frac_needed = 0.01
            else:
                frac_needed = 0.0

        reserve_volume = frac_needed * self.capacity_m3
        # ensure not to supply below reserve
        max_out_vol_after_reserve = max(0.0, self.stored_volume_m3 - reserve_volume)
        desired_out_vol = min(desired_out_vol, max_out_vol_after_reserve)

        # perform outflow
        self.stored_volume_m3 -= desired_out_vol
        supplied_this_step_m3_per_day = desired_out_vol / max(dt, 1e-12)  # convert back to m3/day

        print("storage tank:", supplied_this_step_m3_per_day, self.stored_volume_m3)

        # After flows, update pressure and check safety
        current_pressure = self._estimate_pressure_bar()

        # Safety: overpressure -> vent to flare
        if current_pressure > self.p_max_bar:
            # compute how much volume must be removed to reach p_max
            # invert our pressure models approximately by reducing stored fraction until p <= p_max
            # perform simple linear backoff: remove fraction proportional to pressure exceedance
            if self.storage_type in ("membrane", "dome"):
                # p = p_atm + frac*(p_max - p_atm), want frac_target = (p_max - p_atm)/(p_max - p_atm) = 1.0
                # but if p_max_bar is absolute safety we want to bring pressure to p_max_bar.
                # compute fraction that yields p_max_bar
                target_frac = max(
                    0.0,
                    min(
                        1.0,
                        (self.p_max_bar - self.p_atm_bar) / max(1e-9, (self.p_max_bar - self.p_atm_bar)),
                    ),
                )
                # if current fraction > 1.0 (theory), set to 1.0
                target_frac = min(target_frac, 1.0)
                target_volume = target_frac * self.capacity_m3
                if self.stored_volume_m3 > target_volume:
                    vent = self.stored_volume_m3 - target_volume
                else:
                    vent = 0.0
            else:  # compressed
                # reduce to fraction that yields p_max_bar approx equal to 1.0 fraction
                # simpler: vent until stored <= capacity * 0.999 (small safety)
                target_volume = min(self.capacity_m3, self.capacity_m3 * 0.999)
                vent = max(0.0, self.stored_volume_m3 - target_volume)

            # Vent if allowed (or always vent for safety)
            if allow_vent and vent > 0.0:
                vented_this_step += vent
                self.stored_volume_m3 -= vent
                self._cum_vented_m3 += vent

            # recompute pressure
            current_pressure = self._estimate_pressure_bar()

        # Clamp stored volume
        self.stored_volume_m3 = max(0.0, min(self.stored_volume_m3, self.capacity_m3))

        # Update outputs_data
        self.outputs_data = {
            "stored_volume_m3": float(self.stored_volume_m3),
            "pressure_bar": float(current_pressure),
            "utilization": float(self.stored_volume_m3 / max(1e-9, self.capacity_m3)),
            "vented_volume_m3": float(vented_this_step),
            "Q_gas_supplied_m3_per_day": float(supplied_this_step_m3_per_day),
            "cumulative_vented_m3": float(self._cum_vented_m3),
            "pressure_setpoint_bar": self.pressure_setpoint_bar,
        }

        return self.outputs_data

    # ------------------------------
    # Serialization
    # ------------------------------
    def to_dict(self) -> Dict[str, Any]:
        """Serialize configuration + current state."""
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "storage_type": self.storage_type,
            "capacity_m3": self.capacity_m3,
            "p_atm_bar": self.p_atm_bar,
            "p_min_bar": self.p_min_bar,
            "p_max_bar": self.p_max_bar,
            "stored_volume_m3": self.stored_volume_m3,
            "pressure_setpoint_bar": self.pressure_setpoint_bar,
            "outputs_data": self.outputs_data,
            "inputs": self.inputs,
            "outputs": self.outputs,
        }

    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> "GasStorage":
        """Create GasStorage from dict produced by to_dict."""
        gs = cls(
            component_id=config["component_id"],
            storage_type=config.get("storage_type", "membrane"),
            capacity_m3=config.get("capacity_m3", 1000.0),
            p_min_bar=config.get("p_min_bar", 0.95),
            p_max_bar=config.get("p_max_bar", 1.05),
            initial_fill_fraction=0.0,
            name=config.get("name"),
        )
        # restore state if present
        if "stored_volume_m3" in config:
            try:
                gs.stored_volume_m3 = float(config["stored_volume_m3"])
            except Exception as e:
                print(e)

        if "pressure_setpoint_bar" in config:
            gs.pressure_setpoint_bar = config.get("pressure_setpoint_bar")

        gs.initialize(
            {
                "stored_volume_m3": gs.stored_volume_m3,
                "pressure_setpoint_bar": gs.pressure_setpoint_bar,
            }
        )

        return gs

Functions

__init__(component_id, storage_type='membrane', capacity_m3=1000.0, p_min_bar=0.95, p_max_bar=1.05, initial_fill_fraction=0.1, name=None)

Parameters:

Name	Type	Description	Default
component_id	str	unique id	required
storage_type	str	'membrane' | 'dome' | 'compressed'	'membrane'
capacity_m3	float	usable gas volume at STP (m^3)	1000.0
p_min_bar	float	minimum operating pressure (bar)	0.95
p_max_bar	float	maximum safe pressure (bar)	1.05
initial_fill_fraction	float	initial stored fraction of capacity (0-1)	0.1
name	Optional[str]	optional human-readable name	None

Source code in pyadm1/components/energy/gas_storage.py

def __init__(
    self,
    component_id: str,
    storage_type: str = "membrane",
    capacity_m3: float = 1000.0,
    p_min_bar: float = 0.95,
    p_max_bar: float = 1.05,
    initial_fill_fraction: float = 0.1,
    name: Optional[str] = None,
):
    """

    Args:
        component_id: unique id
        storage_type: 'membrane' | 'dome' | 'compressed'
        capacity_m3: usable gas volume at STP (m^3)
        p_min_bar: minimum operating pressure (bar)
        p_max_bar: maximum safe pressure (bar)
        initial_fill_fraction: initial stored fraction of capacity (0-1)
        name: optional human-readable name
    """
    super().__init__(component_id, ComponentType.STORAGE, name)

    # Config
    self.storage_type = storage_type.lower()
    if self.storage_type not in ("membrane", "dome", "compressed"):
        raise ValueError("storage_type must be 'membrane', 'dome' or 'compressed'")

    self.capacity_m3 = float(capacity_m3)
    self.p_atm_bar = 1.01325
    self.p_min_bar = float(p_min_bar)
    self.p_max_bar = float(p_max_bar)

    # State variables
    self.stored_volume_m3 = float(self.capacity_m3) * float(initial_fill_fraction)
    # cumulative vented (for logging)
    self._cum_vented_m3 = 0.0

    # control setpoint (pressure in bar) - None means no active setpoint
    self.pressure_setpoint_bar: Optional[float] = None

    # initialize default state
    self.initialize()

from_dict(config) classmethod

Create GasStorage from dict produced by to_dict.

Source code in pyadm1/components/energy/gas_storage.py

@classmethod
def from_dict(cls, config: Dict[str, Any]) -> "GasStorage":
    """Create GasStorage from dict produced by to_dict."""
    gs = cls(
        component_id=config["component_id"],
        storage_type=config.get("storage_type", "membrane"),
        capacity_m3=config.get("capacity_m3", 1000.0),
        p_min_bar=config.get("p_min_bar", 0.95),
        p_max_bar=config.get("p_max_bar", 1.05),
        initial_fill_fraction=0.0,
        name=config.get("name"),
    )
    # restore state if present
    if "stored_volume_m3" in config:
        try:
            gs.stored_volume_m3 = float(config["stored_volume_m3"])
        except Exception as e:
            print(e)

    if "pressure_setpoint_bar" in config:
        gs.pressure_setpoint_bar = config.get("pressure_setpoint_bar")

    gs.initialize(
        {
            "stored_volume_m3": gs.stored_volume_m3,
            "pressure_setpoint_bar": gs.pressure_setpoint_bar,
        }
    )

    return gs

initialize(initial_state=None)

Initialize storage state; initial_state may contain stored_volume_m3, pressure_setpoint_bar.

Source code in pyadm1/components/energy/gas_storage.py

def initialize(self, initial_state: Optional[Dict[str, Any]] = None) -> None:
    """Initialize storage state; initial_state may contain stored_volume_m3, pressure_setpoint_bar."""
    if initial_state is not None:
        if "stored_volume_m3" in initial_state:
            self.stored_volume_m3 = float(initial_state["stored_volume_m3"])
        if "pressure_setpoint_bar" in initial_state:
            self.pressure_setpoint_bar = initial_state["pressure_setpoint_bar"]

    # clamp stored volume
    self.stored_volume_m3 = max(0.0, min(self.stored_volume_m3, self.capacity_m3))

    self._cum_vented_m3 = 0.0

    # outputs_data template
    self.outputs_data = {
        "stored_volume_m3": self.stored_volume_m3,
        "pressure_bar": self._estimate_pressure_bar(),
        "utilization": self.stored_volume_m3 / max(1e-9, self.capacity_m3),
        "vented_volume_m3": 0.0,
        "Q_gas_supplied_m3_per_day": 0.0,
    }

    self._initialized = True

step(t, dt, inputs)

One simulation step.

Parameters:

Name	Type	Description	Default
t	float		required
dt	float		required
inputs	Dict[str, Any]	Inputs dictionary may contain: - 'Q_gas_in_m3_per_day' : gas inflow from digesters/other sources (m^3/day) - 'Q_gas_out_m3_per_day' : requested gas outflow (demand) (m^3/day) - 'set_pressure' : desired pressure setpoint (bar) (optional) - 'vent_to_flare' : bool, if True allow venting to flare when overpressure (default True)	required

Returns:

Name	Type	Description
object	Dict[str, Any]	Returns outputs_data with keys: - 'stored_volume_m3' - 'pressure_bar' - 'utilization' (0-1) - 'vented_volume_m3' (this timestep) - 'Q_gas_supplied_m3_per_day' (actual supply that was delivered)

Source code in pyadm1/components/energy/gas_storage.py

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

    Args:
        t:
        dt:
        inputs: Inputs dictionary may contain:
            - 'Q_gas_in_m3_per_day'   : gas inflow from digesters/other sources (m^3/day)
            - 'Q_gas_out_m3_per_day'  : requested gas outflow (demand) (m^3/day)
            - 'set_pressure'          : desired pressure setpoint (bar)  (optional)
            - 'vent_to_flare'         : bool, if True allow venting to flare when overpressure (default True)

    Returns:
        object: Returns outputs_data with keys:
            - 'stored_volume_m3'
            - 'pressure_bar'
            - 'utilization' (0-1)
            - 'vented_volume_m3' (this timestep)
            - 'Q_gas_supplied_m3_per_day' (actual supply that was delivered)
    """
    # get flows (units: m^3/day) -> convert to volume for this timestep: m^3 = Q * dt (dt in days)
    Q_in = float(inputs.get("Q_gas_in_m3_per_day", 0.0))
    Q_out_req = float(inputs.get("Q_gas_out_m3_per_day", 0.0))
    allow_vent = bool(inputs.get("vent_to_flare", True))

    print(f"gas storage Q_in {Q_in}, Q_out: {Q_out_req}")

    # pressure setpoint handling (bar) - if provided, store it for internal control
    if "set_pressure" in inputs:
        sp = inputs["set_pressure"]
        try:
            self.pressure_setpoint_bar = None if sp is None else float(sp)
        except Exception as e:
            # ignore invalid setpoint
            print(e)

    # convert to volumes for this dt
    vol_in = Q_in * dt
    vol_out_req = Q_out_req * dt

    vented_this_step = 0.0
    # supplied_this_step = 0.0

    # Inflow: attempt to store incoming gas
    free_capacity_m3 = self.capacity_m3 - self.stored_volume_m3
    if vol_in <= free_capacity_m3 + 1e-12:
        # all incoming stored
        self.stored_volume_m3 += vol_in
        overflow = 0.0
    else:
        # store what fits, overflow is vented (or forwarded externally)
        stored = max(0.0, free_capacity_m3)
        overflow = vol_in - stored
        self.stored_volume_m3 += stored
        if allow_vent:
            vented_this_step += overflow
            self._cum_vented_m3 += overflow
        else:
            # if venting not allowed, assume overflow is rejected (lost) (count as vented for safety)
            vented_this_step += overflow
            self._cum_vented_m3 += overflow

    # Supply (outflow): cannot supply more than stored
    # If pressure below p_min, limit supply (simulate pressure control)
    current_pressure = self._estimate_pressure_bar()

    # If a pressure setpoint is requested and higher than current,
    # we might choose to deny outflow to increase pressure
    restrict_factor = 1.0
    if self.pressure_setpoint_bar is not None:
        # If setpoint is higher than current pressure, prioritize charging (i.e., restrict outflow)
        if self.pressure_setpoint_bar > current_pressure:
            # reduce allowed outflow proportionally
            # factor between 0..1 -> 0 means block outflow, 1 means full
            gap = self.pressure_setpoint_bar - current_pressure
            span = max(1e-6, self.p_max_bar - self.p_atm_bar)
            restrict_factor = max(0.0, 1.0 - gap / span)

    allowed_out_vol = self.stored_volume_m3 * restrict_factor  # simple limit
    desired_out_vol = min(vol_out_req, allowed_out_vol)

    # Also ensure we don't drop below zero or below some minimum reserve needed to maintain p_min:
    # approximate required volume to keep p >= p_min: invert pressure estimate crudely by fraction
    if self.storage_type in ("membrane", "dome"):
        # frac required to maintain p_min: (p_min - p_atm)/(p_max - p_atm)
        if (self.p_max_bar - self.p_atm_bar) > 1e-9:
            frac_needed = max(
                0.0,
                (self.p_min_bar - self.p_atm_bar) / (self.p_max_bar - self.p_atm_bar),
            )
        else:
            frac_needed = 0.0
    else:  # compressed
        # invert nonlinear mapping p = p_min + frac^alpha*(p_max-p_min)
        if (self.p_max_bar - self.p_min_bar) > 1e-9:
            alpha = 2.0
            frac_needed = max(
                0.0,
                ((self.p_min_bar - self.p_min_bar) / (self.p_max_bar - self.p_min_bar)) ** (1.0 / alpha),
            )
            # above formula trivial -> 0, but keep general structure; use small reserve fraction
            frac_needed = 0.01
        else:
            frac_needed = 0.0

    reserve_volume = frac_needed * self.capacity_m3
    # ensure not to supply below reserve
    max_out_vol_after_reserve = max(0.0, self.stored_volume_m3 - reserve_volume)
    desired_out_vol = min(desired_out_vol, max_out_vol_after_reserve)

    # perform outflow
    self.stored_volume_m3 -= desired_out_vol
    supplied_this_step_m3_per_day = desired_out_vol / max(dt, 1e-12)  # convert back to m3/day

    print("storage tank:", supplied_this_step_m3_per_day, self.stored_volume_m3)

    # After flows, update pressure and check safety
    current_pressure = self._estimate_pressure_bar()

    # Safety: overpressure -> vent to flare
    if current_pressure > self.p_max_bar:
        # compute how much volume must be removed to reach p_max
        # invert our pressure models approximately by reducing stored fraction until p <= p_max
        # perform simple linear backoff: remove fraction proportional to pressure exceedance
        if self.storage_type in ("membrane", "dome"):
            # p = p_atm + frac*(p_max - p_atm), want frac_target = (p_max - p_atm)/(p_max - p_atm) = 1.0
            # but if p_max_bar is absolute safety we want to bring pressure to p_max_bar.
            # compute fraction that yields p_max_bar
            target_frac = max(
                0.0,
                min(
                    1.0,
                    (self.p_max_bar - self.p_atm_bar) / max(1e-9, (self.p_max_bar - self.p_atm_bar)),
                ),
            )
            # if current fraction > 1.0 (theory), set to 1.0
            target_frac = min(target_frac, 1.0)
            target_volume = target_frac * self.capacity_m3
            if self.stored_volume_m3 > target_volume:
                vent = self.stored_volume_m3 - target_volume
            else:
                vent = 0.0
        else:  # compressed
            # reduce to fraction that yields p_max_bar approx equal to 1.0 fraction
            # simpler: vent until stored <= capacity * 0.999 (small safety)
            target_volume = min(self.capacity_m3, self.capacity_m3 * 0.999)
            vent = max(0.0, self.stored_volume_m3 - target_volume)

        # Vent if allowed (or always vent for safety)
        if allow_vent and vent > 0.0:
            vented_this_step += vent
            self.stored_volume_m3 -= vent
            self._cum_vented_m3 += vent

        # recompute pressure
        current_pressure = self._estimate_pressure_bar()

    # Clamp stored volume
    self.stored_volume_m3 = max(0.0, min(self.stored_volume_m3, self.capacity_m3))

    # Update outputs_data
    self.outputs_data = {
        "stored_volume_m3": float(self.stored_volume_m3),
        "pressure_bar": float(current_pressure),
        "utilization": float(self.stored_volume_m3 / max(1e-9, self.capacity_m3)),
        "vented_volume_m3": float(vented_this_step),
        "Q_gas_supplied_m3_per_day": float(supplied_this_step_m3_per_day),
        "cumulative_vented_m3": float(self._cum_vented_m3),
        "pressure_setpoint_bar": self.pressure_setpoint_bar,
    }

    return self.outputs_data

to_dict()

Serialize configuration + current state.

Source code in pyadm1/components/energy/gas_storage.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize configuration + current state."""
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "storage_type": self.storage_type,
        "capacity_m3": self.capacity_m3,
        "p_atm_bar": self.p_atm_bar,
        "p_min_bar": self.p_min_bar,
        "p_max_bar": self.p_max_bar,
        "stored_volume_m3": self.stored_volume_m3,
        "pressure_setpoint_bar": self.pressure_setpoint_bar,
        "outputs_data": self.outputs_data,
        "inputs": self.inputs,
        "outputs": self.outputs,
    }

pyadm1.components.energy.flare.Flare

Bases: Component

Flare component for combusting vented biogas.

The flare accepts an input Q_gas_in_m3_per_day and will combust it. It reports vented_volume_m3 for the current timestep and cumulative_vented_m3.

Parameters

component_id : str Unique id for the flare component. destruction_efficiency : float Fraction of methane destroyed (0..1). Default 0.98. name : Optional[str] Human readable name.

Source code in pyadm1/components/energy/flare.py

class Flare(Component):
    """Flare component for combusting vented biogas.

    The flare accepts an input `Q_gas_in_m3_per_day` and will combust it.
    It reports `vented_volume_m3` for the current timestep and `cumulative_vented_m3`.

    Parameters
    ----------
    component_id : str
        Unique id for the flare component.
    destruction_efficiency : float
        Fraction of methane destroyed (0..1). Default 0.98.
    name : Optional[str]
        Human readable name.
    """

    def __init__(
        self,
        component_id: str,
        destruction_efficiency: float = 0.98,
        name: Optional[str] = None,
    ):
        super().__init__(component_id, ComponentType.STORAGE, name)
        self.destruction_efficiency: float = float(destruction_efficiency)
        self._cum_vented_m3: float = 0.0
        self.initialize()

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

        Args:
            initial_state: optional dict with 'cumulative_vented_m3' to restore state.
        """
        if initial_state and "cumulative_vented_m3" in initial_state:
            try:
                self._cum_vented_m3 = float(initial_state["cumulative_vented_m3"])
            except Exception:
                self._cum_vented_m3 = 0.0
        else:
            self._cum_vented_m3 = 0.0

        self.outputs_data = {
            "vented_volume_m3": 0.0,
            "cumulative_vented_m3": self._cum_vented_m3,
            "CH4_destroyed_m3": 0.0,
        }
        self._initialized = True

    def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """Process one timestep and combust incoming gas.

        Args:
            t: current simulation time [days]
            dt: timestep length [days]
            inputs: dictionary that may contain:
                - 'Q_gas_in_m3_per_day': inflow (m³/day)
                - 'CH4_fraction': methane fraction in the gas (0..1). Default 0.6

        Returns:
            outputs_data dict with keys:
                - 'vented_volume_m3' (this timestep)
                - 'cumulative_vented_m3'
                - 'CH4_destroyed_m3' (m³ of CH4 destroyed this step)
        """
        Q_in = float(inputs.get("Q_gas_in_m3_per_day", 0.0))
        ch4_frac = float(inputs.get("CH4_fraction", 0.6))

        # convert to volume in this timestep
        vol_in = Q_in * dt

        # combustion: a flare destroys a fraction of incoming CH4
        ch4_volume = vol_in * ch4_frac
        ch4_destroyed = ch4_volume * self.destruction_efficiency

        # record vented (treated) volume
        self._cum_vented_m3 += vol_in

        self.outputs_data = {
            "vented_volume_m3": float(vol_in),
            "cumulative_vented_m3": float(self._cum_vented_m3),
            "CH4_destroyed_m3": float(ch4_destroyed),
        }

        return self.outputs_data

    def to_dict(self) -> Dict[str, Any]:
        """Serialize flare configuration and state."""
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "destruction_efficiency": self.destruction_efficiency,
            "cumulative_vented_m3": self._cum_vented_m3,
        }

    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> "Flare":
        """Instantiate Flare from dict created by `to_dict`."""
        flare = cls(
            component_id=config["component_id"],
            destruction_efficiency=config.get("destruction_efficiency", 0.98),
            name=config.get("name"),
        )
        flare.initialize({"cumulative_vented_m3": config.get("cumulative_vented_m3", 0.0)})
        return flare

Functions

from_dict(config) classmethod

Instantiate Flare from dict created by to_dict.

Source code in pyadm1/components/energy/flare.py

@classmethod
def from_dict(cls, config: Dict[str, Any]) -> "Flare":
    """Instantiate Flare from dict created by `to_dict`."""
    flare = cls(
        component_id=config["component_id"],
        destruction_efficiency=config.get("destruction_efficiency", 0.98),
        name=config.get("name"),
    )
    flare.initialize({"cumulative_vented_m3": config.get("cumulative_vented_m3", 0.0)})
    return flare

initialize(initial_state=None)

Initialize flare internal state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	optional dict with 'cumulative_vented_m3' to restore state.	None

Source code in pyadm1/components/energy/flare.py

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

    Args:
        initial_state: optional dict with 'cumulative_vented_m3' to restore state.
    """
    if initial_state and "cumulative_vented_m3" in initial_state:
        try:
            self._cum_vented_m3 = float(initial_state["cumulative_vented_m3"])
        except Exception:
            self._cum_vented_m3 = 0.0
    else:
        self._cum_vented_m3 = 0.0

    self.outputs_data = {
        "vented_volume_m3": 0.0,
        "cumulative_vented_m3": self._cum_vented_m3,
        "CH4_destroyed_m3": 0.0,
    }
    self._initialized = True

step(t, dt, inputs)

Process one timestep and combust incoming gas.

Parameters:

Name	Type	Description	Default
t	float	current simulation time [days]	required
dt	float	timestep length [days]	required
inputs	Dict[str, Any]	dictionary that may contain: - 'Q_gas_in_m3_per_day': inflow (m³/day) - 'CH4_fraction': methane fraction in the gas (0..1). Default 0.6	required

Returns:

Type	Description
Dict[str, Any]	outputs_data dict with keys: - 'vented_volume_m3' (this timestep) - 'cumulative_vented_m3' - 'CH4_destroyed_m3' (m³ of CH4 destroyed this step)

Source code in pyadm1/components/energy/flare.py

def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
    """Process one timestep and combust incoming gas.

    Args:
        t: current simulation time [days]
        dt: timestep length [days]
        inputs: dictionary that may contain:
            - 'Q_gas_in_m3_per_day': inflow (m³/day)
            - 'CH4_fraction': methane fraction in the gas (0..1). Default 0.6

    Returns:
        outputs_data dict with keys:
            - 'vented_volume_m3' (this timestep)
            - 'cumulative_vented_m3'
            - 'CH4_destroyed_m3' (m³ of CH4 destroyed this step)
    """
    Q_in = float(inputs.get("Q_gas_in_m3_per_day", 0.0))
    ch4_frac = float(inputs.get("CH4_fraction", 0.6))

    # convert to volume in this timestep
    vol_in = Q_in * dt

    # combustion: a flare destroys a fraction of incoming CH4
    ch4_volume = vol_in * ch4_frac
    ch4_destroyed = ch4_volume * self.destruction_efficiency

    # record vented (treated) volume
    self._cum_vented_m3 += vol_in

    self.outputs_data = {
        "vented_volume_m3": float(vol_in),
        "cumulative_vented_m3": float(self._cum_vented_m3),
        "CH4_destroyed_m3": float(ch4_destroyed),
    }

    return self.outputs_data

to_dict()

Serialize flare configuration and state.

Source code in pyadm1/components/energy/flare.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize flare configuration and state."""
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "destruction_efficiency": self.destruction_efficiency,
        "cumulative_vented_m3": self._cum_vented_m3,
    }

pyadm1.components.energy.boiler.Boiler

Bases: Component

Auxiliary boiler component (stub for future implementation).

Models a boiler that provides heat to the system when CHP heat is insufficient.

Attributes:

Name	Type	Description
component_id		Unique identifier.
P_th_nom		Nominal thermal power [kW].
efficiency		Thermal efficiency (0-1).
name		Optional name.

Source code in pyadm1/components/energy/boiler.py

class Boiler(Component):
    """
    Auxiliary boiler component (stub for future implementation).

    Models a boiler that provides heat to the system when CHP heat is
    insufficient.

    Attributes:
        component_id: Unique identifier.
        P_th_nom: Nominal thermal power [kW].
        efficiency: Thermal efficiency (0-1).
        name: Optional name.
    """

    def __init__(
        self,
        component_id: str,
        P_th_nom: float = 500.0,
        efficiency: float = 0.9,
        name: Optional[str] = None,
    ):
        """
        Initialize the Boiler.

        Args:
            component_id: Unique identifier.
            P_th_nom: Nominal power [kW].
            efficiency: Boiler efficiency.
            name: Optional name.
        """
        super().__init__(component_id, ComponentType.BOILER, name)
        self.P_th_nom = P_th_nom
        self.efficiency = efficiency

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

        Args:
            initial_state: Optional initial state.
        """
        self.state = {}

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

        Args:
            t: Current time [days].
            dt: Time step [days].
            inputs: Input dictionary.

        Returns:
            Output dictionary.
        """
        return {}

    def to_dict(self) -> Dict[str, Any]:
        """
        Serialize to dictionary.

        Returns:
            Configuration dictionary.
        """
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
        }

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

        Args:
            config: Configuration dictionary.

        Returns:
            New Boiler instance.
        """
        return cls(config["component_id"])

Functions

__init__(component_id, P_th_nom=500.0, efficiency=0.9, name=None)

Initialize the Boiler.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier.	required
P_th_nom	float	Nominal power [kW].	500.0
efficiency	float	Boiler efficiency.	0.9
name	Optional[str]	Optional name.	None

Source code in pyadm1/components/energy/boiler.py

def __init__(
    self,
    component_id: str,
    P_th_nom: float = 500.0,
    efficiency: float = 0.9,
    name: Optional[str] = None,
):
    """
    Initialize the Boiler.

    Args:
        component_id: Unique identifier.
        P_th_nom: Nominal power [kW].
        efficiency: Boiler efficiency.
        name: Optional name.
    """
    super().__init__(component_id, ComponentType.BOILER, name)
    self.P_th_nom = P_th_nom
    self.efficiency = efficiency

from_dict(config) classmethod

Create instance from dictionary.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Configuration dictionary.	required

Returns:

Type	Description
Boiler	New Boiler instance.

Source code in pyadm1/components/energy/boiler.py

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

    Args:
        config: Configuration dictionary.

    Returns:
        New Boiler instance.
    """
    return cls(config["component_id"])

initialize(initial_state=None)

Initialize component state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Optional initial state.	None

Source code in pyadm1/components/energy/boiler.py

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

    Args:
        initial_state: Optional initial state.
    """
    self.state = {}

step(t, dt, inputs)

Execute one simulation step.

Parameters:

Name	Type	Description	Default
t	float	Current time [days].	required
dt	float	Time step [days].	required
inputs	Dict[str, Any]	Input dictionary.	required

Returns:

Type	Description
Dict[str, Any]	Output dictionary.

Source code in pyadm1/components/energy/boiler.py

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

    Args:
        t: Current time [days].
        dt: Time step [days].
        inputs: Input dictionary.

    Returns:
        Output dictionary.
    """
    return {}

to_dict()

Serialize to dictionary.

Returns:

Type	Description
Dict[str, Any]	Configuration dictionary.

Source code in pyadm1/components/energy/boiler.py

def to_dict(self) -> Dict[str, Any]:
    """
    Serialize to dictionary.

    Returns:
        Configuration dictionary.
    """
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
    }