Feeding Components API

pyadm1.components.feeding.substrate_storage.SubstrateStorage

Bases: Component

Storage facility component for biogas plant substrates.

Models storage of different substrate types with inventory tracking, quality degradation, and capacity management. Supports both solid (silage, solid manure) and liquid (liquid manure, slurry) substrates.

Attributes:

Name	Type	Description
storage_type		Type of storage facility
substrate_type		Category of substrate stored
capacity		Maximum storage capacity [t or m³]
current_level		Current inventory level [t or m³]
quality_factor		Current quality relative to fresh (0-1)
degradation_rate		Quality degradation rate [1/d]
density		Substrate bulk density [kg/m³]
dry_matter		Dry matter content [%]
vs_content		Volatile solids [% of DM]

Example

storage = SubstrateStorage( ... "silo1", ... storage_type="vertical_silo", ... substrate_type="corn_silage", ... capacity=1000, ... initial_level=600 ... ) storage.initialize() outputs = storage.step(0, 1, {'withdrawal_rate': 15})

Source code in pyadm1/components/feeding/substrate_storage.py

class SubstrateStorage(Component):
    """
    Storage facility component for biogas plant substrates.

    Models storage of different substrate types with inventory tracking,
    quality degradation, and capacity management. Supports both solid
    (silage, solid manure) and liquid (liquid manure, slurry) substrates.

    Attributes:
        storage_type: Type of storage facility
        substrate_type: Category of substrate stored
        capacity: Maximum storage capacity [t or m³]
        current_level: Current inventory level [t or m³]
        quality_factor: Current quality relative to fresh (0-1)
        degradation_rate: Quality degradation rate [1/d]
        density: Substrate bulk density [kg/m³]
        dry_matter: Dry matter content [%]
        vs_content: Volatile solids [% of DM]

    Example:
        >>> storage = SubstrateStorage(
        ...     "silo1",
        ...     storage_type="vertical_silo",
        ...     substrate_type="corn_silage",
        ...     capacity=1000,
        ...     initial_level=600
        ... )
        >>> storage.initialize()
        >>> outputs = storage.step(0, 1, {'withdrawal_rate': 15})
    """

    def __init__(
        self,
        component_id: str,
        storage_type: str = "vertical_silo",
        substrate_type: str = "corn_silage",
        capacity: float = 1000.0,
        initial_level: float = 0.0,
        degradation_rate: Optional[float] = None,
        temperature: float = 288.15,  # 15°C
        name: Optional[str] = None,
    ):
        """
        Initialize substrate storage component.

        Args:
            component_id: Unique identifier
            storage_type: Type of storage ("vertical_silo", "tank", etc.)
            substrate_type: Substrate category ("corn_silage", "manure_liquid", etc.)
            capacity: Maximum capacity [t or m³]
            initial_level: Initial inventory [t or m³]
            degradation_rate: Quality degradation rate [1/d] (auto-calculated if None)
            temperature: Storage temperature [K]
            name: Human-readable name
        """
        super().__init__(component_id, ComponentType.STORAGE, name)

        # Configuration
        self.storage_type = StorageType(storage_type.lower())
        self.substrate_type = SubstrateType(substrate_type.lower())
        self.capacity = float(capacity)
        self.temperature = float(temperature)

        # Inventory
        self.current_level = float(min(initial_level, capacity))
        self.quality_factor = 1.0  # 1.0 = fresh, 0.0 = completely degraded

        # Substrate properties (typical values, can be updated)
        self._set_substrate_properties()

        # Degradation parameters
        self.degradation_rate = degradation_rate or self._estimate_degradation_rate()

        # Storage losses
        self.cumulative_losses = 0.0  # Total mass lost [t or m³]
        self.cumulative_withdrawals = 0.0  # Total withdrawn [t or m³]

        # Tracking
        self.storage_time = 0.0  # Time substrate has been stored [days]
        self.n_refills = 0

        # Initialize
        self.initialize()

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

        Args:
            initial_state: Optional initial state with keys:
                - 'current_level': Inventory level [t or m³]
                - 'quality_factor': Quality factor (0-1)
                - 'storage_time': Time stored [days]
                - 'cumulative_losses': Total losses [t or m³]
        """
        if initial_state:
            self.current_level = float(initial_state.get("current_level", self.current_level))
            # current_level cannot be larger than capacity
            self.current_level = min(self.current_level, self.capacity)
            self.quality_factor = float(initial_state.get("quality_factor", 1.0))
            self.storage_time = float(initial_state.get("storage_time", 0.0))
            self.cumulative_losses = float(initial_state.get("cumulative_losses", 0.0))

        self.state = {
            "current_level": self.current_level,
            "quality_factor": self.quality_factor,
            "storage_time": self.storage_time,
            "cumulative_losses": self.cumulative_losses,
            "cumulative_withdrawals": self.cumulative_withdrawals,
            "n_refills": self.n_refills,
        }

        self.outputs_data = {
            "current_level": self.current_level,
            "utilization": self.current_level / max(1e-6, self.capacity),
            "quality_factor": self.quality_factor,
            "available_mass": self.current_level * self.quality_factor,
            "degradation_rate": self.degradation_rate,
            "is_empty": self.current_level < 1e-3,
            "is_full": self.current_level > 0.95 * self.capacity,
        }

        self._initialized = True

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

        Args:
            t: Current time [days]
            dt: Time step [days]
            inputs: Input data with optional keys:
                - 'withdrawal_rate': Withdrawal rate [t/d or m³/d]
                - 'refill_amount': Amount to add [t or m³]
                - 'refill_quality': Quality of refill (0-1)
                - 'temperature': Ambient/storage temperature [K]

        Returns:
            Dict with keys:
                - 'current_level': Current inventory [t or m³]
                - 'utilization': Fill level (0-1)
                - 'quality_factor': Current quality (0-1)
                - 'available_mass': Usable inventory [t or m³]
                - 'degradation_rate': Current degradation rate [1/d]
                - 'losses_this_step': Mass lost this timestep [t or m³]
                - 'withdrawn_this_step': Mass withdrawn [t or m³]
                - 'is_empty': Storage empty flag
                - 'is_full': Storage full flag
        """
        # Update temperature if provided
        if "temperature" in inputs:
            self.temperature = float(inputs["temperature"])
            self.degradation_rate = self._estimate_degradation_rate()

        # Handle refilling
        refill_amount = float(inputs.get("refill_amount", 0.0))
        if refill_amount > 0:
            self._add_substrate(refill_amount, inputs.get("refill_quality", 1.0))

        # Calculate quality degradation
        losses_this_step = self._calculate_degradation_losses(dt)

        # Handle withdrawal
        withdrawal_rate = float(inputs.get("withdrawal_rate", 0.0))
        withdrawn_this_step = self._withdraw_substrate(withdrawal_rate, dt)

        # Update storage time
        self.storage_time += dt

        # Update state
        self.state.update(
            {
                "current_level": self.current_level,
                "quality_factor": self.quality_factor,
                "storage_time": self.storage_time,
                "cumulative_losses": self.cumulative_losses,
                "cumulative_withdrawals": self.cumulative_withdrawals,
                "n_refills": self.n_refills,
            }
        )

        # Prepare outputs
        self.outputs_data = {
            "current_level": float(self.current_level),
            "utilization": float(self.current_level / max(1e-6, self.capacity)),
            "quality_factor": float(self.quality_factor),
            "available_mass": float(self.current_level * self.quality_factor),
            "degradation_rate": float(self.degradation_rate),
            "losses_this_step": float(losses_this_step),
            "withdrawn_this_step": float(withdrawn_this_step),
            "is_empty": bool(self.current_level < 1e-3),
            "is_full": bool(self.current_level > 0.95 * self.capacity),
            "storage_time": float(self.storage_time),
            "dry_matter": float(self.dry_matter),
            "vs_content": float(self.vs_content),
        }

        return self.outputs_data

    def _add_substrate(self, amount: float, quality: float = 1.0) -> None:
        """
        Add substrate to storage.

        Args:
            amount: Amount to add [t or m³]
            quality: Quality of added substrate (0-1)
        """
        # Calculate new weighted quality
        total_mass = self.current_level + amount
        if total_mass > 0:
            new_quality = (self.current_level * self.quality_factor + amount * quality) / total_mass
            self.quality_factor = min(1.0, new_quality)

        # Add to inventory (respect capacity)
        available_space = self.capacity - self.current_level
        added = min(amount, available_space)
        self.current_level += added

        # Reset storage time on refill
        if added > 0:
            self.storage_time = 0.0
            self.n_refills += 1

    def _withdraw_substrate(self, rate: float, dt: float) -> float:
        """
        Withdraw substrate from storage.

        Args:
            rate: Withdrawal rate [t/d or m³/d]
            dt: Time step [days]

        Returns:
            Actual amount withdrawn [t or m³]
        """
        requested = rate * dt
        actual = min(requested, self.current_level)

        self.current_level -= actual
        self.cumulative_withdrawals += actual

        return actual

    def _calculate_degradation_losses(self, dt: float) -> float:
        """
        Calculate substrate degradation losses.

        Models aerobic degradation and dry matter losses during storage.

        Args:
            dt: Time step [days]

        Returns:
            Mass lost [t or m³]
        """
        if self.current_level < 1e-6:
            return 0.0

        # Exponential quality decay
        old_quality = self.quality_factor
        self.quality_factor *= np.exp(-self.degradation_rate * dt)
        self.quality_factor = max(0.0, self.quality_factor)

        # Calculate mass loss
        quality_loss = old_quality - self.quality_factor
        mass_loss = self.current_level * quality_loss

        # Remove lost mass from inventory
        self.current_level -= mass_loss
        self.cumulative_losses += mass_loss

        return mass_loss

    def _estimate_degradation_rate(self) -> float:
        """
        Estimate degradation rate based on storage type and conditions.

        Returns:
            Degradation rate [1/d]
        """
        # TODO: get a source confirming those values
        # Base rates for different storage types [1/d]
        # Reduced to more realistic values
        base_rates = {
            StorageType.VERTICAL_SILO: 0.0005,  # Reduced from 0.001 - Well sealed, very low losses
            StorageType.HORIZONTAL_SILO: 0.0008,  # Reduced from 0.002 - Good sealing
            StorageType.BUNKER_SILO: 0.001,  # Reduced from 0.003 - Moderate losses
            StorageType.CLAMP: 0.0025,  # Reduced from 0.005 - Higher losses
            StorageType.PILE: 0.004,  # Reduced from 0.008 - Highest losses
            StorageType.ABOVE_GROUND_TANK: 0.0002,  # Reduced from 0.0005 - Very low for liquids
            StorageType.BELOW_GROUND_TANK: 0.0001,  # Reduced from 0.0003 - Lowest losses
        }

        base_rate = base_rates.get(self.storage_type, 0.001)

        # Temperature correction (Q10 = 2)
        T_ref = 288.15  # 15°C reference
        T_factor = 2.0 ** ((self.temperature - T_ref) / 10.0)

        # Storage type modifier
        if self.storage_type in [
            StorageType.ABOVE_GROUND_TANK,
            StorageType.BELOW_GROUND_TANK,
        ]:
            # Liquid storage has minimal degradation if sealed
            # Further limit temperature effect for liquid storage
            T_factor = min(T_factor, 1.2)  # Changed from 1.5

        return base_rate * T_factor

    def _set_substrate_properties(self) -> None:
        """Set typical substrate properties based on substrate type."""
        # Default properties (density, DM, VS)
        properties = {
            SubstrateType.CORN_SILAGE: (650, 35, 95),
            SubstrateType.GRASS_SILAGE: (700, 30, 92),
            SubstrateType.WHOLE_CROP_SILAGE: (680, 32, 94),
            SubstrateType.MANURE_LIQUID: (1020, 8, 80),
            SubstrateType.MANURE_SOLID: (850, 25, 75),
            SubstrateType.BIOWASTE: (900, 35, 85),
            SubstrateType.FOOD_WASTE: (1000, 20, 90),
            SubstrateType.GENERIC_SOLID: (700, 30, 90),
            SubstrateType.GENERIC_LIQUID: (1000, 10, 80),
        }

        self.density, self.dry_matter, self.vs_content = properties.get(self.substrate_type, (800, 25, 85))

    def to_dict(self) -> Dict[str, Any]:
        """Serialize storage to dictionary."""
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "storage_type": self.storage_type.value,
            "substrate_type": self.substrate_type.value,
            "capacity": self.capacity,
            "temperature": self.temperature,
            "degradation_rate": self.degradation_rate,
            "density": self.density,
            "dry_matter": self.dry_matter,
            "vs_content": self.vs_content,
            "state": self.state,
            "inputs": self.inputs,
            "outputs": self.outputs,
        }

    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> "SubstrateStorage":
        """Create storage from dictionary."""
        storage = cls(
            component_id=config["component_id"],
            storage_type=config.get("storage_type", "vertical_silo"),
            substrate_type=config.get("substrate_type", "corn_silage"),
            capacity=config.get("capacity", 1000.0),
            initial_level=0.0,
            degradation_rate=config.get("degradation_rate"),
            temperature=config.get("temperature", 288.15),
            name=config.get("name"),
        )

        # Restore state if present
        if "state" in config:
            storage.initialize(config["state"])

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

        return storage

Functions

__init__(component_id, storage_type='vertical_silo', substrate_type='corn_silage', capacity=1000.0, initial_level=0.0, degradation_rate=None, temperature=288.15, name=None)

Initialize substrate storage component.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier	required
storage_type	str	Type of storage ("vertical_silo", "tank", etc.)	'vertical_silo'
substrate_type	str	Substrate category ("corn_silage", "manure_liquid", etc.)	'corn_silage'
capacity	float	Maximum capacity [t or m³]	1000.0
initial_level	float	Initial inventory [t or m³]	0.0
degradation_rate	Optional[float]	Quality degradation rate [1/d] (auto-calculated if None)	None
temperature	float	Storage temperature [K]	288.15
name	Optional[str]	Human-readable name	None

Source code in pyadm1/components/feeding/substrate_storage.py

def __init__(
    self,
    component_id: str,
    storage_type: str = "vertical_silo",
    substrate_type: str = "corn_silage",
    capacity: float = 1000.0,
    initial_level: float = 0.0,
    degradation_rate: Optional[float] = None,
    temperature: float = 288.15,  # 15°C
    name: Optional[str] = None,
):
    """
    Initialize substrate storage component.

    Args:
        component_id: Unique identifier
        storage_type: Type of storage ("vertical_silo", "tank", etc.)
        substrate_type: Substrate category ("corn_silage", "manure_liquid", etc.)
        capacity: Maximum capacity [t or m³]
        initial_level: Initial inventory [t or m³]
        degradation_rate: Quality degradation rate [1/d] (auto-calculated if None)
        temperature: Storage temperature [K]
        name: Human-readable name
    """
    super().__init__(component_id, ComponentType.STORAGE, name)

    # Configuration
    self.storage_type = StorageType(storage_type.lower())
    self.substrate_type = SubstrateType(substrate_type.lower())
    self.capacity = float(capacity)
    self.temperature = float(temperature)

    # Inventory
    self.current_level = float(min(initial_level, capacity))
    self.quality_factor = 1.0  # 1.0 = fresh, 0.0 = completely degraded

    # Substrate properties (typical values, can be updated)
    self._set_substrate_properties()

    # Degradation parameters
    self.degradation_rate = degradation_rate or self._estimate_degradation_rate()

    # Storage losses
    self.cumulative_losses = 0.0  # Total mass lost [t or m³]
    self.cumulative_withdrawals = 0.0  # Total withdrawn [t or m³]

    # Tracking
    self.storage_time = 0.0  # Time substrate has been stored [days]
    self.n_refills = 0

    # Initialize
    self.initialize()

from_dict(config) classmethod

Create storage from dictionary.

Source code in pyadm1/components/feeding/substrate_storage.py

@classmethod
def from_dict(cls, config: Dict[str, Any]) -> "SubstrateStorage":
    """Create storage from dictionary."""
    storage = cls(
        component_id=config["component_id"],
        storage_type=config.get("storage_type", "vertical_silo"),
        substrate_type=config.get("substrate_type", "corn_silage"),
        capacity=config.get("capacity", 1000.0),
        initial_level=0.0,
        degradation_rate=config.get("degradation_rate"),
        temperature=config.get("temperature", 288.15),
        name=config.get("name"),
    )

    # Restore state if present
    if "state" in config:
        storage.initialize(config["state"])

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

    return storage

initialize(initial_state=None)

Initialize storage state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Optional initial state with keys: - 'current_level': Inventory level [t or m³] - 'quality_factor': Quality factor (0-1) - 'storage_time': Time stored [days] - 'cumulative_losses': Total losses [t or m³]	None

Source code in pyadm1/components/feeding/substrate_storage.py

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

    Args:
        initial_state: Optional initial state with keys:
            - 'current_level': Inventory level [t or m³]
            - 'quality_factor': Quality factor (0-1)
            - 'storage_time': Time stored [days]
            - 'cumulative_losses': Total losses [t or m³]
    """
    if initial_state:
        self.current_level = float(initial_state.get("current_level", self.current_level))
        # current_level cannot be larger than capacity
        self.current_level = min(self.current_level, self.capacity)
        self.quality_factor = float(initial_state.get("quality_factor", 1.0))
        self.storage_time = float(initial_state.get("storage_time", 0.0))
        self.cumulative_losses = float(initial_state.get("cumulative_losses", 0.0))

    self.state = {
        "current_level": self.current_level,
        "quality_factor": self.quality_factor,
        "storage_time": self.storage_time,
        "cumulative_losses": self.cumulative_losses,
        "cumulative_withdrawals": self.cumulative_withdrawals,
        "n_refills": self.n_refills,
    }

    self.outputs_data = {
        "current_level": self.current_level,
        "utilization": self.current_level / max(1e-6, self.capacity),
        "quality_factor": self.quality_factor,
        "available_mass": self.current_level * self.quality_factor,
        "degradation_rate": self.degradation_rate,
        "is_empty": self.current_level < 1e-3,
        "is_full": self.current_level > 0.95 * self.capacity,
    }

    self._initialized = True

step(t, dt, inputs)

Perform one simulation time step.

Parameters:

Name	Type	Description	Default
t	float	Current time [days]	required
dt	float	Time step [days]	required
inputs	Dict[str, Any]	Input data with optional keys: - 'withdrawal_rate': Withdrawal rate [t/d or m³/d] - 'refill_amount': Amount to add [t or m³] - 'refill_quality': Quality of refill (0-1) - 'temperature': Ambient/storage temperature [K]	required

Returns:

Type

Description

Dict[str, Any]

Dict with keys: - 'current_level': Current inventory [t or m³] - 'utilization': Fill level (0-1) - 'quality_factor': Current quality (0-1) - 'available_mass': Usable inventory [t or m³] - 'degradation_rate': Current degradation rate [1/d] - 'losses_this_step': Mass lost this timestep [t or m³] - 'withdrawn_this_step': Mass withdrawn [t or m³] - 'is_empty': Storage empty flag - 'is_full': Storage full flag

Source code in pyadm1/components/feeding/substrate_storage.py

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

    Args:
        t: Current time [days]
        dt: Time step [days]
        inputs: Input data with optional keys:
            - 'withdrawal_rate': Withdrawal rate [t/d or m³/d]
            - 'refill_amount': Amount to add [t or m³]
            - 'refill_quality': Quality of refill (0-1)
            - 'temperature': Ambient/storage temperature [K]

    Returns:
        Dict with keys:
            - 'current_level': Current inventory [t or m³]
            - 'utilization': Fill level (0-1)
            - 'quality_factor': Current quality (0-1)
            - 'available_mass': Usable inventory [t or m³]
            - 'degradation_rate': Current degradation rate [1/d]
            - 'losses_this_step': Mass lost this timestep [t or m³]
            - 'withdrawn_this_step': Mass withdrawn [t or m³]
            - 'is_empty': Storage empty flag
            - 'is_full': Storage full flag
    """
    # Update temperature if provided
    if "temperature" in inputs:
        self.temperature = float(inputs["temperature"])
        self.degradation_rate = self._estimate_degradation_rate()

    # Handle refilling
    refill_amount = float(inputs.get("refill_amount", 0.0))
    if refill_amount > 0:
        self._add_substrate(refill_amount, inputs.get("refill_quality", 1.0))

    # Calculate quality degradation
    losses_this_step = self._calculate_degradation_losses(dt)

    # Handle withdrawal
    withdrawal_rate = float(inputs.get("withdrawal_rate", 0.0))
    withdrawn_this_step = self._withdraw_substrate(withdrawal_rate, dt)

    # Update storage time
    self.storage_time += dt

    # Update state
    self.state.update(
        {
            "current_level": self.current_level,
            "quality_factor": self.quality_factor,
            "storage_time": self.storage_time,
            "cumulative_losses": self.cumulative_losses,
            "cumulative_withdrawals": self.cumulative_withdrawals,
            "n_refills": self.n_refills,
        }
    )

    # Prepare outputs
    self.outputs_data = {
        "current_level": float(self.current_level),
        "utilization": float(self.current_level / max(1e-6, self.capacity)),
        "quality_factor": float(self.quality_factor),
        "available_mass": float(self.current_level * self.quality_factor),
        "degradation_rate": float(self.degradation_rate),
        "losses_this_step": float(losses_this_step),
        "withdrawn_this_step": float(withdrawn_this_step),
        "is_empty": bool(self.current_level < 1e-3),
        "is_full": bool(self.current_level > 0.95 * self.capacity),
        "storage_time": float(self.storage_time),
        "dry_matter": float(self.dry_matter),
        "vs_content": float(self.vs_content),
    }

    return self.outputs_data

to_dict()

Serialize storage to dictionary.

Source code in pyadm1/components/feeding/substrate_storage.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize storage to dictionary."""
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "storage_type": self.storage_type.value,
        "substrate_type": self.substrate_type.value,
        "capacity": self.capacity,
        "temperature": self.temperature,
        "degradation_rate": self.degradation_rate,
        "density": self.density,
        "dry_matter": self.dry_matter,
        "vs_content": self.vs_content,
        "state": self.state,
        "inputs": self.inputs,
        "outputs": self.outputs,
    }

pyadm1.components.feeding.feeder.Feeder

Bases: Component

Feeder component for automated substrate dosing.

Models feeding systems that transfer substrates from storage to digesters. Includes realistic operational characteristics like dosing accuracy, capacity limits, and power consumption.

Attributes:

Name	Type	Description
feeder_type		Type of feeding system
Q_max		Maximum flow rate [m³/d or t/d]
substrate_type		Physical category of substrate
dosing_accuracy		Accuracy of flow control (std dev as fraction)
power_installed		Installed motor power [kW]
current_flow		Current actual flow rate [m³/d or t/d]
is_running		Operating state

Example

feeder = Feeder( ... "feed1", ... feeder_type="screw", ... Q_max=20, ... substrate_type="solid" ... ) feeder.initialize() result = feeder.step(0, 1/24, {'Q_setpoint': 15})

Source code in pyadm1/components/feeding/feeder.py

class Feeder(Component):
    """
    Feeder component for automated substrate dosing.

    Models feeding systems that transfer substrates from storage to digesters.
    Includes realistic operational characteristics like dosing accuracy,
    capacity limits, and power consumption.

    Attributes:
        feeder_type: Type of feeding system
        Q_max: Maximum flow rate [m³/d or t/d]
        substrate_type: Physical category of substrate
        dosing_accuracy: Accuracy of flow control (std dev as fraction)
        power_installed: Installed motor power [kW]
        current_flow: Current actual flow rate [m³/d or t/d]
        is_running: Operating state

    Example:
        >>> feeder = Feeder(
        ...     "feed1",
        ...     feeder_type="screw",
        ...     Q_max=20,
        ...     substrate_type="solid"
        ... )
        >>> feeder.initialize()
        >>> result = feeder.step(0, 1/24, {'Q_setpoint': 15})
    """

    def __init__(
        self,
        component_id: str,
        feeder_type: Optional[str] = None,
        Q_max: float = 20.0,
        substrate_type: str = "solid",
        dosing_accuracy: Optional[float] = None,
        power_installed: Optional[float] = None,
        enable_dosing_noise: bool = True,
        name: Optional[str] = None,
    ):
        """
        Initialize feeder component.

        Args:
            component_id: Unique identifier
            feeder_type: Type of feeder ("screw", "progressive_cavity", etc.)
            Q_max: Maximum flow rate [m³/d or t/d]
            substrate_type: Substrate category ("solid", "slurry", "liquid", "fibrous")
            dosing_accuracy: Standard deviation of flow as fraction (auto if None)
            power_installed: Installed power [kW] (auto-calculated if None)
            enable_dosing_noise: Add realistic dosing variance
            name: Human-readable name
        """
        super().__init__(component_id, ComponentType.MIXER, name)  # Use MIXER as closest type

        if feeder_type is None:
            # Default to screw feeder for solid substrates
            if substrate_type.lower() in ["solid", "fibrous"]:
                feeder_type = "screw"
            else:
                feeder_type = "centrifugal_pump"

        # Configuration
        self.feeder_type = FeederType(feeder_type.lower())
        self.substrate_type = SubstrateCategory(substrate_type.lower())
        # TODO: check whether substrate type fits to feeder_type. E.g. liquid substrate should not be transported
        #  with a screw
        self.Q_max = float(Q_max)
        self.enable_dosing_noise = enable_dosing_noise

        # Dosing characteristics
        self.dosing_accuracy = dosing_accuracy or self._estimate_dosing_accuracy()
        self.power_installed = power_installed or self._estimate_power_requirement()

        # Operating state
        self.current_flow = 0.0
        self.is_running = False
        self.blockage_detected = False

        # Performance tracking
        self.operating_hours = 0.0
        self.energy_consumed = 0.0  # kWh
        self.total_mass_fed = 0.0  # t or m³
        self.n_starts = 0
        self.n_blockages = 0

        # Speed control (for variable speed feeders)
        self.speed_fraction = 1.0  # Fraction of nominal speed (0-1)

        # Initialize
        self.initialize()

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

        Args:
            initial_state: Optional initial state with keys:
                - 'is_running': Initial operating state
                - 'current_flow': Initial flow rate [m³/d or t/d]
                - 'operating_hours': Cumulative operating hours
                - 'energy_consumed': Cumulative energy [kWh]
                - 'total_mass_fed': Cumulative mass [t or m³]
        """
        if initial_state:
            self.is_running = initial_state.get("is_running", False)
            self.current_flow = float(initial_state.get("current_flow", 0.0))
            self.operating_hours = float(initial_state.get("operating_hours", 0.0))
            self.energy_consumed = float(initial_state.get("energy_consumed", 0.0))
            self.total_mass_fed = float(initial_state.get("total_mass_fed", 0.0))

        self.state = {
            "is_running": self.is_running,
            "current_flow": self.current_flow,
            "operating_hours": self.operating_hours,
            "energy_consumed": self.energy_consumed,
            "total_mass_fed": self.total_mass_fed,
            "blockage_detected": self.blockage_detected,
            "n_starts": self.n_starts,
            "n_blockages": self.n_blockages,
        }

        self.outputs_data = {
            "Q_actual": 0.0,
            "is_running": False,
            "load_factor": 0.0,
            "P_consumed": 0.0,
            "blockage_detected": False,
        }

        self._initialized = True

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

        Args:
            t: Current time [days]
            dt: Time step [days]
            inputs: Input data with optional keys:
                - 'Q_setpoint': Desired flow rate [m³/d or t/d]
                - 'enable_feeding': Enable/disable feeder
                - 'substrate_available': Amount available in storage [t or m³]
                - 'speed_setpoint': Desired speed fraction (0-1)

        Returns:
            Dict with keys:
                - 'Q_actual': Actual flow rate [m³/d or t/d]
                - 'is_running': Current operating state
                - 'load_factor': Operating load (0-1)
                - 'P_consumed': Power consumption [kW]
                - 'blockage_detected': Blockage alarm
                - 'dosing_error': Deviation from setpoint [%]
                - 'speed_fraction': Current speed fraction
        """
        # Get control inputs
        enable_feeding = inputs.get("enable_feeding", True)
        Q_setpoint = float(inputs.get("Q_setpoint", 0.0))
        substrate_available = float(inputs.get("substrate_available", float("inf")))
        speed_setpoint = float(inputs.get("speed_setpoint", 1.0))

        # Determine operating state
        should_run = enable_feeding and Q_setpoint > 0.01

        # Track starts
        if should_run and not self.is_running:
            self.n_starts += 1

        self.is_running = should_run

        if not self.is_running:
            # Feeder is off
            self.current_flow = 0.0
            self.speed_fraction = 0.0
            P_consumed = 0.0
            self.blockage_detected = False

        else:
            # Feeder is running

            # Update speed (for variable speed feeders)
            self.speed_fraction = min(1.0, speed_setpoint)

            # Calculate target flow
            Q_target = min(Q_setpoint, self.Q_max * self.speed_fraction)

            # Check substrate availability
            max_available = substrate_available / dt  # Convert to daily rate
            Q_target = min(Q_target, max_available)

            # Apply dosing noise (realistic variance)
            if self.enable_dosing_noise and Q_target > 0:
                noise = np.random.normal(0, self.dosing_accuracy * Q_target)
                Q_actual = max(0.0, Q_target + noise)
            else:
                Q_actual = Q_target

            self.current_flow = Q_actual

            # Random blockage simulation (very low probability)
            if np.random.random() < 0.0001 * dt:  # ~0.01% per day
                self.blockage_detected = True
                self.n_blockages += 1
                self.current_flow *= 0.1  # Reduced flow during blockage
            else:
                self.blockage_detected = False

            # Calculate power consumption
            P_consumed = self._calculate_power_consumption()

        # Update cumulative values
        dt_hours = dt * 24.0
        if self.is_running:
            self.operating_hours += dt_hours
            self.total_mass_fed += self.current_flow * dt

        self.energy_consumed += P_consumed * dt_hours

        # Calculate load factor
        load_factor = self.current_flow / max(1e-6, self.Q_max) if self.is_running else 0.0

        # Calculate dosing error
        if Q_setpoint > 0:
            dosing_error = abs(self.current_flow - Q_setpoint) / Q_setpoint * 100
        else:
            dosing_error = 0.0

        # Update state
        self.state.update(
            {
                "is_running": self.is_running,
                "current_flow": self.current_flow,
                "operating_hours": self.operating_hours,
                "energy_consumed": self.energy_consumed,
                "total_mass_fed": self.total_mass_fed,
                "blockage_detected": self.blockage_detected,
                "n_starts": self.n_starts,
                "n_blockages": self.n_blockages,
            }
        )

        # Prepare outputs
        self.outputs_data = {
            "Q_actual": float(self.current_flow),
            "is_running": bool(self.is_running),
            "load_factor": float(load_factor),
            "P_consumed": float(P_consumed),
            "blockage_detected": bool(self.blockage_detected),
            "dosing_error": float(dosing_error),
            "speed_fraction": float(self.speed_fraction),
            "dosing_accuracy": float(self.dosing_accuracy),
            "total_mass_fed": float(self.total_mass_fed),
        }

        return self.outputs_data

    def _calculate_power_consumption(self) -> float:
        """
        Calculate power consumption based on load.

        Returns:
            Power consumption [kW]
        """
        if not self.is_running or self.current_flow < 1e-6:
            return 0.0

        # Base power (idling)
        P_base = self.power_installed * 0.2

        # Load-dependent power
        load_factor = self.current_flow / max(1e-6, self.Q_max)
        P_load = self.power_installed * 0.8 * load_factor

        # Blockage increases power consumption
        if self.blockage_detected:
            P_load *= 1.5

        return P_base + P_load

    def _estimate_dosing_accuracy(self) -> float:
        """
        Estimate dosing accuracy based on feeder type.

        Returns:
            Standard deviation as fraction of flow rate
        """
        # Typical dosing accuracy (std dev as fraction)
        accuracies = {
            FeederType.SCREW: 0.05,  # ±5%
            FeederType.TWIN_SCREW: 0.03,  # ±3% (better control)
            FeederType.PROGRESSIVE_CAVITY: 0.02,  # ±2% (volumetric)
            FeederType.PISTON: 0.01,  # ±1% (most accurate)
            FeederType.CENTRIFUGAL_PUMP: 0.08,  # ±8% (less precise)
            FeederType.MIXER_WAGON: 0.10,  # ±10% (batch feeding)
        }

        return accuracies.get(self.feeder_type, 0.05)

    def _estimate_power_requirement(self) -> float:
        """
        Estimate power requirement based on feeder type and capacity.

        TODO: Check these publications that provide numbers:

        Frey, J., Grüssing, F., Nägele, H. J., & Oechsner, H. (2013). Eigenstromverbrauch an Biogasanlagen senken:
        Der Einfluss neuer Techniken. agricultural engineering. eu, 68(1), 58-63.

         Naegele, H.-J., Lemmer, A., Oechsner, H., & Jungbluth, T. (2012). Electric Energy Consumption of the Full
         Scale Research Biogas Plant “Unterer Lindenhof”: Results of Longterm and Full Detail Measurements. Energies,
         5(12), 5198-5214. https://doi.org/10.3390/en5125198

        Returns:
            Power requirement [kW]
        """
        # Specific power requirements [kW per m³/h]
        specific_powers = {
            FeederType.SCREW: 0.8,  # Increased from 0.5
            FeederType.TWIN_SCREW: 1.0,  # Increased from 0.7
            FeederType.PROGRESSIVE_CAVITY: 1.2,  # Increased from 0.8
            FeederType.PISTON: 1.5,  # Increased from 1.0
            FeederType.CENTRIFUGAL_PUMP: 0.5,  # Increased from 0.3
            FeederType.MIXER_WAGON: 2.0,  # Increased from 1.5
        }

        specific_power = specific_powers.get(self.feeder_type, 0.8)

        # Convert Q_max from m³/d to m³/h
        Q_max_per_hour = self.Q_max / 24.0

        # Base power calculation
        power = specific_power * Q_max_per_hour

        # Substrate type modifier
        if self.substrate_type == SubstrateCategory.FIBROUS:
            power *= 1.8  # Increased from 1.5 - Fibrous materials need more power
        elif self.substrate_type == SubstrateCategory.SOLID:
            power *= 1.4  # Increased from 1.2
        elif self.substrate_type == SubstrateCategory.LIQUID:
            power *= 0.7  # Decreased from 0.8 - Liquids easier to pump

        # Add safety margin
        power *= 1.3  # Increased from 1.2

        return max(1.0, power)

    def to_dict(self) -> Dict[str, Any]:
        """Serialize feeder to dictionary."""
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "feeder_type": self.feeder_type.value,
            "substrate_type": self.substrate_type.value,
            "Q_max": self.Q_max,
            "dosing_accuracy": self.dosing_accuracy,
            "power_installed": self.power_installed,
            "enable_dosing_noise": self.enable_dosing_noise,
            "state": self.state,
            "inputs": self.inputs,
            "outputs": self.outputs,
        }

    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> "Feeder":
        """Create feeder from dictionary."""
        feeder = cls(
            component_id=config["component_id"],
            feeder_type=config.get("feeder_type", "screw"),
            Q_max=config.get("Q_max", 20.0),
            substrate_type=config.get("substrate_type", "solid"),
            dosing_accuracy=config.get("dosing_accuracy"),
            power_installed=config.get("power_installed"),
            enable_dosing_noise=config.get("enable_dosing_noise", True),
            name=config.get("name"),
        )

        # Restore state if present
        if "state" in config:
            feeder.initialize(config["state"])

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

        return feeder

Functions

__init__(component_id, feeder_type=None, Q_max=20.0, substrate_type='solid', dosing_accuracy=None, power_installed=None, enable_dosing_noise=True, name=None)

Initialize feeder component.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier	required
feeder_type	Optional[str]	Type of feeder ("screw", "progressive_cavity", etc.)	None
Q_max	float	Maximum flow rate [m³/d or t/d]	20.0
substrate_type	str	Substrate category ("solid", "slurry", "liquid", "fibrous")	'solid'
dosing_accuracy	Optional[float]	Standard deviation of flow as fraction (auto if None)	None
power_installed	Optional[float]	Installed power [kW] (auto-calculated if None)	None
enable_dosing_noise	bool	Add realistic dosing variance	True
name	Optional[str]	Human-readable name	None

Source code in pyadm1/components/feeding/feeder.py

def __init__(
    self,
    component_id: str,
    feeder_type: Optional[str] = None,
    Q_max: float = 20.0,
    substrate_type: str = "solid",
    dosing_accuracy: Optional[float] = None,
    power_installed: Optional[float] = None,
    enable_dosing_noise: bool = True,
    name: Optional[str] = None,
):
    """
    Initialize feeder component.

    Args:
        component_id: Unique identifier
        feeder_type: Type of feeder ("screw", "progressive_cavity", etc.)
        Q_max: Maximum flow rate [m³/d or t/d]
        substrate_type: Substrate category ("solid", "slurry", "liquid", "fibrous")
        dosing_accuracy: Standard deviation of flow as fraction (auto if None)
        power_installed: Installed power [kW] (auto-calculated if None)
        enable_dosing_noise: Add realistic dosing variance
        name: Human-readable name
    """
    super().__init__(component_id, ComponentType.MIXER, name)  # Use MIXER as closest type

    if feeder_type is None:
        # Default to screw feeder for solid substrates
        if substrate_type.lower() in ["solid", "fibrous"]:
            feeder_type = "screw"
        else:
            feeder_type = "centrifugal_pump"

    # Configuration
    self.feeder_type = FeederType(feeder_type.lower())
    self.substrate_type = SubstrateCategory(substrate_type.lower())
    # TODO: check whether substrate type fits to feeder_type. E.g. liquid substrate should not be transported
    #  with a screw
    self.Q_max = float(Q_max)
    self.enable_dosing_noise = enable_dosing_noise

    # Dosing characteristics
    self.dosing_accuracy = dosing_accuracy or self._estimate_dosing_accuracy()
    self.power_installed = power_installed or self._estimate_power_requirement()

    # Operating state
    self.current_flow = 0.0
    self.is_running = False
    self.blockage_detected = False

    # Performance tracking
    self.operating_hours = 0.0
    self.energy_consumed = 0.0  # kWh
    self.total_mass_fed = 0.0  # t or m³
    self.n_starts = 0
    self.n_blockages = 0

    # Speed control (for variable speed feeders)
    self.speed_fraction = 1.0  # Fraction of nominal speed (0-1)

    # Initialize
    self.initialize()

from_dict(config) classmethod

Create feeder from dictionary.

Source code in pyadm1/components/feeding/feeder.py

@classmethod
def from_dict(cls, config: Dict[str, Any]) -> "Feeder":
    """Create feeder from dictionary."""
    feeder = cls(
        component_id=config["component_id"],
        feeder_type=config.get("feeder_type", "screw"),
        Q_max=config.get("Q_max", 20.0),
        substrate_type=config.get("substrate_type", "solid"),
        dosing_accuracy=config.get("dosing_accuracy"),
        power_installed=config.get("power_installed"),
        enable_dosing_noise=config.get("enable_dosing_noise", True),
        name=config.get("name"),
    )

    # Restore state if present
    if "state" in config:
        feeder.initialize(config["state"])

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

    return feeder

initialize(initial_state=None)

Initialize feeder state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Optional initial state with keys: - 'is_running': Initial operating state - 'current_flow': Initial flow rate [m³/d or t/d] - 'operating_hours': Cumulative operating hours - 'energy_consumed': Cumulative energy [kWh] - 'total_mass_fed': Cumulative mass [t or m³]	None

Source code in pyadm1/components/feeding/feeder.py

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

    Args:
        initial_state: Optional initial state with keys:
            - 'is_running': Initial operating state
            - 'current_flow': Initial flow rate [m³/d or t/d]
            - 'operating_hours': Cumulative operating hours
            - 'energy_consumed': Cumulative energy [kWh]
            - 'total_mass_fed': Cumulative mass [t or m³]
    """
    if initial_state:
        self.is_running = initial_state.get("is_running", False)
        self.current_flow = float(initial_state.get("current_flow", 0.0))
        self.operating_hours = float(initial_state.get("operating_hours", 0.0))
        self.energy_consumed = float(initial_state.get("energy_consumed", 0.0))
        self.total_mass_fed = float(initial_state.get("total_mass_fed", 0.0))

    self.state = {
        "is_running": self.is_running,
        "current_flow": self.current_flow,
        "operating_hours": self.operating_hours,
        "energy_consumed": self.energy_consumed,
        "total_mass_fed": self.total_mass_fed,
        "blockage_detected": self.blockage_detected,
        "n_starts": self.n_starts,
        "n_blockages": self.n_blockages,
    }

    self.outputs_data = {
        "Q_actual": 0.0,
        "is_running": False,
        "load_factor": 0.0,
        "P_consumed": 0.0,
        "blockage_detected": False,
    }

    self._initialized = True

step(t, dt, inputs)

Perform one simulation time step.

Parameters:

Name	Type	Description	Default
t	float	Current time [days]	required
dt	float	Time step [days]	required
inputs	Dict[str, Any]	Input data with optional keys: - 'Q_setpoint': Desired flow rate [m³/d or t/d] - 'enable_feeding': Enable/disable feeder - 'substrate_available': Amount available in storage [t or m³] - 'speed_setpoint': Desired speed fraction (0-1)	required

Returns:

Type	Description
Dict[str, Any]	Dict with keys: - 'Q_actual': Actual flow rate [m³/d or t/d] - 'is_running': Current operating state - 'load_factor': Operating load (0-1) - 'P_consumed': Power consumption [kW] - 'blockage_detected': Blockage alarm - 'dosing_error': Deviation from setpoint [%] - 'speed_fraction': Current speed fraction

Source code in pyadm1/components/feeding/feeder.py

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

    Args:
        t: Current time [days]
        dt: Time step [days]
        inputs: Input data with optional keys:
            - 'Q_setpoint': Desired flow rate [m³/d or t/d]
            - 'enable_feeding': Enable/disable feeder
            - 'substrate_available': Amount available in storage [t or m³]
            - 'speed_setpoint': Desired speed fraction (0-1)

    Returns:
        Dict with keys:
            - 'Q_actual': Actual flow rate [m³/d or t/d]
            - 'is_running': Current operating state
            - 'load_factor': Operating load (0-1)
            - 'P_consumed': Power consumption [kW]
            - 'blockage_detected': Blockage alarm
            - 'dosing_error': Deviation from setpoint [%]
            - 'speed_fraction': Current speed fraction
    """
    # Get control inputs
    enable_feeding = inputs.get("enable_feeding", True)
    Q_setpoint = float(inputs.get("Q_setpoint", 0.0))
    substrate_available = float(inputs.get("substrate_available", float("inf")))
    speed_setpoint = float(inputs.get("speed_setpoint", 1.0))

    # Determine operating state
    should_run = enable_feeding and Q_setpoint > 0.01

    # Track starts
    if should_run and not self.is_running:
        self.n_starts += 1

    self.is_running = should_run

    if not self.is_running:
        # Feeder is off
        self.current_flow = 0.0
        self.speed_fraction = 0.0
        P_consumed = 0.0
        self.blockage_detected = False

    else:
        # Feeder is running

        # Update speed (for variable speed feeders)
        self.speed_fraction = min(1.0, speed_setpoint)

        # Calculate target flow
        Q_target = min(Q_setpoint, self.Q_max * self.speed_fraction)

        # Check substrate availability
        max_available = substrate_available / dt  # Convert to daily rate
        Q_target = min(Q_target, max_available)

        # Apply dosing noise (realistic variance)
        if self.enable_dosing_noise and Q_target > 0:
            noise = np.random.normal(0, self.dosing_accuracy * Q_target)
            Q_actual = max(0.0, Q_target + noise)
        else:
            Q_actual = Q_target

        self.current_flow = Q_actual

        # Random blockage simulation (very low probability)
        if np.random.random() < 0.0001 * dt:  # ~0.01% per day
            self.blockage_detected = True
            self.n_blockages += 1
            self.current_flow *= 0.1  # Reduced flow during blockage
        else:
            self.blockage_detected = False

        # Calculate power consumption
        P_consumed = self._calculate_power_consumption()

    # Update cumulative values
    dt_hours = dt * 24.0
    if self.is_running:
        self.operating_hours += dt_hours
        self.total_mass_fed += self.current_flow * dt

    self.energy_consumed += P_consumed * dt_hours

    # Calculate load factor
    load_factor = self.current_flow / max(1e-6, self.Q_max) if self.is_running else 0.0

    # Calculate dosing error
    if Q_setpoint > 0:
        dosing_error = abs(self.current_flow - Q_setpoint) / Q_setpoint * 100
    else:
        dosing_error = 0.0

    # Update state
    self.state.update(
        {
            "is_running": self.is_running,
            "current_flow": self.current_flow,
            "operating_hours": self.operating_hours,
            "energy_consumed": self.energy_consumed,
            "total_mass_fed": self.total_mass_fed,
            "blockage_detected": self.blockage_detected,
            "n_starts": self.n_starts,
            "n_blockages": self.n_blockages,
        }
    )

    # Prepare outputs
    self.outputs_data = {
        "Q_actual": float(self.current_flow),
        "is_running": bool(self.is_running),
        "load_factor": float(load_factor),
        "P_consumed": float(P_consumed),
        "blockage_detected": bool(self.blockage_detected),
        "dosing_error": float(dosing_error),
        "speed_fraction": float(self.speed_fraction),
        "dosing_accuracy": float(self.dosing_accuracy),
        "total_mass_fed": float(self.total_mass_fed),
    }

    return self.outputs_data

to_dict()

Serialize feeder to dictionary.

Source code in pyadm1/components/feeding/feeder.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize feeder to dictionary."""
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "feeder_type": self.feeder_type.value,
        "substrate_type": self.substrate_type.value,
        "Q_max": self.Q_max,
        "dosing_accuracy": self.dosing_accuracy,
        "power_installed": self.power_installed,
        "enable_dosing_noise": self.enable_dosing_noise,
        "state": self.state,
        "inputs": self.inputs,
        "outputs": self.outputs,
    }