Mechanical Components API

pyadm1.components.mechanical.pump.Pump

Bases: Component

Pump component for material handling in biogas plants.

Models different pump types for substrate feeding, recirculation, and digestate transfer. Calculates power consumption based on flow rate, pressure head, and pump efficiency.

Attributes:

Name	Type	Description
pump_type		Type of pump (centrifugal, progressive_cavity, piston)
Q_nom		Nominal flow rate [m³/h]
pressure_head		Pressure head [m] or [bar]
efficiency		Pump efficiency at nominal point (0-1)
motor_efficiency		Motor efficiency (0-1)
fluid_density		Fluid density [kg/m³]
speed_control		Enable variable speed drive (VSD)
current_flow		Current flow rate [m³/h]
is_running		Pump operating state

Example

pump = Pump( ... "feed_pump", ... pump_type="progressive_cavity", ... Q_nom=10.0, ... pressure_head=50.0 ... ) pump.initialize() result = pump.step(0, 1/24, {'Q_setpoint': 8.0})

Source code in pyadm1/components/mechanical/pump.py

class Pump(Component):
    """
    Pump component for material handling in biogas plants.

    Models different pump types for substrate feeding, recirculation, and
    digestate transfer. Calculates power consumption based on flow rate,
    pressure head, and pump efficiency.

    Attributes:
        pump_type: Type of pump (centrifugal, progressive_cavity, piston)
        Q_nom: Nominal flow rate [m³/h]
        pressure_head: Pressure head [m] or [bar]
        efficiency: Pump efficiency at nominal point (0-1)
        motor_efficiency: Motor efficiency (0-1)
        fluid_density: Fluid density [kg/m³]
        speed_control: Enable variable speed drive (VSD)
        current_flow: Current flow rate [m³/h]
        is_running: Pump operating state

    Example:
        >>> pump = Pump(
        ...     "feed_pump",
        ...     pump_type="progressive_cavity",
        ...     Q_nom=10.0,
        ...     pressure_head=50.0
        ... )
        >>> pump.initialize()
        >>> result = pump.step(0, 1/24, {'Q_setpoint': 8.0})
    """

    def __init__(
        self,
        component_id: str,
        pump_type: str = "progressive_cavity",
        Q_nom: float = 10.0,
        pressure_head: float = 50.0,
        efficiency: Optional[float] = None,
        motor_efficiency: float = 0.90,
        fluid_density: float = 1020.0,
        speed_control: bool = True,
        name: Optional[str] = None,
    ):
        """
        Initialize pump component.

        Args:
            component_id: Unique identifier
            pump_type: Type of pump ("centrifugal", "progressive_cavity", "piston")
            Q_nom: Nominal flow rate [m³/h]
            pressure_head: Design pressure head [m]
            efficiency: Pump efficiency (0-1), calculated if None
            motor_efficiency: Motor efficiency (0-1)
            fluid_density: Fluid density [kg/m³]
            speed_control: Enable variable speed drive
            name: Human-readable name
        """
        super().__init__(component_id, ComponentType.MIXER, name)  # Use MIXER as closest type

        # Pump configuration
        self.pump_type = PumpType(pump_type.lower())
        self.Q_nom = Q_nom
        self.pressure_head = pressure_head
        self.motor_efficiency = motor_efficiency
        self.fluid_density = fluid_density
        self.speed_control = speed_control

        # Calculate default efficiency based on pump type and size
        self.efficiency = efficiency or self._estimate_pump_efficiency()

        # Operating state
        self.current_flow = 0.0
        self.is_running = False
        self.operating_hours = 0.0
        self.energy_consumed = 0.0
        self.total_volume_pumped = 0.0

        # Performance tracking
        self.actual_efficiency = self.efficiency
        self.speed_fraction = 1.0

        # Initialize state
        self.initialize()

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

        Args:
            initial_state: Optional initial state dictionary with keys:
                - 'is_running': Initial pump state
                - 'current_flow': Initial flow rate [m³/h]
                - 'operating_hours': Cumulative operating hours
                - 'energy_consumed': Cumulative energy [kWh]
                - 'total_volume_pumped': Cumulative volume [m³]
        """
        if initial_state:
            self.is_running = initial_state.get("is_running", False)
            self.current_flow = initial_state.get("current_flow", 0.0)
            self.operating_hours = initial_state.get("operating_hours", 0.0)
            self.energy_consumed = initial_state.get("energy_consumed", 0.0)
            self.total_volume_pumped = initial_state.get("total_volume_pumped", 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_volume_pumped": self.total_volume_pumped,
            "efficiency": self.efficiency,
            "speed_fraction": self.speed_fraction,
        }

        self.outputs_data = {
            "P_consumed": 0.0,
            "Q_actual": 0.0,
            "is_running": self.is_running,
            "efficiency": self.efficiency,
            "pressure_actual": 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: Current time [days]
            dt: Time step [days]
            inputs: Input data with optional keys:
                - 'Q_setpoint': Desired flow rate [m³/h]
                - 'enable_pump': Enable/disable pump
                - 'fluid_density': Fluid density [kg/m³]
                - 'fluid_viscosity': Fluid viscosity [Pa·s]
                - 'pressure_head': Required pressure head [m]

        Returns:
            Dict with keys:
                - 'P_consumed': Power consumption [kW]
                - 'Q_actual': Actual flow rate [m³/h]
                - 'is_running': Current running state
                - 'efficiency': Current operating efficiency
                - 'pressure_actual': Actual pressure head [m]
                - 'speed_fraction': Speed as fraction of nominal
        """
        # Update fluid properties if provided
        if "fluid_density" in inputs:
            self.fluid_density = inputs["fluid_density"]

        # Update operating state
        enable_pump = inputs.get("enable_pump", True)
        Q_setpoint = inputs.get("Q_setpoint", self.Q_nom)
        pressure_head_req = inputs.get("pressure_head", self.pressure_head)

        # Determine if pump should run
        self.is_running = enable_pump and Q_setpoint > 0

        if not self.is_running:
            # Pump is off
            self.current_flow = 0.0
            self.speed_fraction = 0.0
            P_consumed = 0.0
            pressure_actual = 0.0

        else:
            # Calculate operating point
            if self.speed_control:
                # Variable speed: adjust speed to match flow
                self.speed_fraction = min(1.2, Q_setpoint / self.Q_nom)  # Allow 20% overload
                Q_actual = min(Q_setpoint, self.Q_nom * 1.2)
            else:
                # Fixed speed: flow is nominal
                self.speed_fraction = 1.0
                Q_actual = self.Q_nom

            self.current_flow = Q_actual

            # Calculate efficiency at operating point
            self.actual_efficiency = self._calculate_efficiency_at_operating_point(Q_actual, pressure_head_req)

            # Calculate actual pressure head (may differ from required)
            pressure_actual = self._calculate_pressure_head(Q_actual)

            # Calculate power consumption
            P_consumed = self._calculate_power_consumption(Q_actual, pressure_head_req)

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

        self.energy_consumed += P_consumed * dt_hours

        # 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_volume_pumped": self.total_volume_pumped,
                "efficiency": self.actual_efficiency,
                "speed_fraction": self.speed_fraction,
            }
        )

        # Prepare outputs
        self.outputs_data = {
            "P_consumed": float(P_consumed),
            "Q_actual": float(self.current_flow),
            "is_running": bool(self.is_running),
            "efficiency": float(self.actual_efficiency),
            "pressure_actual": float(pressure_actual),
            "speed_fraction": float(self.speed_fraction),
            "specific_energy": float(P_consumed / max(self.current_flow, 1e-6)),  # kWh/m³
        }

        return self.outputs_data

    def _estimate_pump_efficiency(self) -> float:
        """
        Estimate pump efficiency based on type and size.

        Uses empirical correlations from pump handbooks.

        Returns:
            Estimated pump efficiency (0-1)
        """
        # Efficiency increases with pump size
        # Based on Gülich (2014) correlations

        if self.pump_type == PumpType.CENTRIFUGAL:
            # Centrifugal pumps: 65-85% for typical biogas applications
            if self.Q_nom < 10:
                eta = 0.65
            elif self.Q_nom < 50:
                eta = 0.70
            else:
                eta = 0.75

        elif self.pump_type == PumpType.PROGRESSIVE_CAVITY:
            # Progressive cavity: 50-75% (volumetric pumps are less efficient)
            if self.Q_nom < 10:
                eta = 0.50
            elif self.Q_nom < 50:
                eta = 0.60
            else:
                eta = 0.70

        else:  # PISTON
            # Piston pumps: 70-85%
            if self.Q_nom < 10:
                eta = 0.70
            elif self.Q_nom < 50:
                eta = 0.75
            else:
                eta = 0.80

        return eta

    def _calculate_efficiency_at_operating_point(self, Q: float, H: float) -> float:
        """
        Calculate pump efficiency at current operating point.

        Efficiency varies with flow rate and head. Maximum efficiency
        occurs at design point (Q_nom, H_nom).

        Args:
            Q: Flow rate [m³/h]
            H: Pressure head [m]

        Returns:
            Operating efficiency (0-1)
        """
        if Q <= 0:
            return 0.0

        # Calculate relative flow
        Q_rel = Q / self.Q_nom

        # Efficiency curve (parabolic approximation)
        # Maximum at Q_opt ≈ 1.0 × Q_nom
        Q_opt = 1.0

        if self.pump_type == PumpType.CENTRIFUGAL:
            # Centrifugal pumps have broader efficiency curve
            # eta(Q) = eta_max * (1 - a*(Q/Q_opt - 1)²)
            a = 0.3
            eta = self.efficiency * (1 - a * (Q_rel / Q_opt - 1) ** 2)

        else:  # Volumetric pumps (PC, piston)
            # Volumetric pumps maintain efficiency better at part load
            # but drop off at overload
            if Q_rel <= 1.0:
                # Slight increase at part load due to reduced slip
                eta = self.efficiency * (0.95 + 0.05 * Q_rel)
            else:
                # Efficiency drops at overload
                a = 0.5
                eta = self.efficiency * (1 - a * (Q_rel - 1) ** 2)

        # Ensure reasonable bounds
        eta = max(0.1, min(eta, 0.95))

        return eta

    def _calculate_pressure_head(self, Q: float) -> float:
        """
        Calculate actual pressure head at given flow rate.

        Uses pump characteristic curve. For centrifugal pumps, head
        decreases with flow. For volumetric pumps, head is nearly constant.

        Args:
            Q: Flow rate [m³/h]

        Returns:
            Pressure head [m]
        """
        if self.pump_type == PumpType.CENTRIFUGAL:
            # Centrifugal pump curve: H = H0 - k*Q²
            # At Q_nom: H = H_nom
            # Estimate H0 ≈ 1.2 * H_nom
            H0 = 1.2 * self.pressure_head
            k = (H0 - self.pressure_head) / (self.Q_nom**2)
            H = H0 - k * Q**2

        else:  # Volumetric pumps
            # Nearly constant head (slight drop due to slip at high pressure)
            H = self.pressure_head * 0.98

        return max(0.0, H)

    def _calculate_power_consumption(self, Q: float, H: float) -> float:
        """
        Calculate electrical power consumption.

        Uses hydraulic power formula with efficiency corrections.

        Args:
            Q: Flow rate [m³/h]
            H: Pressure head [m]

        Returns:
            Power consumption [kW]
        """
        if Q <= 0:
            return 0.0

        # Convert flow to m³/s
        Q_m3_per_s = Q / 3600.0

        # Hydraulic power: P_hyd = ρ * g * Q * H [W]
        g = 9.81  # m/s²
        P_hydraulic = self.fluid_density * g * Q_m3_per_s * H / 1000.0  # kW

        # Shaft power (accounting for pump efficiency)
        P_shaft = P_hydraulic / max(self.actual_efficiency, 0.01)

        # Electrical power (accounting for motor efficiency)
        P_electrical = P_shaft / max(self.motor_efficiency, 0.01)

        return P_electrical

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

        Returns:
            Dictionary representation
        """
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "pump_type": self.pump_type.value,
            "Q_nom": self.Q_nom,
            "pressure_head": self.pressure_head,
            "efficiency": self.efficiency,
            "motor_efficiency": self.motor_efficiency,
            "fluid_density": self.fluid_density,
            "speed_control": self.speed_control,
            "state": self.state,
            "inputs": self.inputs,
            "outputs": self.outputs,
        }

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

        Args:
            config: Configuration dictionary

        Returns:
            Pump instance
        """
        pump = cls(
            component_id=config["component_id"],
            pump_type=config.get("pump_type", "progressive_cavity"),
            Q_nom=config.get("Q_nom", 10.0),
            pressure_head=config.get("pressure_head", 50.0),
            efficiency=config.get("efficiency"),
            motor_efficiency=config.get("motor_efficiency", 0.90),
            fluid_density=config.get("fluid_density", 1020.0),
            speed_control=config.get("speed_control", True),
            name=config.get("name"),
        )

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

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

        return pump

Functions

__init__(component_id, pump_type='progressive_cavity', Q_nom=10.0, pressure_head=50.0, efficiency=None, motor_efficiency=0.9, fluid_density=1020.0, speed_control=True, name=None)

Initialize pump component.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier	required
pump_type	str	Type of pump ("centrifugal", "progressive_cavity", "piston")	'progressive_cavity'
Q_nom	float	Nominal flow rate [m³/h]	10.0
pressure_head	float	Design pressure head [m]	50.0
efficiency	Optional[float]	Pump efficiency (0-1), calculated if None	None
motor_efficiency	float	Motor efficiency (0-1)	0.9
fluid_density	float	Fluid density [kg/m³]	1020.0
speed_control	bool	Enable variable speed drive	True
name	Optional[str]	Human-readable name	None

Source code in pyadm1/components/mechanical/pump.py

def __init__(
    self,
    component_id: str,
    pump_type: str = "progressive_cavity",
    Q_nom: float = 10.0,
    pressure_head: float = 50.0,
    efficiency: Optional[float] = None,
    motor_efficiency: float = 0.90,
    fluid_density: float = 1020.0,
    speed_control: bool = True,
    name: Optional[str] = None,
):
    """
    Initialize pump component.

    Args:
        component_id: Unique identifier
        pump_type: Type of pump ("centrifugal", "progressive_cavity", "piston")
        Q_nom: Nominal flow rate [m³/h]
        pressure_head: Design pressure head [m]
        efficiency: Pump efficiency (0-1), calculated if None
        motor_efficiency: Motor efficiency (0-1)
        fluid_density: Fluid density [kg/m³]
        speed_control: Enable variable speed drive
        name: Human-readable name
    """
    super().__init__(component_id, ComponentType.MIXER, name)  # Use MIXER as closest type

    # Pump configuration
    self.pump_type = PumpType(pump_type.lower())
    self.Q_nom = Q_nom
    self.pressure_head = pressure_head
    self.motor_efficiency = motor_efficiency
    self.fluid_density = fluid_density
    self.speed_control = speed_control

    # Calculate default efficiency based on pump type and size
    self.efficiency = efficiency or self._estimate_pump_efficiency()

    # Operating state
    self.current_flow = 0.0
    self.is_running = False
    self.operating_hours = 0.0
    self.energy_consumed = 0.0
    self.total_volume_pumped = 0.0

    # Performance tracking
    self.actual_efficiency = self.efficiency
    self.speed_fraction = 1.0

    # Initialize state
    self.initialize()

from_dict(config) classmethod

Create pump from dictionary.

Parameters:

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

Returns:

Type	Description
Pump	Pump instance

Source code in pyadm1/components/mechanical/pump.py

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

    Args:
        config: Configuration dictionary

    Returns:
        Pump instance
    """
    pump = cls(
        component_id=config["component_id"],
        pump_type=config.get("pump_type", "progressive_cavity"),
        Q_nom=config.get("Q_nom", 10.0),
        pressure_head=config.get("pressure_head", 50.0),
        efficiency=config.get("efficiency"),
        motor_efficiency=config.get("motor_efficiency", 0.90),
        fluid_density=config.get("fluid_density", 1020.0),
        speed_control=config.get("speed_control", True),
        name=config.get("name"),
    )

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

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

    return pump

initialize(initial_state=None)

Initialize pump state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Optional initial state dictionary with keys: - 'is_running': Initial pump state - 'current_flow': Initial flow rate [m³/h] - 'operating_hours': Cumulative operating hours - 'energy_consumed': Cumulative energy [kWh] - 'total_volume_pumped': Cumulative volume [m³]	None

Source code in pyadm1/components/mechanical/pump.py

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

    Args:
        initial_state: Optional initial state dictionary with keys:
            - 'is_running': Initial pump state
            - 'current_flow': Initial flow rate [m³/h]
            - 'operating_hours': Cumulative operating hours
            - 'energy_consumed': Cumulative energy [kWh]
            - 'total_volume_pumped': Cumulative volume [m³]
    """
    if initial_state:
        self.is_running = initial_state.get("is_running", False)
        self.current_flow = initial_state.get("current_flow", 0.0)
        self.operating_hours = initial_state.get("operating_hours", 0.0)
        self.energy_consumed = initial_state.get("energy_consumed", 0.0)
        self.total_volume_pumped = initial_state.get("total_volume_pumped", 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_volume_pumped": self.total_volume_pumped,
        "efficiency": self.efficiency,
        "speed_fraction": self.speed_fraction,
    }

    self.outputs_data = {
        "P_consumed": 0.0,
        "Q_actual": 0.0,
        "is_running": self.is_running,
        "efficiency": self.efficiency,
        "pressure_actual": 0.0,
    }

    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³/h] - 'enable_pump': Enable/disable pump - 'fluid_density': Fluid density [kg/m³] - 'fluid_viscosity': Fluid viscosity [Pa·s] - 'pressure_head': Required pressure head [m]	required

Returns:

Type	Description
Dict[str, Any]	Dict with keys: - 'P_consumed': Power consumption [kW] - 'Q_actual': Actual flow rate [m³/h] - 'is_running': Current running state - 'efficiency': Current operating efficiency - 'pressure_actual': Actual pressure head [m] - 'speed_fraction': Speed as fraction of nominal

Source code in pyadm1/components/mechanical/pump.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³/h]
            - 'enable_pump': Enable/disable pump
            - 'fluid_density': Fluid density [kg/m³]
            - 'fluid_viscosity': Fluid viscosity [Pa·s]
            - 'pressure_head': Required pressure head [m]

    Returns:
        Dict with keys:
            - 'P_consumed': Power consumption [kW]
            - 'Q_actual': Actual flow rate [m³/h]
            - 'is_running': Current running state
            - 'efficiency': Current operating efficiency
            - 'pressure_actual': Actual pressure head [m]
            - 'speed_fraction': Speed as fraction of nominal
    """
    # Update fluid properties if provided
    if "fluid_density" in inputs:
        self.fluid_density = inputs["fluid_density"]

    # Update operating state
    enable_pump = inputs.get("enable_pump", True)
    Q_setpoint = inputs.get("Q_setpoint", self.Q_nom)
    pressure_head_req = inputs.get("pressure_head", self.pressure_head)

    # Determine if pump should run
    self.is_running = enable_pump and Q_setpoint > 0

    if not self.is_running:
        # Pump is off
        self.current_flow = 0.0
        self.speed_fraction = 0.0
        P_consumed = 0.0
        pressure_actual = 0.0

    else:
        # Calculate operating point
        if self.speed_control:
            # Variable speed: adjust speed to match flow
            self.speed_fraction = min(1.2, Q_setpoint / self.Q_nom)  # Allow 20% overload
            Q_actual = min(Q_setpoint, self.Q_nom * 1.2)
        else:
            # Fixed speed: flow is nominal
            self.speed_fraction = 1.0
            Q_actual = self.Q_nom

        self.current_flow = Q_actual

        # Calculate efficiency at operating point
        self.actual_efficiency = self._calculate_efficiency_at_operating_point(Q_actual, pressure_head_req)

        # Calculate actual pressure head (may differ from required)
        pressure_actual = self._calculate_pressure_head(Q_actual)

        # Calculate power consumption
        P_consumed = self._calculate_power_consumption(Q_actual, pressure_head_req)

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

    self.energy_consumed += P_consumed * dt_hours

    # 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_volume_pumped": self.total_volume_pumped,
            "efficiency": self.actual_efficiency,
            "speed_fraction": self.speed_fraction,
        }
    )

    # Prepare outputs
    self.outputs_data = {
        "P_consumed": float(P_consumed),
        "Q_actual": float(self.current_flow),
        "is_running": bool(self.is_running),
        "efficiency": float(self.actual_efficiency),
        "pressure_actual": float(pressure_actual),
        "speed_fraction": float(self.speed_fraction),
        "specific_energy": float(P_consumed / max(self.current_flow, 1e-6)),  # kWh/m³
    }

    return self.outputs_data

to_dict()

Serialize pump to dictionary.

Returns:

Type	Description
Dict[str, Any]	Dictionary representation

Source code in pyadm1/components/mechanical/pump.py

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

    Returns:
        Dictionary representation
    """
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "pump_type": self.pump_type.value,
        "Q_nom": self.Q_nom,
        "pressure_head": self.pressure_head,
        "efficiency": self.efficiency,
        "motor_efficiency": self.motor_efficiency,
        "fluid_density": self.fluid_density,
        "speed_control": self.speed_control,
        "state": self.state,
        "inputs": self.inputs,
        "outputs": self.outputs,
    }

pyadm1.components.mechanical.mixer.Mixer

Bases: Component

Mixer/agitator component for biogas digesters. Models mechanical or hydraulic mixing systems that maintain homogeneity in anaerobic digesters. Calculates power consumption based on mixer type, operating conditions, and fluid properties.

Attributes:

Name	Type	Description
mixer_type		Type of mixer (propeller, paddle, jet)
tank_volume		Tank volume [m³]
tank_diameter		Tank diameter [m]
tank_height		Tank height [m]
mixing_intensity		Mixing intensity level
power_installed		Installed mixer power [kW]
impeller_diameter		Impeller diameter [m]
operating_speed		Mixer rotational speed [rpm]
intermittent		Intermittent operation mode
on_time_fraction		Fraction of time mixer is on (0-1)

Example

mixer = Mixer( ... "mix1", ... mixer_type="propeller", ... tank_volume=2000, ... mixing_intensity="medium" ... ) mixer.initialize() result = mixer.step(0, 1/24, {})

Source code in pyadm1/components/mechanical/mixer.py

class Mixer(Component):
    """
    Mixer/agitator component for biogas digesters.
    Models mechanical or hydraulic mixing systems that maintain homogeneity
    in anaerobic digesters. Calculates power consumption based on mixer type,
    operating conditions, and fluid properties.

    Attributes:
        mixer_type: Type of mixer (propeller, paddle, jet)
        tank_volume: Tank volume [m³]
        tank_diameter: Tank diameter [m]
        tank_height: Tank height [m]
        mixing_intensity: Mixing intensity level
        power_installed: Installed mixer power [kW]
        impeller_diameter: Impeller diameter [m]
        operating_speed: Mixer rotational speed [rpm]
        intermittent: Intermittent operation mode
        on_time_fraction: Fraction of time mixer is on (0-1)

    Example:
        >>> mixer = Mixer(
        ...     "mix1",
        ...     mixer_type="propeller",
        ...     tank_volume=2000,
        ...     mixing_intensity="medium"
        ... )
        >>> mixer.initialize()
        >>> result = mixer.step(0, 1/24, {})
    """

    def __init__(
        self,
        component_id: str,
        mixer_type: str = "propeller",
        tank_volume: float = 2000.0,
        tank_diameter: Optional[float] = None,
        tank_height: Optional[float] = None,
        mixing_intensity: str = "medium",
        power_installed: Optional[float] = None,
        impeller_diameter: Optional[float] = None,
        operating_speed: Optional[float] = None,
        intermittent: bool = True,
        on_time_fraction: float = 0.25,
        name: Optional[str] = None,
    ):
        """
        Initialize mixer component.

        Args:
            component_id: Unique identifier
            mixer_type: Type of mixer ("propeller", "paddle", "jet")
            tank_volume: Tank liquid volume [m³]
            tank_diameter: Tank diameter [m] (calculated if None)
            tank_height: Tank height [m] (calculated if None)
            mixing_intensity: Intensity level ("low", "medium", "high")
            power_installed: Installed power [kW] (calculated if None)
            impeller_diameter: Impeller diameter [m] (calculated if None)
            operating_speed: Rotational speed [rpm] (calculated if None)
            intermittent: Enable intermittent operation
            on_time_fraction: Fraction of time mixer is on (0-1)
            name: Human-readable name
        """
        super().__init__(component_id, ComponentType.MIXER, name)

        # Mixer configuration
        self.mixer_type = MixerType(mixer_type.lower())
        self.mixing_intensity = MixingIntensity(mixing_intensity.lower())
        self.intermittent = intermittent
        self.on_time_fraction = min(1.0, max(0.0, on_time_fraction))

        # Tank geometry
        self.tank_volume = tank_volume
        self.tank_diameter = tank_diameter or self._estimate_tank_diameter(tank_volume)
        self.tank_height = tank_height or self._estimate_tank_height(tank_volume, self.tank_diameter)

        # Mixer geometry and power
        self.impeller_diameter = impeller_diameter or self._estimate_impeller_diameter()
        self.operating_speed = operating_speed or self._estimate_operating_speed()
        self.power_installed = power_installed or self._estimate_power_requirement()

        # Fluid properties (typical biogas substrate)
        self.fluid_density = 1020.0  # kg/m³
        self.fluid_viscosity = 0.050  # Pa·s (50 mPa·s, typical for biogas substrate)

        # Operating state
        self.is_running = True
        self.current_speed_fraction = 1.0  # Fraction of nominal speed
        self.operating_hours = 0.0
        self.energy_consumed = 0.0

        # Performance tracking
        self.mixing_time = 0.0  # Time to achieve homogeneity [min]
        self.power_number = 0.0  # Dimensionless power number
        self.reynolds_number = 0.0  # Reynolds number for mixing

        # Initialize state
        self.initialize()

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

        Args:
            initial_state: Optional initial state dictionary with keys:
                - 'is_running': Mixer running state
                - 'current_speed_fraction': Speed fraction (0-1)
                - 'operating_hours': Cumulative operating hours
                - 'energy_consumed': Cumulative energy [kWh]
        """
        if initial_state:
            self.is_running = initial_state.get("is_running", True)
            self.current_speed_fraction = initial_state.get("current_speed_fraction", 1.0)
            self.operating_hours = initial_state.get("operating_hours", 0.0)
            self.energy_consumed = initial_state.get("energy_consumed", 0.0)

        # Calculate initial performance parameters
        self._calculate_mixing_parameters()

        self.state = {
            "is_running": self.is_running,
            "current_speed_fraction": self.current_speed_fraction,
            "operating_hours": self.operating_hours,
            "energy_consumed": self.energy_consumed,
            "power_number": self.power_number,
            "reynolds_number": self.reynolds_number,
            "mixing_time": self.mixing_time,
        }

        self.outputs_data = {
            "P_consumed": 0.0,
            "is_running": self.is_running,
            "mixing_quality": 1.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: Current time [days]
            dt: Time step [days]
            inputs: Input data with optional keys:
                - 'speed_setpoint': Desired speed fraction (0-1)
                - 'enable_mixing': Enable/disable mixer
                - 'fluid_viscosity': Fluid viscosity [Pa·s]
                - 'temperature': Fluid temperature [K]

        Returns:
            Dict with keys:
                - 'P_consumed': Power consumption [kW]
                - 'P_average': Time-averaged power [kW]
                - 'is_running': Current running state
                - 'mixing_quality': Mixing quality index (0-1)
                - 'reynolds_number': Reynolds number
                - 'power_number': Power number
                - 'mixing_time': Mixing time [min]
                - 'shear_rate': Average shear rate [1/s]
        """
        # Update fluid properties if provided
        if "fluid_viscosity" in inputs:
            self.fluid_viscosity = inputs["fluid_viscosity"]

        if "temperature" in inputs:
            self._update_fluid_properties(inputs["temperature"])

        # Update operating state
        enable_mixing = inputs.get("enable_mixing", True)
        speed_setpoint = inputs.get("speed_setpoint", 1.0)

        # Intermittent operation logic
        if self.intermittent:
            # Simple on/off cycling based on on_time_fraction
            cycle_time = 1.0  # 1 hour cycle
            time_in_cycle = (t % cycle_time) / cycle_time
            self.is_running = time_in_cycle < self.on_time_fraction and enable_mixing
        else:
            self.is_running = enable_mixing

        # Update speed
        # take max of speed_setpoint, 0.0 - to avoid wrong calculations if accidently a negative setpoint is given
        self.current_speed_fraction = max(speed_setpoint, 0.0) if self.is_running else 0.0

        # Calculate mixing parameters
        self._calculate_mixing_parameters()

        # Calculate power consumption
        P_consumed = self._calculate_power_consumption()

        # Calculate time-averaged power (accounting for intermittent operation)
        if self.intermittent:
            P_average = P_consumed * self.on_time_fraction
        else:
            P_average = P_consumed

        # Calculate mixing quality
        mixing_quality = self._calculate_mixing_quality()

        # Calculate average shear rate
        shear_rate = self._calculate_shear_rate()

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

        # Update state
        self.state.update(
            {
                "is_running": self.is_running,
                "current_speed_fraction": self.current_speed_fraction,
                "operating_hours": self.operating_hours,
                "energy_consumed": self.energy_consumed,
                "power_number": self.power_number,
                "reynolds_number": self.reynolds_number,
                "mixing_time": self.mixing_time,
            }
        )

        # Prepare outputs
        self.outputs_data = {
            "P_consumed": float(P_consumed),
            "P_average": float(P_average),
            "is_running": bool(self.is_running),
            "mixing_quality": float(mixing_quality),
            "reynolds_number": float(self.reynolds_number),
            "power_number": float(self.power_number),
            "mixing_time": float(self.mixing_time),
            "shear_rate": float(shear_rate),
            "specific_power": float(P_consumed / self.tank_volume),  # kW/m³
            "tip_speed": float(self._calculate_tip_speed()),  # m/s
        }

        return self.outputs_data

    def _calculate_mixing_parameters(self) -> None:
        """Calculate mixing performance parameters."""
        # Current rotational speed
        N = self.operating_speed * self.current_speed_fraction / 60.0  # Hz (rev/s)
        D = self.impeller_diameter

        # Reynolds number for mixing
        # Re = ρ * N * D² / μ
        self.reynolds_number = self.fluid_density * N * D**2 / self.fluid_viscosity

        # Power number (depends on mixer type and Reynolds number)
        self.power_number = self._calculate_power_number()

        # Mixing time estimation (Nienow correlation)
        # θ_mix = C * (D_T/D)^α * (H/D_T)^β / N
        # where C, α, β depend on mixer type
        D_T = self.tank_diameter
        H = self.tank_height

        if self.mixer_type == MixerType.PROPELLER:
            C, alpha, beta = 5.3, 2.0, 0.5
        elif self.mixer_type == MixerType.PADDLE:
            C, alpha, beta = 6.5, 2.5, 0.7
        else:  # JET
            C, alpha, beta = 4.0, 1.5, 0.3

        if N > 1e-6:  # Only calculate if mixer is actually running
            self.mixing_time = C * (D_T / D) ** alpha * (H / D_T) ** beta / (N * 60.0)  # minutes
        else:
            self.mixing_time = float("inf")

    def _calculate_power_number(self) -> float:
        """
        Calculate power number based on Reynolds number and mixer type.

        When mixer is not running (Re < 1e-6), returns a safe default value.

        Returns:
            Power number (dimensionless)
        """
        Re = self.reynolds_number

        # Handle zero Reynolds number (mixer not running)
        if Re < 1e-6:
            # Return a reasonable default for the mixer type
            if self.mixer_type == MixerType.PROPELLER:
                return 0.32  # Turbulent regime value
            elif self.mixer_type == MixerType.PADDLE:
                return 5.0
            else:  # JET
                return 0.1

        if self.mixer_type == MixerType.PROPELLER:
            # Propeller: transition from laminar to turbulent
            if Re < 100:
                Np = 14.0 * Re ** (-0.67)  # Laminar regime
            elif Re < 10000:
                Np = 1.2 * Re ** (-0.15)  # Transition regime
            else:
                Np = 0.32  # Turbulent regime (constant)

        elif self.mixer_type == MixerType.PADDLE:
            # Paddle mixer
            if Re < 10:
                Np = 300.0 / Re
            elif Re < 10000:
                Np = 8.0 * Re ** (-0.25)
            else:
                Np = 5.0

        else:  # JET
            # Jet mixer (based on jet momentum)
            # Power number is not directly applicable, use empirical value
            Np = 0.1

        return Np

    def _calculate_power_consumption(self) -> float:
        """
        Calculate actual power consumption.

        Returns:
            Power consumption [kW]
        """
        if not self.is_running:
            return 0.0

        # Mechanical power from power number correlation
        # P = Np * ρ * N³ * D⁵
        N = self.operating_speed * self.current_speed_fraction / 60.0  # Hz
        D = self.impeller_diameter

        P_mech = self.power_number * self.fluid_density * N**3 * D**5 / 1000.0  # Convert W to kW

        P_electrical = P_mech / MOTOR_EFFICIENCY

        # Part-load electrical limit should scale with speed^3
        speed = max(self.current_speed_fraction, 0.0)
        if speed <= 1.0:
            dynamic_limit = self.power_installed * speed**3
        else:
            dynamic_limit = self.power_installed * min(speed**3, 1.2)

        # Limit to installed power
        P_actual = min(P_electrical, dynamic_limit)

        return P_actual

    def _calculate_mixing_quality(self) -> float:
        """
        Calculate mixing quality index based on mixing time and intensity.

        Returns:
            Mixing quality (0-1, where 1 is perfect mixing)
        """
        if not self.is_running:
            return 0.0

        # Quality based on mixing time
        # Good mixing: < 5 min, Poor mixing: > 30 min
        if self.mixing_time < 5.0:
            quality = 1.0
        elif self.mixing_time > 30.0:
            quality = 0.3
        else:
            quality = 1.0 - 0.7 * (self.mixing_time - 5.0) / 25.0

        # Adjust for speed fraction
        quality *= self.current_speed_fraction

        # Adjust for Reynolds number (laminar flow reduces quality)
        if self.reynolds_number < 1000:
            quality *= self.reynolds_number / 1000.0

        return min(1.0, quality)

    def _calculate_shear_rate(self) -> float:
        """
        Calculate average shear rate in the tank.

        Returns:
            Average shear rate [1/s]
        """
        # Average shear rate estimation
        # γ̇ ≈ k * N * (D/D_T)
        # where k is a constant (typically 10-15 for propeller mixers)

        N = self.operating_speed * self.current_speed_fraction / 60.0  # Hz
        D = self.impeller_diameter
        D_T = self.tank_diameter

        if self.mixer_type == MixerType.PROPELLER:
            k = 13.0
        elif self.mixer_type == MixerType.PADDLE:
            k = 11.0
        else:  # JET
            k = 8.0

        shear_rate = k * N * (D / D_T)

        return shear_rate

    def _calculate_tip_speed(self) -> float:
        """
        Calculate impeller tip speed.

        Returns:
            Tip speed [m/s]
        """
        N = self.operating_speed * self.current_speed_fraction / 60.0  # Hz
        D = self.impeller_diameter

        tip_speed = np.pi * N * D

        return tip_speed

    def _update_fluid_properties(self, temperature: float) -> None:
        """
        Update fluid properties based on temperature.

        Args:
            temperature: Fluid temperature [K]
        """
        # Viscosity temperature dependence (Arrhenius-type)
        # μ(T) = μ₀ * exp(Ea/R * (1/T - 1/T₀))
        T_0 = 308.15  # Reference temperature (35°C)
        mu_0 = 0.050  # Reference viscosity [Pa·s]
        Ea_R = 2000.0  # Activation energy / gas constant [K]

        self.fluid_viscosity = mu_0 * np.exp(Ea_R * (1 / temperature - 1 / T_0))

    @staticmethod
    def _estimate_tank_diameter(volume: float) -> float:
        """
        Estimate tank diameter from volume assuming cylindrical tank.

        Args:
            volume: Tank volume [m³]

        Returns:
            Estimated diameter [m]
        """
        # Assume H/D = 1.5 (typical for digesters)
        # V = π/4 * D² * H = π/4 * D² * 1.5*D = 1.178 * D³
        diameter = (volume / 1.178) ** (1 / 3)
        return diameter

    @staticmethod
    def _estimate_tank_height(volume: float, diameter: float) -> float:
        """
        Estimate tank height from volume and diameter.

        Args:
            volume: Tank volume [m³]
            diameter: Tank diameter [m]

        Returns:
            Estimated height [m]
        """
        # V = π/4 * D² * H
        height = volume / (np.pi / 4 * diameter**2)
        return height

    def _estimate_impeller_diameter(self) -> float:
        """
        Estimate impeller diameter based on tank size and mixer type.

        Returns:
            Impeller diameter [m]
        """
        D_T = self.tank_diameter

        # D/D_T ratio depends on mixer type
        if self.mixer_type == MixerType.PROPELLER:
            ratio = 0.33  # Typically 1/3 tank diameter
        elif self.mixer_type == MixerType.PADDLE:
            ratio = 0.50  # Larger for paddles
        else:  # JET
            ratio = 0.10  # Jet nozzle diameter

        return ratio * D_T

    def _estimate_operating_speed(self) -> float:
        """
        Estimate operating speed based on mixer type and tank size.

        Returns:
            Operating speed [rpm]
        """
        if self.mixer_type == MixerType.PROPELLER:
            # Propellers: typically 40-100 rpm for large digesters
            speed = 60.0
        elif self.mixer_type == MixerType.PADDLE:
            # Paddles: typically 20-60 rpm
            speed = 40.0
        else:  # JET
            # Jet mixers: recirculation pump speed
            speed = 1450.0  # Typical pump speed

        # Scale with tank size (smaller tanks → higher speed)
        scale_factor = (2000.0 / self.tank_volume) ** (1 / 3)
        speed *= scale_factor

        return speed

    def _estimate_power_requirement(self) -> float:
        """
        Estimate power requirement based on mixing intensity and tank volume.

        Returns:
            Power requirement [kW]
        """
        # Specific power input [W/m³] depends on intensity
        if self.mixing_intensity == MixingIntensity.LOW:
            specific_power = 3.0  # W/m³
        elif self.mixing_intensity == MixingIntensity.MEDIUM:
            specific_power = 5.0  # W/m³
        else:  # HIGH
            specific_power = 8.0  # W/m³

        # Total power
        power = specific_power * self.tank_volume / 1000.0  # kW

        # Adjust for mixer type
        if self.mixer_type == MixerType.PROPELLER:
            power *= 1.0  # Baseline
        elif self.mixer_type == MixerType.PADDLE:
            power *= 1.2  # Paddles typically need more power
        else:  # JET
            power *= 1.5  # Jet mixers include pump power

        return power

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

        Returns:
            Dictionary representation
        """
        return {
            "component_id": self.component_id,
            "component_type": self.component_type.value,
            "name": self.name,
            "mixer_type": self.mixer_type.value,
            "mixing_intensity": self.mixing_intensity.value,
            "tank_volume": self.tank_volume,
            "tank_diameter": self.tank_diameter,
            "tank_height": self.tank_height,
            "impeller_diameter": self.impeller_diameter,
            "operating_speed": self.operating_speed,
            "power_installed": self.power_installed,
            "intermittent": self.intermittent,
            "on_time_fraction": self.on_time_fraction,
            "fluid_density": self.fluid_density,
            "fluid_viscosity": self.fluid_viscosity,
            "state": self.state,
            "inputs": self.inputs,
            "outputs": self.outputs,
        }

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

        Args:
            config: Configuration dictionary

        Returns:
            Mixer instance
        """
        mixer = cls(
            component_id=config["component_id"],
            mixer_type=config.get("mixer_type", "propeller"),
            tank_volume=config.get("tank_volume", 2000.0),
            tank_diameter=config.get("tank_diameter"),
            tank_height=config.get("tank_height"),
            mixing_intensity=config.get("mixing_intensity", "medium"),
            power_installed=config.get("power_installed"),
            impeller_diameter=config.get("impeller_diameter"),
            operating_speed=config.get("operating_speed"),
            intermittent=config.get("intermittent", True),
            on_time_fraction=config.get("on_time_fraction", 0.25),
            name=config.get("name"),
        )

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

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

        return mixer

Functions

__init__(component_id, mixer_type='propeller', tank_volume=2000.0, tank_diameter=None, tank_height=None, mixing_intensity='medium', power_installed=None, impeller_diameter=None, operating_speed=None, intermittent=True, on_time_fraction=0.25, name=None)

Initialize mixer component.

Parameters:

Name	Type	Description	Default
component_id	str	Unique identifier	required
mixer_type	str	Type of mixer ("propeller", "paddle", "jet")	'propeller'
tank_volume	float	Tank liquid volume [m³]	2000.0
tank_diameter	Optional[float]	Tank diameter [m] (calculated if None)	None
tank_height	Optional[float]	Tank height [m] (calculated if None)	None
mixing_intensity	str	Intensity level ("low", "medium", "high")	'medium'
power_installed	Optional[float]	Installed power [kW] (calculated if None)	None
impeller_diameter	Optional[float]	Impeller diameter [m] (calculated if None)	None
operating_speed	Optional[float]	Rotational speed [rpm] (calculated if None)	None
intermittent	bool	Enable intermittent operation	True
on_time_fraction	float	Fraction of time mixer is on (0-1)	0.25
name	Optional[str]	Human-readable name	None

Source code in pyadm1/components/mechanical/mixer.py

def __init__(
    self,
    component_id: str,
    mixer_type: str = "propeller",
    tank_volume: float = 2000.0,
    tank_diameter: Optional[float] = None,
    tank_height: Optional[float] = None,
    mixing_intensity: str = "medium",
    power_installed: Optional[float] = None,
    impeller_diameter: Optional[float] = None,
    operating_speed: Optional[float] = None,
    intermittent: bool = True,
    on_time_fraction: float = 0.25,
    name: Optional[str] = None,
):
    """
    Initialize mixer component.

    Args:
        component_id: Unique identifier
        mixer_type: Type of mixer ("propeller", "paddle", "jet")
        tank_volume: Tank liquid volume [m³]
        tank_diameter: Tank diameter [m] (calculated if None)
        tank_height: Tank height [m] (calculated if None)
        mixing_intensity: Intensity level ("low", "medium", "high")
        power_installed: Installed power [kW] (calculated if None)
        impeller_diameter: Impeller diameter [m] (calculated if None)
        operating_speed: Rotational speed [rpm] (calculated if None)
        intermittent: Enable intermittent operation
        on_time_fraction: Fraction of time mixer is on (0-1)
        name: Human-readable name
    """
    super().__init__(component_id, ComponentType.MIXER, name)

    # Mixer configuration
    self.mixer_type = MixerType(mixer_type.lower())
    self.mixing_intensity = MixingIntensity(mixing_intensity.lower())
    self.intermittent = intermittent
    self.on_time_fraction = min(1.0, max(0.0, on_time_fraction))

    # Tank geometry
    self.tank_volume = tank_volume
    self.tank_diameter = tank_diameter or self._estimate_tank_diameter(tank_volume)
    self.tank_height = tank_height or self._estimate_tank_height(tank_volume, self.tank_diameter)

    # Mixer geometry and power
    self.impeller_diameter = impeller_diameter or self._estimate_impeller_diameter()
    self.operating_speed = operating_speed or self._estimate_operating_speed()
    self.power_installed = power_installed or self._estimate_power_requirement()

    # Fluid properties (typical biogas substrate)
    self.fluid_density = 1020.0  # kg/m³
    self.fluid_viscosity = 0.050  # Pa·s (50 mPa·s, typical for biogas substrate)

    # Operating state
    self.is_running = True
    self.current_speed_fraction = 1.0  # Fraction of nominal speed
    self.operating_hours = 0.0
    self.energy_consumed = 0.0

    # Performance tracking
    self.mixing_time = 0.0  # Time to achieve homogeneity [min]
    self.power_number = 0.0  # Dimensionless power number
    self.reynolds_number = 0.0  # Reynolds number for mixing

    # Initialize state
    self.initialize()

from_dict(config) classmethod

Create mixer from dictionary.

Parameters:

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

Returns:

Type	Description
Mixer	Mixer instance

Source code in pyadm1/components/mechanical/mixer.py

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

    Args:
        config: Configuration dictionary

    Returns:
        Mixer instance
    """
    mixer = cls(
        component_id=config["component_id"],
        mixer_type=config.get("mixer_type", "propeller"),
        tank_volume=config.get("tank_volume", 2000.0),
        tank_diameter=config.get("tank_diameter"),
        tank_height=config.get("tank_height"),
        mixing_intensity=config.get("mixing_intensity", "medium"),
        power_installed=config.get("power_installed"),
        impeller_diameter=config.get("impeller_diameter"),
        operating_speed=config.get("operating_speed"),
        intermittent=config.get("intermittent", True),
        on_time_fraction=config.get("on_time_fraction", 0.25),
        name=config.get("name"),
    )

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

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

    return mixer

initialize(initial_state=None)

Initialize mixer state.

Parameters:

Name	Type	Description	Default
initial_state	Optional[Dict[str, Any]]	Optional initial state dictionary with keys: - 'is_running': Mixer running state - 'current_speed_fraction': Speed fraction (0-1) - 'operating_hours': Cumulative operating hours - 'energy_consumed': Cumulative energy [kWh]	None

Source code in pyadm1/components/mechanical/mixer.py

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

    Args:
        initial_state: Optional initial state dictionary with keys:
            - 'is_running': Mixer running state
            - 'current_speed_fraction': Speed fraction (0-1)
            - 'operating_hours': Cumulative operating hours
            - 'energy_consumed': Cumulative energy [kWh]
    """
    if initial_state:
        self.is_running = initial_state.get("is_running", True)
        self.current_speed_fraction = initial_state.get("current_speed_fraction", 1.0)
        self.operating_hours = initial_state.get("operating_hours", 0.0)
        self.energy_consumed = initial_state.get("energy_consumed", 0.0)

    # Calculate initial performance parameters
    self._calculate_mixing_parameters()

    self.state = {
        "is_running": self.is_running,
        "current_speed_fraction": self.current_speed_fraction,
        "operating_hours": self.operating_hours,
        "energy_consumed": self.energy_consumed,
        "power_number": self.power_number,
        "reynolds_number": self.reynolds_number,
        "mixing_time": self.mixing_time,
    }

    self.outputs_data = {
        "P_consumed": 0.0,
        "is_running": self.is_running,
        "mixing_quality": 1.0,
    }

    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: - 'speed_setpoint': Desired speed fraction (0-1) - 'enable_mixing': Enable/disable mixer - 'fluid_viscosity': Fluid viscosity [Pa·s] - 'temperature': Fluid temperature [K]	required

Returns:

Type	Description
Dict[str, Any]	Dict with keys: - 'P_consumed': Power consumption [kW] - 'P_average': Time-averaged power [kW] - 'is_running': Current running state - 'mixing_quality': Mixing quality index (0-1) - 'reynolds_number': Reynolds number - 'power_number': Power number - 'mixing_time': Mixing time [min] - 'shear_rate': Average shear rate [1/s]

Source code in pyadm1/components/mechanical/mixer.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:
            - 'speed_setpoint': Desired speed fraction (0-1)
            - 'enable_mixing': Enable/disable mixer
            - 'fluid_viscosity': Fluid viscosity [Pa·s]
            - 'temperature': Fluid temperature [K]

    Returns:
        Dict with keys:
            - 'P_consumed': Power consumption [kW]
            - 'P_average': Time-averaged power [kW]
            - 'is_running': Current running state
            - 'mixing_quality': Mixing quality index (0-1)
            - 'reynolds_number': Reynolds number
            - 'power_number': Power number
            - 'mixing_time': Mixing time [min]
            - 'shear_rate': Average shear rate [1/s]
    """
    # Update fluid properties if provided
    if "fluid_viscosity" in inputs:
        self.fluid_viscosity = inputs["fluid_viscosity"]

    if "temperature" in inputs:
        self._update_fluid_properties(inputs["temperature"])

    # Update operating state
    enable_mixing = inputs.get("enable_mixing", True)
    speed_setpoint = inputs.get("speed_setpoint", 1.0)

    # Intermittent operation logic
    if self.intermittent:
        # Simple on/off cycling based on on_time_fraction
        cycle_time = 1.0  # 1 hour cycle
        time_in_cycle = (t % cycle_time) / cycle_time
        self.is_running = time_in_cycle < self.on_time_fraction and enable_mixing
    else:
        self.is_running = enable_mixing

    # Update speed
    # take max of speed_setpoint, 0.0 - to avoid wrong calculations if accidently a negative setpoint is given
    self.current_speed_fraction = max(speed_setpoint, 0.0) if self.is_running else 0.0

    # Calculate mixing parameters
    self._calculate_mixing_parameters()

    # Calculate power consumption
    P_consumed = self._calculate_power_consumption()

    # Calculate time-averaged power (accounting for intermittent operation)
    if self.intermittent:
        P_average = P_consumed * self.on_time_fraction
    else:
        P_average = P_consumed

    # Calculate mixing quality
    mixing_quality = self._calculate_mixing_quality()

    # Calculate average shear rate
    shear_rate = self._calculate_shear_rate()

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

    # Update state
    self.state.update(
        {
            "is_running": self.is_running,
            "current_speed_fraction": self.current_speed_fraction,
            "operating_hours": self.operating_hours,
            "energy_consumed": self.energy_consumed,
            "power_number": self.power_number,
            "reynolds_number": self.reynolds_number,
            "mixing_time": self.mixing_time,
        }
    )

    # Prepare outputs
    self.outputs_data = {
        "P_consumed": float(P_consumed),
        "P_average": float(P_average),
        "is_running": bool(self.is_running),
        "mixing_quality": float(mixing_quality),
        "reynolds_number": float(self.reynolds_number),
        "power_number": float(self.power_number),
        "mixing_time": float(self.mixing_time),
        "shear_rate": float(shear_rate),
        "specific_power": float(P_consumed / self.tank_volume),  # kW/m³
        "tip_speed": float(self._calculate_tip_speed()),  # m/s
    }

    return self.outputs_data

to_dict()

Serialize mixer to dictionary.

Returns:

Type	Description
Dict[str, Any]	Dictionary representation

Source code in pyadm1/components/mechanical/mixer.py

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

    Returns:
        Dictionary representation
    """
    return {
        "component_id": self.component_id,
        "component_type": self.component_type.value,
        "name": self.name,
        "mixer_type": self.mixer_type.value,
        "mixing_intensity": self.mixing_intensity.value,
        "tank_volume": self.tank_volume,
        "tank_diameter": self.tank_diameter,
        "tank_height": self.tank_height,
        "impeller_diameter": self.impeller_diameter,
        "operating_speed": self.operating_speed,
        "power_installed": self.power_installed,
        "intermittent": self.intermittent,
        "on_time_fraction": self.on_time_fraction,
        "fluid_density": self.fluid_density,
        "fluid_viscosity": self.fluid_viscosity,
        "state": self.state,
        "inputs": self.inputs,
        "outputs": self.outputs,
    }