Source code for cratepy.clustering.clusteringphase

"""Cluster-Reduced Material Phase.

This module includes the interface to implement any Cluster-Reduced Material
Phase and several cluster-reduced material phases, namely the most basic Static
Cluster-Reduced Material Phase (SCRMP). Each class contains all the required
methods to manage the material phase clustering.

The concept of Cluster-Reduced Material Phase was coined by
Ferreira et. al (2022) [#]_ and arises in the context of clustering-based
reduced order modeling (see Chapter 4 of Ferreira (2022) [#]_).

.. [#] Ferreira, B.P., Andrade Pires, F.M. and Bessa, M.A. (2022).
       *Adaptivity for clustering-based reduced-order modeling of
       localized history-dependent phenomena.* Comp Methods Appl M, 393
       (see `here <https://www.sciencedirect.com/science/article/pii/
       S0045782522000895?via%3Dihub>`_)

.. [#] Ferreira, B.P. (2022). *Towards Data-driven Multi-scale
       Optimization of Thermoplastic Blends: Microstructural
       Generation, Constitutive Development and Clustering-based
       Reduced-Order Modeling.* PhD Thesis, University of Porto
       (see `here <https://repositorio-aberto.up.pt/handle/10216/
       146900?locale=en>`_)

Classes
-------
CRMP
    Cluster-Reduced Material Phase interface.
ACRMP
    Adaptive Cluster-Reduced Material Phase interface.
SCRMP
    Static Cluster-Reduced Material Phase (SCRMP).
GACRMP
    Generalized Adaptive Cluster-Reduced Material Phase.
HAACRMP
    Hierarchical Agglomerative Adaptive Cluster-Reduced Material Phase.
"""
#
#                                                                       Modules
# =============================================================================
# Standard
from abc import ABC, abstractmethod
import time
import copy
# Third-party
import numpy as np
import scipy.cluster.hierarchy as sciclst
import anytree
# import anytree.exporter
# Local
import clustering.clusteringalgs as clstalgs
from clustering.clusteringalgs import ClusterAnalysis
import tensor.matrixoperations as mop
#
#                                                          Authorship & Credits
# =============================================================================
__author__ = 'Bernardo Ferreira (bernardo_ferreira@brown.edu)'
__credits__ = ['Bernardo Ferreira', ]
__status__ = 'Stable'
# =============================================================================
#
# =============================================================================
#
#                                     Interface: Cluster-reduced material phase
# =============================================================================
[docs]class CRMP(ABC): """Cluster-Reduced Material Phase interface. Methods ------- perform_base_clustering(self) *abstract*: Perform CRMP base clustering. get_n_clusters(self) *abstract*: Get current number of clusters. get_clustering_type(self) *abstract*: Get cluster-reduced material phase adaptivity type. get_valid_clust_algs() *abstract*: Get valid clustering algorithms to compute the CRMP. _update_cluster_labels(labels, min_label=0) Update cluster labels starting with the provided minimum label. """
[docs] @abstractmethod def __init__(self): """Constructor.""" pass
# -------------------------------------------------------------------------
[docs] @abstractmethod def perform_base_clustering(self): """Perform CRMP base clustering.""" pass
# -------------------------------------------------------------------------
[docs] @abstractmethod def get_n_clusters(self): """Get current number of clusters. Returns ------- n_clusters : int Number of material phase clusters. """ pass
# -------------------------------------------------------------------------
[docs] @abstractmethod def get_clustering_type(self): """Get cluster-reduced material phase adaptivity type. Returns ------- clustering_type : str Type of cluster-reduced material phase. """ pass
# -------------------------------------------------------------------------
[docs] @staticmethod @abstractmethod def get_valid_clust_algs(): """Get valid clustering algorithms to compute the CRMP. Returns ---------- clust_algs : list[str] Clustering algorithms identifiers (str). """ pass
# -------------------------------------------------------------------------
[docs] @staticmethod def _update_cluster_labels(labels, min_label=0): """Update cluster labels starting with the provided minimum label. Parameters ---------- labels : numpy.ndarray (1d) Cluster labels (numpy.ndarray[int] of shape (n_items,)). min_label : int, default=0 Minimum cluster label. Returns ------- new_labels : numpy.ndarray (1d) Updated cluster labels (numpy.ndarray[int] of shape (n_items,)). max_label : int Maximum cluster label. """ # Get sorted set of original labels unique_labels = sorted(list(set(labels))) # Initialize updated labels array new_labels = np.full(len(labels), -1, dtype=int) # Initialize new label new_label = min_label # Loop over sorted original labels for label in unique_labels: # Get original labels indexes indexes = np.where(labels == label) # Set updated labels new_labels[indexes] = new_label # Increment new label new_label += 1 # Get maximum cluster label max_label = max(new_labels) # Check cluster updated labels if len(unique_labels) != len(set(new_labels)): raise RuntimeError('Number of clusters differs between original ' 'and updated labels.') # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Return return new_labels, max_label
# =============================================================================
[docs]class ACRMP(CRMP): """Adaptive Cluster-Reduced Material Phase interface. Methods ------- perform_adaptive_clustering(self, target_clusters, target_clusters_data) *abstract*: Perform ACRMP adaptive clustering step. get_adaptive_output(self) *abstract*: Get adaptivity metrics for clustering adaptivity output. _check_adaptivity_lock(self) *abstract*: Check ACRMP adaptivity locking conditions. get_adaptivity_type_parameters() *abstract*: Get ACRMP mandatory and optional adaptivity type parameters. _set_adaptivity_type_parameters(self, adaptivity_type) *abstract*: Set clustering adaptivity parameters. _dynamic_split_factor(ref_split_factor, adapt_trigger_ratio, magnitude, \ dynamic_amp=0) Compute dynamic adaptive clustering split factor. """
[docs] @abstractmethod def perform_adaptive_clustering(self, target_clusters, target_clusters_data): """Perform ACRMP adaptive clustering step. Parameters ---------- target_clusters : list[int] List with the labels (int) of clusters to be adapted. target_clusters_data : dict For each target cluster (key, str), store dictionary (item, dict) containing cluster associated parameters relevant for the adaptive procedures. Returns ------- adaptive_clustering_map : dict List of new cluster labels (item, list[int] resulting from the adaptivity of each target cluster (key, str). """ pass
# -------------------------------------------------------------------------
[docs] @abstractmethod def get_adaptive_output(self): """Get adaptivity metrics for clustering adaptivity output. Returns ------- adaptivity_output : list[int or float] List containing the adaptivity metrics associated with the clustering adaptivity output file. """ pass
# -------------------------------------------------------------------------
[docs] @abstractmethod def _check_adaptivity_lock(self): """Check ACRMP adaptivity locking conditions. Check conditions that may deactivate the adaptive procedures in the ACRMP. Once the ACRMP adaptivity is locked, it is treated as a SCRMP for the remainder of the problem numerical solution. """ pass
# -------------------------------------------------------------------------
[docs] @staticmethod @abstractmethod def get_adaptivity_type_parameters(): """Get ACRMP mandatory and optional adaptivity type parameters. Besides returning the ACRMP mandatory and optional adaptivity type parameters, this method establishes the default values for the optional parameters. Returns ---------- adapt_type_man_parameters : list[str] Mandatory adaptivity type parameters (str). adapt_type_opt_parameters : dict Optional adaptivity type parameters (key, str) and associated default value (item). """ pass
# -------------------------------------------------------------------------
[docs] @abstractmethod def _set_adaptivity_type_parameters(self, adaptivity_type): """Set clustering adaptivity parameters. Parameters ---------- adaptivity_type : dict Clustering adaptivity parameters. """ pass
# -------------------------------------------------------------------------
[docs] @staticmethod def _dynamic_split_factor(ref_split_factor, adapt_trigger_ratio, magnitude, dynamic_amp=0): """Compute dynamic adaptive clustering split factor. A detailed description of the dynamic adaptive clustering split factor can be found in Ferreira et. al (2022) [#]_. .. [#] Ferreira, B.P., Andrade Pires, F.M. and Bessa, M.A. (2022). *Adaptivity for clustering-based reduced-order modeling of localized history-dependent phenomena.* Comp Methods Appl M, 393 (see `here <https://www.sciencedirect.com/science/article/pii/ S0045782522000895?via%3Dihub>`_) ---- Parameters ---------- ref_split_factor : float Reference (centered) adaptive clustering split factor. The adaptive clustering split factor must be contained between 0 and 1 (included). The lower bound (0) enforces a single split, while the upper bound (1) performs the maximum number splits of each cluster (leading to single-voxel clusters). adapt_trigger_ratio : float Threshold associated with the adaptivity trigger condition. magnitude : float Difference between cluster ratio and adaptive trigger ratio. Given that the cluster ratio ranges between 0 and 1 and only clusters with a ratio greater or equal than the adaptive trigger ratio are targeted, the magnitude ranges between 0 and 1 - trigger ratio. dynamic_amp : float, default=0 Dynamic split factor amplitude centered around the reference adaptive clustering split factor. Returns ------- adapt_split_factor : float Adaptive clustering split factor. The adaptive clustering split factor must be contained between 0 and 1 (included). The lower bound (0) enforces a single split, i.e., 2 new clusters, while the upper bound (1) is associated with a maximum defined number of new voxels. """ # Check provided parameters if not (ref_split_factor >= 0 and ref_split_factor <= 1): raise RuntimeError('Invalid reference adaptive clustering split ' 'factor.') if not (adapt_trigger_ratio >= 0 and adapt_trigger_ratio <= 1): raise RuntimeError('Invalid adaptive trigger ratio.') if not (dynamic_amp >= 0): raise RuntimeError('Invalid dynamic split factor amplitude.') # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # If the dynamic split factor amplitude is null, skip computations and # return reference adaptive clustering split factor. Otherwise, compute # adaptive clustering split factor lower and upper bounds if abs(dynamic_amp) < 1e-10: return ref_split_factor else: lower_bound = max(0, ref_split_factor - 0.5*dynamic_amp) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Choose dynamic function type dynamic_type = 'power' # Set dynamic function if dynamic_type == 'power': # Set dynamic function power n = 1.0 # Check power admissibility if n < 0: raise RuntimeError('Dynamic function power must be greater or ' 'equal than zero.') # Power dynamic function dynamic_function = \ lambda magnitude: (1/((1 - adapt_trigger_ratio)**n))*( magnitude**n) else: # Linear dynamic function dynamic_function = \ lambda magnitude: (1/(1 - adapt_trigger_ratio))*magnitude # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Compute adaptive clustering split factor adapt_split_factor = min(1, lower_bound + dynamic_function(magnitude)*dynamic_amp) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Return return adapt_split_factor
# # Cluster-Reduced Material Phases # =============================================================================
[docs]class SCRMP(CRMP): """Static Cluster-Reduced Material Phase (SCRMP). Attributes ---------- _clustering_type : str Type of cluster-reduced material phase. max_label : int Clustering maximum label. cluster_labels : numpy.ndarray (1d) Material phase cluster labels (numpy.ndarray[int] of shape (n_phase_voxels,)). Methods ------- perform_base_clustering(self, base_clustering_scheme, min_label=0) Perform SCRMP base clustering. get_n_clusters(self) Get current number of clusters. get_clustering_type(self) Get cluster-reduced material phase adaptivity type. get_valid_clust_algs() Get valid clustering algorithms to compute the CRMP. """
[docs] def __init__(self, mat_phase, cluster_data_matrix, n_clusters): """Constructor. Parameters ---------- mat_phase : str Material phase label. cluster_data_matrix: numpy.ndarray (2d) Data matrix (numpy.ndarray of shape (n_phase_voxels, n_features_dims)) containing the clustering features data to perform the material phase cluster analyses. n_clusters : int Number of material phase clusters. """ self._mat_phase = mat_phase self._cluster_data_matrix = cluster_data_matrix self._clustering_type = 'static' self._n_clusters = n_clusters self.max_label = 0 self.cluster_labels = None
# -------------------------------------------------------------------------
[docs] def perform_base_clustering(self, base_clustering_scheme, min_label=0): """Perform SCRMP base clustering. Parameters ---------- base_clustering_scheme : dict Prescribed base clustering scheme (item, numpy.ndarray of shape (n_clusterings, 3)) for each material phase (key, str). Each row is associated with a unique clustering characterized by a clustering algorithm (col 1, int), a list of features (col 2, list[int]) and a list of the features data matrix' indexes (col 3, list[int]). min_label : int, default=0 Minimum cluster label. """ # Get number of material phase voxels n_phase_voxels = self._cluster_data_matrix.shape[0] # Get number of prescribed clusterings n_clusterings = base_clustering_scheme.shape[0] # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Initialize collection of clustering solutions clustering_solutions = [] # Loop over prescribed clustering solutions for i in range(n_clusterings): # Get base clustering algorithm and check validity clust_alg_id = str(base_clustering_scheme[i, 0]) if clust_alg_id not in self.get_valid_clust_algs(): raise RuntimeError('Invalid base clustering algorithm.') # Get clustering features' column indexes indexes = base_clustering_scheme[i, 2] # Get base clustering data matrix data_matrix = mop.get_condensed_matrix( self._cluster_data_matrix, list(range(n_phase_voxels)), indexes) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Perform cluster analysis cluster_labels, _, is_n_clusters_satisfied = \ ClusterAnalysis().get_fitted_estimator(data_matrix, clust_alg_id, self._n_clusters) # Check if prescribed number of clusters is satisfied if not is_n_clusters_satisfied: raise RuntimeError('The number of clusters (' + str(len(set(cluster_labels))) + ') obtained is different from the ' + 'prescribed number of clusters (' + str(self._n_clusters) + ').') # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Add clustering to collection of clustering solutions clustering_solutions.append(cluster_labels) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Get consensus clustering if n_clusterings > 1: raise RuntimeError('No clustering ensemble method has been ' 'implemented yet.') else: self.cluster_labels = cluster_labels # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update cluster labels self.cluster_labels, self.max_label = \ self._update_cluster_labels(self.cluster_labels, min_label)
# -------------------------------------------------------------------------
[docs] def get_n_clusters(self): """Get current number of clusters. Returns ------- n_clusters : int Number of material phase clusters. """ return self._n_clusters
# -------------------------------------------------------------------------
[docs] def get_clustering_type(self): """Get cluster-reduced material phase adaptivity type. Returns ------- clustering_type : str Type of cluster-reduced material phase. """ return self._clustering_type
# -------------------------------------------------------------------------
[docs] @staticmethod def get_valid_clust_algs(): """Get valid clustering algorithms to compute the CRMP. Returns ---------- clust_algs : list Clustering algorithms identifiers (str). """ return list(ClusterAnalysis.available_clustering_alg.keys())
# =============================================================================
[docs]class GACRMP(ACRMP): """Generalized Adaptive Cluster-Reduced Material Phase. A detailed description of a Generalized Adaptive Cluster-Reduced Material Phase can be found in Ferreira et. al (2022) [#]_. .. [#] Ferreira, B.P., Andrade Pires, F.M. and Bessa, M.A. (2022). *Adaptivity for clustering-based reduced-order modeling of localized history-dependent phenomena.* Comp Methods Appl M, 393 (see `here <https://www.sciencedirect.com/science/article/pii/ S0045782522000895?via%3Dihub>`_) Attributes ---------- _clustering_type : str Type of cluster-reduced material phase. _adaptive_step : int Counter of adaptive clustering steps, with 0 associated with the base clustering. _adapt_split_factor : float Adaptive clustering split factor. The adaptive clustering split factor must be contained between 0 and 1 (included). The lower bound (0) enforces a single split, i.e., 2 new clusters, while the upper bound (1) is associated with a maximum defined number of new voxels. _threshold_n_clusters : int Threshold number of adaptive material phase number of clusters. Once this threshold is surpassed, the adaptive procedures of the adaptive material phase are deactivated. _is_dynamic_split_factor : bool True if adaptive clustering split factor is to be computed dynamically. Otherwise, the adaptive clustering split factor is always set equal to `_adapt_split_factor`. _clustering_tree_nodes : dict Clustering tree node (item, anytree.Node) associated with each material cluster (key, str). _root_cluster_node : anytree.Node Clustering tree root node. max_label : int Clustering maximum label. cluster_labels : numpy.ndarray (1d) Material phase cluster labels (numpy.ndarray[int] of shape (n_phase_voxels,)). adaptive_time : float Total amount of time (s) spent in the adaptive procedures. adaptivity_lock : bool True if the adaptive procedures are deactivated, False otherwise. Methods ------- perform_base_clustering(self, base_clustering_scheme, min_label=0) Perform GACRMP base clustering. get_valid_clust_algs(): Get valid clustering algorithms to compute the CRMP. perform_adaptive_clustering(self, target_clusters, target_clusters_data, \ adaptive_clustering_scheme=None, min_label=0) Perform GACRMP adaptive clustering step. _check_adaptivity_lock(self) Check ACRMP adaptivity locking conditions. get_n_clusters(self) Get current number of clusters. get_clustering_type(self) Get cluster-reduced material phase adaptivity type. get_clustering_tree_nodes(self) Get clustering tree nodes. get_adaptivity_type_parameters() Get ACRMP mandatory and optional adaptivity type parameters. _set_adaptivity_type_parameters(self, adaptivity_type) Set clustering adaptivity parameters. get_adaptive_output(self) Get adaptivity metrics for clustering adaptivity output. reset_adaptive_parameters(self) Reset clustering adaptive progress parameters. update_adaptivity_type(self, adaptivity_type) Update clustering adaptivity parameters. """
[docs] def __init__(self, mat_phase, cluster_data_matrix, n_clusters, adaptivity_type): """Constructor. Parameters ---------- mat_phase : str Material phase label. cluster_data_matrix: numpy.ndarray (2d) Data matrix (numpy.ndarray of shape (n_phase_voxels, n_features_dims)) containing the clustering features data to perform the material phase cluster analyses. n_clusters : int Number of material phase clusters. adaptivity_type : dict Clustering adaptivity parameters. """ self._mat_phase = mat_phase self._cluster_data_matrix = cluster_data_matrix self._clustering_type = 'adaptive' self._n_clusters = n_clusters self._adaptivity_type = adaptivity_type self._adaptive_step = 0 self._clustering_tree_nodes = {} self.max_label = 0 self.cluster_labels = None self.adaptive_time = 0 self.adaptivity_lock = False # Set clustering adaptivity parameters self._set_adaptivity_type_parameters(self._adaptivity_type) # Set dynamic adaptive split factor if abs(self._dynamic_split_factor_amp) < 1e-10: self._is_dynamic_split_factor = False else: self._is_dynamic_split_factor = True # Set clustering tree root node root_cluster = -1 self._root_cluster_node = anytree.Node(-1) self._clustering_tree_nodes[str(root_cluster)] = \ self._root_cluster_node
# -------------------------------------------------------------------------
[docs] def perform_base_clustering(self, base_clustering_scheme, min_label=0): """Perform GACRMP base clustering. Parameters ---------- base_clustering_scheme : dict Prescribed base clustering scheme (item, numpy.ndarray of shape (n_clusterings, 3)) for each material phase (key, str). Each row is associated with a unique clustering characterized by a clustering algorithm (col 1, int), a list of features (col 2, list[int]) and a list of the features data matrix' indexes (col 3, list[int]). min_label : int, default=0 Minimum cluster label. """ # Get number of material phase voxels n_phase_voxels = self._cluster_data_matrix.shape[0] # Get number of prescribed clusterings n_clusterings = base_clustering_scheme.shape[0] # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Initialize collection of clustering solutions clustering_solutions = [] # Loop over prescribed clustering solutions for i in range(n_clusterings): # Get base clustering algorithm and check validity clust_alg_id = str(base_clustering_scheme[i, 0]) if clust_alg_id not in self.get_valid_clust_algs(): raise RuntimeError('Invalid base clustering algorithm.') # Get clustering features' column indexes indexes = base_clustering_scheme[i, 2] # Get base clustering data matrix data_matrix = mop.get_condensed_matrix( self._cluster_data_matrix, list(range(n_phase_voxels)), indexes) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Perform cluster analysis cluster_labels, _, is_n_clusters_satisfied = \ ClusterAnalysis().get_fitted_estimator(data_matrix, clust_alg_id, self._n_clusters) # Check if prescribed number of clusters is satisfied if not is_n_clusters_satisfied: raise RuntimeError('The number of clusters (' + str(len(set(cluster_labels))) + ') obtained is different from the ' + 'prescribed number of clusters (' + str(self._n_clusters) + ').') # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Add clustering to collection of clustering solutions clustering_solutions.append(cluster_labels) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Get consensus clustering if n_clusterings > 1: raise RuntimeError('No clustering ensemble method has been ' 'implemented yet.') else: self.cluster_labels = cluster_labels # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update cluster labels self.cluster_labels, self.max_label = \ self._update_cluster_labels(self.cluster_labels, min_label) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Loop over base clustering clusters for cluster in set(self.cluster_labels): # Update clustering tree self._clustering_tree_nodes[str(cluster)] = \ anytree.Node(cluster, parent=self._root_cluster_node)
# -------------------------------------------------------------------------
[docs] @staticmethod def get_valid_clust_algs(): """Get valid clustering algorithms to compute the CRMP. Returns ---------- clust_algs : list Clustering algorithms identifiers (str). """ return list(ClusterAnalysis.available_clustering_alg.keys())
# -------------------------------------------------------------------------
[docs] def perform_adaptive_clustering( self, target_clusters, target_clusters_data, adaptive_clustering_scheme=None, min_label=0): """Perform GACRMP adaptive clustering step. Refine the provided target clusters by splitting them according to the prescribed adaptive clustering scheme. ---- Parameters ---------- target_clusters : list[int] List with the labels (int) of clusters to be adapted. target_clusters_data : dict For each target cluster (key, str), store dictionary (item, dict) containing cluster associated parameters required for the adaptive procedures. adaptive_clustering_scheme : dict Prescribed adaptive clustering scheme (item, numpy.ndarray of shape (n_clusterings, 3)) for each material phase (key, str). Each row is associated with a unique clustering characterized by a clustering algorithm (col 1, int), a list of features (col 2, list[int]) and a list of the features data matrix' indexes (col 3, list[int]). min_label : int, default=0 Minimum cluster label. Returns ------- adaptive_clustering_map : dict List of new cluster labels (item, list[int]) resulting from the adaptivity of each target cluster (key, str). """ # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Check for duplicated target clusters if len(target_clusters) != len(np.unique(target_clusters)): raise RuntimeError('List of target clusters contains duplicated ' 'labels.') # Check for unexistent target clusters for target_cluster in target_clusters: if target_cluster not in self.cluster_labels: raise RuntimeError('Target cluster ' + str(target_cluster) + ' does not exist in material phase ' + str(self._mat_phase)) # Check adaptive clustering scheme if adaptive_clustering_scheme is None: raise RuntimeError('An adaptive clustering scheme must be ' 'prescribed to perform the GACRMP clustering ' 'adaptivity.') # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ init_time = time.time() # Increment adaptive clustering step counter self._adaptive_step += 1 # Initialize adaptive clustering mapping dictionary adaptive_clustering_map = {} # Get material phase original clustering original_cluster_labels = copy.deepcopy(self.cluster_labels) # Initialize new cluster label new_cluster_label = min_label # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Initialize sorted target clusters flag is_sorted_target_clusters = False # Check if target clusters magnitude is available if 'max_magnitude' in \ target_clusters_data[str(target_clusters[0])].keys(): # Set sorted target clusters flag is_sorted_target_clusters = True # Get target clusters magnitude target_clusters_magnitude = { str(cluster): target_clusters_data[str(cluster)]['max_magnitude'] for cluster in target_clusters} # Get target clusters in descending order of magnitude target_clusters_sorted = \ [int(x[0]) for x in sorted(target_clusters_magnitude.items(), key=lambda x: x[1], reverse=True)] # Set sorted target clusters if set(target_clusters) == set(target_clusters_sorted): target_clusters = copy.deepcopy(target_clusters_sorted) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Loop over target clusters for i in range(len(target_clusters)): # Get target cluster label target_cluster = target_clusters[i] # Get total number of voxels associated with target cluster. # If target cluster is already single-voxeled, skip to the next # target cluster. Otherwise compute the number of child clusters. # If the target cluster number of voxels is lower or equal than the # number of child clusters, skip to the next target cluster. n_cluster_voxels = \ np.count_nonzero(original_cluster_labels == target_cluster) if n_cluster_voxels == 1: continue else: # Get referece adaptive clustering split factor ref_split_factor = self._adapt_split_factor # Get target cluster dynamic split factor ability is_cluster_dynamic_split_factor = target_clusters_data[ str(target_cluster)]['is_dynamic_split_factor'] # Set adaptive clustering split factor if self._is_dynamic_split_factor \ and is_cluster_dynamic_split_factor: # Get adaptive trigger ratio and magnitude adapt_trigger_ratio = target_clusters_data[ str(target_cluster)]['adapt_trigger_ratio'] magnitude = target_clusters_data[ str(target_cluster)]['max_magnitude'] # Compute dynamic adaptive clustering split factor adapt_split_factor = super()._dynamic_split_factor( ref_split_factor, adapt_trigger_ratio, magnitude, dynamic_amp=self._dynamic_split_factor_amp) else: adapt_split_factor = ref_split_factor # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Compute number of child clusters, enforcing at least two # clusters n_new_clusters = max( 2, int(round(adapt_split_factor*int( round(1.0/self._child_cluster_vol_fraction))))) # Compare number of child clusters and number of target cluster # number of voxels if n_cluster_voxels <= n_new_clusters: continue # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Initialize target cluster mapping adaptive_clustering_map[str(target_cluster)] = [] # Get target cluster indexes target_cluster_idxs = \ list(*np.nonzero(original_cluster_labels == target_cluster)) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Get number of prescribed clusterings n_clusterings = adaptive_clustering_scheme.shape[0] # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Initialize collection of clustering solutions clustering_solutions = [] # Loop over prescribed clustering solutions for i in range(n_clusterings): # Get adaptive clustering algorithm and check validity clust_alg_id = str(adaptive_clustering_scheme[i, 0]) if clust_alg_id not in self.get_valid_clust_algs(): raise RuntimeError('Invalid adaptive clustering ' 'algorithm.') # Get clustering features' column indexes indexes = adaptive_clustering_scheme[i, 2] # Get adaptive clustering data matrix data_matrix = mop.get_condensed_matrix( self._cluster_data_matrix, target_cluster_idxs, indexes) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Perform cluster analysis cluster_labels, _, is_n_clusters_satisfied = \ ClusterAnalysis().get_fitted_estimator(data_matrix, clust_alg_id, n_new_clusters) # Check if prescribed number of clusters is satisfied if not is_n_clusters_satisfied: # If the prescribed number of clusters is not satisfied, # proceed with the number of clusters obtained pass # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Add clustering to collection of clustering solutions clustering_solutions.append(cluster_labels) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Get consensus clustering if n_clusterings > 1: raise RuntimeError('No clustering ensemble method has been ' 'implemented yet.') else: child_cluster_labels = cluster_labels # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update cluster labels child_cluster_labels, max_label = self._update_cluster_labels( child_cluster_labels, new_cluster_label) # Update new cluster label new_cluster_label = max_label + 1 # Add new clusters to target cluster mapping adaptive_clustering_map[str(target_cluster)] += \ list(set(child_cluster_labels)) # Update material phase clustering self.cluster_labels[target_cluster_idxs] = child_cluster_labels # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Check if number of clusters threshold has been surpassed if is_sorted_target_clusters: if len(set(self.cluster_labels)) > self._threshold_n_clusters: break # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update number of material phase clusters self._n_clusters = len(set(self.cluster_labels)) # Update material phase maximum cluster label self.max_label = max(self.cluster_labels) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Loop over target clusters for target_cluster in adaptive_clustering_map.keys(): # Get target cluster node parent_node = self._clustering_tree_nodes[str(target_cluster)] # Loop over child clusters for child_cluster in adaptive_clustering_map[target_cluster]: # Set child cluster tree node self._clustering_tree_nodes[str(child_cluster)] = \ anytree.Node(child_cluster, parent=parent_node) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update total amount of time spent in the adaptive procedures self.adaptive_time += time.time() - init_time # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Check adaptivity threshold conditions self._check_adaptivity_lock() # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Return return adaptive_clustering_map
# -------------------------------------------------------------------------
[docs] def _check_adaptivity_lock(self): """Check ACRMP adaptivity locking conditions. Check conditions that may deactivate the adaptive procedures in the ACRMP. Once the ACRMP adaptivity is locked, it is treated as a SCRMP for the remainder of the problem numerical solution. """ # Check if the number of clusters threshold as been surpassed if self._n_clusters > self._threshold_n_clusters: self.adaptivity_lock = True
# -------------------------------------------------------------------------
[docs] def get_n_clusters(self): """Get current number of clusters. Returns ------- n_clusters : int Number of material phase clusters. """ return self._n_clusters
# -------------------------------------------------------------------------
[docs] def get_clustering_type(self): """Get cluster-reduced material phase adaptivity type. Returns ------- clustering_type : str Type of cluster-reduced material phase. """ return self._clustering_type
# -------------------------------------------------------------------------
[docs] def get_clustering_tree_nodes(self): """Get clustering tree nodes. Returns ------- clustering_tree_nodes : dict Clustering tree node (item, anytree.Node) associated with each material cluster (key, str). root_cluster_node : anytree.Node Clustering tree root node. """ # Output clustering tree # anytree.exporter.DotExporter(self._root_cluster_node).to_picture( # 'clustering_tree_nodes_phase_' + self._mat_phase + '.png') return self._clustering_tree_nodes, self._root_cluster_node
# -------------------------------------------------------------------------
[docs] @staticmethod def get_adaptivity_type_parameters(): """Get ACRMP mandatory and optional adaptivity type parameters. Besides returning the ACRMP mandatory and optional adaptivity type parameters, this method establishes the default values for the optional parameters. ---- Returns ---------- mandatory_parameters : dict Mandatory adaptivity type parameters (str) and associated type (item, type). optional_parameters : dict Optional adaptivity type parameters (key, str) and associated default value (item). """ # Set mandatory adaptivity type parameters mandatory_parameters = {} # Set optional adaptivity type parameters and associated default values optional_parameters = {'adapt_split_factor': 0.01, 'child_cluster_vol_fraction': 0.5, 'dynamic_split_factor_amp': 0.0, 'threshold_n_clusters': 10**6} # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Return return mandatory_parameters, optional_parameters
# -------------------------------------------------------------------------
[docs] def _set_adaptivity_type_parameters(self, adaptivity_type): """Set clustering adaptivity parameters. Parameters ---------- adaptivity_type : dict Clustering adaptivity parameters. """ # Set mandatory adaptivity type parameters pass # Set optional adaptivity type parameters self._adapt_split_factor = adaptivity_type['adapt_split_factor'] self._child_cluster_vol_fraction = \ adaptivity_type['child_cluster_vol_fraction'] self._dynamic_split_factor_amp = \ adaptivity_type['dynamic_split_factor_amp'] self._threshold_n_clusters = adaptivity_type['threshold_n_clusters'] # Set dynamic adaptive split factor if abs(self._dynamic_split_factor_amp) < 1e-10: self._is_dynamic_split_factor = False else: self._is_dynamic_split_factor = True
# -------------------------------------------------------------------------
[docs] def get_adaptive_output(self): """Get adaptivity metrics for clustering adaptivity output. Returns ------- adaptivity_output : list List containing the adaptivity metrics associated with the clustering adaptivity output file. """ # Build adaptivity output adaptivity_output = [self._n_clusters, self._adaptive_step, self.adaptive_time] # Return return adaptivity_output
# -------------------------------------------------------------------------
[docs] def reset_adaptive_parameters(self): """Reset clustering adaptive progress parameters.""" # Reset counter of adaptive clustering steps self._adaptive_step = 0 # Reset time spent in adaptive procedures self.adaptive_time = 0
# -------------------------------------------------------------------------
[docs] def update_adaptivity_type(self, adaptivity_type): """Update clustering adaptivity parameters. Parameters ---------- adaptivity_type : dict Clustering adaptivity parameters. """ # Update clustering adaptivity parameters self._adaptivity_type = adaptivity_type # Set clustering adaptivity parameters self._set_adaptivity_type_parameters(self._adaptivity_type)
# =============================================================================
[docs]class HAACRMP(ACRMP): """Hierarchical Agglomerative Adaptive Cluster-Reduced Material Phase. Attributes ---------- _clustering_type : str Type of cluster-reduced material phase. _linkage_matrix : numpy.ndarray (2d) Linkage matrix associated with the hierarchical agglomerative clustering (numpy.ndarray of shape (n_phase_voxels - 1, 4)). _cluster_node_map : dict Tree node id (item, int) associated with each cluster label (key, str). _adaptive_step : int Counter of adaptive clustering steps, with 0 associated with the base clustering. _adapt_split_factor : float Adaptive clustering split factor. The adaptive clustering split factor must be contained between 0 and 1 (included). The lower bound (0) enforces a single split, i.e., 2 new clusters, while the upper bound (1) is associated with a maximum defined number of new voxels. _threshold_n_clusters : int Threshold number of adaptive material phase number of clusters. Once this threshold is surpassed, the adaptive procedures of the adaptive material phase are deactivated. _is_dynamic_split_factor : bool True if adaptive clustering split factor is to be computed dynamically. Otherwise, the adaptive clustering split factor is always set equal to `_adapt_split_factor`. max_label : int Clustering maximum label. cluster_labels : numpy.ndarray (1d) Material phase cluster labels (numpy.ndarray[int] of shape (n_phase_voxels,)). adaptive_time : float Total amount of time (s) spent in the adaptive procedures. adaptivity_lock : bool True if the adaptive procedures are deactivated, False otherwise. Methods ------- perform_base_clustering(self, base_clustering_scheme, min_label=0) Perform HAACRMP base clustering. perform_adaptive_clustering(self, target_clusters, target_clusters_data, \ adaptive_clustering_scheme=None, \ min_label=0) Perform HAACRMP adaptive clustering step. add_to_tree_node_list(node_list, node) Add node to tree node list and sort by descending linkage distance. _check_adaptivity_lock(self) Check ACRMP adaptivity locking conditions. print_adaptive_clustering(self, adaptive_clustering_map, \ adaptive_tree_node_map) get_valid_clust_algs() Get valid clustering algorithms to compute the CRMP. get_n_clusters(self) Get current number of clusters. get_clustering_type(self) Get cluster-reduced material phase adaptivity type. get_adaptivity_type_parameters() Get ACRMP mandatory and optional adaptivity type parameters. _set_adaptivity_type_parameters(self, adaptivity_type) Set clustering adaptivity parameters. get_adaptive_output(self) Get adaptivity metrics for clustering adaptivity output. """
[docs] def __init__(self, mat_phase, cluster_data_matrix, n_clusters, adaptivity_type): """Constructor. Parameters ---------- mat_phase : str Material phase label. cluster_data_matrix: numpy.ndarray (2d) Data matrix (numpy.ndarray of shape (n_phase_voxels, n_features_dims)) containing the clustering features data to perform the material phase cluster analyses. n_clusters : int Number of material phase clusters. adaptivity_type : dict Clustering adaptivity parameters. """ self._mat_phase = mat_phase self._cluster_data_matrix = cluster_data_matrix self._clustering_type = 'adaptive' self._n_clusters = n_clusters self._adaptivity_type = adaptivity_type self._linkage_matrix = None self._cluster_node_map = None self._adaptive_step = 0 self.max_label = 0 self.cluster_labels = None self.adaptive_time = 0 self.adaptivity_lock = False # Set clustering adaptivity parameters self._set_adaptivity_type_parameters(self._adaptivity_type) # Set dynamic adaptive split factor if abs(self._dynamic_split_factor_amp) < 1e-10: self._is_dynamic_split_factor = False else: self._is_dynamic_split_factor = True
# -------------------------------------------------------------------------
[docs] def perform_base_clustering(self, base_clustering_scheme, min_label=0): """Perform HAACRMP base clustering. Parameters ---------- base_clustering_scheme : dict Prescribed base clustering scheme (item, numpy.ndarray of shape (n_clusterings, 3)) for each material phase (key, str). Each row is associated with a unique clustering characterized by a clustering algorithm (col 1, int), a list of features (col 2, list[int]) and a list of the features data matrix' indexes (col 3, list[int]). min_label : int, default=0 Minimum cluster label. """ # Get number of prescribed clusterings n_clusterings = base_clustering_scheme.shape[0] if n_clusterings > 1: raise RuntimeError('A HAACRMP only accepts a single ' 'hierarchical agglomerative prescribed ' 'clustering.') # Get number of material phase voxels n_phase_voxels = self._cluster_data_matrix.shape[0] # Get clustering algorithm and check validity clust_alg_id = str(base_clustering_scheme[0, 0]) if clust_alg_id not in self.get_valid_clust_algs(): raise RuntimeError('An invalid clustering algorithm has been ' 'prescribed.') # Get base clustering features' column indexes indexes = base_clustering_scheme[0, 2] # Get base clustering data matrix data_matrix = mop.get_condensed_matrix(self._cluster_data_matrix, list(range(n_phase_voxels)), indexes) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Perform cluster analysis cluster_analysis = clstalgs.ClusterAnalysis() cluster_labels, clust_alg, is_n_clusters_satisfied = \ cluster_analysis.get_fitted_estimator(data_matrix, clust_alg_id, self._n_clusters) # Check if prescribed number of clusters is satisfied if not is_n_clusters_satisfied: raise RuntimeError('The number of clusters (' + str(len(set(cluster_labels))) + ') obtained is different from the ' 'prescribed number of clusters (' + str(self._n_clusters) + ').') # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Set cluster labels self.cluster_labels = cluster_labels # Get hierarchical agglomerative base clustering linkage matrix self._linkage_matrix = clust_alg.get_linkage_matrix() # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update cluster labels self.cluster_labels, self.max_label = \ self._update_cluster_labels(self.cluster_labels, min_label)
# -------------------------------------------------------------------------
[docs] def perform_adaptive_clustering(self, target_clusters, target_clusters_data, adaptive_clustering_scheme=None, min_label=0): """Perform HAACRMP adaptive clustering step. Refine the provided target clusters by splitting them according to the hierarchical agglomerative tree, prioritizing child nodes by descending order of linkage distance. ---- Parameters ---------- target_clusters : list[int] List with the labels (int) of clusters to be adapted. target_clusters_data : dict For each target cluster (key, str), store dictionary (item, dict) containing cluster associated parameters relevant for the adaptive procedures. adaptive_clustering_scheme : dict Prescribed adaptive clustering scheme (item, numpy.ndarray of shape (n_clusterings, 3)) for each material phase (key, str). Each row is associated with a unique clustering characterized by a clustering algorithm (col 1, int), a list of features (col 2, list[int]) and a list of the features data matrix' indexes (col 3, list[int]). min_label : int, default=0 Minimum cluster label. Returns ------- adaptive_clustering_map : dict List of new cluster labels (item, list[int]) resulting from the adaptivity of each target cluster (key, str). adaptive_tree_node_map : dict List of new cluster tree node ids (item, list[int]) resulting from the split of each target cluster tree node id (key, str). Validation purposes only (not returned otherwise). """ # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Check for duplicated target clusters if len(target_clusters) != len(np.unique(target_clusters)): raise RuntimeError('List of target clusters contains duplicated ' 'labels.') # Check for unexistent target clusters for target_cluster in target_clusters: if target_cluster not in self.cluster_labels: raise RuntimeError('Target cluster ' + str(target_cluster) + ' does not exist in material phase ' + str(self._mat_phase)) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ init_time = time.time() # Increment adaptive clustering step counter self._adaptive_step += 1 # Initialize adaptive clustering mapping dictionary adaptive_clustering_map = {} # Initialize adaptive tree node mapping dictionary (validation purposes # only) adaptive_tree_node_map = {} # Get current hierarchical agglomerative clustering cluster_labels = copy.deepcopy(self.cluster_labels) # Initialize new cluster label new_cluster_label = min_label # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Get cluster labels (conversion to int32 is required to avoid raising # a TypeError in scipy hierarchy leaders function) labels = cluster_labels.astype('int32') # Convert hierarchical agglomerative base clustering linkage matrix # into tree object rootnode, nodelist = sciclst.to_tree(self._linkage_matrix, rd=True) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Build initial cluster-node mapping between cluster labels and tree # nodes associated with the hierarchical agglomerative base clustering if self._cluster_node_map is None: # Get root nodes of hierarchical clustering corresponding to an # horizontal cut defined by a flat clustering assignment vector. # L contains the tree nodes ids while M contains the corresponding # cluster labels L, M = sciclst.leaders(self._linkage_matrix, labels) # Build initial cluster-node mapping self._cluster_node_map = dict(zip([str(x) for x in M], L)) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Loop over target clusters for i in range(len(target_clusters)): # Get target cluster label target_cluster = target_clusters[i] # Get target cluster tree node instance target_node = nodelist[self._cluster_node_map[str(target_cluster)]] # Get total number of leaf nodes associated with target node. If # target node is a leaf itself (not splitable), skip to the next # target cluster if target_node.is_leaf(): continue # else: # n_leaves = target_node.get_count() # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Get referece adaptive clustering split factor ref_split_factor = self._adapt_split_factor # Get target cluster dynamic split factor ability is_cluster_dynamic_split_factor = target_clusters_data[ str(target_cluster)]['is_dynamic_split_factor'] # Set adaptive clustering split factor if self._is_dynamic_split_factor \ and is_cluster_dynamic_split_factor: # Get adaptive trigger ratio and magnitude adapt_trigger_ratio = target_clusters_data[ str(target_cluster)]['adapt_trigger_ratio'] magnitude = target_clusters_data[ str(target_cluster)]['max_magnitude'] # Compute dynamic adaptive clustering split factor adapt_split_factor = super()._dynamic_split_factor( ref_split_factor, adapt_trigger_ratio, magnitude, dynamic_amp=self._dynamic_split_factor_amp) else: adapt_split_factor = ref_split_factor # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Compute total number of tree node splits, enforcing at least one # split n_splits = max( 1, int(round(adapt_split_factor*int(int( round(1.0/self._child_cluster_vol_fraction)) - 1)))) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Initialize child nodes list child_nodes = [] # Initialize child target nodes list child_target_nodes = [] # Split loop for i_split in range(n_splits): # Set node to be splitted if i_split == 0: # In the first split operation, the node to be splitted is # the target cluster tree node node_to_split = target_node else: # Get maximum linkage distance child target node and remove # it from the child target nodes list node_to_split = child_target_nodes[0] child_target_nodes.pop(0) # Loop over child target node's left and right child nodes for node in [node_to_split.get_left(), node_to_split.get_right()]: if node.is_leaf(): # Append to child nodes list if leaf node child_nodes.append(node) else: # Append to child target nodes list if non-leaf node child_target_nodes = self.add_to_tree_node_list( child_target_nodes, node) # Add remaining child target nodes to child nodes list child_nodes += child_target_nodes # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Initialize target cluster mapping adaptive_clustering_map[str(target_cluster)] = [] # Remove target cluster from cluster node mapping self._cluster_node_map.pop(str(target_cluster)) # Update target cluster mapping and flat clustering labels for node in child_nodes: # Add new cluster to target cluster mapping adaptive_clustering_map[str(target_cluster)].append( new_cluster_label) # Update flat clustering labels labels[node.pre_order()] = new_cluster_label # Update cluster-node mapping self._cluster_node_map[str(new_cluster_label)] = node.id # Increment new cluster label new_cluster_label += 1 # Update adaptive tree node mapping dictionary (validation purposes # only) adaptive_tree_node_map[str(target_node.id)] = \ [x.id for x in child_nodes] # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update RVE hierarchical agglomerative clustering self.cluster_labels = labels # Update number of material phase clusters self._n_clusters = len(set(self.cluster_labels)) # Update clustering maximum label self.max_label = max(self.cluster_labels) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Update total amount of time spent in the adaptive procedures self.adaptive_time += time.time() - init_time # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Check adaptivity threshold conditions self._check_adaptivity_lock() # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Return return adaptive_clustering_map
# -------------------------------------------------------------------------
[docs] @staticmethod def add_to_tree_node_list(node_list, node): """Add node to tree node list and sort by descending linkage distance. Parameters ---------- node_list : list[ClusterNode] List of ClusterNode instances. node : ClusterNode ClusterNode to be added to list[ClusterNode]. """ # Check parameters if not isinstance(node, sciclst.ClusterNode): raise TypeError('Node must be of type ClusterNode, not ' + str(type(node)) + '.') if any([not isinstance(node, sciclst.ClusterNode) for node in node_list]): raise TypeError('Node list can only contain elements of the type ' 'ClusterNode.') # Append tree node to node list node_list.append(node) # Sort tree node list by descending order of linkage distance node_list = sorted(node_list, reverse=True, key=lambda x: x.dist) # Return sorted tree node list return node_list
# -------------------------------------------------------------------------
[docs] def _check_adaptivity_lock(self): """Check ACRMP adaptivity locking conditions. Check conditions that may deactivate the adaptive procedures in the ACRMP. Once the ACRMP adaptivity is locked, it is treated as a SCRMP for the remainder of the problem numerical solution. """ # Check if the number of clusters threshold as been surpassed if self._n_clusters > self._threshold_n_clusters: self.adaptivity_lock = True
# -------------------------------------------------------------------------
[docs] def print_adaptive_clustering(self, adaptive_clustering_map, adaptive_tree_node_map): """Print hierarchical adaptive clustering report (validation).""" # Print report header print(3*'\n' + 'Hierarchical adaptive clustering report\n' + 80*'-' + '\n\n' + 'Material phase: ' + str(self._mat_phase)) # Print adaptive clustering adaptive step print('\nAdaptive refinement step: ', self._adaptive_step) # Print hierarchical adaptive CRVE print('\n\n' + 'Adaptive clustering: ' + '(' + str(len(np.unique(self.cluster_labels))) + ' clusters)' + '\n\n', self.cluster_labels) # Print adaptive clustering mapping print('\n\n' + 'Adaptive cluster mapping: ') for old_cluster in adaptive_clustering_map.keys(): print(' Old cluster: ' + '{:>4s}'.format(old_cluster) + ' -> ' + 'New clusters: ', adaptive_clustering_map[str(old_cluster)]) # Print adaptive tree node mapping print('\n\n' + 'Adaptive tree node mapping (validation): ') for old_node in adaptive_tree_node_map.keys(): print(' Old node: ' + '{:>4s}'.format(old_node) + ' -> ' + 'New nodes: ', adaptive_tree_node_map[str(old_node)]) # Print cluster-node mapping print('\n\n' + 'Cluster-Node mapping: ') for new_cluster in self._cluster_node_map.keys(): print(' Cluster: ' + '{:>4s}'.format(new_cluster) + ' -> ' + 'Tree node: ', self._cluster_node_map[str(new_cluster)])
# -------------------------------------------------------------------------
[docs] @staticmethod def get_valid_clust_algs(): """Get valid clustering algorithms to compute the CRMP. Returns ---------- clust_algs : list[str] Clustering algorithms identifiers (str). """ return ['3', ]
# -------------------------------------------------------------------------
[docs] def get_n_clusters(self): """Get current number of clusters. Returns ------- n_clusters : int Number of material phase clusters. """ return self._n_clusters
# -------------------------------------------------------------------------
[docs] def get_clustering_type(self): """Get cluster-reduced material phase adaptivity type. Returns ------- clustering_type : str Type of cluster-reduced material phase. """ return self._clustering_type
# -------------------------------------------------------------------------
[docs] @staticmethod def get_adaptivity_type_parameters(): """Get ACRMP mandatory and optional adaptivity type parameters. Besides returning the ACRMP mandatory and optional adaptivity type parameters, this method establishes the default values for the optional parameters. ---- Returns ---------- mandatory_parameters : dict Mandatory adaptivity type parameters (str) and associated type (item, type). optional_parameters : dict Optional adaptivity type parameters (key, str) and associated default value (item). """ # Set mandatory adaptivity type parameters mandatory_parameters = {} # Set optional adaptivity type parameters and associated default values optional_parameters = {'adapt_split_factor': 0.01, 'child_cluster_vol_fraction': 0.5, 'dynamic_split_factor_amp': 0.0, 'threshold_n_clusters': 10**6} # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Return return mandatory_parameters, optional_parameters
# -------------------------------------------------------------------------
[docs] def _set_adaptivity_type_parameters(self, adaptivity_type): """Set clustering adaptivity parameters. Parameters ---------- adaptivity_type : dict Clustering adaptivity parameters. """ # Set mandatory adaptivity type parameters pass # Set optional adaptivity type parameters self._adapt_split_factor = adaptivity_type['adapt_split_factor'] self._child_cluster_vol_fraction = \ adaptivity_type['child_cluster_vol_fraction'] self._dynamic_split_factor_amp = \ adaptivity_type['dynamic_split_factor_amp'] self._threshold_n_clusters = adaptivity_type['threshold_n_clusters']
# -------------------------------------------------------------------------
[docs] def get_adaptive_output(self): """Get adaptivity metrics for clustering adaptivity output. Returns ------- adaptivity_output : list List containing the adaptivity metrics associated with the clustering adaptivity output file. """ # Build adaptivity output adaptivity_output = [self._n_clusters, self._adaptive_step, self.adaptive_time] # Return return adaptivity_output