← Back to CoordinatorHistoryEntry documentation

CoordinatorHistoryEntry - Source Code

File: Distributed_Design_Optimizer/coordination/CoordinatorHistoryEntry.py

# Copyright (C) The DistributedDesignOptimizer Contributors
# Licensed under the GNU General Public License v3.0. See LICENSE file for details.
"""Coordinator history entry module.

This module provides the history log entry for the coordinator. An entry is a
snapshot of selected coordinator-level state captured at one inner loop iteration.
"""

from typing import Dict, Tuple
import networkx as nx
from Distributed_Design_Optimizer.subsystem.historyentry import HistoryEntry
from Distributed_Design_Optimizer.postprocess import PerformanceMetrics


class CoordinatorHistoryEntry(HistoryEntry):
    """History log entry for the coordinator.

    Extends HistoryEntry with the system-wide metrics and analysis summaries that
    the coordinator records for each inner loop iteration.
    """

    def __init__(self,
                 outerloop_itr: int,
                 innerloop_itr: int,
                 innerloop_itr_runtime: float | None,
                 innerloop_itr_numberofdesignvariableevaluations: int | None,
                 maxinconsistencyvalue: float,
                 maxinconsistencyID: str,
                 maxratioofactiveconstraints: float,
                 maxratioofactiveconstraintsID: str,
                 performancemetrics: PerformanceMetrics,
                 centralitymeasures: Dict,
                 compromisemeasures: Dict,
                 inconsistencyoscillationindex: Dict,
                 clusteranalysis: Dict,
                 couplingstrength: Dict[Tuple[str, str], float],
                 mastergraph: nx.MultiDiGraph
                 ) -> None:
        """Create a new CoordinatorHistoryEntry.

        Args:
            outerloop_itr: Outer loop iteration number of this snapshot.
            innerloop_itr: Inner loop iteration number of this snapshot.
            innerloop_itr_runtime: Runtime of the inner loop iteration.
            innerloop_itr_numberofdesignvariableevaluations: Number of design
                variable evaluations of the inner loop iteration.
            maxinconsistencyvalue: Maximum inconsistency value across all subsystems.
            maxinconsistencyID: Identifier of the subsystem with the maximum inconsistency.
            maxratioofactiveconstraints: Maximum ratio of active constraints across all subsystems.
            maxratioofactiveconstraintsID: Identifier of the subsystem with the maximum ratio of active constraints.
            performancemetrics: Snapshot of the coordinator performance metrics.
            centralitymeasures: Snapshot of the centrality measures summary.
            compromisemeasures: Snapshot of the compromise measures summary.
            inconsistencyoscillationindex: Snapshot of the inconsistency oscillation index summary.
            clusteranalysis: Snapshot of the cluster analysis summary.
            couplingstrength: Snapshot of the coupling strength summary.
            mastergraph: Snapshot of the master graph.
        """
        super().__init__(outerloop_itr=outerloop_itr,
                         innerloop_itr=innerloop_itr,
                         innerloop_itr_runtime=innerloop_itr_runtime,
                         innerloop_itr_numberofdesignvariableevaluations=innerloop_itr_numberofdesignvariableevaluations)

        self._maxinconsistencyvalue: float = maxinconsistencyvalue
        self._maxinconsistencyID: str = maxinconsistencyID
        self._maxratioofactiveconstraints: float = maxratioofactiveconstraints
        self._maxratioofactiveconstraintsID: str = maxratioofactiveconstraintsID
        self._performancemetrics: PerformanceMetrics = performancemetrics
        self._centralitymeasures: Dict = centralitymeasures
        self._compromisemeasures: Dict = compromisemeasures
        self._inconsistencyoscillationindex: Dict = inconsistencyoscillationindex
        self._clusteranalysis: Dict = clusteranalysis
        self._couplingstrength: Dict[Tuple[str, str], float] = couplingstrength
        self._mastergraph: nx.MultiDiGraph = mastergraph

    def get_MaxInconsistencyValue(self) -> float:
        """Return the maximum inconsistency value across all subsystems.

        Returns:
            The maximum inconsistency value.
        """
        # No copy needed - float is a primitive/immutable type.
        return self._maxinconsistencyvalue

    def get_MaxInconsistencyValueSubsystemID(self) -> str:
        """Return the identifier of the subsystem with the maximum inconsistency.

        Returns:
            The subsystem identifier.
        """
        # No copy needed - str is a primitive/immutable type.
        return self._maxinconsistencyID

    def get_MaxRatioOfActiveConstraints(self) -> float:
        """Return the maximum ratio of active constraints across all subsystems.

        Returns:
            The maximum ratio of active constraints.
        """
        # No copy needed - float is a primitive/immutable type.
        return self._maxratioofactiveconstraints

    def get_MaxRatioOfActiveConstraintsSubsystemID(self) -> str:
        """Return the identifier of the subsystem with the maximum ratio of active constraints.

        Returns:
            The subsystem identifier.
        """
        # No copy needed - str is a primitive/immutable type.
        return self._maxratioofactiveconstraintsID

    def get_PerformanceMetrics(self) -> PerformanceMetrics:
        """Return the snapshot of the coordinator performance metrics.

        Returns:
            The performance metrics of this snapshot.
        """
        # No copy - the stored object is the snapshot itself, returned by reference intentionally.
        return self._performancemetrics

    def get_CentralityMeasures(self) -> Dict:
        """Return the snapshot of the centrality measures summary.

        Returns:
            The centrality measures summary of this snapshot.
        """
        # No copy - the stored object is the snapshot itself, returned by reference intentionally.
        return self._centralitymeasures

    def get_CompromiseMeasures(self) -> Dict:
        """Return the snapshot of the compromise measures summary.

        Returns:
            The compromise measures summary of this snapshot.
        """
        # No copy - the stored object is the snapshot itself, returned by reference intentionally.
        return self._compromisemeasures

    def get_InConsistencyOscillationIndex(self) -> Dict:
        """Return the snapshot of the inconsistency oscillation index summary.

        Returns:
            The inconsistency oscillation index summary of this snapshot.
        """
        # No copy - the stored object is the snapshot itself, returned by reference intentionally.
        return self._inconsistencyoscillationindex

    def get_ClusterAnalysis(self) -> Dict:
        """Return the snapshot of the cluster analysis summary.

        Returns:
            The cluster analysis summary of this snapshot.
        """
        # No copy - the stored object is the snapshot itself, returned by reference intentionally.
        return self._clusteranalysis

    def get_CouplingStrength(self) -> Dict[Tuple[str, str], float]:
        """Return the snapshot of the coupling strength summary.

        Returns:
            The coupling strength summary of this snapshot.
        """
        # No copy - the stored object is the snapshot itself, returned by reference intentionally.
        return self._couplingstrength

    def get_MasterGraph(self) -> nx.MultiDiGraph:
        """Return the snapshot of the master graph.

        Returns:
            The master graph of this snapshot.
        """
        # No copy - the stored object is the snapshot itself, returned by reference intentionally.
        return self._mastergraph