← Back to Consensus_ALC documentation
Consensus_ALC - Source Code¶
File: Distributed_Design_Optimizer/coordination/coordinationmethod/Consensus_ALC.py
# Copyright (C) The DistributedDesignOptimizer Contributors
# Licensed under the GNU General Public License v3.0. See LICENSE file for details.
"""Consensus-based ALC coordination method module.
This module implements consensus-based Augmented Lagrangian Coordination
for distributed multidisciplinary design optimization.
"""
import copy
from typing import List, Type
from Distributed_Design_Optimizer.postprocess.terminal_print_tools import DDO_Color, Reset, ddo_print, ddo_print_border
from Distributed_Design_Optimizer.coordination.coordinationmethod import CoordinationMethodBasis
from Distributed_Design_Optimizer.coordination.innerloop_iterationscheme import (IterationSchemeInterface,
Parallel,
ParallelPerLevelIncreasing,
SequentialForward,
SequentialBackward,
ParallelEvenThenOddLevels,
ParallelOddThenEvenLevels
)
from Distributed_Design_Optimizer.coordination.updatecouplingparametermethod import UpdateCouplingParameterMethodInterface
from Distributed_Design_Optimizer.coordination.convergence import (ConvergenceIndicator_Innerloop_Interface,
ConvergenceIndicator_Outerloop_Interface
)
from Distributed_Design_Optimizer.subsystem.optimization import AnalysisInterface, OptimizationInterface
from Distributed_Design_Optimizer.subsystem.optimization.designproblem import LocalObjectiveInterface, LocalConstraintsInterface
from Distributed_Design_Optimizer.subsystem import SubSystemInterface, LocalSubSystemConsensusALC
from Distributed_Design_Optimizer.middlelevel.consensus_alc import MiddleLevelDataStorageConsensusALC
class Consensus_ALC(CoordinationMethodBasis):
"""Consensus ALC coordination method for distributed optimization.
This coordination method implements consensus-based Augmented Lagrangian
Coordination. Each subsystem maintains its own copy of the shared coupling
variables and drives them towards a common consensus value, allowing the
independent subproblems to be solved fully in parallel.
"""
def __init__(self,
convergence_indicator_innerloop: ConvergenceIndicator_Innerloop_Interface,
convergence_indicator_outerloop: ConvergenceIndicator_Outerloop_Interface,
updatecouplingparametermethod_outerloop: UpdateCouplingParameterMethodInterface,
iterationscheme: IterationSchemeInterface) -> None:
"""Initialize the Consensus_ALC coordination method.
Args:
convergence_indicator_innerloop: Convergence indicator for inner loop (e.g., ConvergenceIndicator_Innerloop_DeWit).
convergence_indicator_outerloop: Convergence indicator for outer loop (e.g., ConvergenceIndicator_Outerloop_DeWit).
updatecouplingparametermethod_outerloop: Method for updating coupling parameters in outer loop.
iterationscheme: Iteration scheme for inner loop execution.
"""
super().__init__()
# Allowed iteration schemes for Consensus ALC (no controller)
self._allowediterationschemes: List[Type[IterationSchemeInterface]] = [
Parallel, ParallelPerLevelIncreasing, SequentialForward,
SequentialBackward, ParallelEvenThenOddLevels, ParallelOddThenEvenLevels
]
# Unrecommended schemes - sequential execution is inefficient for independent subproblems
self._unrecommendediterationschemes: List[Type[IterationSchemeInterface]] = [
ParallelPerLevelIncreasing, SequentialForward, SequentialBackward,
ParallelEvenThenOddLevels, ParallelOddThenEvenLevels
]
# Recommended scheme - parallel execution is most efficient for independent subproblems
self._recommendediterationschemes: List[Type[IterationSchemeInterface]] = [
Parallel
]
# Set inputs
self._convergence_indicator_innerloop: ConvergenceIndicator_Innerloop_Interface = convergence_indicator_innerloop
self._convergence_indicator_outerloop: ConvergenceIndicator_Outerloop_Interface = convergence_indicator_outerloop
self._updatecouplingparametermethod_outerloop: UpdateCouplingParameterMethodInterface = updatecouplingparametermethod_outerloop
self._iterationscheme: IterationSchemeInterface = iterationscheme
# Validate inputs
self.validate_inputs()
def validate_inputs(self) -> None:
"""Validate the inputs provided to the Consensus_ALC coordination method.
Validates that the iteration scheme is compatible with Consensus_ALC. Validation of
the update coupling parameter method and convergence indicators is delegated to
LocalSubSystemConsensusALC.validate_inputs, since those components are handed to the
subsystems.
Raises:
ValueError: If any parameter is outside the allowed range.
"""
# ===== Iteration Scheme Validation =====
if type(self._iterationscheme) not in self._allowediterationschemes:
raise ValueError(
f"{DDO_Color}Iteration scheme '{type(self._iterationscheme).__name__}' is not compatible with Consensus_ALC. "
f"Consensus_ALC does not use a controller, so controller-based schemes are not supported. "
f"Please choose one of the compatible schemes: {[s.__name__ for s in self._allowediterationschemes]}{Reset}"
)
elif type(self._iterationscheme) in self._unrecommendediterationschemes:
ddo_print_border()
ddo_print(f"{type(self).__name__}: WARNING: Iteration scheme '{type(self._iterationscheme).__name__}' is not recommended for Consensus_ALC.")
ddo_print(f"{type(self).__name__}: Subproblems are independent, so parallel execution is more efficient.")
ddo_print(f"{type(self).__name__}: Recommended schemes for Consensus_ALC: {[s.__name__ for s in self._recommendediterationschemes]}")
ddo_print_border()
def createSubSystems(self,
id_list: List[str],
level_list: List[int],
neighborid_list: List[List[str]],
analysis_list: List[AnalysisInterface],
localobjective_list: List[LocalObjectiveInterface],
localconstraints_list: List[LocalConstraintsInterface],
optimization_list: List[OptimizationInterface]) -> List[LocalSubSystemConsensusALC]:
"""Create subsystems for consensus-based augmented Lagrangian coordination.
Args:
id_list: List of unique identifiers for each subsystem.
level_list: List of hierarchy levels for each subsystem.
neighborid_list: List of neighbor subsystem IDs for each subsystem.
analysis_list: List of analysis objects for each subsystem.
localobjective_list: List of local objective functions for each subsystem.
localconstraints_list: List of local constraints for each subsystem.
optimization_list: List of optimization objects for each subsystem.
Returns:
List of initialized LocalSubSystemConsensusALC objects.
"""
if not (len(id_list) == len(level_list) == len(neighborid_list) == len(analysis_list) == len(localobjective_list) == len(localconstraints_list) == len(optimization_list)):
raise ValueError(f"{DDO_Color}The length of the provided list does not match in Consensus_ALC.createSubSystems{Reset}")
subsystems = [LocalSubSystemConsensusALC(id=id_list[i],
level=level_list[i],
neighborid=neighborid_list[i],
analysis=analysis_list[i],
localobjective=localobjective_list[i],
localconstraints=localconstraints_list[i],
optimization=optimization_list[i],
local_convergenceindicator_innerloop=self._convergence_indicator_innerloop.createLocalConvergenceIndicator(),
local_convergenceindicator_outerloop=self._convergence_indicator_outerloop.createLocalConvergenceIndicator(),
# deepcopy() to strengthen distributed character - only middlelevels are shared recourses between subsystems
updatecouplingparametermethod_outerloop=copy.deepcopy(self._updatecouplingparametermethod_outerloop)
)
for i in range(len(id_list))]
return subsystems
def createControllerSubSystem(self, subsystemsIn: List[LocalSubSystemConsensusALC]) -> SubSystemInterface | None:
"""Create a controller subsystem for coordinating local subsystems.
Args:
subsystemsIn: List of local subsystems to be coordinated.
Returns:
None, as Consensus ALC operates without a central controller.
"""
# Consensus_ALC works without a controller
subsystemcontroller = None
return subsystemcontroller
def createMiddleLevel(self, idparent: str, idchild: str, multiprocessing_lock: object = None) -> MiddleLevelDataStorageConsensusALC:
"""Create a middle level data storage for coupling between subsystems.
Args:
idparent: Identifier of the parent subsystem.
idchild: Identifier of the child subsystem.
multiprocessing_lock: Optional lock for thread-safe access in multiprocessing.
Returns:
MiddleLevelDataStorage instance for managing coupling data.
"""
return MiddleLevelDataStorageConsensusALC(idparent, idchild, multiprocessing_lock)
def createControllerMiddleLevel(self, idparent: str, idchild: str, local_neighbors_list: List[str], multiprocessing_lock: object = None) -> MiddleLevelDataStorageConsensusALC | None:
"""Create middle level between controller and local subsystem.
Args:
idparent: Identifier of the parent (controller) subsystem.
idchild: Identifier of the child subsystem.
local_neighbors_list: List of neighbor IDs for the local subsystem (unused).
multiprocessing_lock: Optional lock for thread-safe access in multiprocessing.
Returns:
None, as Consensus ALC operates without a central controller.
"""
# No controller, hence return None
return None
def centralized_prepare_updateCouplingParameters(self, subsystemsIn: List[LocalSubSystemConsensusALC]) -> None:
"""Prepare centralized update of coupling parameters for all subsystems.
For Consensus ALC, no centralized operation is required.
Args:
subsystemsIn: List of subsystems to prepare coupling parameters for.
"""
pass
def print_beginning_of_centralized_prepare_updateCouplingParameters(self) -> None:
"""Nothing to print."""
# Nothing to print
pass