Configurator 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 import Feedstock, BiogasPlant from pyadm1.components.biological import Digester

feedstock = Feedstock(["maize_silage_milk_ripeness", "swine_manure"], feeding_freq=24) plant = BiogasPlant("My Plant") digester = Digester("dig1", feedstock, V_liq=1200, V_gas=216, T_ad=315.15) 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 import Feedstock, BiogasPlant
        >>> from pyadm1.components.biological import Digester
        >>>
        >>> feedstock = Feedstock(["maize_silage_milk_ripeness", "swine_manure"], feeding_freq=24)
        >>> plant = BiogasPlant("My Plant")
        >>> digester = Digester("dig1", feedstock, V_liq=1200, V_gas=216, T_ad=315.15)
        >>> 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, non-heating components
        # ========================================================================
        # Heaters are also deferred (until after Pass 3) because their CHP
        # input ``P_th_available`` is only meaningful once the CHP has run
        # in Pass 3 with the actual gas supply. Stepping a heater here would
        # see a stale/zero ``P_th`` and (a) charge the auxiliary boiler for
        # heat the CHP would otherwise have supplied and (b) double-count the
        # heater's ``energy_consumed`` if we then re-stepped it in Pass 3.
        for component_id in execution_order:
            component = self.components[component_id]

            # Skip storages in first pass
            if component.component_type.value == "storage":
                continue
            # Skip heaters in first pass (see comment above)
            if component.component_type.value == "heating":
                continue

            # Gather inputs from connected components. For "liquid"
            # cascade connections (e.g. primary digester -> post-digester),
            # the upstream component writes its effluent as Q_out /
            # state_out but the downstream digester reads its inflow as
            # Q_in / state_in -- remap on the fly so the cascade actually
            # carries flow. Other connection types (gas, heat, ...) keep
            # their keys verbatim.
            inputs = {}
            for input_id in component.inputs:
                if input_id not in self.components:
                    continue
                input_comp = self.components[input_id]
                out = input_comp.outputs_data
                conn_type = None
                for conn in self.connections:
                    if conn.from_component == input_id and conn.to_component == component_id:
                        conn_type = conn.connection_type
                        break
                if conn_type == "liquid":
                    if "Q_out" in out:
                        inputs["Q_in"] = out["Q_out"]
                    if "state_out" in out:
                        inputs["state_in"] = out["state_out"]
                else:
                    inputs.update(out)

            # 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.
            #
            # IMPORTANT: do not pass ``Q_gas_in_m3_per_day`` again here.
            # The digester's production for this step was already routed
            # into the storage in Pass 2 and is now sitting in
            # ``storage.stored_volume_m3``. The storage's ``step()`` adds
            # ``Q_in * dt`` to ``stored_volume_m3`` on every call, so
            # re-passing the production would double-count it -- inflating
            # both inventory growth and (when the storage is saturated)
            # vented gas, producing the "vented > produced" symptom in
            # whole-run mass balances.
            #
            # We also need to preserve any vent that happened in Pass 2,
            # because the storage resets ``vented_volume_m3`` to 0 at the
            # start of every ``step()`` call. After this Pass-3 call, the
            # output dict's ``vented_volume_m3`` would otherwise only
            # contain Pass 3's vent (typically 0; Pass 2 carries the
            # overflow vents). Carry the Pass-2 value through manually.
            for storage_id in connected_storages:
                storage = self.components[storage_id]

                pass2_vented = float(results.get(storage_id, {}).get("vented_volume_m3", 0.0))

                storage_inputs = {
                    "Q_gas_in_m3_per_day": 0.0,
                    "Q_gas_out_m3_per_day": demand_per_storage,
                    "vent_to_flare": True,
                }

                storage_output = storage.step(self.simulation_time, dt, storage_inputs)

                # Merge Pass-2 vent volume into the final per-step report.
                # ``cumulative_vented_m3`` already tracks all vents across
                # both passes inside the storage component itself, so we
                # only need to repair the per-step field here.
                storage_output["vented_volume_m3"] = float(storage_output.get("vented_volume_m3", 0.0)) + pass2_vented
                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

            # ============================================================
            # PASS 3b: Re-execute heaters connected to this CHP with the
            # CHP's now-current thermal output. Without this, heaters
            # always see ``P_th_available = 0`` from the CHP's
            # initialised/stale outputs and charge all heat demand to the
            # auxiliary boiler -- even when the CHP has plenty of free
            # thermal power.
            #
            # When multiple heaters share one CHP, split the available
            # thermal power equally. This is a defensible-but-simple
            # allocation that preserves energy conservation
            # (``sum(P_th_used) <= chp.P_th``). A demand-proportional split
            # would require a separate iteration.
            # ============================================================
            connected_heaters = [
                conn.to_component
                for conn in self.connections
                if conn.from_component == component_id
                and conn.connection_type == "heat"
                and conn.to_component in self.components
                and self.components[conn.to_component].component_type.value == "heating"
            ]

            if connected_heaters:
                P_th_per_heater = chp_output.get("P_th", 0.0) / len(connected_heaters)
                for h_id in connected_heaters:
                    heater = self.components[h_id]
                    heater_inputs: Dict[str, Any] = {"P_th_available": P_th_per_heater}
                    # Forward any non-heat upstream inputs the heater may
                    # need (e.g. T_digester from a connected digester).
                    for conn in self.connections:
                        if conn.to_component == h_id and conn.connection_type != "heat":
                            source = self.components.get(conn.from_component)
                            if source is not None:
                                heater_inputs.update(source.outputs_data)
                    results[h_id] = heater.step(self.simulation_time, dt, heater_inputs)

        # ============================================================
        # Catch any heaters NOT connected to a CHP. They were skipped in
        # Pass 1 to avoid the stale-CHP-output problem, so they need to
        # run here with their full upstream inputs but no CHP heat.
        # ============================================================
        stepped_heaters = {cid for cid in results if cid in self.components}
        for component_id in execution_order:
            component = self.components[component_id]
            if component.component_type.value != "heating":
                continue
            if component_id in stepped_heaters:
                continue
            heater_inputs = {"P_th_available": 0.0}
            for conn in self.connections:
                if conn.to_component == component_id:
                    source = self.components.get(conn.from_component)
                    if source is not None:
                        heater_inputs.update(source.outputs_data)
            results[component_id] = component.step(self.simulation_time, dt, heater_inputs)

        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:
            """Recursive DFS helper: visit dependencies first, then append this component."""
            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.  Digesters need the supplied feedstock; for every
        # other component type we delegate to the class's ``from_dict`` via the
        # global component registry so all auto-attached children (gas storage,
        # flares, mixers, sensors, …) round-trip without bespoke branches.
        from pyadm1.components.registry import get_registry

        registry = get_registry()
        type_to_registry_key = {
            ComponentType.DIGESTER: "Digester",
            ComponentType.CHP: "CHP",
            ComponentType.HEATING: "HeatingSystem",
            ComponentType.STORAGE: "GasStorage",
            ComponentType.FLARE: "Flare",
            ComponentType.BOILER: "Boiler",
            ComponentType.SEPARATOR: "Separator",
        }

        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 in type_to_registry_key:
                cls_obj = registry.get_registered_components().get(type_to_registry_key[comp_type])
                if cls_obj is None:
                    raise ValueError(f"No registered class for component type: {comp_type}")
                component = cls_obj.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.  Digesters need the supplied feedstock; for every
    # other component type we delegate to the class's ``from_dict`` via the
    # global component registry so all auto-attached children (gas storage,
    # flares, mixers, sensors, …) round-trip without bespoke branches.
    from pyadm1.components.registry import get_registry

    registry = get_registry()
    type_to_registry_key = {
        ComponentType.DIGESTER: "Digester",
        ComponentType.CHP: "CHP",
        ComponentType.HEATING: "HeatingSystem",
        ComponentType.STORAGE: "GasStorage",
        ComponentType.FLARE: "Flare",
        ComponentType.BOILER: "Boiler",
        ComponentType.SEPARATOR: "Separator",
    }

    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 in type_to_registry_key:
            cls_obj = registry.get_registered_components().get(type_to_registry_key[comp_type])
            if cls_obj is None:
                raise ValueError(f"No registered class for component type: {comp_type}")
            component = cls_obj.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, non-heating components
    # ========================================================================
    # Heaters are also deferred (until after Pass 3) because their CHP
    # input ``P_th_available`` is only meaningful once the CHP has run
    # in Pass 3 with the actual gas supply. Stepping a heater here would
    # see a stale/zero ``P_th`` and (a) charge the auxiliary boiler for
    # heat the CHP would otherwise have supplied and (b) double-count the
    # heater's ``energy_consumed`` if we then re-stepped it in Pass 3.
    for component_id in execution_order:
        component = self.components[component_id]

        # Skip storages in first pass
        if component.component_type.value == "storage":
            continue
        # Skip heaters in first pass (see comment above)
        if component.component_type.value == "heating":
            continue

        # Gather inputs from connected components. For "liquid"
        # cascade connections (e.g. primary digester -> post-digester),
        # the upstream component writes its effluent as Q_out /
        # state_out but the downstream digester reads its inflow as
        # Q_in / state_in -- remap on the fly so the cascade actually
        # carries flow. Other connection types (gas, heat, ...) keep
        # their keys verbatim.
        inputs = {}
        for input_id in component.inputs:
            if input_id not in self.components:
                continue
            input_comp = self.components[input_id]
            out = input_comp.outputs_data
            conn_type = None
            for conn in self.connections:
                if conn.from_component == input_id and conn.to_component == component_id:
                    conn_type = conn.connection_type
                    break
            if conn_type == "liquid":
                if "Q_out" in out:
                    inputs["Q_in"] = out["Q_out"]
                if "state_out" in out:
                    inputs["state_in"] = out["state_out"]
            else:
                inputs.update(out)

        # 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.
        #
        # IMPORTANT: do not pass ``Q_gas_in_m3_per_day`` again here.
        # The digester's production for this step was already routed
        # into the storage in Pass 2 and is now sitting in
        # ``storage.stored_volume_m3``. The storage's ``step()`` adds
        # ``Q_in * dt`` to ``stored_volume_m3`` on every call, so
        # re-passing the production would double-count it -- inflating
        # both inventory growth and (when the storage is saturated)
        # vented gas, producing the "vented > produced" symptom in
        # whole-run mass balances.
        #
        # We also need to preserve any vent that happened in Pass 2,
        # because the storage resets ``vented_volume_m3`` to 0 at the
        # start of every ``step()`` call. After this Pass-3 call, the
        # output dict's ``vented_volume_m3`` would otherwise only
        # contain Pass 3's vent (typically 0; Pass 2 carries the
        # overflow vents). Carry the Pass-2 value through manually.
        for storage_id in connected_storages:
            storage = self.components[storage_id]

            pass2_vented = float(results.get(storage_id, {}).get("vented_volume_m3", 0.0))

            storage_inputs = {
                "Q_gas_in_m3_per_day": 0.0,
                "Q_gas_out_m3_per_day": demand_per_storage,
                "vent_to_flare": True,
            }

            storage_output = storage.step(self.simulation_time, dt, storage_inputs)

            # Merge Pass-2 vent volume into the final per-step report.
            # ``cumulative_vented_m3`` already tracks all vents across
            # both passes inside the storage component itself, so we
            # only need to repair the per-step field here.
            storage_output["vented_volume_m3"] = float(storage_output.get("vented_volume_m3", 0.0)) + pass2_vented
            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

        # ============================================================
        # PASS 3b: Re-execute heaters connected to this CHP with the
        # CHP's now-current thermal output. Without this, heaters
        # always see ``P_th_available = 0`` from the CHP's
        # initialised/stale outputs and charge all heat demand to the
        # auxiliary boiler -- even when the CHP has plenty of free
        # thermal power.
        #
        # When multiple heaters share one CHP, split the available
        # thermal power equally. This is a defensible-but-simple
        # allocation that preserves energy conservation
        # (``sum(P_th_used) <= chp.P_th``). A demand-proportional split
        # would require a separate iteration.
        # ============================================================
        connected_heaters = [
            conn.to_component
            for conn in self.connections
            if conn.from_component == component_id
            and conn.connection_type == "heat"
            and conn.to_component in self.components
            and self.components[conn.to_component].component_type.value == "heating"
        ]

        if connected_heaters:
            P_th_per_heater = chp_output.get("P_th", 0.0) / len(connected_heaters)
            for h_id in connected_heaters:
                heater = self.components[h_id]
                heater_inputs: Dict[str, Any] = {"P_th_available": P_th_per_heater}
                # Forward any non-heat upstream inputs the heater may
                # need (e.g. T_digester from a connected digester).
                for conn in self.connections:
                    if conn.to_component == h_id and conn.connection_type != "heat":
                        source = self.components.get(conn.from_component)
                        if source is not None:
                            heater_inputs.update(source.outputs_data)
                results[h_id] = heater.step(self.simulation_time, dt, heater_inputs)

    # ============================================================
    # Catch any heaters NOT connected to a CHP. They were skipped in
    # Pass 1 to avoid the stale-CHP-output problem, so they need to
    # run here with their full upstream inputs but no CHP heat.
    # ============================================================
    stepped_heaters = {cid for cid in results if cid in self.components}
    for component_id in execution_order:
        component = self.components[component_id]
        if component.component_type.value != "heating":
            continue
        if component_id in stepped_heaters:
            continue
        heater_inputs = {"P_th_available": 0.0}
        for conn in self.connections:
            if conn.to_component == component_id:
                source = self.components.get(conn.from_component)
                if source is not None:
                    heater_inputs.update(source.outputs_data)
        results[component_id] = component.step(self.simulation_time, dt, heater_inputs)

    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.

Provides convenient methods for adding components with sensible defaults and automatic setup of common configurations (gas storage attached to digesters, flare attached to CHP, etc.).

Source code in pyadm1/configurator/plant_configurator.py

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

    Provides convenient methods for adding components with sensible defaults
    and automatic setup of common configurations (gas storage attached to
    digesters, flare attached to CHP, etc.).
    """

    def __init__(self, plant: BiogasPlant, feedstock: Feedstock):
        """
        Parameters
        ----------
        plant : BiogasPlant
            Plant instance to configure.
        feedstock : Feedstock
            Feedstock used by all digesters added through this configurator.
        """
        self.plant = plant
        self.feedstock = feedstock

    def add_digester(
        self,
        digester_id: str,
        V_liq: float = 1050.0,
        V_gas: float = 150.0,
        T_ad: float = 315.15,
        name: Optional[str] = None,
        Q_substrates: Optional[list] = None,
        k_L_a: Optional[float] = None,
        adm1_state: Optional[list] = None,
        dynamic_volume: bool = False,
        initial_fill_fraction: float = 1.0,
        outflow_time_constant: float = 1.0,
    ) -> "tuple[Digester, str]":
        """
        Add an ADM1da digester to the plant.

        The digester's influent DataFrame, density, and steady-state initial
        state are wired automatically from the attached :class:`Feedstock`.
        A gas storage is auto-created and connected.

        Parameters
        ----------
        digester_id : str
            Unique identifier for this digester.
        V_liq : float
            Liquid volume [m³] (default 1050).
        V_gas : float
            Gas headspace volume [m³] (default 150).
        T_ad : float
            Operating temperature [K] (default 315.15 = 42 °C).
        name : str, optional
        Q_substrates : list of float, optional
            Substrate feed rates [m³/d], one entry per substrate slot
            (up to 10 slots).
        k_L_a : float, optional
            Override of the gas–liquid mass-transfer coefficient [1/d].
        adm1_state : list of float, optional
            41-element initial state vector.  When supplied, replaces the
            auto-built steady-state vector.
        dynamic_volume : bool, default False
            Enable a dynamic sludge-volume balance
            ``dV/dt = Q_in − Q_out − q_S,loss``, with ``Q_out`` from an
            overflow weir at ``V_liq``. When False, sludge volume stays
            constant.
        initial_fill_fraction : float, default 1.0
            Starting sludge fill as a fraction of ``V_liq``. Only used when
            ``dynamic_volume=True``. Set below 1.0 to simulate a partially-
            filled startup transient.
        outflow_time_constant : float, default 1.0
            Overflow-weir time constant ``τ_out`` [d]. Only used when
            ``dynamic_volume=True``.

        Returns
        -------
        (Digester, str)
            The created digester and a one-line description of how the
            initial state was determined.
        """
        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,
            dynamic_volume=dynamic_volume,
            initial_fill_fraction=initial_fill_fraction,
            outflow_time_constant=outflow_time_constant,
        )

        if k_L_a is not None:
            digester.adm1.set_calibration_parameters({"k_L_a": float(k_L_a)})

        if Q_substrates is None:
            Q_substrates = [0.0] * 10

        init_kwargs: Dict[str, Any] = {"Q_substrates": Q_substrates}
        if adm1_state is not None:
            init_kwargs["adm1_state"] = list(adm1_state)
            state_info = "  - Initial state: User-supplied 41-element ADM1 vector\n"
        else:
            state_info = "  - Initial state: Auto-built steady-state from feedstock\n"
        digester.initialize(init_kwargs)

        self.plant.add_component(digester)

        # Automatic gas storage + connection
        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)
        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.

        Automatically creates and connects a safety flare downstream of the CHP.
        """
        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)

        flare_id = f"{chp_id}_flare"
        flare = Flare(component_id=flare_id, name=f"{chp_id}_flare")
        self.plant.add_component(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."""
        heating = HeatingSystem(
            component_id=heating_id,
            target_temperature=target_temperature,
            heat_loss_coefficient=heat_loss_coefficient,
            name=name or heating_id,
            feedstock=self.feedstock,
        )

        self.plant.add_component(heating)

        return heating

    def connect(self, from_component: str, to_component: str, connection_type: str = "default") -> Connection:
        """Connect two components."""
        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:
        """Connect digester → gas_storage → chp."""
        storage_id = f"{digester_id}_storage"

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

        self.connect(storage_id, chp_id, "gas")

    def auto_connect_chp_to_heating(self, chp_id: str, heating_id: str) -> None:
        """Connect CHP → heating with heat flow."""
        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."""
        digester_config = digester_config or {}
        chp_config = chp_config or {}
        heating_config = heating_config or {}

        digester_config.setdefault("digester_id", "main_digester")
        chp_config.setdefault("chp_id", "chp_main")
        heating_config.setdefault("heating_id", "heating_main")

        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 two-stage plant: hydrolysis pre-tank → main fermenter.

        The hydrolysis stage is just another :class:`Digester` instance with
        a higher temperature and shorter HRT — there is no separate
        ``Hydrolysis`` class.
        """
        hydrolysis_config = hydrolysis_config or {}
        digester_config = digester_config or {}
        chp_config = chp_config or {}

        hydrolysis_config.setdefault("digester_id", "hydrolysis_tank")
        hydrolysis_config.setdefault("name", "Hydrolysis Tank")
        hydrolysis_config.setdefault("T_ad", 318.15)  # 45 °C
        hydrolysis_config.setdefault("V_liq", 500.0)
        hydrolysis_config.setdefault("V_gas", 75.0)

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

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

        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",
        }

        if auto_connect:
            self.connect(hydrolysis.component_id, digester.component_id, "liquid")

        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(hydrolysis.component_id, chp.component_id)
                self.auto_connect_digester_to_chp(digester.component_id, chp.component_id)

        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)

Parameters

plant : BiogasPlant Plant instance to configure. feedstock : Feedstock Feedstock used by all digesters added through this configurator.

Source code in pyadm1/configurator/plant_configurator.py

def __init__(self, plant: BiogasPlant, feedstock: Feedstock):
    """
    Parameters
    ----------
    plant : BiogasPlant
        Plant instance to configure.
    feedstock : Feedstock
        Feedstock used by all digesters added through this configurator.
    """
    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.

Automatically creates and connects a safety flare downstream of the CHP.

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.

    Automatically creates and connects a safety flare downstream of the CHP.
    """
    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)

    flare_id = f"{chp_id}_flare"
    flare = Flare(component_id=flare_id, name=f"{chp_id}_flare")
    self.plant.add_component(flare)
    self.connect(chp_id, flare_id, "gas")

    return chp

add_digester(digester_id, V_liq=1050.0, V_gas=150.0, T_ad=315.15, name=None, Q_substrates=None, k_L_a=None, adm1_state=None, dynamic_volume=False, initial_fill_fraction=1.0, outflow_time_constant=1.0)

Add an ADM1da digester to the plant.

The digester's influent DataFrame, density, and steady-state initial state are wired automatically from the attached :class:Feedstock. A gas storage is auto-created and connected.

Parameters

digester_id : str Unique identifier for this digester. V_liq : float Liquid volume [m³] (default 1050). V_gas : float Gas headspace volume [m³] (default 150). T_ad : float Operating temperature [K] (default 315.15 = 42 °C). name : str, optional Q_substrates : list of float, optional Substrate feed rates [m³/d], one entry per substrate slot (up to 10 slots). k_L_a : float, optional Override of the gas–liquid mass-transfer coefficient [1/d]. adm1_state : list of float, optional 41-element initial state vector. When supplied, replaces the auto-built steady-state vector. dynamic_volume : bool, default False Enable a dynamic sludge-volume balance dV/dt = Q_in − Q_out − q_S,loss, with Q_out from an overflow weir at V_liq. When False, sludge volume stays constant. initial_fill_fraction : float, default 1.0 Starting sludge fill as a fraction of V_liq. Only used when dynamic_volume=True. Set below 1.0 to simulate a partially- filled startup transient. outflow_time_constant : float, default 1.0 Overflow-weir time constant τ_out [d]. Only used when dynamic_volume=True.

Returns

(Digester, str) The created digester and a one-line description of how the initial state was determined.

Source code in pyadm1/configurator/plant_configurator.py

def add_digester(
    self,
    digester_id: str,
    V_liq: float = 1050.0,
    V_gas: float = 150.0,
    T_ad: float = 315.15,
    name: Optional[str] = None,
    Q_substrates: Optional[list] = None,
    k_L_a: Optional[float] = None,
    adm1_state: Optional[list] = None,
    dynamic_volume: bool = False,
    initial_fill_fraction: float = 1.0,
    outflow_time_constant: float = 1.0,
) -> "tuple[Digester, str]":
    """
    Add an ADM1da digester to the plant.

    The digester's influent DataFrame, density, and steady-state initial
    state are wired automatically from the attached :class:`Feedstock`.
    A gas storage is auto-created and connected.

    Parameters
    ----------
    digester_id : str
        Unique identifier for this digester.
    V_liq : float
        Liquid volume [m³] (default 1050).
    V_gas : float
        Gas headspace volume [m³] (default 150).
    T_ad : float
        Operating temperature [K] (default 315.15 = 42 °C).
    name : str, optional
    Q_substrates : list of float, optional
        Substrate feed rates [m³/d], one entry per substrate slot
        (up to 10 slots).
    k_L_a : float, optional
        Override of the gas–liquid mass-transfer coefficient [1/d].
    adm1_state : list of float, optional
        41-element initial state vector.  When supplied, replaces the
        auto-built steady-state vector.
    dynamic_volume : bool, default False
        Enable a dynamic sludge-volume balance
        ``dV/dt = Q_in − Q_out − q_S,loss``, with ``Q_out`` from an
        overflow weir at ``V_liq``. When False, sludge volume stays
        constant.
    initial_fill_fraction : float, default 1.0
        Starting sludge fill as a fraction of ``V_liq``. Only used when
        ``dynamic_volume=True``. Set below 1.0 to simulate a partially-
        filled startup transient.
    outflow_time_constant : float, default 1.0
        Overflow-weir time constant ``τ_out`` [d]. Only used when
        ``dynamic_volume=True``.

    Returns
    -------
    (Digester, str)
        The created digester and a one-line description of how the
        initial state was determined.
    """
    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,
        dynamic_volume=dynamic_volume,
        initial_fill_fraction=initial_fill_fraction,
        outflow_time_constant=outflow_time_constant,
    )

    if k_L_a is not None:
        digester.adm1.set_calibration_parameters({"k_L_a": float(k_L_a)})

    if Q_substrates is None:
        Q_substrates = [0.0] * 10

    init_kwargs: Dict[str, Any] = {"Q_substrates": Q_substrates}
    if adm1_state is not None:
        init_kwargs["adm1_state"] = list(adm1_state)
        state_info = "  - Initial state: User-supplied 41-element ADM1 vector\n"
    else:
        state_info = "  - Initial state: Auto-built steady-state from feedstock\n"
    digester.initialize(init_kwargs)

    self.plant.add_component(digester)

    # Automatic gas storage + connection
    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)
    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.

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."""
    heating = HeatingSystem(
        component_id=heating_id,
        target_temperature=target_temperature,
        heat_loss_coefficient=heat_loss_coefficient,
        name=name or heating_id,
        feedstock=self.feedstock,
    )

    self.plant.add_component(heating)

    return heating

auto_connect_chp_to_heating(chp_id, heating_id)

Connect CHP → heating with heat flow.

Source code in pyadm1/configurator/plant_configurator.py

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

auto_connect_digester_to_chp(digester_id, chp_id)

Connect digester → gas_storage → chp.

Source code in pyadm1/configurator/plant_configurator.py

def auto_connect_digester_to_chp(self, digester_id: str, chp_id: str) -> None:
    """Connect digester → gas_storage → chp."""
    storage_id = f"{digester_id}_storage"

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

    self.connect(storage_id, chp_id, "gas")

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

Connect two components.

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."""
    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.

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."""
    digester_config = digester_config or {}
    chp_config = chp_config or {}
    heating_config = heating_config or {}

    digester_config.setdefault("digester_id", "main_digester")
    chp_config.setdefault("chp_id", "chp_main")
    heating_config.setdefault("heating_id", "heating_main")

    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 two-stage plant: hydrolysis pre-tank → main fermenter.

The hydrolysis stage is just another :class:Digester instance with a higher temperature and shorter HRT — there is no separate Hydrolysis class.

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 two-stage plant: hydrolysis pre-tank → main fermenter.

    The hydrolysis stage is just another :class:`Digester` instance with
    a higher temperature and shorter HRT — there is no separate
    ``Hydrolysis`` class.
    """
    hydrolysis_config = hydrolysis_config or {}
    digester_config = digester_config or {}
    chp_config = chp_config or {}

    hydrolysis_config.setdefault("digester_id", "hydrolysis_tank")
    hydrolysis_config.setdefault("name", "Hydrolysis Tank")
    hydrolysis_config.setdefault("T_ad", 318.15)  # 45 °C
    hydrolysis_config.setdefault("V_liq", 500.0)
    hydrolysis_config.setdefault("V_gas", 75.0)

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

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

    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",
    }

    if auto_connect:
        self.connect(hydrolysis.component_id, digester.component_id, "liquid")

    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(hydrolysis.component_id, chp.component_id)
            self.auto_connect_digester_to_chp(digester.component_id, chp.component_id)

    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