Skip to content

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 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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_available is the input thermal power from methane; multiply by eta_el
        # to get the maximum possible electrical output power.
        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,
            "P_th_available": P_th,  # Key expected by HeatingSystem
            "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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
@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 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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_available is the input thermal power from methane; multiply by eta_el
    # to get the maximum possible electrical output power.
    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,
        "P_th_available": P_th,  # Key expected by HeatingSystem
        "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 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
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.
feedstock

Feedstock used to derive per-substrate density and c_p for the sensible-heat term. Optional — when omitted, only the UAΔT loss term contributes to demand.
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 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
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.
        feedstock: Feedstock used to derive per-substrate density and c_p
            for the sensible-heat term. Optional — when omitted, only
            the UAΔT loss term contributes to demand.

    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,
        feedstock=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.
            feedstock: Optional :class:`Feedstock` for sensible-heat calculation.
        """
        super().__init__(component_id, ComponentType.HEATING, name)

        self.target_temperature = target_temperature
        self.heat_loss_coefficient = heat_loss_coefficient
        self.feedstock = feedstock

        # 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]
                - 'Q_substrates': Substrate feed rates [m³/d]

        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.feedstock, self.target_temperature, T_ambient)

        # 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, feedstock=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
feedstock

Optional :class:Feedstock for sensible-heat calculation.

 None
Source code in pyadm1/components/energy/heating.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def __init__(
    self,
    component_id: str,
    target_temperature: float = 308.15,
    heat_loss_coefficient: float = 0.5,
    name: Optional[str] = None,
    feedstock=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.
        feedstock: Optional :class:`Feedstock` for sensible-heat calculation.
    """
    super().__init__(component_id, ComponentType.HEATING, name)

    self.target_temperature = target_temperature
    self.heat_loss_coefficient = heat_loss_coefficient
    self.feedstock = feedstock

    # 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 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@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 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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] - 'Q_substrates': Substrate feed rates [m³/d]

 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 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
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]
            - 'Q_substrates': Substrate feed rates [m³/d]

    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.feedstock, self.target_temperature, T_ambient)

    # 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 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
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 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
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))

        # 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

        # 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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
@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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
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))

    # 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

    # 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 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
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 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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.FLARE, 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 
110
111
112
113
114
115
116
117
118
119
@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 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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 
100
101
102
103
104
105
106
107
108
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 gas boiler for backup and peak heating.

Covers the residual heat demand (P_aux_heat from :class:HeatingSystem) that CHP waste heat cannot supply. Supports biogas-only, natural-gas-only, or dual-fuel operation.

Attributes:

Name Type Description
P_th_nom float

Nominal thermal output [kW].
efficiency float

Rated thermal efficiency at full load (0-1).
fuel_type str

Fuel mode: "biogas" | "natural_gas" | "dual".
lhv_biogas float

Lower heating value of biogas [kWh/m3].
lhv_natural_gas float

LHV of natural gas [kWh/m3].
Example

boiler = Boiler("boiler_1", P_th_nom=200.0, fuel_type="dual", ... efficiency=0.92) boiler.initialize() out = boiler.step(t=0, dt=1/24, ... inputs={"P_th_demand": 80.0, ... "Q_gas_available_m3_per_day": 50.0})

Source code in pyadm1/components/energy/boiler.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
class Boiler(Component):
    """
    Auxiliary gas boiler for backup and peak heating.

    Covers the residual heat demand (``P_aux_heat`` from :class:`HeatingSystem`)
    that CHP waste heat cannot supply.  Supports biogas-only, natural-gas-only,
    or dual-fuel operation.

    Attributes:
        P_th_nom (float):    Nominal thermal output [kW].
        efficiency (float):  Rated thermal efficiency at full load (0-1).
        fuel_type (str):     Fuel mode: ``"biogas"`` | ``"natural_gas"`` | ``"dual"``.
        lhv_biogas (float):  Lower heating value of biogas [kWh/m3].
        lhv_natural_gas (float): LHV of natural gas [kWh/m3].

    Example:
        >>> boiler = Boiler("boiler_1", P_th_nom=200.0, fuel_type="dual",
        ...                 efficiency=0.92)
        >>> boiler.initialize()
        >>> out = boiler.step(t=0, dt=1/24,
        ...     inputs={"P_th_demand": 80.0,
        ...             "Q_gas_available_m3_per_day": 50.0})
    """

    def __init__(
        self,
        component_id: str,
        P_th_nom: float = 200.0,
        efficiency: float = 0.90,
        fuel_type: str = "dual",
        lhv_biogas: float = _LHV_BIOGAS,
        lhv_natural_gas: float = _LHV_NATURAL_GAS,
        name: Optional[str] = None,
    ):
        """
        Initialize boiler.

        Args:
            component_id:    Unique identifier.
            P_th_nom:        Nominal thermal power [kW].  Default 200.
            efficiency:      Rated full-load thermal efficiency (0-1).  Default 0.90.
            fuel_type:       Fuel mode — ``"biogas"``, ``"natural_gas"``,
                             or ``"dual"`` (biogas first, then natural gas).
                             Default ``"dual"``.
            lhv_biogas:      Lower heating value of biogas [kWh/m3].
                             Default 6.0 (KTBL 2013 at 60% CH4).
            lhv_natural_gas: Lower heating value of natural gas [kWh/m3].
                             Default 10.0 (DVGW G 260 H-gas).
            name:            Human-readable display name.
        """
        super().__init__(component_id, ComponentType.BOILER, name)

        self.P_th_nom = float(P_th_nom)
        self.efficiency = float(efficiency)

        fuel_type = fuel_type.lower()
        if fuel_type not in ("biogas", "natural_gas", "dual"):
            raise ValueError("fuel_type must be 'biogas', 'natural_gas', or 'dual'")
        self.fuel_type = fuel_type

        self.lhv_biogas = float(lhv_biogas)
        self.lhv_natural_gas = float(lhv_natural_gas)

        # Cumulative tracking
        self.energy_supplied = 0.0  # kWh
        self.gas_consumed_total = 0.0  # m3 biogas
        self.ng_consumed_total = 0.0  # m3 natural gas
        self.operating_hours = 0.0  # h

        self.initialize()

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

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

        Args:
            initial_state: Optional dict with keys:
                - ``energy_supplied``:    cumulative heat output [kWh]
                - ``gas_consumed_total``: cumulative biogas consumed [m3]
                - ``ng_consumed_total``:  cumulative natural gas consumed [m3]
                - ``operating_hours``:    cumulative run time [h]
        """
        if initial_state:
            self.energy_supplied = float(initial_state.get("energy_supplied", 0.0))
            self.gas_consumed_total = float(initial_state.get("gas_consumed_total", 0.0))
            self.ng_consumed_total = float(initial_state.get("ng_consumed_total", 0.0))
            self.operating_hours = float(initial_state.get("operating_hours", 0.0))

        self.state = {
            "energy_supplied": self.energy_supplied,
            "gas_consumed_total": self.gas_consumed_total,
            "ng_consumed_total": self.ng_consumed_total,
            "operating_hours": self.operating_hours,
        }

        self.outputs_data = {
            "P_th_supplied": 0.0,
            "P_th_available": 0.0,  # alias — HeatingSystem reads this key
            "Q_gas_consumed_m3_per_day": 0.0,
            "Q_natural_gas_m3_per_day": 0.0,
            "is_running": False,
            "load_fraction": 0.0,
            "efficiency_actual": self.efficiency,
        }

        self._initialized = True

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

        The boiler fires to satisfy ``P_th_demand`` up to its nominal
        capacity.  Gas is drawn first from available biogas (if provided),
        then from the natural gas grid if the fuel mode allows it.

        Args:
            t:  Current simulation time [days].
            dt: Time step [days].
            inputs: Dict with optional keys:
                - ``P_th_demand`` [kW]:
                    Heat demand to be covered.  Accepts ``P_aux_heat``
                    (the key emitted by :class:`HeatingSystem`) as alias.
                - ``Q_gas_available_m3_per_day`` [m3/d]:
                    Biogas available from gas storage.  Only relevant for
                    ``fuel_type`` ``"biogas"`` or ``"dual"``.
                - ``enable`` [bool]:
                    If *False*, boiler is forced off.  Default *True*.

        Returns:
            Dict with keys:
                - ``P_th_supplied`` [kW]:
                    Actual thermal output delivered.
                - ``P_th_available`` [kW]:
                    Alias for ``P_th_supplied`` (for HeatingSystem input).
                - ``Q_gas_consumed_m3_per_day`` [m3/d]:
                    Biogas consumption rate.
                - ``Q_natural_gas_m3_per_day`` [m3/d]:
                    Natural gas consumption rate.
                - ``is_running`` [bool]:
                    Whether the boiler is currently firing.
                - ``load_fraction`` [float]:
                    Current output as fraction of nominal (0-1).
                - ``efficiency_actual`` [float]:
                    Effective efficiency at this load point.
        """
        enable = bool(inputs.get("enable", True))

        # Accept both P_th_demand and P_aux_heat (HeatingSystem output key)
        P_demand = float(inputs.get("P_th_demand", inputs.get("P_aux_heat", 0.0)))
        Q_gas_avail = float(inputs.get("Q_gas_available_m3_per_day", 0.0))

        if not enable or P_demand <= 0.0:
            self._update_outputs(0.0, 0.0, 0.0, dt)
            return self.outputs_data

        # Clamp demand to nominal capacity
        P_th_supplied = min(P_demand, self.P_th_nom)
        load_fraction = P_th_supplied / self.P_th_nom

        # Part-load efficiency (linear correction, VDI 4631)
        eta_actual = self.efficiency * (1.0 - _PART_LOAD_PENALTY * (1.0 - load_fraction))
        eta_actual = max(0.01, eta_actual)

        # --- Fuel allocation ------------------------------------------
        Q_gas_consumed = 0.0  # m3/d biogas
        Q_ng_consumed = 0.0  # m3/d natural gas

        if self.fuel_type == "natural_gas":
            # Burn natural gas only
            Q_ng_consumed = P_th_supplied * 24.0 / (eta_actual * self.lhv_natural_gas)

        elif self.fuel_type == "biogas":
            # Burn biogas only (even if unavailable — caller must manage supply)
            Q_gas_consumed = P_th_supplied * 24.0 / (eta_actual * self.lhv_biogas)

        else:  # dual fuel: biogas first, natural gas for remainder
            # Maximum biogas combustion that would cover the demand
            Q_gas_needed_total = P_th_supplied * 24.0 / (eta_actual * self.lhv_biogas)

            if Q_gas_avail >= Q_gas_needed_total:
                # Sufficient biogas: no natural gas needed
                Q_gas_consumed = Q_gas_needed_total
            else:
                # Burn all available biogas, supplement with natural gas
                Q_gas_consumed = Q_gas_avail
                P_from_biogas = Q_gas_consumed * eta_actual * self.lhv_biogas / 24.0
                P_from_ng = P_th_supplied - P_from_biogas
                Q_ng_consumed = P_from_ng * 24.0 / (eta_actual * self.lhv_natural_gas)

        self._update_outputs(P_th_supplied, Q_gas_consumed, Q_ng_consumed, dt)
        return self.outputs_data

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

    def to_dict(self) -> Dict[str, Any]:
        """Serialize configuration and cumulative state to dictionary."""
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "P_th_nom": self.P_th_nom,
            "efficiency": self.efficiency,
            "fuel_type": self.fuel_type,
            "lhv_biogas": self.lhv_biogas,
            "lhv_natural_gas": self.lhv_natural_gas,
            "inputs": self.inputs,
            "outputs": self.outputs,
            "state": self.state,
        }

    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> "Boiler":
        """
        Create Boiler from dictionary (produced by :meth:`to_dict`).

        Args:
            config: Configuration dictionary.

        Returns:
            Initialized Boiler instance.
        """
        boiler = cls(
            component_id=config["component_id"],
            P_th_nom=config.get("P_th_nom", 200.0),
            efficiency=config.get("efficiency", 0.90),
            fuel_type=config.get("fuel_type", "dual"),
            lhv_biogas=config.get("lhv_biogas", _LHV_BIOGAS),
            lhv_natural_gas=config.get("lhv_natural_gas", _LHV_NATURAL_GAS),
            name=config.get("name"),
        )

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

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

        return boiler

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

    def _update_outputs(
        self,
        P_th_supplied: float,
        Q_gas_consumed: float,
        Q_ng_consumed: float,
        dt: float,
    ) -> None:
        """Update outputs_data and cumulative state variables."""
        is_running = P_th_supplied > 0.0
        load_fraction = P_th_supplied / self.P_th_nom if self.P_th_nom > 0 else 0.0
        eta_actual = self.efficiency * (1.0 - _PART_LOAD_PENALTY * (1.0 - load_fraction)) if is_running else self.efficiency

        dt_hours = dt * 24.0
        self.energy_supplied += P_th_supplied * dt_hours  # kWh
        self.gas_consumed_total += Q_gas_consumed * dt  # m3
        self.ng_consumed_total += Q_ng_consumed * dt  # m3
        if is_running:
            self.operating_hours += dt_hours

        self.state.update(
            {
                "energy_supplied": self.energy_supplied,
                "gas_consumed_total": self.gas_consumed_total,
                "ng_consumed_total": self.ng_consumed_total,
                "operating_hours": self.operating_hours,
            }
        )

        self.outputs_data = {
            "P_th_supplied": float(P_th_supplied),
            "P_th_available": float(P_th_supplied),  # HeatingSystem input key
            "Q_gas_consumed_m3_per_day": float(Q_gas_consumed),
            "Q_natural_gas_m3_per_day": float(Q_ng_consumed),
            "is_running": bool(is_running),
            "load_fraction": float(load_fraction),
            "efficiency_actual": float(eta_actual),
            # Cumulative
            "energy_supplied_kwh": float(self.energy_supplied),
            "gas_consumed_total_m3": float(self.gas_consumed_total),
            "ng_consumed_total_m3": float(self.ng_consumed_total),
            "operating_hours": float(self.operating_hours),
        }

Functions

__init__(component_id, P_th_nom=200.0, efficiency=0.9, fuel_type='dual', lhv_biogas=_LHV_BIOGAS, lhv_natural_gas=_LHV_NATURAL_GAS, name=None)

Initialize boiler.

Parameters:

Name Type Description Default
component_id str

Unique identifier.

 required
P_th_nom float

Nominal thermal power [kW]. Default 200.

 200.0
efficiency float

Rated full-load thermal efficiency (0-1). Default 0.90.

 0.9
fuel_type str

Fuel mode — "biogas", "natural_gas", or "dual" (biogas first, then natural gas). Default "dual".

 'dual'
lhv_biogas float

Lower heating value of biogas [kWh/m3]. Default 6.0 (KTBL 2013 at 60% CH4).

 _LHV_BIOGAS
lhv_natural_gas float

Lower heating value of natural gas [kWh/m3]. Default 10.0 (DVGW G 260 H-gas).

 _LHV_NATURAL_GAS
name Optional[str]

Human-readable display name.

 None
Source code in pyadm1/components/energy/boiler.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def __init__(
    self,
    component_id: str,
    P_th_nom: float = 200.0,
    efficiency: float = 0.90,
    fuel_type: str = "dual",
    lhv_biogas: float = _LHV_BIOGAS,
    lhv_natural_gas: float = _LHV_NATURAL_GAS,
    name: Optional[str] = None,
):
    """
    Initialize boiler.

    Args:
        component_id:    Unique identifier.
        P_th_nom:        Nominal thermal power [kW].  Default 200.
        efficiency:      Rated full-load thermal efficiency (0-1).  Default 0.90.
        fuel_type:       Fuel mode — ``"biogas"``, ``"natural_gas"``,
                         or ``"dual"`` (biogas first, then natural gas).
                         Default ``"dual"``.
        lhv_biogas:      Lower heating value of biogas [kWh/m3].
                         Default 6.0 (KTBL 2013 at 60% CH4).
        lhv_natural_gas: Lower heating value of natural gas [kWh/m3].
                         Default 10.0 (DVGW G 260 H-gas).
        name:            Human-readable display name.
    """
    super().__init__(component_id, ComponentType.BOILER, name)

    self.P_th_nom = float(P_th_nom)
    self.efficiency = float(efficiency)

    fuel_type = fuel_type.lower()
    if fuel_type not in ("biogas", "natural_gas", "dual"):
        raise ValueError("fuel_type must be 'biogas', 'natural_gas', or 'dual'")
    self.fuel_type = fuel_type

    self.lhv_biogas = float(lhv_biogas)
    self.lhv_natural_gas = float(lhv_natural_gas)

    # Cumulative tracking
    self.energy_supplied = 0.0  # kWh
    self.gas_consumed_total = 0.0  # m3 biogas
    self.ng_consumed_total = 0.0  # m3 natural gas
    self.operating_hours = 0.0  # h

    self.initialize()

from_dict(config) classmethod

Create Boiler from dictionary (produced by :meth:to_dict).

Parameters:

Name Type Description Default
config Dict[str, Any]

Configuration dictionary.

 required

Returns:

Type Description
Boiler

Initialized Boiler instance.
Source code in pyadm1/components/energy/boiler.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
@classmethod
def from_dict(cls, config: Dict[str, Any]) -> "Boiler":
    """
    Create Boiler from dictionary (produced by :meth:`to_dict`).

    Args:
        config: Configuration dictionary.

    Returns:
        Initialized Boiler instance.
    """
    boiler = cls(
        component_id=config["component_id"],
        P_th_nom=config.get("P_th_nom", 200.0),
        efficiency=config.get("efficiency", 0.90),
        fuel_type=config.get("fuel_type", "dual"),
        lhv_biogas=config.get("lhv_biogas", _LHV_BIOGAS),
        lhv_natural_gas=config.get("lhv_natural_gas", _LHV_NATURAL_GAS),
        name=config.get("name"),
    )

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

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

    return boiler

initialize(initial_state=None)

Initialize boiler state.

Parameters:

Name Type Description Default
initial_state Optional[Dict[str, Any]]

Optional dict with keys: - energy_supplied: cumulative heat output [kWh] - gas_consumed_total: cumulative biogas consumed [m3] - ng_consumed_total: cumulative natural gas consumed [m3] - operating_hours: cumulative run time [h]

 None
Source code in pyadm1/components/energy/boiler.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def initialize(self, initial_state: Optional[Dict[str, Any]] = None) -> None:
    """
    Initialize boiler state.

    Args:
        initial_state: Optional dict with keys:
            - ``energy_supplied``:    cumulative heat output [kWh]
            - ``gas_consumed_total``: cumulative biogas consumed [m3]
            - ``ng_consumed_total``:  cumulative natural gas consumed [m3]
            - ``operating_hours``:    cumulative run time [h]
    """
    if initial_state:
        self.energy_supplied = float(initial_state.get("energy_supplied", 0.0))
        self.gas_consumed_total = float(initial_state.get("gas_consumed_total", 0.0))
        self.ng_consumed_total = float(initial_state.get("ng_consumed_total", 0.0))
        self.operating_hours = float(initial_state.get("operating_hours", 0.0))

    self.state = {
        "energy_supplied": self.energy_supplied,
        "gas_consumed_total": self.gas_consumed_total,
        "ng_consumed_total": self.ng_consumed_total,
        "operating_hours": self.operating_hours,
    }

    self.outputs_data = {
        "P_th_supplied": 0.0,
        "P_th_available": 0.0,  # alias — HeatingSystem reads this key
        "Q_gas_consumed_m3_per_day": 0.0,
        "Q_natural_gas_m3_per_day": 0.0,
        "is_running": False,
        "load_fraction": 0.0,
        "efficiency_actual": self.efficiency,
    }

    self._initialized = True

step(t, dt, inputs)

Perform one simulation time step.

The boiler fires to satisfy P_th_demand up to its nominal capacity. Gas is drawn first from available biogas (if provided), then from the natural gas grid if the fuel mode allows it.

Parameters:

Name Type Description Default
t float

Current simulation time [days].

 required
dt float

Time step [days].

 required
inputs Dict[str, Any]

Dict with optional keys: - P_th_demand [kW]: Heat demand to be covered. Accepts P_aux_heat (the key emitted by :class:HeatingSystem) as alias. - Q_gas_available_m3_per_day [m3/d]: Biogas available from gas storage. Only relevant for fuel_type "biogas" or "dual". - enable [bool]: If False, boiler is forced off. Default True.

 required

Returns:

Type Description
Dict[str, Any]

Dict with keys: - P_th_supplied [kW]: Actual thermal output delivered. - P_th_available [kW]: Alias for P_th_supplied (for HeatingSystem input). - Q_gas_consumed_m3_per_day [m3/d]: Biogas consumption rate. - Q_natural_gas_m3_per_day [m3/d]: Natural gas consumption rate. - is_running [bool]: Whether the boiler is currently firing. - load_fraction [float]: Current output as fraction of nominal (0-1). - efficiency_actual [float]: Effective efficiency at this load point.
Source code in pyadm1/components/energy/boiler.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def step(self, t: float, dt: float, inputs: Dict[str, Any]) -> Dict[str, Any]:
    """
    Perform one simulation time step.

    The boiler fires to satisfy ``P_th_demand`` up to its nominal
    capacity.  Gas is drawn first from available biogas (if provided),
    then from the natural gas grid if the fuel mode allows it.

    Args:
        t:  Current simulation time [days].
        dt: Time step [days].
        inputs: Dict with optional keys:
            - ``P_th_demand`` [kW]:
                Heat demand to be covered.  Accepts ``P_aux_heat``
                (the key emitted by :class:`HeatingSystem`) as alias.
            - ``Q_gas_available_m3_per_day`` [m3/d]:
                Biogas available from gas storage.  Only relevant for
                ``fuel_type`` ``"biogas"`` or ``"dual"``.
            - ``enable`` [bool]:
                If *False*, boiler is forced off.  Default *True*.

    Returns:
        Dict with keys:
            - ``P_th_supplied`` [kW]:
                Actual thermal output delivered.
            - ``P_th_available`` [kW]:
                Alias for ``P_th_supplied`` (for HeatingSystem input).
            - ``Q_gas_consumed_m3_per_day`` [m3/d]:
                Biogas consumption rate.
            - ``Q_natural_gas_m3_per_day`` [m3/d]:
                Natural gas consumption rate.
            - ``is_running`` [bool]:
                Whether the boiler is currently firing.
            - ``load_fraction`` [float]:
                Current output as fraction of nominal (0-1).
            - ``efficiency_actual`` [float]:
                Effective efficiency at this load point.
    """
    enable = bool(inputs.get("enable", True))

    # Accept both P_th_demand and P_aux_heat (HeatingSystem output key)
    P_demand = float(inputs.get("P_th_demand", inputs.get("P_aux_heat", 0.0)))
    Q_gas_avail = float(inputs.get("Q_gas_available_m3_per_day", 0.0))

    if not enable or P_demand <= 0.0:
        self._update_outputs(0.0, 0.0, 0.0, dt)
        return self.outputs_data

    # Clamp demand to nominal capacity
    P_th_supplied = min(P_demand, self.P_th_nom)
    load_fraction = P_th_supplied / self.P_th_nom

    # Part-load efficiency (linear correction, VDI 4631)
    eta_actual = self.efficiency * (1.0 - _PART_LOAD_PENALTY * (1.0 - load_fraction))
    eta_actual = max(0.01, eta_actual)

    # --- Fuel allocation ------------------------------------------
    Q_gas_consumed = 0.0  # m3/d biogas
    Q_ng_consumed = 0.0  # m3/d natural gas

    if self.fuel_type == "natural_gas":
        # Burn natural gas only
        Q_ng_consumed = P_th_supplied * 24.0 / (eta_actual * self.lhv_natural_gas)

    elif self.fuel_type == "biogas":
        # Burn biogas only (even if unavailable — caller must manage supply)
        Q_gas_consumed = P_th_supplied * 24.0 / (eta_actual * self.lhv_biogas)

    else:  # dual fuel: biogas first, natural gas for remainder
        # Maximum biogas combustion that would cover the demand
        Q_gas_needed_total = P_th_supplied * 24.0 / (eta_actual * self.lhv_biogas)

        if Q_gas_avail >= Q_gas_needed_total:
            # Sufficient biogas: no natural gas needed
            Q_gas_consumed = Q_gas_needed_total
        else:
            # Burn all available biogas, supplement with natural gas
            Q_gas_consumed = Q_gas_avail
            P_from_biogas = Q_gas_consumed * eta_actual * self.lhv_biogas / 24.0
            P_from_ng = P_th_supplied - P_from_biogas
            Q_ng_consumed = P_from_ng * 24.0 / (eta_actual * self.lhv_natural_gas)

    self._update_outputs(P_th_supplied, Q_gas_consumed, Q_ng_consumed, dt)
    return self.outputs_data

to_dict()

Serialize configuration and cumulative state to dictionary.

Source code in pyadm1/components/energy/boiler.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def to_dict(self) -> Dict[str, Any]:
    """Serialize configuration and cumulative state to dictionary."""
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "P_th_nom": self.P_th_nom,
        "efficiency": self.efficiency,
        "fuel_type": self.fuel_type,
        "lhv_biogas": self.lhv_biogas,
        "lhv_natural_gas": self.lhv_natural_gas,
        "inputs": self.inputs,
        "outputs": self.outputs,
        "state": self.state,
    }