KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _remove_cluster method in localstack

Best Python code snippet using localstack_python

aggregate.py

Source:aggregate.py

...76        return self.active_pairs77    '''78    section for merge nodes79    '''80    def _remove_cluster(self, cluster):81        '''82        :param cluster: cluster(node) to be deleted83        :return:84        '''85        # cluster list86        self.clusters.remove(cluster)87        # similar pairs88        delete_set = set()89        for key_tuple in self.similarity_scores:90            if self._contains(key_tuple, cluster):91                delete_set.add(key_tuple)92        for key_tuple in delete_set:93            del self.similarity_scores[key_tuple]94        #distance map95        del self.distance_map[cluster]96        for nd in self.distance_map:97            self.distance_map[nd] = list(filter(lambda x: x[0] != cluster, self.distance_map[nd]))98        #active_pairs will be updated after adding the cluster got from merge99    def _add_cluster(self, cluster):100        '''101        :param cluster: after merging, new cluster comes102        :return:103        '''104        # cluster list105        self.clusters.append(cluster)106        # similar scores107        for node in self.clusters:108            pair = (node, cluster)109            dist = AggregateLocalSL._similarity_between_coefs(node.get_coefs(), cluster.get_coefs())110            self.similarity_scores[pair] = dist111            pair = (cluster, node)112            self.similarity_scores[pair] = dist113        # distance map114        for node in self.distance_map:115            self.distance_map[node].append((cluster, np.linalg.norm(cluster.get_centroid()-node.get_centroid())))116            self.distance_map[node] = sorted(self.distance_map[node], key=lambda x: x[1])117        self.distance_map[cluster] = []118        for node in self.clusters:119            if node == cluster:120                continue121            self.distance_map[cluster].append((node, np.linalg.norm(cluster.get_centroid()-node.get_centroid())))122        self.distance_map[node] = sorted(self.distance_map[node], key=lambda x: x[1])123    # active_pairs124        self._update_active_pairs()125    def _aggregate_two_nodes(self, nd1, nd2):126        merged_index = nd1.get_index()127        merged_index.extend(nd2.get_index())128        merged_node = Node(merged_index)129        self._update_centroid(merged_node)130        coefs = self._extract_coefs(merged_node.get_centroid())131        merged_node.update_coefs(coefs)132        self._remove_cluster(nd1)133        self._remove_cluster(nd2)134        self._add_cluster(merged_node)135    def _find_pair_to_merge(self):136        '''137        find the pair in active pairs with most similar coefs138        :return: None if all large errors, else the pair139        '''140        final_s_score = float('-inf')141        final_pair = None142        print(len(self.active_pairs))143        for pair in self.active_pairs:144            #print(pair)145            if self.similarity_scores[pair] == constant.LARGEERROR:146                print("yes")147                continue...

simulation.py

Source:simulation.py

...54         Remove linkage if asked in configuration55         :param pid2source2pages:56         :return:57         """58         self._suspended_linkages = _remove_cluster(59             pid2source2pages, _config_.get_linkage_removal_rule(), _config_.get_percentage_linkage_kept())60    def check_and_add_product_page(self, bdsa_data, page:Page, key2values:dict, pids):61        """62        Check whether page can be added or not, according to simulation configuration.63        Currently the only case is when do_delete_pages_without_linkage is activated, and page64        is not associated with any product ID.65        If true, add it immediately. Otherwise, put the operation in "delay" and add them66        :param pids:67        :param product_page_adder:68        :return:69        """70        if not _config_.do_delete_pages_without_linkage() or len(pids) > 0:71            bdsa_data.add_page(page, key2values, pids)72        else:73            self._suspended_pages.append(SuspendedPage(page, pids, key2values))74    def do_add_source(self, source):75        """76        Check if a source has to be added. According to configuration, one may remove random sources, or sources77        with most or less linkage78        :param source:79        :return:80        """81        if _config_.get_source_removal_rule() == SourceRemovalRule.NONE:82            return True83        return source in self.sources_to_keep84    def add_wrong_linkages(self, bdsa_data):85        percent_added = _config_.get_ratio_linkage_added()86        if percent_added > 0:87            # Save old linkages so we can retrieve them later88            self._old_linkage = copy.deepcopy(bdsa_data.pid2source2pages)89            # This is current nb of pairs of specifications in linkage90            current_pairs = sum(sum(len(pgs[0]) * len(pgs[1]) for pgs in itertools.combinations(source2pages.values(), 2))91                                for source2pages in bdsa_data.pid2source2pages.values())92            new_expected_pairs = current_pairs * percent_added93            new_pairs = 094            nb_iterations = 095            pids = list(bdsa_data.pid2source2pages.keys())96            pages = list(bdsa_data.page2sa2value.keys())97            # Add new IDs to pages until we reach enough linkages OR too many iterations98            while new_pairs < new_expected_pairs and nb_iterations < len(bdsa_data.page2sa2value.keys()) * 5:99                pid = random.choice(pids)100                page = random.choice(pages)101                source2pages_pid = bdsa_data.pid2source2pages[pid]102                if page in source2pages_pid[page.source]:103                    continue104                new_pairs += sum(len(pages) for source, pages in source2pages_pid.items() if source != page.source)105                source2pages_pid[page.source].add(page)106def generate_random_pairs(max):107    """108    Generate random pair of number without repetitions109    :param max: 110    :return: 111    """112    already_seen = set()113    size = 0114    while size < max * (max - 1) / 2:115        res = tuple(sorted([random.randint(0, max), random.randint(0, max)]))116        if res[0] != res[1] and res not in already_seen:117            size += 1118            already_seen.add(res)119            yield res120        # def all_pairs(pids, pages, choice_function):121#     for i in pids:122#         for j in pages:123#             if choice_function(i,j):124#                 yield (i,j)125#126# def choose_pairs(pids, pages, sample):127#     total_pairs = len(pids) * len(pages)128#     random.sample()129def _remove_cluster(linkage_clusters: dict, removal_rule: LinkageRemovalRule, ratio_linkage_keep: float):130    """131    Remove linkage cluster to simulate different conditions of input132    :return:133    >>> linkage_clusters_orig = {10: {1: [1,2,3], 2: [2,3]}, 20: {1:[1,2], 2: [3]}, 30: {1: [1,2,3,4]}}134    >>> linkage_clusters_copy = linkage_clusters_orig.copy()135    >>> _remove_cluster(linkage_clusters_copy, LinkageRemovalRule.KEEP_SMALL_CLUSTERS, 0.7)136    {10: {1: [1, 2, 3], 2: [2, 3]}}137    >>> linkage_clusters_copy138    {20: {1: [1, 2], 2: [3]}, 30: {1: [1, 2, 3, 4]}}139    >>> linkage_clusters_copy = linkage_clusters_orig.copy()140    >>> _remove_cluster(linkage_clusters_copy,  LinkageRemovalRule.KEEP_BIG_CLUSTERS, 0.7)141    {20: {1: [1, 2], 2: [3]}}142    >>> linkage_clusters_copy143    {10: {1: [1, 2, 3], 2: [2, 3]}, 30: {1: [1, 2, 3, 4]}}144    >>> linkage_clusters_copy = linkage_clusters_orig.copy()145    >>> _remove_cluster(linkage_clusters_copy, LinkageRemovalRule.KEEP_SMALL_CLUSTERS, 0.4)146    {10: {1: [1, 2, 3], 2: [2, 3]}, 30: {1: [1, 2, 3, 4]}}147    >>> linkage_clusters_copy148    {20: {1: [1, 2], 2: [3]}}149    >>> linkage_clusters_copy = linkage_clusters_orig.copy()150    >>> _remove_cluster(linkage_clusters_copy, LinkageRemovalRule.KEEP_BIG_CLUSTERS, 0.4)151    {20: {1: [1, 2], 2: [3]}, 30: {1: [1, 2, 3, 4]}}152    >>> linkage_clusters_copy153    {10: {1: [1, 2, 3], 2: [2, 3]}}154    >>> linkage_clusters_copy = linkage_clusters_orig.copy()155    >>> sorted(_remove_cluster(linkage_clusters_copy, LinkageRemovalRule.RANDOM, 0.1).items())156    [(10, {1: [1, 2, 3], 2: [2, 3]}), (20, {1: [1, 2], 2: [3]}), (30, {1: [1, 2, 3, 4]})]157    >>> linkage_clusters_copy158    {}159    >>> linkage_clusters_copy = linkage_clusters_orig.copy()160    >>> ignore = _remove_cluster(linkage_clusters_copy, LinkageRemovalRule.RANDOM, 0.4)161    >>> len(linkage_clusters_copy)162    1163    """164    total_linked = sum(len(pages) for source2pages in linkage_clusters.values() for pages in source2pages.values())165    elements_to_remove = round(total_linked * (1 - ratio_linkage_keep))166    removed = {}167    nb_removed = 0168    pids_to_remove = []169    if removal_rule in {LinkageRemovalRule.RANDOM, LinkageRemovalRule.KEEP_SMALL_CLUSTERS, LinkageRemovalRule.KEEP_BIG_CLUSTERS}:170        if removal_rule == LinkageRemovalRule.RANDOM:171            pids_sorted = list(linkage_clusters.keys())172            random.shuffle(pids_sorted)173        elif removal_rule in {LinkageRemovalRule.KEEP_SMALL_CLUSTERS, LinkageRemovalRule.KEEP_BIG_CLUSTERS}:174            remove_biggest_first = removal_rule == LinkageRemovalRule.KEEP_SMALL_CLUSTERS...

remove_cluster.py

Source:remove_cluster.py

...21    def __init__(self, **data: Any) -> None:22        super().__init__(23            command_name=CommandName.REMOVE_CLUSTER, version=1, **data24        )25    def _remove_cluster(self, study_data: FileStudyTreeConfig) -> None:26        study_data.areas[self.area_id].thermals = [27            cluster28            for cluster in study_data.areas[self.area_id].thermals29            if cluster.id != self.cluster_id.lower()30        ]31    def _apply_config(32        self, study_data_config: FileStudyTreeConfig33    ) -> Tuple[CommandOutput, Dict[str, Any]]:34        if self.area_id not in study_data_config.areas:35            return (36                CommandOutput(37                    status=False,38                    message=f"Area '{self.area_id}' does not exist",39                ),40                dict(),41            )42        if (43            len(44                [45                    cluster46                    for cluster in study_data_config.areas[47                        self.area_id48                    ].thermals49                    if cluster.id == self.cluster_id50                ]51            )52            == 053        ):54            return (55                CommandOutput(56                    status=False,57                    message=f"Cluster '{self.cluster_id}' does not exist",58                ),59                dict(),60            )61        self._remove_cluster(study_data_config)62        remove_area_cluster_from_binding_constraints(63            study_data_config, self.area_id, self.cluster_id64        )65        return (66            CommandOutput(67                status=True,68                message=f"Cluster '{self.cluster_id}' removed from area '{self.area_id}'",69            ),70            dict(),71        )72    def _apply(self, study_data: FileStudy) -> CommandOutput:73        if self.area_id not in study_data.config.areas:74            return CommandOutput(75                status=False,76                message=f"Area '{self.area_id}' does not exist",77            )78        if (79            len(80                [81                    cluster82                    for cluster in study_data.config.areas[83                        self.area_id84                    ].thermals85                    if cluster.id == self.cluster_id86                ]87            )88            == 089        ):90            return CommandOutput(91                status=False,92                message=f"Cluster '{self.cluster_id}' does not exist",93            )94        study_data.tree.delete(95            [96                "input",97                "thermal",98                "clusters",99                self.area_id,100                "list",101                self.cluster_id,102            ]103        )104        study_data.tree.delete(105            [106                "input",107                "thermal",108                "prepro",109                self.area_id,110                self.cluster_id,111            ]112        )113        study_data.tree.delete(114            [115                "input",116                "thermal",117                "series",118                self.area_id,119                self.cluster_id,120            ]121        )122        self._remove_cluster(study_data.config)123        self._remove_cluster_from_binding_constraints(study_data)124        return CommandOutput(125            status=True,126            message=f"Cluster '{self.cluster_id}' removed from area '{self.area_id}'",127        )128    def to_dto(self) -> CommandDTO:129        return CommandDTO(130            action=CommandName.REMOVE_CLUSTER.value,131            args={"area_id": self.area_id, "cluster_id": self.cluster_id},132        )133    def match_signature(self) -> str:134        return str(135            self.command_name.value136            + MATCH_SIGNATURE_SEPARATOR...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.