Konfigurator-API

pyadm1.configurator.plant_builder.BiogasPlant

Complete biogas plant model with multiple components.

Manages component lifecycle, connections, and simulation. Supports JSON-based configuration.

Attributes:

Name	Type	Description
plant_name	str	Name of the biogas plant.
components	Dict[str, Component]	Dictionary of all plant components.
connections	List[Connection]	List of connections between components.
simulation_time	float	Current simulation time in days.

Example

from pyadm1.substrates.feedstock import Feedstock from pyadm1.configurator.plant_builder import BiogasPlant from pyadm1.components.biological.digester import Digester

feedstock = Feedstock(feeding_freq=48) plant = BiogasPlant("My Plant") digester = Digester("dig1", feedstock, V_liq=2000) plant.add_component(digester) plant.initialize()

Source code in pyadm1/configurator/plant_builder.py

class BiogasPlant:
    """
    Complete biogas plant model with multiple components.

    Manages component lifecycle, connections, and simulation.
    Supports JSON-based configuration.

    Attributes:
        plant_name (str): Name of the biogas plant.
        components (Dict[str, Component]): Dictionary of all plant components.
        connections (List[Connection]): List of connections between components.
        simulation_time (float): Current simulation time in days.

    Example:
        >>> from pyadm1.substrates.feedstock import Feedstock
        >>> from pyadm1.configurator.plant_builder import BiogasPlant
        >>> from pyadm1.components.biological.digester import Digester
        >>>
        >>> feedstock = Feedstock(feeding_freq=48)
        >>> plant = BiogasPlant("My Plant")
        >>> digester = Digester("dig1", feedstock, V_liq=2000)
        >>> plant.add_component(digester)
        >>> plant.initialize()
    """

    def __init__(self, plant_name: str = "Biogas Plant"):
        """
        Initialize biogas plant.

        Args:
            plant_name (str): Name of the plant. Defaults to "Biogas Plant".
        """
        self.plant_name = plant_name
        self.components: Dict[str, Component] = {}
        self.connections: List[Connection] = []
        self.simulation_time = 0.0

    def add_component(self, component: Component) -> None:
        """
        Add a component to the plant.

        Args:
            component (Component): Component to add to the plant.

        Raises:
            ValueError: If component with same ID already exists.
        """
        if component.component_id in self.components:
            raise ValueError(f"Component with ID '{component.component_id}' already exists")
        self.components[component.component_id] = component

    def add_connection(self, connection: Connection) -> None:
        """
        Add a connection between components.

        Args:
            connection (Connection): Connection to add.

        Raises:
            ValueError: If source or target component not found.
        """
        # Verify components exist
        if connection.from_component not in self.components:
            raise ValueError(f"Source component '{connection.from_component}' not found")
        if connection.to_component not in self.components:
            raise ValueError(f"Target component '{connection.to_component}' not found")

        # Update component connections
        from_comp = self.components[connection.from_component]
        to_comp = self.components[connection.to_component]

        from_comp.add_output(connection.to_component)
        to_comp.add_input(connection.from_component)

        self.connections.append(connection)

    def initialize(self) -> None:
        """
        Initialize all components.

        Note: Most components auto-initialize in their constructor.
        This method is kept for compatibility and to ensure any
        components that need explicit initialization are handled.
        """
        for component in self.components.values():
            if not component._initialized:
                component.initialize()

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

        This uses a three-pass execution model:
        1. Execute digesters to produce gas → storages
        2. Execute CHPs to determine gas demand → storages
        3. Execute storages to supply gas → CHPs (re-execute with actual supply)

        Args:
            dt (float): Time step in days.

        Returns:
            Dict[str, Dict[str, Any]]: Results from all components.
        """
        results = {}

        # Build dependency graph
        execution_order = self._get_execution_order()

        # ========================================================================
        # PASS 1: Execute all non-storage components
        # ========================================================================
        for component_id in execution_order:
            component = self.components[component_id]

            # Skip storages in first pass
            if component.component_type.value == "storage":
                continue

            # Gather inputs from connected components
            inputs = {}
            for input_id in component.inputs:
                if input_id in self.components:
                    input_comp = self.components[input_id]
                    inputs.update(input_comp.outputs_data)

            # Execute component
            output = component.step(self.simulation_time, dt, inputs)
            results[component_id] = output

        # ========================================================================
        # PASS 2: Execute gas storages with gas production from digesters
        # ========================================================================
        for component_id in execution_order:
            component = self.components[component_id]

            if component.component_type.value != "storage":
                continue

            # Get gas input from connected digesters
            gas_input = 0.0
            for conn in self.connections:
                if conn.to_component == component_id and conn.connection_type == "gas":
                    source_comp = self.components.get(conn.from_component)
                    if source_comp and source_comp.component_type.value == "digester":
                        gas_input += source_comp.outputs_data.get("Q_gas", 0.0)

            # Execute storage with production only (no demand yet)
            storage_inputs = {
                "Q_gas_in_m3_per_day": gas_input,
                "Q_gas_out_m3_per_day": 0.0,  # No demand yet
                "vent_to_flare": True,
            }

            output = component.step(self.simulation_time, dt, storage_inputs)
            results[component_id] = output

        # ========================================================================
        # PASS 3: Handle gas demand from CHPs
        # ========================================================================
        for component_id, component in self.components.items():
            if component.component_type.value != "chp":
                continue

            # Calculate CHP gas demand based on current operating point
            P_el_nom = component.P_el_nom
            eta_el = component.eta_el
            load_setpoint = 1.0  # Full load by default

            # Calculate required gas (m³/d biogas at 60% CH4)
            E_ch4 = 10.0  # kWh/m³ CH4
            CH4_content = 0.60
            P_required = load_setpoint * P_el_nom  # kW
            Q_ch4_required = (P_required / eta_el) * 24.0 / E_ch4  # m³/d CH4
            Q_gas_required = Q_ch4_required / CH4_content  # m³/d biogas

            # Find all gas storages connected to this CHP
            connected_storages = []
            for conn in self.connections:
                if conn.to_component == component_id and conn.connection_type == "gas":
                    storage_id = conn.from_component
                    if storage_id in self.components:
                        storage_comp = self.components[storage_id]
                        if storage_comp.component_type.value == "storage":
                            connected_storages.append(storage_id)

            if not connected_storages:
                continue

            # Distribute demand equally among storages
            demand_per_storage = Q_gas_required / len(connected_storages)

            total_supplied = 0.0

            # Re-execute storages with gas demand
            for storage_id in connected_storages:
                storage = self.components[storage_id]

                # Get gas input from pass 2
                gas_input = 0.0
                for conn in self.connections:
                    if conn.to_component == storage_id and conn.connection_type == "gas":
                        source_comp = self.components.get(conn.from_component)
                        if source_comp and source_comp.component_type.value == "digester":
                            gas_input += source_comp.outputs_data.get("Q_gas", 0.0)

                # Re-execute with demand
                storage_inputs = {
                    "Q_gas_in_m3_per_day": gas_input,
                    "Q_gas_out_m3_per_day": demand_per_storage,
                    "vent_to_flare": True,
                }

                storage_output = storage.step(self.simulation_time, dt, storage_inputs)
                results[storage_id] = storage_output

                # Accumulate supplied gas
                supplied = storage_output.get("Q_gas_supplied_m3_per_day", 0.0)
                total_supplied += supplied

            # Re-execute CHP with actual gas supply
            chp_inputs = {
                "Q_gas_supplied_m3_per_day": total_supplied,
                "load_setpoint": load_setpoint,
            }

            chp_output = component.step(self.simulation_time, dt, chp_inputs)
            results[component_id] = chp_output

        self.simulation_time += dt

        return results

    def simulate(
        self,
        duration: float,
        dt: float = 1.0 / 24.0,
        save_interval: Optional[float] = None,
    ) -> List[Dict[str, Any]]:
        """
        Run simulation for specified duration.

        Args:
            duration (float): Simulation duration in days.
            dt (float): Time step in days. Defaults to 1 hour (1/24 day).
            save_interval (Optional[float]): Interval for saving results in days.
                If None, saves every step.

        Returns:
            List[Dict[str, Any]]: Simulation results at each saved time point.
                Each entry contains 'time' and 'components' with component results.

        Example:
            >>> results = plant.simulate(duration=30, dt=1/24, save_interval=1.0)
            >>> print(f"Simulated {len(results)} time points")
        """
        if save_interval is None:
            save_interval = dt

        results = []
        next_save_time = self.simulation_time

        n_steps = int(duration / dt)

        for i in range(n_steps):
            step_result = self.step(dt)

            # Save results at specified interval
            if self.simulation_time >= next_save_time:
                results.append(
                    {
                        "time": self.simulation_time,
                        "components": step_result,
                    }
                )
                next_save_time += save_interval

            if (i + 1) % 100 == 0:
                print(f"Simulated {i + 1}/{n_steps} steps")

        return results

    def _get_execution_order(self) -> List[str]:
        """
        Determine execution order based on component dependencies.

        Returns:
            List[str]: Topologically sorted list of component IDs.
        """
        # Simple topological sort
        visited = set()
        order = []

        def visit(comp_id: str) -> None:
            if comp_id in visited:
                return
            visited.add(comp_id)

            component = self.components[comp_id]
            for input_id in component.inputs:
                if input_id in self.components:
                    visit(input_id)

            order.append(comp_id)

        for comp_id in self.components:
            visit(comp_id)

        return order

    def to_json(self, filepath: str) -> None:
        """
        Save plant configuration to JSON file.

        Args:
            filepath (str): Path to JSON file.
        """
        config = {
            "plant_name": self.plant_name,
            "simulation_time": self.simulation_time,
            "components": [comp.to_dict() for comp in self.components.values()],
            "connections": [conn.to_dict() for conn in self.connections],
        }

        with open(filepath, "w") as f:
            json.dump(config, f, indent=2)

        print(f"Plant configuration saved to {filepath}")

    @classmethod
    def from_json(cls, filepath: str, feedstock: Optional[Feedstock] = None) -> "BiogasPlant":
        """
        Load plant configuration from JSON file.

        Args:
            filepath (str): Path to JSON file.
            feedstock (Optional[Feedstock]): Feedstock object for digesters.
                Required if plant has digesters.

        Returns:
            BiogasPlant: Loaded plant model.

        Raises:
            ValueError: If feedstock is None but plant contains digesters.
        """
        with open(filepath, "r") as f:
            config = json.load(f)

        plant = cls(config.get("plant_name", "Biogas Plant"))
        plant.simulation_time = config.get("simulation_time", 0.0)

        # Create components
        for comp_config in config.get("components", []):
            comp_type = ComponentType(comp_config["component_type"])

            if comp_type == ComponentType.DIGESTER:
                if feedstock is None:
                    raise ValueError("Feedstock required for loading plant with digesters")
                component = Digester.from_dict(comp_config, feedstock)
            elif comp_type == ComponentType.CHP:
                component = CHP.from_dict(comp_config)
            elif comp_type == ComponentType.HEATING:
                component = HeatingSystem.from_dict(comp_config)
            else:
                raise ValueError(f"Unknown component type: {comp_type}")

            plant.add_component(component)

        # Create connections
        for conn_config in config.get("connections", []):
            connection = Connection.from_dict(conn_config)
            plant.add_connection(connection)

        print(f"Plant configuration loaded from {filepath}")

        return plant

    def get_summary(self) -> str:
        """
        Get human-readable summary of plant configuration.

        Returns:
            str: Summary text with components and connections.
        """
        lines = [
            f"=== {self.plant_name} ===",
            f"Simulation time: {self.simulation_time:.2f} days",
            f"\nComponents ({len(self.components)}):",
        ]

        for comp in self.components.values():
            lines.append(f"  - {comp.name} ({comp.component_type.value})")

        lines.append(f"\nConnections ({len(self.connections)}):")
        for conn in self.connections:
            from_name = self.components[conn.from_component].name
            to_name = self.components[conn.to_component].name
            lines.append(f"  - {from_name} -> {to_name} ({conn.connection_type})")

        return "\n".join(lines)

Functions

__init__(plant_name='Biogas Plant')

Initialize biogas plant.

Parameters:

Name	Type	Description	Default
plant_name	str	Name of the plant. Defaults to "Biogas Plant".	'Biogas Plant'

Source code in pyadm1/configurator/plant_builder.py

def __init__(self, plant_name: str = "Biogas Plant"):
    """
    Initialize biogas plant.

    Args:
        plant_name (str): Name of the plant. Defaults to "Biogas Plant".
    """
    self.plant_name = plant_name
    self.components: Dict[str, Component] = {}
    self.connections: List[Connection] = []
    self.simulation_time = 0.0

add_component(component)

Add a component to the plant.

Parameters:

Name	Type	Description	Default
component	Component	Component to add to the plant.	required

Raises:

Type	Description
ValueError	If component with same ID already exists.

Source code in pyadm1/configurator/plant_builder.py

def add_component(self, component: Component) -> None:
    """
    Add a component to the plant.

    Args:
        component (Component): Component to add to the plant.

    Raises:
        ValueError: If component with same ID already exists.
    """
    if component.component_id in self.components:
        raise ValueError(f"Component with ID '{component.component_id}' already exists")
    self.components[component.component_id] = component

add_connection(connection)

Add a connection between components.

Parameters:

Name	Type	Description	Default
connection	Connection	Connection to add.	required

Raises:

Type	Description
ValueError	If source or target component not found.

Source code in pyadm1/configurator/plant_builder.py

def add_connection(self, connection: Connection) -> None:
    """
    Add a connection between components.

    Args:
        connection (Connection): Connection to add.

    Raises:
        ValueError: If source or target component not found.
    """
    # Verify components exist
    if connection.from_component not in self.components:
        raise ValueError(f"Source component '{connection.from_component}' not found")
    if connection.to_component not in self.components:
        raise ValueError(f"Target component '{connection.to_component}' not found")

    # Update component connections
    from_comp = self.components[connection.from_component]
    to_comp = self.components[connection.to_component]

    from_comp.add_output(connection.to_component)
    to_comp.add_input(connection.from_component)

    self.connections.append(connection)

from_json(filepath, feedstock=None) classmethod

Load plant configuration from JSON file.

Parameters:

Name	Type	Description	Default
filepath	str	Path to JSON file.	required
feedstock	Optional[Feedstock]	Feedstock object for digesters. Required if plant has digesters.	None

Returns:

Name	Type	Description
BiogasPlant	BiogasPlant	Loaded plant model.

Raises:

Type	Description
ValueError	If feedstock is None but plant contains digesters.

Source code in pyadm1/configurator/plant_builder.py

@classmethod
def from_json(cls, filepath: str, feedstock: Optional[Feedstock] = None) -> "BiogasPlant":
    """
    Load plant configuration from JSON file.

    Args:
        filepath (str): Path to JSON file.
        feedstock (Optional[Feedstock]): Feedstock object for digesters.
            Required if plant has digesters.

    Returns:
        BiogasPlant: Loaded plant model.

    Raises:
        ValueError: If feedstock is None but plant contains digesters.
    """
    with open(filepath, "r") as f:
        config = json.load(f)

    plant = cls(config.get("plant_name", "Biogas Plant"))
    plant.simulation_time = config.get("simulation_time", 0.0)

    # Create components
    for comp_config in config.get("components", []):
        comp_type = ComponentType(comp_config["component_type"])

        if comp_type == ComponentType.DIGESTER:
            if feedstock is None:
                raise ValueError("Feedstock required for loading plant with digesters")
            component = Digester.from_dict(comp_config, feedstock)
        elif comp_type == ComponentType.CHP:
            component = CHP.from_dict(comp_config)
        elif comp_type == ComponentType.HEATING:
            component = HeatingSystem.from_dict(comp_config)
        else:
            raise ValueError(f"Unknown component type: {comp_type}")

        plant.add_component(component)

    # Create connections
    for conn_config in config.get("connections", []):
        connection = Connection.from_dict(conn_config)
        plant.add_connection(connection)

    print(f"Plant configuration loaded from {filepath}")

    return plant

get_summary()

Get human-readable summary of plant configuration.

Returns:

Name	Type	Description
str	str	Summary text with components and connections.

Source code in pyadm1/configurator/plant_builder.py

def get_summary(self) -> str:
    """
    Get human-readable summary of plant configuration.

    Returns:
        str: Summary text with components and connections.
    """
    lines = [
        f"=== {self.plant_name} ===",
        f"Simulation time: {self.simulation_time:.2f} days",
        f"\nComponents ({len(self.components)}):",
    ]

    for comp in self.components.values():
        lines.append(f"  - {comp.name} ({comp.component_type.value})")

    lines.append(f"\nConnections ({len(self.connections)}):")
    for conn in self.connections:
        from_name = self.components[conn.from_component].name
        to_name = self.components[conn.to_component].name
        lines.append(f"  - {from_name} -> {to_name} ({conn.connection_type})")

    return "\n".join(lines)

initialize()

Initialize all components.

Note: Most components auto-initialize in their constructor. This method is kept for compatibility and to ensure any components that need explicit initialization are handled.

Source code in pyadm1/configurator/plant_builder.py

def initialize(self) -> None:
    """
    Initialize all components.

    Note: Most components auto-initialize in their constructor.
    This method is kept for compatibility and to ensure any
    components that need explicit initialization are handled.
    """
    for component in self.components.values():
        if not component._initialized:
            component.initialize()

simulate(duration, dt=1.0 / 24.0, save_interval=None)

Run simulation for specified duration.

Parameters:

Name	Type	Description	Default
duration	float	Simulation duration in days.	required
dt	float	Time step in days. Defaults to 1 hour (1/24 day).	1.0 / 24.0
save_interval	Optional[float]	Interval for saving results in days. If None, saves every step.	None

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: Simulation results at each saved time point. Each entry contains 'time' and 'components' with component results.

Example

results = plant.simulate(duration=30, dt=1/24, save_interval=1.0) print(f"Simulated {len(results)} time points")

Source code in pyadm1/configurator/plant_builder.py

def simulate(
    self,
    duration: float,
    dt: float = 1.0 / 24.0,
    save_interval: Optional[float] = None,
) -> List[Dict[str, Any]]:
    """
    Run simulation for specified duration.

    Args:
        duration (float): Simulation duration in days.
        dt (float): Time step in days. Defaults to 1 hour (1/24 day).
        save_interval (Optional[float]): Interval for saving results in days.
            If None, saves every step.

    Returns:
        List[Dict[str, Any]]: Simulation results at each saved time point.
            Each entry contains 'time' and 'components' with component results.

    Example:
        >>> results = plant.simulate(duration=30, dt=1/24, save_interval=1.0)
        >>> print(f"Simulated {len(results)} time points")
    """
    if save_interval is None:
        save_interval = dt

    results = []
    next_save_time = self.simulation_time

    n_steps = int(duration / dt)

    for i in range(n_steps):
        step_result = self.step(dt)

        # Save results at specified interval
        if self.simulation_time >= next_save_time:
            results.append(
                {
                    "time": self.simulation_time,
                    "components": step_result,
                }
            )
            next_save_time += save_interval

        if (i + 1) % 100 == 0:
            print(f"Simulated {i + 1}/{n_steps} steps")

    return results

step(dt)

Perform one simulation time step for all components.

This uses a three-pass execution model: 1. Execute digesters to produce gas → storages 2. Execute CHPs to determine gas demand → storages 3. Execute storages to supply gas → CHPs (re-execute with actual supply)

Parameters:

Name	Type	Description	Default
dt	float	Time step in days.	required

Returns:

Type	Description
Dict[str, Dict[str, Any]]	Dict[str, Dict[str, Any]]: Results from all components.

Source code in pyadm1/configurator/plant_builder.py

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

    This uses a three-pass execution model:
    1. Execute digesters to produce gas → storages
    2. Execute CHPs to determine gas demand → storages
    3. Execute storages to supply gas → CHPs (re-execute with actual supply)

    Args:
        dt (float): Time step in days.

    Returns:
        Dict[str, Dict[str, Any]]: Results from all components.
    """
    results = {}

    # Build dependency graph
    execution_order = self._get_execution_order()

    # ========================================================================
    # PASS 1: Execute all non-storage components
    # ========================================================================
    for component_id in execution_order:
        component = self.components[component_id]

        # Skip storages in first pass
        if component.component_type.value == "storage":
            continue

        # Gather inputs from connected components
        inputs = {}
        for input_id in component.inputs:
            if input_id in self.components:
                input_comp = self.components[input_id]
                inputs.update(input_comp.outputs_data)

        # Execute component
        output = component.step(self.simulation_time, dt, inputs)
        results[component_id] = output

    # ========================================================================
    # PASS 2: Execute gas storages with gas production from digesters
    # ========================================================================
    for component_id in execution_order:
        component = self.components[component_id]

        if component.component_type.value != "storage":
            continue

        # Get gas input from connected digesters
        gas_input = 0.0
        for conn in self.connections:
            if conn.to_component == component_id and conn.connection_type == "gas":
                source_comp = self.components.get(conn.from_component)
                if source_comp and source_comp.component_type.value == "digester":
                    gas_input += source_comp.outputs_data.get("Q_gas", 0.0)

        # Execute storage with production only (no demand yet)
        storage_inputs = {
            "Q_gas_in_m3_per_day": gas_input,
            "Q_gas_out_m3_per_day": 0.0,  # No demand yet
            "vent_to_flare": True,
        }

        output = component.step(self.simulation_time, dt, storage_inputs)
        results[component_id] = output

    # ========================================================================
    # PASS 3: Handle gas demand from CHPs
    # ========================================================================
    for component_id, component in self.components.items():
        if component.component_type.value != "chp":
            continue

        # Calculate CHP gas demand based on current operating point
        P_el_nom = component.P_el_nom
        eta_el = component.eta_el
        load_setpoint = 1.0  # Full load by default

        # Calculate required gas (m³/d biogas at 60% CH4)
        E_ch4 = 10.0  # kWh/m³ CH4
        CH4_content = 0.60
        P_required = load_setpoint * P_el_nom  # kW
        Q_ch4_required = (P_required / eta_el) * 24.0 / E_ch4  # m³/d CH4
        Q_gas_required = Q_ch4_required / CH4_content  # m³/d biogas

        # Find all gas storages connected to this CHP
        connected_storages = []
        for conn in self.connections:
            if conn.to_component == component_id and conn.connection_type == "gas":
                storage_id = conn.from_component
                if storage_id in self.components:
                    storage_comp = self.components[storage_id]
                    if storage_comp.component_type.value == "storage":
                        connected_storages.append(storage_id)

        if not connected_storages:
            continue

        # Distribute demand equally among storages
        demand_per_storage = Q_gas_required / len(connected_storages)

        total_supplied = 0.0

        # Re-execute storages with gas demand
        for storage_id in connected_storages:
            storage = self.components[storage_id]

            # Get gas input from pass 2
            gas_input = 0.0
            for conn in self.connections:
                if conn.to_component == storage_id and conn.connection_type == "gas":
                    source_comp = self.components.get(conn.from_component)
                    if source_comp and source_comp.component_type.value == "digester":
                        gas_input += source_comp.outputs_data.get("Q_gas", 0.0)

            # Re-execute with demand
            storage_inputs = {
                "Q_gas_in_m3_per_day": gas_input,
                "Q_gas_out_m3_per_day": demand_per_storage,
                "vent_to_flare": True,
            }

            storage_output = storage.step(self.simulation_time, dt, storage_inputs)
            results[storage_id] = storage_output

            # Accumulate supplied gas
            supplied = storage_output.get("Q_gas_supplied_m3_per_day", 0.0)
            total_supplied += supplied

        # Re-execute CHP with actual gas supply
        chp_inputs = {
            "Q_gas_supplied_m3_per_day": total_supplied,
            "load_setpoint": load_setpoint,
        }

        chp_output = component.step(self.simulation_time, dt, chp_inputs)
        results[component_id] = chp_output

    self.simulation_time += dt

    return results

to_json(filepath)

Save plant configuration to JSON file.

Parameters:

Name	Type	Description	Default
filepath	str	Path to JSON file.	required

Source code in pyadm1/configurator/plant_builder.py

def to_json(self, filepath: str) -> None:
    """
    Save plant configuration to JSON file.

    Args:
        filepath (str): Path to JSON file.
    """
    config = {
        "plant_name": self.plant_name,
        "simulation_time": self.simulation_time,
        "components": [comp.to_dict() for comp in self.components.values()],
        "connections": [conn.to_dict() for conn in self.connections],
    }

    with open(filepath, "w") as f:
        json.dump(config, f, indent=2)

    print(f"Plant configuration saved to {filepath}")

pyadm1.configurator.plant_configurator.PlantConfigurator

High-level configurator for building biogas plants.

This class provides convenient methods for adding components with sensible defaults and automatic setup of common configurations.

Source code in pyadm1/configurator/plant_configurator.py

class PlantConfigurator:
    """
    High-level configurator for building biogas plants.

    This class provides convenient methods for adding components with
    sensible defaults and automatic setup of common configurations.
    """

    def __init__(self, plant: BiogasPlant, feedstock: Feedstock):
        """
        Initialize configurator.

        Args:
            plant: BiogasPlant instance to configure
            feedstock: Feedstock instance for digesters
        """
        self.plant = plant
        self.feedstock = feedstock

    def add_digester(
        self,
        digester_id: str,
        V_liq: float = 1977.0,
        V_gas: float = 304.0,
        T_ad: float = 308.15,
        name: Optional[str] = None,
        load_initial_state: bool = True,
        initial_state_file: Optional[str] = None,
        Q_substrates: Optional[list] = None,
    ) -> (Digester, str):
        """
        Add a digester component to the plant.

        Automatically creates and connects a gas storage for the digester.

        Args:
            digester_id: Unique identifier for this digester
            V_liq: Liquid volume in m³ (default: 1977.0)
            V_gas: Gas volume in m³ (default: 304.0)
            T_ad: Operating temperature in K (default: 308.15 = 35°C)
            name: Human-readable name (optional)
            load_initial_state: Load default initial state (default: True)
            initial_state_file: Path to custom initial state CSV (optional)
            Q_substrates: Initial substrate feed rates [m³/d] (optional)

        Returns:
            Created Digester component

        Example:
            >>> config = PlantConfigurator(plant, feedstock)
            >>> digester = config.add_digester(
            ...     "main_digester",
            ...     V_liq=2000,
            ...     Q_substrates=[15, 10, 0, 0, 0, 0, 0, 0, 0, 0]
            ... )
        """
        # Create digester
        digester = Digester(
            component_id=digester_id,
            feedstock=self.feedstock,
            V_liq=V_liq,
            V_gas=V_gas,
            T_ad=T_ad,
            name=name or digester_id,
        )

        # Initialize with state
        if load_initial_state:
            if initial_state_file:
                # Load from custom file
                adm1_state = get_state_zero_from_initial_state(initial_state_file)
                state_info = f"  - Initial state: Loaded from {initial_state_file}\n"
            else:
                # Load from default file
                try:
                    data_path = Path(__file__).parent.parent.parent / "data" / "initial_states"
                    default_file = data_path / "digester_initial8.csv"

                    if default_file.exists():
                        adm1_state = get_state_zero_from_initial_state(str(default_file))
                        state_info = f"  - Initial state: Loaded from {default_file.name}\n"
                    else:
                        adm1_state = None
                        state_info = "  - Initial state: Default initialization\n"
                except Exception as e:
                    adm1_state = None
                    state_info = f"  - Initial state: Default (error loading file: {str(e)})\n"

            # Set substrate feeds
            if Q_substrates is None:
                Q_substrates = [0] * 10

            digester.initialize({"adm1_state": adm1_state, "Q_substrates": Q_substrates})
        else:
            digester.initialize()
            state_info = "  - Initial state: Not initialized\n"

        # Add to plant
        self.plant.add_component(digester)

        # ----------------------------------------------------------
        # AUTOMATIC GAS STORAGE FOR DIGESTER
        # ----------------------------------------------------------

        storage_id = f"{digester_id}_storage"
        storage = GasStorage(
            component_id=storage_id,
            storage_type="membrane",
            capacity_m3=max(50.0, V_gas),
            name=f"{name or digester_id} Gas Storage",
        )
        self.plant.add_component(storage)

        # Connect: digester → storage
        self.connect(digester_id, storage_id, "gas")

        return digester, state_info

    def add_chp(
        self,
        chp_id: str,
        P_el_nom: float = 500.0,
        eta_el: float = 0.40,
        eta_th: float = 0.45,
        name: Optional[str] = None,
    ) -> CHP:
        """
        Add a CHP unit to the plant.

        Args:
            chp_id: Unique identifier for this CHP unit
            P_el_nom: Nominal electrical power in kW (default: 500.0)
            eta_el: Electrical efficiency 0-1 (default: 0.40)
            eta_th: Thermal efficiency 0-1 (default: 0.45)
            name: Human-readable name (optional)

        Returns:
            Created CHP component

        Example:
            >>> chp = config.add_chp("chp_main", P_el_nom=500)
        """
        chp = CHP(
            component_id=chp_id,
            P_el_nom=P_el_nom,
            eta_el=eta_el,
            eta_th=eta_th,
            name=name or chp_id,
        )

        self.plant.add_component(chp)

        # ----------------------------------------------------------
        # AUTOMATIC CHP FLARE
        # ----------------------------------------------------------
        flare_id = f"{chp_id}_flare"
        flare = Flare(component_id=flare_id, name=f"{chp_id}_flare")
        self.plant.add_component(flare)

        # Connect: CHP → flare
        self.connect(chp_id, flare_id, "gas")

        return chp

    def add_heating(
        self,
        heating_id: str,
        target_temperature: float = 308.15,
        heat_loss_coefficient: float = 0.5,
        name: Optional[str] = None,
    ) -> HeatingSystem:
        """
        Add a heating system to the plant.

        Args:
            heating_id: Unique identifier for heating system
            target_temperature: Target temperature in K (default: 308.15 = 35°C)
            heat_loss_coefficient: Heat loss in kW/K (default: 0.5)
            name: Human-readable name (optional)

        Returns:
            Created HeatingSystem component

        Example:
            >>> heating = config.add_heating("heating_main")
        """
        heating = HeatingSystem(
            component_id=heating_id,
            target_temperature=target_temperature,
            heat_loss_coefficient=heat_loss_coefficient,
            name=name or heating_id,
        )

        self.plant.add_component(heating)

        return heating

    def connect(self, from_component: str, to_component: str, connection_type: str = "default") -> Connection:
        """
        Connect two components.

        Args:
            from_component: Source component ID
            to_component: Target component ID
            connection_type: Type of connection ('liquid', 'gas', 'heat', etc.)

        Returns:
            Created Connection

        Example:
            >>> config.connect("digester_1", "chp_1", "gas")
        """
        connection = Connection(from_component, to_component, connection_type)
        self.plant.add_connection(connection)
        return connection

    def auto_connect_digester_to_chp(self, digester_id: str, chp_id: str) -> None:
        """
        Automatically connect digester to CHP through gas storage.

        Creates the connection chain: digester -> gas_storage -> chp

        Args:
            digester_id: Digester component ID
            chp_id: CHP component ID

        Raises:
            ValueError: If gas storage for digester is not found
        """
        # Gas storage is created with pattern: {digester_id}_storage
        storage_id = f"{digester_id}_storage"

        # Verify the storage exists
        if storage_id not in self.plant.components:
            raise ValueError(
                f"Gas storage '{storage_id}' not found for digester '{digester_id}'. "
                f"Ensure digester was added via PlantConfigurator.add_digester()"
            )

        # Connect: digester -> storage (already done in add_digester)
        # Connect: storage -> chp
        self.connect(storage_id, chp_id, "gas")

    def auto_connect_chp_to_heating(self, chp_id: str, heating_id: str) -> None:
        """
        Automatically connect CHP to heating with heat flow.

        Args:
            chp_id: CHP component ID
            heating_id: Heating component ID
        """
        self.connect(chp_id, heating_id, "heat")

    def create_single_stage_plant(
        self,
        digester_config: Optional[Dict[str, Any]] = None,
        chp_config: Optional[Dict[str, Any]] = None,
        heating_config: Optional[Dict[str, Any]] = None,
        auto_connect: bool = True,
    ) -> Dict[str, Any]:
        """
        Create a complete single-stage plant configuration.

        Args:
            digester_config: Configuration for digester (optional)
            chp_config: Configuration for CHP (optional)
            heating_config: Configuration for heating (optional)
            auto_connect: Automatically connect components (default: True)

        Returns:
            Dictionary with created component IDs

        Example:
            >>> components = config.create_single_stage_plant(
            ...     digester_config={'V_liq': 2000},
            ...     chp_config={'P_el_nom': 500}
            ... )
        """
        # Default configs
        digester_config = digester_config or {}
        chp_config = chp_config or {}
        heating_config = heating_config or {}

        # Set default IDs if not provided
        digester_config.setdefault("digester_id", "main_digester")
        chp_config.setdefault("chp_id", "chp_main")
        heating_config.setdefault("heating_id", "heating_main")

        # Create components
        digester, _ = self.add_digester(**digester_config)

        components = {
            "digester": digester.component_id,
            "storage": f"{digester.component_id}_storage",
        }

        if chp_config:
            chp = self.add_chp(**chp_config)
            components["chp"] = chp.component_id
            components["flare"] = f"{chp.component_id}_flare"

            if auto_connect:
                self.auto_connect_digester_to_chp(digester.component_id, chp.component_id)

        if heating_config:
            heating = self.add_heating(**heating_config)
            components["heating"] = heating.component_id

            if auto_connect and "chp" in components:
                self.auto_connect_chp_to_heating(chp.component_id, heating.component_id)

        return components

    def create_two_stage_plant(
        self,
        hydrolysis_config: Optional[Dict[str, Any]] = None,
        digester_config: Optional[Dict[str, Any]] = None,
        chp_config: Optional[Dict[str, Any]] = None,
        heating_configs: Optional[list] = None,
        auto_connect: bool = True,
    ) -> Dict[str, Any]:
        """
        Create a complete two-stage plant configuration.

        Args:
            hydrolysis_config: Configuration for hydrolysis tank (optional)
            digester_config: Configuration for main digester (optional)
            chp_config: Configuration for CHP (optional)
            heating_configs: List of heating configurations (optional)
            auto_connect: Automatically connect components (default: True)

        Returns:
            Dictionary with created component IDs

        Example:
            >>> components = config.create_two_stage_plant(
            ...     hydrolysis_config={'V_liq': 500, 'T_ad': 318.15},
            ...     digester_config={'V_liq': 1500}
            ... )
        """
        # Default configs
        hydrolysis_config = hydrolysis_config or {}
        digester_config = digester_config or {}
        chp_config = chp_config or {}

        # Set default IDs
        hydrolysis_config.setdefault("digester_id", "hydrolysis_tank")
        hydrolysis_config.setdefault("name", "Hydrolysis Tank")
        hydrolysis_config.setdefault("T_ad", 318.15)  # 45°C for thermophilic

        digester_config.setdefault("digester_id", "main_digester")
        digester_config.setdefault("name", "Main Digester")

        chp_config.setdefault("chp_id", "chp_main")

        # Create components
        hydrolysis, _ = self.add_digester(**hydrolysis_config)
        digester, _ = self.add_digester(**digester_config)

        components = {
            "hydrolysis": hydrolysis.component_id,
            "hydrolysis_storage": f"{hydrolysis.component_id}_storage",
            "digester": digester.component_id,
            "digester_storage": f"{digester.component_id}_storage",
        }

        # Connect digesters in series
        if auto_connect:
            self.connect(hydrolysis.component_id, digester.component_id, "liquid")

        # Add CHP
        if chp_config:
            chp = self.add_chp(**chp_config)
            components["chp"] = chp.component_id
            components["flare"] = f"{chp.component_id}_flare"

            if auto_connect:
                # Connect both digesters to CHP
                self.auto_connect_digester_to_chp(hydrolysis.component_id, chp.component_id)
                self.auto_connect_digester_to_chp(digester.component_id, chp.component_id)

        # Add heating systems
        if heating_configs:
            components["heating"] = []
            for i, heating_cfg in enumerate(heating_configs):
                heating_cfg.setdefault("heating_id", f"heating_{i + 1}")
                heating = self.add_heating(**heating_cfg)
                components["heating"].append(heating.component_id)

                if auto_connect and "chp" in components:
                    self.auto_connect_chp_to_heating(chp.component_id, heating.component_id)

        return components

Functions

__init__(plant, feedstock)

Initialize configurator.

Parameters:

Name	Type	Description	Default
plant	BiogasPlant	BiogasPlant instance to configure	required
feedstock	Feedstock	Feedstock instance for digesters	required

Source code in pyadm1/configurator/plant_configurator.py

def __init__(self, plant: BiogasPlant, feedstock: Feedstock):
    """
    Initialize configurator.

    Args:
        plant: BiogasPlant instance to configure
        feedstock: Feedstock instance for digesters
    """
    self.plant = plant
    self.feedstock = feedstock

add_chp(chp_id, P_el_nom=500.0, eta_el=0.4, eta_th=0.45, name=None)

Add a CHP unit to the plant.

Parameters:

Name	Type	Description	Default
chp_id	str	Unique identifier for this CHP unit	required
P_el_nom	float	Nominal electrical power in kW (default: 500.0)	500.0
eta_el	float	Electrical efficiency 0-1 (default: 0.40)	0.4
eta_th	float	Thermal efficiency 0-1 (default: 0.45)	0.45
name	Optional[str]	Human-readable name (optional)	None

Returns:

Type	Description
CHP	Created CHP component

Example

chp = config.add_chp("chp_main", P_el_nom=500)

Source code in pyadm1/configurator/plant_configurator.py

def add_chp(
    self,
    chp_id: str,
    P_el_nom: float = 500.0,
    eta_el: float = 0.40,
    eta_th: float = 0.45,
    name: Optional[str] = None,
) -> CHP:
    """
    Add a CHP unit to the plant.

    Args:
        chp_id: Unique identifier for this CHP unit
        P_el_nom: Nominal electrical power in kW (default: 500.0)
        eta_el: Electrical efficiency 0-1 (default: 0.40)
        eta_th: Thermal efficiency 0-1 (default: 0.45)
        name: Human-readable name (optional)

    Returns:
        Created CHP component

    Example:
        >>> chp = config.add_chp("chp_main", P_el_nom=500)
    """
    chp = CHP(
        component_id=chp_id,
        P_el_nom=P_el_nom,
        eta_el=eta_el,
        eta_th=eta_th,
        name=name or chp_id,
    )

    self.plant.add_component(chp)

    # ----------------------------------------------------------
    # AUTOMATIC CHP FLARE
    # ----------------------------------------------------------
    flare_id = f"{chp_id}_flare"
    flare = Flare(component_id=flare_id, name=f"{chp_id}_flare")
    self.plant.add_component(flare)

    # Connect: CHP → flare
    self.connect(chp_id, flare_id, "gas")

    return chp

add_digester(digester_id, V_liq=1977.0, V_gas=304.0, T_ad=308.15, name=None, load_initial_state=True, initial_state_file=None, Q_substrates=None)

Add a digester component to the plant.

Automatically creates and connects a gas storage for the digester.

Parameters:

Name	Type	Description	Default
digester_id	str	Unique identifier for this digester	required
V_liq	float	Liquid volume in m³ (default: 1977.0)	1977.0
V_gas	float	Gas volume in m³ (default: 304.0)	304.0
T_ad	float	Operating temperature in K (default: 308.15 = 35°C)	308.15
name	Optional[str]	Human-readable name (optional)	None
load_initial_state	bool	Load default initial state (default: True)	True
initial_state_file	Optional[str]	Path to custom initial state CSV (optional)	None
Q_substrates	Optional[list]	Initial substrate feed rates [m³/d] (optional)	None

Returns:

Type	Description
(Digester, str)	Created Digester component

Example

config = PlantConfigurator(plant, feedstock) digester = config.add_digester( ... "main_digester", ... V_liq=2000, ... Q_substrates=[15, 10, 0, 0, 0, 0, 0, 0, 0, 0] ... )

Source code in pyadm1/configurator/plant_configurator.py

def add_digester(
    self,
    digester_id: str,
    V_liq: float = 1977.0,
    V_gas: float = 304.0,
    T_ad: float = 308.15,
    name: Optional[str] = None,
    load_initial_state: bool = True,
    initial_state_file: Optional[str] = None,
    Q_substrates: Optional[list] = None,
) -> (Digester, str):
    """
    Add a digester component to the plant.

    Automatically creates and connects a gas storage for the digester.

    Args:
        digester_id: Unique identifier for this digester
        V_liq: Liquid volume in m³ (default: 1977.0)
        V_gas: Gas volume in m³ (default: 304.0)
        T_ad: Operating temperature in K (default: 308.15 = 35°C)
        name: Human-readable name (optional)
        load_initial_state: Load default initial state (default: True)
        initial_state_file: Path to custom initial state CSV (optional)
        Q_substrates: Initial substrate feed rates [m³/d] (optional)

    Returns:
        Created Digester component

    Example:
        >>> config = PlantConfigurator(plant, feedstock)
        >>> digester = config.add_digester(
        ...     "main_digester",
        ...     V_liq=2000,
        ...     Q_substrates=[15, 10, 0, 0, 0, 0, 0, 0, 0, 0]
        ... )
    """
    # Create digester
    digester = Digester(
        component_id=digester_id,
        feedstock=self.feedstock,
        V_liq=V_liq,
        V_gas=V_gas,
        T_ad=T_ad,
        name=name or digester_id,
    )

    # Initialize with state
    if load_initial_state:
        if initial_state_file:
            # Load from custom file
            adm1_state = get_state_zero_from_initial_state(initial_state_file)
            state_info = f"  - Initial state: Loaded from {initial_state_file}\n"
        else:
            # Load from default file
            try:
                data_path = Path(__file__).parent.parent.parent / "data" / "initial_states"
                default_file = data_path / "digester_initial8.csv"

                if default_file.exists():
                    adm1_state = get_state_zero_from_initial_state(str(default_file))
                    state_info = f"  - Initial state: Loaded from {default_file.name}\n"
                else:
                    adm1_state = None
                    state_info = "  - Initial state: Default initialization\n"
            except Exception as e:
                adm1_state = None
                state_info = f"  - Initial state: Default (error loading file: {str(e)})\n"

        # Set substrate feeds
        if Q_substrates is None:
            Q_substrates = [0] * 10

        digester.initialize({"adm1_state": adm1_state, "Q_substrates": Q_substrates})
    else:
        digester.initialize()
        state_info = "  - Initial state: Not initialized\n"

    # Add to plant
    self.plant.add_component(digester)

    # ----------------------------------------------------------
    # AUTOMATIC GAS STORAGE FOR DIGESTER
    # ----------------------------------------------------------

    storage_id = f"{digester_id}_storage"
    storage = GasStorage(
        component_id=storage_id,
        storage_type="membrane",
        capacity_m3=max(50.0, V_gas),
        name=f"{name or digester_id} Gas Storage",
    )
    self.plant.add_component(storage)

    # Connect: digester → storage
    self.connect(digester_id, storage_id, "gas")

    return digester, state_info

add_heating(heating_id, target_temperature=308.15, heat_loss_coefficient=0.5, name=None)

Add a heating system to the plant.

Parameters:

Name	Type	Description	Default
heating_id	str	Unique identifier for heating system	required
target_temperature	float	Target temperature in K (default: 308.15 = 35°C)	308.15
heat_loss_coefficient	float	Heat loss in kW/K (default: 0.5)	0.5
name	Optional[str]	Human-readable name (optional)	None

Returns:

Type	Description
HeatingSystem	Created HeatingSystem component

Example

heating = config.add_heating("heating_main")

Source code in pyadm1/configurator/plant_configurator.py

def add_heating(
    self,
    heating_id: str,
    target_temperature: float = 308.15,
    heat_loss_coefficient: float = 0.5,
    name: Optional[str] = None,
) -> HeatingSystem:
    """
    Add a heating system to the plant.

    Args:
        heating_id: Unique identifier for heating system
        target_temperature: Target temperature in K (default: 308.15 = 35°C)
        heat_loss_coefficient: Heat loss in kW/K (default: 0.5)
        name: Human-readable name (optional)

    Returns:
        Created HeatingSystem component

    Example:
        >>> heating = config.add_heating("heating_main")
    """
    heating = HeatingSystem(
        component_id=heating_id,
        target_temperature=target_temperature,
        heat_loss_coefficient=heat_loss_coefficient,
        name=name or heating_id,
    )

    self.plant.add_component(heating)

    return heating

auto_connect_chp_to_heating(chp_id, heating_id)

Automatically connect CHP to heating with heat flow.

Parameters:

Name	Type	Description	Default
chp_id	str	CHP component ID	required
heating_id	str	Heating component ID	required

Source code in pyadm1/configurator/plant_configurator.py

def auto_connect_chp_to_heating(self, chp_id: str, heating_id: str) -> None:
    """
    Automatically connect CHP to heating with heat flow.

    Args:
        chp_id: CHP component ID
        heating_id: Heating component ID
    """
    self.connect(chp_id, heating_id, "heat")

auto_connect_digester_to_chp(digester_id, chp_id)

Automatically connect digester to CHP through gas storage.

Creates the connection chain: digester -> gas_storage -> chp

Parameters:

Name	Type	Description	Default
digester_id	str	Digester component ID	required
chp_id	str	CHP component ID	required

Raises:

Type	Description
ValueError	If gas storage for digester is not found

Source code in pyadm1/configurator/plant_configurator.py

def auto_connect_digester_to_chp(self, digester_id: str, chp_id: str) -> None:
    """
    Automatically connect digester to CHP through gas storage.

    Creates the connection chain: digester -> gas_storage -> chp

    Args:
        digester_id: Digester component ID
        chp_id: CHP component ID

    Raises:
        ValueError: If gas storage for digester is not found
    """
    # Gas storage is created with pattern: {digester_id}_storage
    storage_id = f"{digester_id}_storage"

    # Verify the storage exists
    if storage_id not in self.plant.components:
        raise ValueError(
            f"Gas storage '{storage_id}' not found for digester '{digester_id}'. "
            f"Ensure digester was added via PlantConfigurator.add_digester()"
        )

    # Connect: digester -> storage (already done in add_digester)
    # Connect: storage -> chp
    self.connect(storage_id, chp_id, "gas")

connect(from_component, to_component, connection_type='default')

Connect two components.

Parameters:

Name	Type	Description	Default
from_component	str	Source component ID	required
to_component	str	Target component ID	required
connection_type	str	Type of connection ('liquid', 'gas', 'heat', etc.)	'default'

Returns:

Type	Description
Connection	Created Connection

Example

config.connect("digester_1", "chp_1", "gas")

Source code in pyadm1/configurator/plant_configurator.py

def connect(self, from_component: str, to_component: str, connection_type: str = "default") -> Connection:
    """
    Connect two components.

    Args:
        from_component: Source component ID
        to_component: Target component ID
        connection_type: Type of connection ('liquid', 'gas', 'heat', etc.)

    Returns:
        Created Connection

    Example:
        >>> config.connect("digester_1", "chp_1", "gas")
    """
    connection = Connection(from_component, to_component, connection_type)
    self.plant.add_connection(connection)
    return connection

create_single_stage_plant(digester_config=None, chp_config=None, heating_config=None, auto_connect=True)

Create a complete single-stage plant configuration.

Parameters:

Name	Type	Description	Default
digester_config	Optional[Dict[str, Any]]	Configuration for digester (optional)	None
chp_config	Optional[Dict[str, Any]]	Configuration for CHP (optional)	None
heating_config	Optional[Dict[str, Any]]	Configuration for heating (optional)	None
auto_connect	bool	Automatically connect components (default: True)	True

Returns:

Type	Description
Dict[str, Any]	Dictionary with created component IDs

Example

components = config.create_single_stage_plant( ... digester_config={'V_liq': 2000}, ... chp_config={'P_el_nom': 500} ... )

Source code in pyadm1/configurator/plant_configurator.py

def create_single_stage_plant(
    self,
    digester_config: Optional[Dict[str, Any]] = None,
    chp_config: Optional[Dict[str, Any]] = None,
    heating_config: Optional[Dict[str, Any]] = None,
    auto_connect: bool = True,
) -> Dict[str, Any]:
    """
    Create a complete single-stage plant configuration.

    Args:
        digester_config: Configuration for digester (optional)
        chp_config: Configuration for CHP (optional)
        heating_config: Configuration for heating (optional)
        auto_connect: Automatically connect components (default: True)

    Returns:
        Dictionary with created component IDs

    Example:
        >>> components = config.create_single_stage_plant(
        ...     digester_config={'V_liq': 2000},
        ...     chp_config={'P_el_nom': 500}
        ... )
    """
    # Default configs
    digester_config = digester_config or {}
    chp_config = chp_config or {}
    heating_config = heating_config or {}

    # Set default IDs if not provided
    digester_config.setdefault("digester_id", "main_digester")
    chp_config.setdefault("chp_id", "chp_main")
    heating_config.setdefault("heating_id", "heating_main")

    # Create components
    digester, _ = self.add_digester(**digester_config)

    components = {
        "digester": digester.component_id,
        "storage": f"{digester.component_id}_storage",
    }

    if chp_config:
        chp = self.add_chp(**chp_config)
        components["chp"] = chp.component_id
        components["flare"] = f"{chp.component_id}_flare"

        if auto_connect:
            self.auto_connect_digester_to_chp(digester.component_id, chp.component_id)

    if heating_config:
        heating = self.add_heating(**heating_config)
        components["heating"] = heating.component_id

        if auto_connect and "chp" in components:
            self.auto_connect_chp_to_heating(chp.component_id, heating.component_id)

    return components

create_two_stage_plant(hydrolysis_config=None, digester_config=None, chp_config=None, heating_configs=None, auto_connect=True)

Create a complete two-stage plant configuration.

Parameters:

Name	Type	Description	Default
hydrolysis_config	Optional[Dict[str, Any]]	Configuration for hydrolysis tank (optional)	None
digester_config	Optional[Dict[str, Any]]	Configuration for main digester (optional)	None
chp_config	Optional[Dict[str, Any]]	Configuration for CHP (optional)	None
heating_configs	Optional[list]	List of heating configurations (optional)	None
auto_connect	bool	Automatically connect components (default: True)	True

Returns:

Type	Description
Dict[str, Any]	Dictionary with created component IDs

Example

components = config.create_two_stage_plant( ... hydrolysis_config={'V_liq': 500, 'T_ad': 318.15}, ... digester_config={'V_liq': 1500} ... )

Source code in pyadm1/configurator/plant_configurator.py

def create_two_stage_plant(
    self,
    hydrolysis_config: Optional[Dict[str, Any]] = None,
    digester_config: Optional[Dict[str, Any]] = None,
    chp_config: Optional[Dict[str, Any]] = None,
    heating_configs: Optional[list] = None,
    auto_connect: bool = True,
) -> Dict[str, Any]:
    """
    Create a complete two-stage plant configuration.

    Args:
        hydrolysis_config: Configuration for hydrolysis tank (optional)
        digester_config: Configuration for main digester (optional)
        chp_config: Configuration for CHP (optional)
        heating_configs: List of heating configurations (optional)
        auto_connect: Automatically connect components (default: True)

    Returns:
        Dictionary with created component IDs

    Example:
        >>> components = config.create_two_stage_plant(
        ...     hydrolysis_config={'V_liq': 500, 'T_ad': 318.15},
        ...     digester_config={'V_liq': 1500}
        ... )
    """
    # Default configs
    hydrolysis_config = hydrolysis_config or {}
    digester_config = digester_config or {}
    chp_config = chp_config or {}

    # Set default IDs
    hydrolysis_config.setdefault("digester_id", "hydrolysis_tank")
    hydrolysis_config.setdefault("name", "Hydrolysis Tank")
    hydrolysis_config.setdefault("T_ad", 318.15)  # 45°C for thermophilic

    digester_config.setdefault("digester_id", "main_digester")
    digester_config.setdefault("name", "Main Digester")

    chp_config.setdefault("chp_id", "chp_main")

    # Create components
    hydrolysis, _ = self.add_digester(**hydrolysis_config)
    digester, _ = self.add_digester(**digester_config)

    components = {
        "hydrolysis": hydrolysis.component_id,
        "hydrolysis_storage": f"{hydrolysis.component_id}_storage",
        "digester": digester.component_id,
        "digester_storage": f"{digester.component_id}_storage",
    }

    # Connect digesters in series
    if auto_connect:
        self.connect(hydrolysis.component_id, digester.component_id, "liquid")

    # Add CHP
    if chp_config:
        chp = self.add_chp(**chp_config)
        components["chp"] = chp.component_id
        components["flare"] = f"{chp.component_id}_flare"

        if auto_connect:
            # Connect both digesters to CHP
            self.auto_connect_digester_to_chp(hydrolysis.component_id, chp.component_id)
            self.auto_connect_digester_to_chp(digester.component_id, chp.component_id)

    # Add heating systems
    if heating_configs:
        components["heating"] = []
        for i, heating_cfg in enumerate(heating_configs):
            heating_cfg.setdefault("heating_id", f"heating_{i + 1}")
            heating = self.add_heating(**heating_cfg)
            components["heating"].append(heating.component_id)

            if auto_connect and "chp" in components:
                self.auto_connect_chp_to_heating(chp.component_id, heating.component_id)

    return components

pyadm1.configurator.connection_manager.ConnectionManager

Manages connections between components in a biogas plant.

The ConnectionManager handles connection validation, dependency resolution, and provides utilities for analyzing component relationships.

Attributes:

Name	Type	Description
connections	List[Connection]	List of all connections.

Example

manager = ConnectionManager() manager.add_connection(Connection("dig1", "chp1", "gas")) deps = manager.get_dependencies("chp1") print(deps) # ['dig1']

Source code in pyadm1/configurator/connection_manager.py

class ConnectionManager:
    """
    Manages connections between components in a biogas plant.

    The ConnectionManager handles connection validation, dependency resolution,
    and provides utilities for analyzing component relationships.

    Attributes:
        connections (List[Connection]): List of all connections.

    Example:
        >>> manager = ConnectionManager()
        >>> manager.add_connection(Connection("dig1", "chp1", "gas"))
        >>> deps = manager.get_dependencies("chp1")
        >>> print(deps)  # ['dig1']
    """

    def __init__(self):
        """Initialize an empty connection manager."""
        self.connections: List[Connection] = []

    def add_connection(self, connection: Connection) -> None:
        """
        Add a connection to the manager.

        Args:
            connection (Connection): Connection to add.

        Raises:
            ValueError: If connection already exists.
        """
        # Check for duplicate
        for conn in self.connections:
            if (
                conn.from_component == connection.from_component
                and conn.to_component == connection.to_component
                and conn.connection_type == connection.connection_type
            ):
                raise ValueError(
                    f"Connection already exists: {connection.from_component} -> "
                    f"{connection.to_component} ({connection.connection_type})"
                )

        self.connections.append(connection)

    def remove_connection(
        self,
        from_component: str,
        to_component: str,
        connection_type: Optional[str] = None,
    ) -> bool:
        """
        Remove a connection.

        Args:
            from_component (str): Source component ID.
            to_component (str): Target component ID.
            connection_type (Optional[str]): Connection type. If None, removes
                all connections between the components.

        Returns:
            bool: True if at least one connection was removed, False otherwise.
        """
        removed = False
        self.connections = [
            conn
            for conn in self.connections
            if not (
                conn.from_component == from_component
                and conn.to_component == to_component
                and (connection_type is None or conn.connection_type == connection_type)
            )
            or not (removed := True)
        ]
        return removed

    def get_connections_from(self, component_id: str) -> List[Connection]:
        """
        Get all connections originating from a component.

        Args:
            component_id (str): Component ID.

        Returns:
            List[Connection]: List of outgoing connections.
        """
        return [conn for conn in self.connections if conn.from_component == component_id]

    def get_connections_to(self, component_id: str) -> List[Connection]:
        """
        Get all connections terminating at a component.

        Args:
            component_id (str): Component ID.

        Returns:
            List[Connection]: List of incoming connections.
        """
        return [conn for conn in self.connections if conn.to_component == component_id]

    def get_dependencies(self, component_id: str) -> List[str]:
        """
        Get all components that the given component depends on.

        Args:
            component_id (str): Component ID.

        Returns:
            List[str]: List of component IDs that this component depends on.
        """
        return [conn.from_component for conn in self.get_connections_to(component_id)]

    def get_dependents(self, component_id: str) -> List[str]:
        """
        Get all components that depend on the given component.

        Args:
            component_id (str): Component ID.

        Returns:
            List[str]: List of component IDs that depend on this component.
        """
        return [conn.to_component for conn in self.get_connections_from(component_id)]

    def get_execution_order(self, component_ids: List[str]) -> List[str]:
        """
        Determine execution order based on dependencies (topological sort).

        Args:
            component_ids (List[str]): List of all component IDs.

        Returns:
            List[str]: Component IDs in execution order.

        Raises:
            ValueError: If circular dependencies are detected.
        """
        # Build adjacency list
        in_degree = {comp_id: 0 for comp_id in component_ids}
        adjacency = {comp_id: [] for comp_id in component_ids}

        for conn in self.connections:
            if conn.from_component in component_ids and conn.to_component in component_ids:
                adjacency[conn.from_component].append(conn.to_component)
                in_degree[conn.to_component] += 1

        # Kahn's algorithm for topological sort
        queue = [comp_id for comp_id in component_ids if in_degree[comp_id] == 0]
        result = []

        while queue:
            current = queue.pop(0)
            result.append(current)

            for neighbor in adjacency[current]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    queue.append(neighbor)

        # Check for cycles
        if len(result) != len(component_ids):
            raise ValueError("Circular dependency detected in component connections")

        return result

    def has_circular_dependency(self, component_ids: List[str]) -> bool:
        """
        Check if there are circular dependencies.

        Args:
            component_ids (List[str]): List of component IDs to check.

        Returns:
            bool: True if circular dependencies exist, False otherwise.
        """
        try:
            self.get_execution_order(component_ids)
            return False
        except ValueError:
            return True

    def get_connected_components(self, component_id: str) -> Set[str]:
        """
        Get all components connected to the given component (directly or indirectly).

        Args:
            component_id (str): Starting component ID.

        Returns:
            Set[str]: Set of connected component IDs.
        """
        visited = set()
        queue = [component_id]

        while queue:
            current = queue.pop(0)
            if current in visited:
                continue

            visited.add(current)

            # Add all connected components
            for conn in self.connections:
                if conn.from_component == current and conn.to_component not in visited:
                    queue.append(conn.to_component)
                elif conn.to_component == current and conn.from_component not in visited:
                    queue.append(conn.from_component)

        visited.discard(component_id)  # Remove the starting component
        return visited

    def validate_connections(self, component_ids: List[str]) -> List[str]:
        """
        Validate all connections and return list of issues.

        Args:
            component_ids (List[str]): List of valid component IDs.

        Returns:
            List[str]: List of validation error messages. Empty if all valid.
        """
        errors = []

        # Check for invalid component references
        for conn in self.connections:
            if conn.from_component not in component_ids:
                errors.append(f"Connection references non-existent source component: " f"{conn.from_component}")
            if conn.to_component not in component_ids:
                errors.append(f"Connection references non-existent target component: " f"{conn.to_component}")

        # Check for circular dependencies
        if self.has_circular_dependency(component_ids):
            errors.append("Circular dependency detected in connections")

        return errors

    def get_all_connections(self) -> List[Connection]:
        """
        Get all connections.

        Returns:
            List[Connection]: List of all connections.
        """
        return self.connections.copy()

    def clear(self) -> None:
        """Remove all connections."""
        self.connections.clear()

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

        Returns:
            Dict[str, Any]: Dictionary with connection data.
        """
        return {"connections": [conn.to_dict() for conn in self.connections]}

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

        Args:
            config (Dict[str, Any]): Dictionary with 'connections' key.

        Returns:
            ConnectionManager: New manager with loaded connections.
        """
        manager = cls()
        for conn_config in config.get("connections", []):
            manager.add_connection(Connection.from_dict(conn_config))
        return manager

    def __len__(self) -> int:
        """Return number of connections."""
        return len(self.connections)

    def __repr__(self) -> str:
        """String representation of manager."""
        return f"ConnectionManager(connections={len(self.connections)})"

Functions

__init__()

Initialize an empty connection manager.

Source code in pyadm1/configurator/connection_manager.py

def __init__(self):
    """Initialize an empty connection manager."""
    self.connections: List[Connection] = []

__len__()

Return number of connections.

Source code in pyadm1/configurator/connection_manager.py

def __len__(self) -> int:
    """Return number of connections."""
    return len(self.connections)

__repr__()

String representation of manager.

Source code in pyadm1/configurator/connection_manager.py

def __repr__(self) -> str:
    """String representation of manager."""
    return f"ConnectionManager(connections={len(self.connections)})"

add_connection(connection)

Add a connection to the manager.

Parameters:

Name	Type	Description	Default
connection	Connection	Connection to add.	required

Raises:

Type	Description
ValueError	If connection already exists.

Source code in pyadm1/configurator/connection_manager.py

def add_connection(self, connection: Connection) -> None:
    """
    Add a connection to the manager.

    Args:
        connection (Connection): Connection to add.

    Raises:
        ValueError: If connection already exists.
    """
    # Check for duplicate
    for conn in self.connections:
        if (
            conn.from_component == connection.from_component
            and conn.to_component == connection.to_component
            and conn.connection_type == connection.connection_type
        ):
            raise ValueError(
                f"Connection already exists: {connection.from_component} -> "
                f"{connection.to_component} ({connection.connection_type})"
            )

    self.connections.append(connection)

clear()

Remove all connections.

Source code in pyadm1/configurator/connection_manager.py

def clear(self) -> None:
    """Remove all connections."""
    self.connections.clear()

from_dict(config) classmethod

Create ConnectionManager from dictionary.

Parameters:

Name	Type	Description	Default
config	Dict[str, Any]	Dictionary with 'connections' key.	required

Returns:

Name	Type	Description
ConnectionManager	ConnectionManager	New manager with loaded connections.

Source code in pyadm1/configurator/connection_manager.py

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

    Args:
        config (Dict[str, Any]): Dictionary with 'connections' key.

    Returns:
        ConnectionManager: New manager with loaded connections.
    """
    manager = cls()
    for conn_config in config.get("connections", []):
        manager.add_connection(Connection.from_dict(conn_config))
    return manager

get_all_connections()

Get all connections.

Returns:

Type	Description
List[Connection]	List[Connection]: List of all connections.

Source code in pyadm1/configurator/connection_manager.py

def get_all_connections(self) -> List[Connection]:
    """
    Get all connections.

    Returns:
        List[Connection]: List of all connections.
    """
    return self.connections.copy()

get_connected_components(component_id)

Get all components connected to the given component (directly or indirectly).

Parameters:

Name	Type	Description	Default
component_id	str	Starting component ID.	required

Returns:

Type	Description
Set[str]	Set[str]: Set of connected component IDs.

Source code in pyadm1/configurator/connection_manager.py

def get_connected_components(self, component_id: str) -> Set[str]:
    """
    Get all components connected to the given component (directly or indirectly).

    Args:
        component_id (str): Starting component ID.

    Returns:
        Set[str]: Set of connected component IDs.
    """
    visited = set()
    queue = [component_id]

    while queue:
        current = queue.pop(0)
        if current in visited:
            continue

        visited.add(current)

        # Add all connected components
        for conn in self.connections:
            if conn.from_component == current and conn.to_component not in visited:
                queue.append(conn.to_component)
            elif conn.to_component == current and conn.from_component not in visited:
                queue.append(conn.from_component)

    visited.discard(component_id)  # Remove the starting component
    return visited

get_connections_from(component_id)

Get all connections originating from a component.

Parameters:

Name	Type	Description	Default
component_id	str	Component ID.	required

Returns:

Type	Description
List[Connection]	List[Connection]: List of outgoing connections.

Source code in pyadm1/configurator/connection_manager.py

def get_connections_from(self, component_id: str) -> List[Connection]:
    """
    Get all connections originating from a component.

    Args:
        component_id (str): Component ID.

    Returns:
        List[Connection]: List of outgoing connections.
    """
    return [conn for conn in self.connections if conn.from_component == component_id]

get_connections_to(component_id)

Get all connections terminating at a component.

Parameters:

Name	Type	Description	Default
component_id	str	Component ID.	required

Returns:

Type	Description
List[Connection]	List[Connection]: List of incoming connections.

Source code in pyadm1/configurator/connection_manager.py

def get_connections_to(self, component_id: str) -> List[Connection]:
    """
    Get all connections terminating at a component.

    Args:
        component_id (str): Component ID.

    Returns:
        List[Connection]: List of incoming connections.
    """
    return [conn for conn in self.connections if conn.to_component == component_id]

get_dependencies(component_id)

Get all components that the given component depends on.

Parameters:

Name	Type	Description	Default
component_id	str	Component ID.	required

Returns:

Type	Description
List[str]	List[str]: List of component IDs that this component depends on.

Source code in pyadm1/configurator/connection_manager.py

def get_dependencies(self, component_id: str) -> List[str]:
    """
    Get all components that the given component depends on.

    Args:
        component_id (str): Component ID.

    Returns:
        List[str]: List of component IDs that this component depends on.
    """
    return [conn.from_component for conn in self.get_connections_to(component_id)]

get_dependents(component_id)

Get all components that depend on the given component.

Parameters:

Name	Type	Description	Default
component_id	str	Component ID.	required

Returns:

Type	Description
List[str]	List[str]: List of component IDs that depend on this component.

Source code in pyadm1/configurator/connection_manager.py

def get_dependents(self, component_id: str) -> List[str]:
    """
    Get all components that depend on the given component.

    Args:
        component_id (str): Component ID.

    Returns:
        List[str]: List of component IDs that depend on this component.
    """
    return [conn.to_component for conn in self.get_connections_from(component_id)]

get_execution_order(component_ids)

Determine execution order based on dependencies (topological sort).

Parameters:

Name	Type	Description	Default
component_ids	List[str]	List of all component IDs.	required

Returns:

Type	Description
List[str]	List[str]: Component IDs in execution order.

Raises:

Type	Description
ValueError	If circular dependencies are detected.

Source code in pyadm1/configurator/connection_manager.py

def get_execution_order(self, component_ids: List[str]) -> List[str]:
    """
    Determine execution order based on dependencies (topological sort).

    Args:
        component_ids (List[str]): List of all component IDs.

    Returns:
        List[str]: Component IDs in execution order.

    Raises:
        ValueError: If circular dependencies are detected.
    """
    # Build adjacency list
    in_degree = {comp_id: 0 for comp_id in component_ids}
    adjacency = {comp_id: [] for comp_id in component_ids}

    for conn in self.connections:
        if conn.from_component in component_ids and conn.to_component in component_ids:
            adjacency[conn.from_component].append(conn.to_component)
            in_degree[conn.to_component] += 1

    # Kahn's algorithm for topological sort
    queue = [comp_id for comp_id in component_ids if in_degree[comp_id] == 0]
    result = []

    while queue:
        current = queue.pop(0)
        result.append(current)

        for neighbor in adjacency[current]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    # Check for cycles
    if len(result) != len(component_ids):
        raise ValueError("Circular dependency detected in component connections")

    return result

has_circular_dependency(component_ids)

Check if there are circular dependencies.

Parameters:

Name	Type	Description	Default
component_ids	List[str]	List of component IDs to check.	required

Returns:

Name	Type	Description
bool	bool	True if circular dependencies exist, False otherwise.

Source code in pyadm1/configurator/connection_manager.py

def has_circular_dependency(self, component_ids: List[str]) -> bool:
    """
    Check if there are circular dependencies.

    Args:
        component_ids (List[str]): List of component IDs to check.

    Returns:
        bool: True if circular dependencies exist, False otherwise.
    """
    try:
        self.get_execution_order(component_ids)
        return False
    except ValueError:
        return True

remove_connection(from_component, to_component, connection_type=None)

Remove a connection.

Parameters:

Name	Type	Description	Default
from_component	str	Source component ID.	required
to_component	str	Target component ID.	required
connection_type	Optional[str]	Connection type. If None, removes all connections between the components.	None

Returns:

Name	Type	Description
bool	bool	True if at least one connection was removed, False otherwise.

Source code in pyadm1/configurator/connection_manager.py

def remove_connection(
    self,
    from_component: str,
    to_component: str,
    connection_type: Optional[str] = None,
) -> bool:
    """
    Remove a connection.

    Args:
        from_component (str): Source component ID.
        to_component (str): Target component ID.
        connection_type (Optional[str]): Connection type. If None, removes
            all connections between the components.

    Returns:
        bool: True if at least one connection was removed, False otherwise.
    """
    removed = False
    self.connections = [
        conn
        for conn in self.connections
        if not (
            conn.from_component == from_component
            and conn.to_component == to_component
            and (connection_type is None or conn.connection_type == connection_type)
        )
        or not (removed := True)
    ]
    return removed

to_dict()

Serialize all connections to dictionary.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Dictionary with connection data.

Source code in pyadm1/configurator/connection_manager.py

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

    Returns:
        Dict[str, Any]: Dictionary with connection data.
    """
    return {"connections": [conn.to_dict() for conn in self.connections]}

validate_connections(component_ids)

Validate all connections and return list of issues.

Parameters:

Name	Type	Description	Default
component_ids	List[str]	List of valid component IDs.	required

Returns:

Type	Description
List[str]	List[str]: List of validation error messages. Empty if all valid.

Source code in pyadm1/configurator/connection_manager.py

def validate_connections(self, component_ids: List[str]) -> List[str]:
    """
    Validate all connections and return list of issues.

    Args:
        component_ids (List[str]): List of valid component IDs.

    Returns:
        List[str]: List of validation error messages. Empty if all valid.
    """
    errors = []

    # Check for invalid component references
    for conn in self.connections:
        if conn.from_component not in component_ids:
            errors.append(f"Connection references non-existent source component: " f"{conn.from_component}")
        if conn.to_component not in component_ids:
            errors.append(f"Connection references non-existent target component: " f"{conn.to_component}")

    # Check for circular dependencies
    if self.has_circular_dependency(component_ids):
        errors.append("Circular dependency detected in connections")

    return errors