From 3443e979056557b4dc94edd042d331c3336dddf0 Mon Sep 17 00:00:00 2001 From: Narges Abbasi Date: Thu, 26 Feb 2026 16:00:43 +0330 Subject: [PATCH 1/2] update metrics computation --- codart/metrics/complexity.py | 152 ++++++++++---- .../metrics/design_testability_prediction2.py | 67 +++--- codart/metrics/testability_prediction2.py | 194 ++++++++++++------ 3 files changed, 281 insertions(+), 132 deletions(-) diff --git a/codart/metrics/complexity.py b/codart/metrics/complexity.py index 5121c160..97b75626 100644 --- a/codart/metrics/complexity.py +++ b/codart/metrics/complexity.py @@ -1,7 +1,11 @@ +import os import time +import traceback from design_4_testability.class_diagram_extraction.class_diagram import ClassDiagram from design_4_testability.config import test_class_diagram +from design_4_testability import config +from design_4_testability.config2 import projects_info import networkx as nx import csv @@ -18,35 +22,58 @@ def __init__(self, class_diagram): for node in candidate_nodes: self.inheritance_complexity_dic[node] = self.__calculate_inheritance_complexity(node) + # NEW FUNCTION — Fast detection whether a use_def relation exists + def has_use_def_relation(self, source, target): + """ + Quickly returns True if there exists ANY path between source → target + that includes at least one use_def edge. + """ + + # Check each use_def edge and see if it can participate in a path + for u, v, data in self.CDG.edges(data=True): + if data.get("relation_type") == "use_def": + # Is source -> u reachable AND v -> target reachable? + if nx.has_path(self.class_diagram, source, u) and nx.has_path(self.class_diagram, v, target): + return True + return False + def calculate_interaction_complexity(self, source, target): + + # Case 1: No path at all → None --- + if not nx.has_path(self.class_diagram, source, target): + return None + + # Case 2: Path exists, but no use_def on any path → complexity = 1 --- + if not self.has_use_def_relation(source, target): + return 1 + + # Case 3: There is use_def on at least one path → compute formula --- complexity = 1 - has_path = False for path in nx.all_simple_paths(self.class_diagram, source=source, target=target): - has_path = True complexity *= self.__calculate_path_complexity(path) - if not has_path: - complexity = None + return complexity def __calculate_path_complexity(self, path): complexity = 1 for i in range(len(path) - 1): - if self.CDG[path[i]][path[i+1]]['relation_type'] == 'use_def': - if path[i] in self.inheritance_complexity_dic: - complexity *= self.inheritance_complexity_dic[path[i]] + u, v = path[i], path[i + 1] + if self.CDG.has_edge(u, v) and self.CDG[u][v].get('relation_type') == 'use_def': + if u in self.inheritance_complexity_dic: + complexity *= self.inheritance_complexity_dic[u] return complexity def __calculate_inheritance_complexity(self, node): complexity = 0 - stack = [] - stack.append(node) + stack = [node] + # stack.append(node) + depth_dic = {node: 1} - depth_dic = {node:1} - while stack != []: + while stack: current_node = stack.pop() is_leave = True for neighbor in self.CDG[current_node]: - if (current_node in self.CDG[neighbor]): + if current_node in self.CDG[neighbor]: if self.CDG[current_node][neighbor]['relation_type'] == 'child' and self.CDG[neighbor][current_node]['relation_type'] == 'parent': is_leave = False stack.append(neighbor) @@ -58,23 +85,25 @@ def __calculate_inheritance_complexity(self, node): def __find_inheritance_candidates(self): candidates = set() - for edge in self.CDG.edges: - if self.CDG.edges[edge]['relation_type'] == 'parent': - candidates.add(edge[1]) + for u, v, d in self.CDG.edges(data=True): + if d.get('relation_type') == 'parent': + candidates.add(v) return candidates def get_matrix(self): start_time = time.time() - node_list = list(self.CDG.nodes) + node_list = sorted(self.CDG.nodes) no_nodes = len(node_list) - node_list.sort() matrix = [] for s in range(no_nodes): matrix.append([]) for d in range(no_nodes): - if self.CDG.nodes[node_list[s]]['type'] == "class" and self.CDG.nodes[node_list[d]]['type'] == "class": - complexity = self.calculate_interaction_complexity(node_list[s], node_list[d]) + src = node_list[s] + dst = node_list[d] + + if self.CDG.nodes[src]['type'] == "class" and self.CDG.nodes[dst]['type'] == "class": + complexity = self.calculate_interaction_complexity(src, dst) matrix[s].append(complexity) else: matrix[s].append(None) @@ -90,16 +119,11 @@ def get_avg_of_matrix(matrix): if j is not None: n += 1 s += j - return s / n + return s / n if n > 0 else 0 @staticmethod def get_sum_of_matrix(matrix): - s = 0 - for i in matrix: - for j in i: - if j is not None: - s += j - return s + return sum(val for row in matrix for val in row if val is not None) def save_csv(self, path): node_list = list(self.CDG.nodes) @@ -109,25 +133,75 @@ def save_csv(self, path): with open(path, 'w', encoding='UTF8') as f: writer = csv.writer(f) + writer.writerow(header) # write the header - no_row = 0 writer.writerow(header) for s in range(no_nodes): for d in range(no_nodes): - print(s, d) - if self.CDG.nodes[node_list[s]]['type'] == "class" and self.CDG.nodes[node_list[d]][ - 'type'] == "class": - complexity = self.calculate_interaction_complexity(str(node_list[s]), str(node_list[d])) - no_row += 1 - -if __name__ == "__main__": - cd = ClassDiagram(java_project_address='', base_dirs='', files=[], index_dic={}) - cd.class_diagram_graph = test_class_diagram - cd.show(cd.class_diagram_graph) - c = Complexity(cd) - print(c.calculate_interaction_complexity(8, 7)) + src = node_list[s] + dst = node_list[d] + if self.CDG.nodes[src]['type'] == "class" and self.CDG.nodes[dst]['type'] == "class": + complexity = self.calculate_interaction_complexity(src, dst) + writer.writerow([src, dst, complexity]) +if __name__ == "__main__": + output_csv = "results_complexity.csv" + if not os.path.exists(output_csv): + with open(output_csv, mode='w', newline='', encoding='utf-8') as f: + writer = csv.writer(f) + writer.writerow(["Project Name", "Address", "Number of Nodes", "Sum of Complexities", "Average Complexity"]) + + for java_project in projects_info: + java_project_address = projects_info[java_project]['path'] + base_dirs = projects_info[java_project]['base_dirs'] + + print(f'\n=== Running project: {java_project} ===') + # print('Project address:', java_project_address) + # print('Base dirs:', base_dirs) + + try: + cd = ClassDiagram(java_project_address=java_project_address, + base_dirs=base_dirs, + files=[], + index_dic={}) + cd.make_class_diagram() + + c = Complexity(cd) + node_list = list(cd.class_diagram_graph.nodes) + node_count = len(node_list) + + total_complexity = 0 + + for source in node_list: + for target in node_list: + if source != target: + complexity = c.calculate_interaction_complexity(source, target) + if complexity is not None: + total_complexity += complexity + print(f"Interaction Complexity between {source} and {target}: {complexity}") + else: + print(f"Warning: None value between {source} and {target}") + + average_complexity = total_complexity / node_count if node_count > 0 else 0 + + print(f"\n Project: {java_project}") + print(f"Nodes: {node_count}") + print(f"Total Complexity: {total_complexity}") + print(f"Average Complexity: {average_complexity:.2f}") + + with open(output_csv, mode='a', newline='', encoding='utf-8') as f: + writer = csv.writer(f) + writer.writerow([ + java_project, + java_project_address, + node_count, + total_complexity, + round(average_complexity, 2) + ]) + except Exception as e: + print(f"Error processing project {java_project}: {e}") + traceback.print_exc() diff --git a/codart/metrics/design_testability_prediction2.py b/codart/metrics/design_testability_prediction2.py index ff1da83d..ccbf00b1 100644 --- a/codart/metrics/design_testability_prediction2.py +++ b/codart/metrics/design_testability_prediction2.py @@ -23,11 +23,15 @@ __author__ = 'Morteza Zakeri' import os +import sys +import gc + import pandas as pd import joblib from joblib import Parallel, delayed import time - +os.add_dll_directory("C:\\Program Files\\SciTools\\bin\\pc-win64\\") +sys.path.append("C:\\Program Files\\SciTools\\bin\\pc-win64\\python") import understand as und from codart.metrics import metrics_names @@ -35,11 +39,11 @@ scaler1 = joblib.load( os.path.join(os.path.dirname(__file__), - '../test_effectiveness/sklearn_models_nodes_regress/DS07710_scaler.joblib') + 'C:/Users/98910/Desktop/dataset/DS07710_scaler.joblib') ) model5 = joblib.load( os.path.join(os.path.dirname(__file__), - '../test_effectiveness/sklearn_models_nodes_regress/VR1_DS7.joblib') + 'C:/Users/98910/Desktop/dataset/VoR1_DS2.joblib') ) @@ -274,10 +278,11 @@ def compute_java_class_metrics2(cls, db=None, entity=None): return class_metrics -def do(class_entity_long_name, project_db_path): - import understand as und - db = und.open(project_db_path) +def do(class_entity_long_name, db): class_entity = UnderstandUtility.get_class_entity_by_name(class_name=class_entity_long_name, db=db) + if class_entity is None: + return None + one_class_metrics_value = [class_entity.longname()] # print('Calculating package metrics') @@ -304,8 +309,6 @@ def do(class_entity_long_name, project_db_path): one_class_metrics_value.extend([package_metrics_dict[metric_name] for metric_name in TestabilityMetrics.get_package_metrics_names()]) - db.close() - del db # print(one_class_metrics_value) # quit() return one_class_metrics_value @@ -329,29 +332,27 @@ def compute_metrics_by_class_list(cls, project_db_path, n_jobs): # class_entities = cls.read_project_classes(db=db, classes_names_list=class_list, ) # print(project_db_path) db = und.open(project_db_path) - class_list = UnderstandUtility.get_project_classes_longnames_java(db=db) - db.close() - # del db - - if n_jobs == 0: # Sequential computing - res = [do(class_entity_long_name, project_db_path) for class_entity_long_name in class_list] - else: # Parallel computing - res = Parallel(n_jobs=n_jobs, )( - delayed(do)(class_entity_long_name, project_db_path) for class_entity_long_name in class_list - ) - res = list(filter(None, res)) - - columns = ['Class'] - columns.extend(TestabilityMetrics.get_all_primary_metrics_names()) - # print('*' * 50) - # print(len(columns), columns) - # print('*' * 50) - - df = pd.DataFrame(data=res, columns=columns) - # print('df for class {0} with shape {1}'.format(project_name, df.shape)) - # df.to_csv(csv_path + project_name + '.csv', index=False) - # print(df) - return df + try: + class_list = UnderstandUtility.get_project_classes_longnames_java(db=db) + + results = [] + for i, class_name in enumerate(class_list): + if i % 100 == 0: + print(f"[INFO] Processed {i}/{len(class_list)} classes...") + gc.collect() # <-- CRITICAL + + res = do(class_name, db) + if res is not None: + results.append(res) + + columns = ['Class'] + columns.extend(TestabilityMetrics.get_all_primary_metrics_names()) + df = pd.DataFrame(data=results, columns=columns) + return df + + finally: + db.close() + gc.collect() class TestabilityModel: @@ -409,8 +410,8 @@ def main(project_db_path, initial_value=1.0, verbose=False, log_path=None): # Test module if __name__ == '__main__': - db_path_ = r'E:/LSSDS/CodART/Experimental1/udbs/jvlt-1.3.2.und' # This path should be replaced for each project - project_name_ = 'jvlt-1.3.2' + db_path_ = r'F:\benchmarks-proposal\java-corpus\bcel-5.2\bcel-5.2.und' # This path should be replaced for each project + project_name_ = 'bcel-5.2' log_path_ = os.path.join(os.path.dirname(__file__), project_name_ + '_testability_s2.csv') diff --git a/codart/metrics/testability_prediction2.py b/codart/metrics/testability_prediction2.py index 807ef9e9..672876b1 100644 --- a/codart/metrics/testability_prediction2.py +++ b/codart/metrics/testability_prediction2.py @@ -23,18 +23,24 @@ __author__ = 'Morteza Zakeri' import os +import sys +import gc +import traceback +import time + import pandas as pd import joblib from joblib import Parallel, delayed - +os.add_dll_directory("C:\\Program Files\\SciTools\\bin\\pc-win64\\") +sys.path.append("C:\\Program Files\\SciTools\\bin\\pc-win64\\python") import understand as und from codart import config from codart.metrics import metrics_names from codart.metrics.metrics_coverability import UnderstandUtility -# scaler1 = joblib.load(os.path.join(os.path.dirname(__file__), 'data_model/DS07510.joblib')) -# model5 = joblib.load(os.path.join(os.path.dirname(__file__), 'data_model/VR1_DS5.joblib')) +scaler1 = joblib.load(os.path.join(os.path.dirname(__file__), 'data_model/DS07510.joblib')) +model5 = joblib.load(os.path.join(os.path.dirname(__file__), 'data_model/VR1_DS5.joblib')) # model_branch = joblib.load(os.path.join(os.path.dirname(__file__), 'sklearn_models6/VR6_DS5_branch.joblib')) # model_line = joblib.load(os.path.join(os.path.dirname(__file__), 'sklearn_models6/VR6_DS5_line.joblib')) @@ -314,41 +320,73 @@ def compute_java_class_metrics2(cls, db=None, entity=None): return class_metrics -def do(class_entity_long_name, project_db_path): - import understand as und - db = und.open(project_db_path) - class_entity = UnderstandUtility.get_class_entity_by_name(class_name=class_entity_long_name, db=db) - one_class_metrics_value = [class_entity.longname()] +def do_with_db(class_entity_long_name, db): + """Process a single class using an already-open Understand DB (sequential mode).""" + try: + class_entity = UnderstandUtility.get_class_entity_by_name(class_name=class_entity_long_name, db=db) + if class_entity is None: + return None - # print('Calculating package metrics') - package_metrics_dict = TestabilityMetrics.compute_java_package_metrics(db=db, entity=class_entity) - if package_metrics_dict is None or len(package_metrics_dict) == 0: - return None + one_class_metrics_value = [class_entity.longname()] - # print('Calculating class lexicon metrics') - class_lexicon_metrics_dict = TestabilityMetrics.compute_java_class_metrics_lexicon(entity=class_entity) - if class_lexicon_metrics_dict is None or len(class_lexicon_metrics_dict) == 0: - return None + package_metrics_dict = TestabilityMetrics.compute_java_package_metrics(db=db, entity=class_entity) + if not package_metrics_dict: + return None + + # Lexicon metrics may be expensive; still compute if you need them + class_lexicon_metrics_dict = TestabilityMetrics.compute_java_class_metrics_lexicon(entity=class_entity) + if class_lexicon_metrics_dict is None: + return None + + class_ordinary_metrics_dict = TestabilityMetrics.compute_java_class_metrics2(db=db, entity=class_entity) + if class_ordinary_metrics_dict is None: + return None - # print('Calculating class ordinary metrics') - class_ordinary_metrics_dict = TestabilityMetrics.compute_java_class_metrics2(db=db, entity=class_entity) - if class_ordinary_metrics_dict is None or len(class_ordinary_metrics_dict) == 0: + one_class_metrics_value.extend([package_metrics_dict[metric_name] for + metric_name in TestabilityMetrics.get_package_metrics_names()]) + + one_class_metrics_value.extend([class_lexicon_metrics_dict[metric_name] for + metric_name in TestabilityMetrics.get_class_lexicon_metrics_names()]) + + one_class_metrics_value.extend([class_ordinary_metrics_dict[metric_name] for + metric_name in TestabilityMetrics.get_class_ordinary_metrics_names()]) + + return one_class_metrics_value + + except Exception: + traceback.print_exc() return None + finally: + # Help Python release references to Understand objects quickly + try: + del class_entity + except Exception: + pass + gc.collect() - one_class_metrics_value.extend([package_metrics_dict[metric_name] for - metric_name in TestabilityMetrics.get_package_metrics_names()]) - one_class_metrics_value.extend([class_lexicon_metrics_dict[metric_name] for - metric_name in TestabilityMetrics.get_class_lexicon_metrics_names()]) +def do_worker(class_entity_long_name, project_db_path): + """ + Worker used in parallel mode. Each worker opens/closes the DB itself. + This avoids sharing Understand objects between processes. + """ + db = None + try: + try: + db = und.open(project_db_path, readonly=True, cache=False) + except TypeError: + db = und.open(project_db_path) - one_class_metrics_value.extend([class_ordinary_metrics_dict[metric_name] for - metric_name in TestabilityMetrics.get_class_ordinary_metrics_names()]) + return do_with_db(class_entity_long_name, db) - db.close() - del db - # print(one_class_metrics_value) - # quit() - return one_class_metrics_value + finally: + try: + if db: + db.close() + except Exception: + pass + del db + gc.collect() # ------------------------------------------------------------------------ @@ -362,31 +400,64 @@ class PreProcess: @classmethod def compute_metrics_by_class_list(cls, project_db_path, n_jobs): """ - - + Safe compute: sequential (n_jobs==0) opens DB once and reuse it. + Parallel (n_jobs>0) uses do_worker where each process opens its DB. """ - - # class_entities = cls.read_project_classes(db=db, classes_names_list=class_list, ) - # print(project_db_path) - db = und.open(project_db_path) - class_list = UnderstandUtility.get_project_classes_longnames_java(db=db) - db.close() - # del db - - if n_jobs == 0: # Sequential computing - res = [do(class_entity_long_name, project_db_path) for class_entity_long_name in class_list] - else: # Parallel computing - res = Parallel(n_jobs=n_jobs, )( - delayed(do)(class_entity_long_name, project_db_path) for class_entity_long_name in class_list + # First open DB to get class list (readonly + cache=False if available) + try: + db = und.open(project_db_path, readonly=True, cache=False) + except TypeError: + db = und.open(project_db_path) + + try: + class_list = UnderstandUtility.get_project_classes_longnames_java(db=db) + finally: + try: + db.close() + except Exception: + pass + del db + gc.collect() + + if not class_list: + return pd.DataFrame(columns=['Class'] + TestabilityMetrics.get_all_primary_metrics_names()) + + if n_jobs == 0: + # Sequential: open DB once and reuse + try: + db_seq = und.open(project_db_path, readonly=True, cache=False) + except TypeError: + db_seq = und.open(project_db_path) + + results = [] + try: + total = len(class_list) + for i, class_entity_long_name in enumerate(class_list, start=1): + if i % 100 == 0: + print(f"[INFO] Processed {i}/{total} classes - running gc.collect()") + gc.collect() + + r = do_with_db(class_entity_long_name, db_seq) + if r: + results.append(r) + finally: + try: + db_seq.close() + except Exception: + pass + del db_seq + gc.collect() + else: + # Parallel: each worker handles opening/closing DB + results = Parallel(n_jobs=n_jobs)( + delayed(do_worker)(class_entity_long_name, project_db_path) for class_entity_long_name in class_list ) - res = list(filter(None, res)) + # results may contain None entries + results = list(filter(None, results)) columns = ['Class'] columns.extend(TestabilityMetrics.get_all_primary_metrics_names()) - df = pd.DataFrame(data=res, columns=columns) - # print('df for class {0} with shape {1}'.format(project_name, df.shape)) - # df.to_csv(csv_path + project_name + '.csv', index=False) - # print(df) + df = pd.DataFrame(data=results, columns=columns) return df @@ -400,8 +471,8 @@ class TestabilityModel: def __init__(self, ): self.scaler = scaler1 self.model = model5 - self.model_branch = model_branch - self.model_line = model_line + # self.model_branch = model_branch + # self.model_line = model_line def inference(self, df_predict_data=None, verbose=False, log_path=None): df_predict_data = df_predict_data.fillna(0) @@ -409,14 +480,14 @@ def inference(self, df_predict_data=None, verbose=False, log_path=None): X_test = self.scaler.transform(X_test1) y_pred = self.model.predict(X_test) - y_pred_branch = self.model_branch.predict(X_test) - y_pred_line = self.model_line.predict(X_test) + # y_pred_branch = self.model_branch.predict(X_test) + # y_pred_line = self.model_line.predict(X_test) df_new = pd.DataFrame(df_predict_data.iloc[:, 0], columns=['Class']) df_new['PredictedTestability'] = list(y_pred) - df_new['BranchCoverage'] = list(y_pred_branch) - df_new['LineCoverage'] = list(y_pred_line) + # df_new['BranchCoverage'] = list(y_pred_branch) + # df_new['LineCoverage'] = list(y_pred_line) if verbose: self.export_class_testability_values(df=df_new, log_path=log_path) @@ -440,8 +511,9 @@ def export_class_testability_values(cls, df=None, log_path=None): df.to_csv(log_path, index=False) + # API -def main(project_db_path, initial_value=1.0, verbose=False, log_path=None): +def main(project_db_path, initial_value=1.0, verbose=False, log_path=None, n_jobs=0): """ testability_prediction module API @@ -450,7 +522,7 @@ def main(project_db_path, initial_value=1.0, verbose=False, log_path=None): df = PreProcess().compute_metrics_by_class_list( project_db_path, - n_jobs=0 # n_job must be set to number of CPU cores, use zero for non-parallel computing of metrics + n_jobs=n_jobs # n_job must be set to number of CPU cores, use zero for non-parallel computing of metrics ) testability_ = TestabilityModel().inference(df_predict_data=df, verbose=verbose, log_path=log_path) # print('testability=', testability_) @@ -463,6 +535,8 @@ def main(project_db_path, initial_value=1.0, verbose=False, log_path=None): # project_path_ = r'../benchmark_projects/JSON/JSON.und' # T=0.4531 # project_path_ = r'D:/IdeaProjects/JSON20201115/JSON20201115.und' # T=0.4749 # project_path_ = r'D:/IdeaProjects/jvlt-1.3.2/src.und' # T=0.3997 - print(f"UDB path: {config.UDB_PATH}") + project_path_ = r'F:\benchmarks-proposal\java-corpus\bcel-5.2\bcel-5.2.und' + # print(f"UDB path: {config.UDB_PATH}") for i in range(0, 1): - print('mean testability2 normalize by 1\t', main(config.UDB_PATH, initial_value=1.0, verbose=False)) + # print('mean testability2 normalize by 1\t', main(config.UDB_PATH, initial_value=1.0, verbose=False)) + print('mean testability2 normalize by 1\t', main(project_path_, initial_value=1.0, verbose=False)) From c295ad2d540b792f3289f6be7b766e690e9ec2b3 Mon Sep 17 00:00:00 2001 From: Narges Abbasi Date: Fri, 27 Feb 2026 22:43:09 +0330 Subject: [PATCH 2/2] added refactoring for hierarchy smells --- ...encapsulate_field_for_extract_hierarchy.py | 146 +++ .../extract_hierarchy_detection.py | 597 +++++++++++++ .../extract_hierarchy_execution_pipeline.py | 171 ++++ .../replace_conditional_with_polymorphism2.py | 413 +++++++++ ...egy_pattern_refactoring_for_switch_case.py | 778 ++++++++++++++++ .../strategy_pattern_refactoring_if.py | 840 ++++++++++++++++++ 6 files changed, 2945 insertions(+) create mode 100644 codart/refactorings/encapsulate_field_for_extract_hierarchy.py create mode 100644 codart/refactorings/extract_hierarchy_detection.py create mode 100644 codart/refactorings/extract_hierarchy_execution_pipeline.py create mode 100644 codart/refactorings/replace_conditional_with_polymorphism2.py create mode 100644 codart/refactorings/strategy_pattern_refactoring_for_switch_case.py create mode 100644 codart/refactorings/strategy_pattern_refactoring_if.py diff --git a/codart/refactorings/encapsulate_field_for_extract_hierarchy.py b/codart/refactorings/encapsulate_field_for_extract_hierarchy.py new file mode 100644 index 00000000..f2ab20b8 --- /dev/null +++ b/codart/refactorings/encapsulate_field_for_extract_hierarchy.py @@ -0,0 +1,146 @@ +from dataclasses import dataclass +from antlr4 import * +from antlr4.TokenStreamRewriter import TokenStreamRewriter +from codart.gen.JavaLexer import JavaLexer +from codart.gen.JavaParserLabeled import JavaParserLabeled +from codart.gen.JavaParserLabeledListener import JavaParserLabeledListener + + +@dataclass +class ClassModel: + name: str + fields: dict + methods: set + type_field: str | None = None + type_field_insert_index: int | None = None + + +class TypeFieldNormalizationListener(JavaParserLabeledListener): + """ + Pass 1 — Class normalization / field encapsulation + Triggered only for classes that have type-checking smells. + """ + + def __init__(self, token_stream: CommonTokenStream, smells_for_class): + self.rewriter = TokenStreamRewriter(token_stream) + # self.current_class: ClassModel | None = None + self.class_stack: list[ClassModel] = [] + self.class_models: dict[str, ClassModel] = {} + self.smells_for_class = smells_for_class + + def enterClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + class_name = ctx.IDENTIFIER().getText() + + model = ClassModel( + name=class_name, + fields={}, + methods=set() + ) + + self.class_models[class_name] = model + self.class_stack.append(model) + + def exitClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + if not self.class_stack: + return + + model = self.class_stack.pop() + + # Only encapsulate if there are smells in this class + if not self.smells_for_class: + return + + field = model.type_field + field_type = model.fields.get(field) + insert_index = model.type_field_insert_index + + if not field or not field_type or insert_index is None: + return + + cap = field[0].upper() + field[1:] + getter = f"get{cap}" + setter = f"set{cap}" + + has_getter = getter in model.methods + has_setter = setter in model.methods + + if has_getter and has_setter: + return + + code = "" + + if not has_getter: + code += ( + f"\n\tpublic {field_type} {getter}() {{\n" + f"\t\treturn {field};\n" + f"\t}}\n" + ) + + if not has_setter: + code += ( + f"\n\tpublic void {setter}({field_type} {field}) {{\n" + f"\t\tthis.{field} = {field};\n" + f"\t}}\n" + ) + + self.rewriter.insertBefore( + program_name=self.rewriter.DEFAULT_PROGRAM_NAME, + index=insert_index, + text=code + ) + + def enterFieldDeclaration(self, ctx: JavaParserLabeled.FieldDeclarationContext): + if not self.current_class: + return + + field_type = ctx.typeType().getText() + for d in ctx.variableDeclarators().variableDeclarator(): + field_name = d.variableDeclaratorId().getText() + self.current_class.fields[field_name] = field_type + + stop_token = ctx.stop + self.current_class.type_field_insert_index = stop_token.tokenIndex + 1 + + def enterMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + if self.current_class: + self.current_class.methods.add(ctx.IDENTIFIER().getText()) + + def enterStatement8(self, ctx: JavaParserLabeled.Statement8Context): + if not self.current_class: + return + + expr = ctx.parExpression().expression() + + # Case: switch(getType()) + if hasattr(expr, "methodCall") and expr.methodCall(): + name = expr.methodCall().IDENTIFIER().getText() + if name.startswith("get") and len(name) > 3: + field = name[3:].lower() + if field in self.current_class.fields: + self.current_class.type_field = field + return + + # Case: switch(type) + field = expr.getText() + if field in self.current_class.fields: + self.current_class.type_field = field + + @property + def current_class(self): + return self.class_stack[-1] if self.class_stack else None + + +def apply_encapsulation(java_file_path: str, smells_for_class): + with open(java_file_path, "r", encoding="utf-8", errors="ignore") as f: + source = f.read() + + lexer = JavaLexer(InputStream(source)) + tokens = CommonTokenStream(lexer) + parser = JavaParserLabeled(tokens) + tree = parser.compilationUnit() + + listener = TypeFieldNormalizationListener(tokens, smells_for_class) + walker = ParseTreeWalker() + walker.walk(listener, tree) + + return listener.rewriter.getDefaultText() diff --git a/codart/refactorings/extract_hierarchy_detection.py b/codart/refactorings/extract_hierarchy_detection.py new file mode 100644 index 00000000..c7e03dc5 --- /dev/null +++ b/codart/refactorings/extract_hierarchy_detection.py @@ -0,0 +1,597 @@ +import os +import re +from antlr4 import * +from codart.gen.JavaParserLabeled import JavaParserLabeled +from codart.gen.JavaLexer import JavaLexer +from codart.gen.JavaParserLabeledListener import JavaParserLabeledListener +from collections import defaultdict +from dataclasses import dataclass + + +@dataclass(frozen=True) +class TypeCheckingSmell: + package: str + class_name: str + method: str + line: int + kind: str + condition: str | None = None + + +class SymbolTableListener(JavaParserLabeledListener): + def __init__(self): + self.class_modifiers = [] + self.abstract_classes = set() + self.subclasses = {} # {superclass: [subclasses]} + self.current_class = None + self.static_constants = {} # {class: [static constant names]} + self.abstract_methods = [] # {class: [abstract method names]} + self.current_is_abstract = False + + def enterClassOrInterfaceModifier(self, ctx: JavaParserLabeled.ClassOrInterfaceModifierContext): + self.class_modifiers = ctx.getText() + + def enterClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + class_name = ctx.IDENTIFIER().getText() + # Track superclass relationships + if ctx.IMPLEMENTS(): + superclass = ctx.typeList().getText() + self.abstract_classes.add(superclass) + self.subclasses.setdefault(superclass, []).append(class_name) + if ctx.EXTENDS(): + superclass = ctx.typeType().getText() + self.abstract_classes.add(superclass) + self.subclasses.setdefault(superclass, []).append(class_name) + + def enterFieldDeclaration(self, ctx: JavaParserLabeled.FieldDeclarationContext): + # modifiers = [m.getText() for m in ctx.parentCtx.modifier()] + # if "static" in modifiers and "final" in modifiers: + # if self.current_class: + # var_name = ctx.variableDeclarators().variableDeclarator(0).variableDeclaratorId().getText() + # self.static_constants[self.current_class].append(var_name) + parent = ctx.parentCtx + grandparent = parent.parentCtx + + modifiers = [] + if hasattr(grandparent, "modifier"): # classBodyDeclaration likely + modifiers = [m.getText() for m in grandparent.modifier()] + elif hasattr(parent, "modifier"): + modifiers = [m.getText() for m in parent.modifier()] + + field_name = ctx.variableDeclarators().variableDeclarator(0).variableDeclaratorId().IDENTIFIER().getText() + + def enterMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + method_name = ctx.IDENTIFIER().getText() + method_body = ctx.methodBody() + # Check if method has Nobody (i.e., abstract methods) + if method_body is None or method_body.getText() == ";": + self.abstract_methods.append(method_name) + + if self.current_is_abstract: + modifiers = [m.getText() for m in ctx.parentCtx.modifier()] + if "abstract" in modifiers: + method_name = ctx.IDENTIFIER().getText() + self.abstract_methods[self.current_class].append(method_name) + + +class MissingHierarchyListener(JavaParserLabeledListener): + def __init__(self, abstract_classes, subclasses, abstract_methods): + self.enter_class = False + self.class_fields = set() + self.static_fields = [] + self.abstract_classes = abstract_classes + self.parameters = [] + self.enter_method = False + self.method_identifier = "" + self.class_identifier = [] + self.package_identifier = "" + self.abstract_method = [] + self.current_abstract = None + self.getters = {} # key = methodName, value = fieldName + self.enum_constants = [] + self.fields = [] + self.literals = [] + self.if_else = False + self.else_count = 0 + self.if_conditions = [] + self.visited_methods = set() + self.operators = ["==", "!=", "<=", ">=", "instanceof"] + self.switch = False + self.switch_condition = [] + self.switch_label = [] + self.subclasses = subclasses + self.abstract_methods = abstract_methods + self.detected_smells = [] + self.rtti_smells = [] + self.abstract_getters = defaultdict(list) + self.abstract_var_to_type = {} + self.method_called = [] + self.type_checking_smells: set[TypeCheckingSmell] = set() + self._if_block_start_line = None + + def enterCompilationUnit(self, ctx: JavaParserLabeled.CompilationUnitContext): + if ctx.packageDeclaration(): + self.package_identifier = ctx.packageDeclaration().getText().replace(";", "").strip() + + def enterClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + self.enter_class = True + class_name = ctx.IDENTIFIER().getText() + self.class_identifier.append(class_name) + + def enterFieldDeclaration(self, ctx: JavaParserLabeled.FieldDeclarationContext): + for declarator in ctx.variableDeclarators().variableDeclarator(): + field_name = declarator.variableDeclaratorId().getText() + self.class_fields.add(field_name) + self.fields.append(field_name) + + parent = ctx.parentCtx # should be memberDeclaration + grandparent = parent.parentCtx if parent else None + if grandparent and hasattr(grandparent, 'modifier'): + modifiers = grandparent.modifier() + is_static = any(m.getText() == 'static' for m in modifiers) + if is_static: + var_type = ctx.typeType().getText() + var_names = [v.variableDeclaratorId().getText() for v in ctx.variableDeclarators().variableDeclarator()] + for name in var_names: + self.static_fields.append((var_type, name)) + self.literals.append(name) + + def enterMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + self.enter_method = True + parameters = [] + + formal_params_ctx = ctx.formalParameters() + if formal_params_ctx: + param_list_ctx = formal_params_ctx.formalParameterList() + if param_list_ctx: + # Process normal parameters (if any) + if hasattr(param_list_ctx, "formalParameter"): + for param in param_list_ctx.formalParameter(): + if param.typeType(): + param_type = param.typeType().getText() + else: + param_type = "Unknown" + + param_name = param.variableDeclaratorId().getText() + # Only track parameters that match abstract classes + for superClass in self.abstract_classes: + if superClass == param_type: + parameters.append((param_type, param_name)) + self.parameters.append(param_name) + + # Process last parameter if it exists (for varargs) + if hasattr(param_list_ctx, "lastFormalParameter") and param_list_ctx.lastFormalParameter(): + param = param_list_ctx.lastFormalParameter() + if param.typeType(): + param_type = param.typeType().getText() + else: + param_type = "Unknown" + + param_name = param.variableDeclaratorId().getText() + for superClass in self.abstract_classes: + if superClass == param_type: + parameters.append((param_type, param_name)) + self.parameters.append(param_name) + + self.method_identifier = ctx.IDENTIFIER().getText() + parameters = ctx.formalParameters().getText() + return_type = ctx.typeTypeOrVoid().getText() + self.abstract_method = ctx.IDENTIFIER().getText() + method_body = ctx.methodBody() + # Check if method has Nobody (i.e., abstract methods) + if method_body is None or method_body.getText() == ";": + if self.current_abstract: + self.abstract_getters[self.current_abstract].append(self.abstract_method) + + if parameters != "()" or return_type == "void": + return + if self.method_identifier.startswith("get") or self.method_identifier.startswith("is"): + method_body = ctx.methodBody().getText() + for field in self.class_fields: + if f"return{field}" in method_body or f"return this.{field}" in method_body: + self.getters[self.method_identifier] = field + + def enterEnumDeclaration(self, ctx: JavaParserLabeled.EnumDeclarationContext): + if ctx.enumConstants(): + for enum_constant in ctx.enumConstants().enumConstant(): + name = enum_constant.IDENTIFIER().getText() + self.enum_constants.append(name) + + def enterEnhancedForControl(self, ctx: JavaParserLabeled.EnhancedForControlContext): + var_type = ctx.typeType().getText() + for superClass in self.abstract_classes: + if superClass == var_type: + # Extract parameter name + var_name = ctx.variableDeclaratorId().getText() + self.parameters.append(var_name) + + def enterMethodCall0(self, ctx: JavaParserLabeled.MethodCall0Context): + called_method = ctx.getText() + if called_method in self.abstract_method: + print("abstract method:", called_method) + + def enterFormalParameter(self, ctx: JavaParserLabeled.FormalParameterContext): + parameter = ctx.variableDeclaratorId().getText() + self.fields.append(parameter) + + # def enterVariableDeclarator(self, ctx: JavaParserLabeled.VariableDeclaratorContext): + # variable = ctx.variableDeclaratorId().getText() + # if ctx.variableInitializer(): + # # self.literals.append(variable) + # self.fields.append(variable) + # else: + # self.fields.append(variable) + + def enterLocalVariableDeclaration(self, ctx: JavaParserLabeled.LocalVariableDeclarationContext): + for dec in ctx.variableDeclarators().variableDeclarator(): + var_name = dec.variableDeclaratorId().getText() + initializer = dec.variableInitializer() + if initializer is None: + continue + init_text = initializer.getText() + root = init_text.split(".")[0] + if not ctx.typeType().primitiveType(): + self.fields.append(var_name) + elif root not in self.class_fields: + continue + elif "Dialog" in init_text: + continue + self.fields.append(var_name) + + # def enterPrimary3(self, ctx: JavaParserLabeled.Primary3Context): + # exp = ctx.getText() + # if exp not in ("0", "0L", "0.0", "false", "null", "", "this"): # skip trivial numeric constants + # self.literals.append(exp) + + def exitExpression1(self, ctx: JavaParserLabeled.Expression1Context): + text = ctx.getText() + if "." in text and text[0].isupper(): # crude but effective enum check + self.literals.append(text) + + def debug_tree(self, ctx, level=0): + print(" " * level, type(ctx), "->", ctx.getText()) + for i in range(ctx.getChildCount()): + self.debug_tree(ctx.getChild(i), level + 1) + + def extract_conditions(self, expr_ctx): + conditions = [] + # Unwrap single-child wrappers + if expr_ctx.getChildCount() == 1: + return self.extract_conditions(expr_ctx.getChild(0)) + + # Unwrap parentheses inside Primary + if "Primary" in type(expr_ctx).__name__: + for i in range(expr_ctx.getChildCount()): + child = expr_ctx.getChild(i) + if child.getText() not in ["(", ")"]: + return self.extract_conditions(child) + + # Split on logical operators + for i in range(expr_ctx.getChildCount()): + child = expr_ctx.getChild(i) + if child.getText() in ["&&", "||"]: + left = expr_ctx.getChild(i - 1) + right = expr_ctx.getChild(i + 1) + + conditions.extend(self.extract_conditions(left)) + conditions.extend(self.extract_conditions(right)) + return conditions + + # Atomic condition + conditions.append(expr_ctx.getText()) + return conditions + + def enterStatement2(self, ctx: JavaParserLabeled.Statement2Context): + self.if_else = True + if ctx.ELSE(): + self.else_count += 1 + # Store the first line of the if-else block + # if not hasattr(self, "_if_block_start_line") or self._if_block_start_line is None: + # self._if_block_start_line = ctx.start.line + + # condition = ctx.parExpression().expression().getText() + # if condition not in self.if_conditions: + # self.if_conditions.append(condition) + expr_ctx = ctx.parExpression().expression() + atomic_conditions = self.extract_conditions(expr_ctx) + + for cond in atomic_conditions: + if cond not in self.if_conditions: + self.if_conditions.append(cond) + + def exitStatement2(self, ctx: JavaParserLabeled.Statement2Context): + """ + Record only one smell per if-else block, using the line of the first 'if'. + """ + if self.if_else and self.else_count > 0: + has_type_check = False + has_rtti = False + for cond in self.if_conditions: + # print("condition", cond) + # Skip pure boolean checks like if(result.isSuccess()) + if self.is_simple_boolean_check(cond): + continue + + # if any(op in cond for op in self.operators): + if re.fullmatch(r'\s*\w+\s*(==|!=|<=|>=|<|>)\s*\w+\s*', cond): + + lhs, rhs = self.split_condition(cond) + lhs = lhs.split('.')[0] + # direct field/literal or getter + for field in self.fields: + for literal in self.literals: + match_direct = (field in lhs and rhs == literal) or (rhs == field and lhs == literal) + match_getter = (self.is_getter_invocation(lhs, self.getters) and rhs == literal) or \ + (self.is_getter_invocation(rhs, self.getters) and lhs == literal) + if match_direct or match_getter: + has_type_check = True + break + + if has_type_check: + self._if_block_start_line = ctx.start.line + break + if has_type_check: + break + # Pattern2: (Shape instanceof Circle) + if "instanceof" in cond: + lhs, rhs = self.split_instanceof(cond) + if self.check_type_check(lhs, rhs, ctx.start.line): + has_rtti = True + break + # Pattern3: getClass() == Subclass.class + if re.search(r"\.getClass\(\)\s*==\s*\w+\.class", cond): + lhs, rhs = self.split_getclass_equality(cond) + if self.check_type_check(lhs, rhs, ctx.start.line): + has_rtti = True + break + # Pattern4: getClass().equals(SubClass.class) + if re.search(r"\.getClass\(\)\.equals\(\s*\w+\.class\s*\)", cond): + lhs, rhs = self.split_getclass_equals(cond) + if self.check_type_check(lhs, rhs, ctx.start.line): + has_rtti = True + break + + # Pattern 5: var.getType() == CONSTANT + if re.search(r"(\w+)\.(\w+)\(\)\s*==\s*(\w+(\.\w+)?)", cond): + lhs, rhs = self.split_condition(cond) + if self.check_encoded_type(lhs, rhs, ctx.start.line): + has_rtti = True + break + + if has_type_check: + self.record_smell( + ctx, + kind="IF_ELSE_TYPE_CHECK", + condition=" | ".join(self.if_conditions), + line=self._if_block_start_line # use first if line + ) + if has_rtti: + self.record_smell( + ctx, + kind="RTTI", + condition=" | ".join(self.if_conditions), + line=self._if_block_start_line # use first if line + ) + + # Reset temporary state + self.if_else = False + self.else_count = 0 + self.if_conditions.clear() + self._if_block_start_line = None + + def enterStatement8(self, ctx: JavaParserLabeled.Statement8Context): + self.switch = True + statement = ctx.SWITCH() + if statement: + condition = ctx.parExpression().expression().getText() + if '.' in condition: + condition = condition.split('.')[-1] + self.switch_condition.append(condition) + + def enterSwitchLabel(self, ctx: JavaParserLabeled.SwitchLabelContext): + label = ctx.getText() + self.switch_label.append(label) + + def exitStatement8(self, ctx: JavaParserLabeled.Statement8Context): + for field in self.fields: + for literal in self.literals: + if field in self.switch_condition and f"case{literal}:" in self.switch_label: + self.switch = False + print("Found Switch statement involving type field!", self.method_identifier) + if field in self.switch_condition and f"case{literal}:" in self.switch_label: + self.record_smell( + ctx, + kind="SWITCH_TYPE_CHECK", + condition=" ".join(self.switch_condition) + ) + + return self.method_identifier + for getter in self.getters: + if f"{getter}()" in self.switch_condition and f"case{literal}:" in self.switch_label: + self.switch = False + print("Found Switch statement involving getter!", self.method_identifier) + if f"{getter}()" in self.switch_condition and f"case{literal}:" in self.switch_label: + self.record_smell( + ctx, + kind="SWITCH_GETTER_TYPE_CHECK", + condition=" ".join(self.switch_condition) + ) + return self.package_identifier, self.class_identifier, self.method_identifier + + def exitMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + self.enter_method = False + + def exitClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + self.class_identifier.pop() + self.enter_class = False + # self.fields.clear() + self.getters.clear() + self.literals.clear() + + def split_condition(self, condition: str): + """ + Split the condition into left-hand side (lhs) and right-hand side (rhs). + """ + for op in ["==", "!=", "<=", ">="]: + if op in condition: + return condition.split(op) + return "", "" + + def is_getter_invocation(self, expr: str, known_getters: dict) -> bool: + """ + Check if the expression looks like a getter invocation. + Accepts forms like: phone.getType(), this.getType(), getType() + """ + for getter in known_getters.keys(): + if expr.endswith(f".{getter}()") or expr == f"{getter}()": + return True + return False + + def split_instanceof(self, condition): + parts = condition.replace("(", "").replace(")", "").split("instanceof") + lhs = parts[0].strip() + rhs = parts[1].strip() + return lhs, rhs + + def split_getclass_equality(self, condition): + # Example: (s.getClass() == Circle.class) + m = re.search(r"(\w+)\.getClass\(\)\s*==\s*(\w+)\.class", condition) + if m: + lhs = m.group(1) + rhs = m.group(2) + return lhs, rhs + return None, None + + def split_getclass_equals(self, condition): + # Example: s.getClass().equals(Circle.class) + m = re.search(r"(\w+)\.getClass\(\)\.equals\(\s*(\w+)\.class\s*\)", condition) + if m: + lhs = m.group(1) + rhs = m.group(2) + return lhs, rhs + return None, None + + def check_type_check(self, lhs, rhs, line) -> bool: + for superClass in self.abstract_classes: + subclasses = self.subclasses.get(superClass, []) + if rhs in subclasses and lhs in self.parameters: + self.detected_smells.append({ + "line": line, + "super": superClass, + "sub": rhs, + "variable": lhs + }) + return True # smell detected + return False + + def check_encoded_type(self, lhs, rhs, line) -> bool: + if "." in lhs and "(" in lhs and lhs.endswith(")"): + lhs_f = lhs.split(".")[0] # 'type' + lhs_l = lhs.split(".")[1].split("(")[0] # 'getType' + for superClass in self.abstract_classes: + if lhs_f in self.parameters and lhs_l in self.abstract_methods: + self.detected_smells.append({ + "line": line, + "super": superClass, + "sub": rhs, + "variable": lhs + }) + return True + return False + + def record_smell(self, ctx, kind: str, condition: str | None = None, line: int | None = None): + current_class = self.class_identifier[-1] if self.class_identifier else None + self.type_checking_smells.add( + TypeCheckingSmell( + package=self.package_identifier, + class_name=current_class, + method=self.method_identifier, + line=line if line is not None else ctx.start.line, # use provided line if given + kind=kind, + condition=condition + ) + ) + + def is_simple_boolean_check(self, cond: str) -> bool: + """ + Returns True if condition is a simple boolean check like: + - result.isSuccess() + - isValid() + - flag + - !result.isSuccess() + """ + + cond = cond.strip() + + # remove surrounding parentheses + if cond.startswith("(") and cond.endswith(")"): + cond = cond[1:-1].strip() + + # remove leading negation + cond = cond.lstrip("!").strip() + + # Case 1: simple variable (flag) + if re.fullmatch(r"\w+", cond): + return True + + # Case 2: simple method call without comparison (obj.isSuccess()) + if re.fullmatch(r"\w+\.\w+\(\)", cond): + return True + + # Case 3: direct method call (isValid()) + if re.fullmatch(r"\w+\(\)", cond): + return True + + return False + + +def get_all_java_files(directory): + java_files = [] + for root, _, files in os.walk(directory): + for file in files: + if file.endswith(".java"): + java_files.append(os.path.join(root, file)) + return java_files + + +def parse_java_file(file_path, symbol_listener, if_listener=None): + input_stream = FileStream(file_path, encoding="utf-8", errors="ignore") + lexer = JavaLexer(input_stream) + stream = CommonTokenStream(lexer) + parser = JavaParserLabeled(stream) + tree = parser.compilationUnit() + # listener = listener_class() + walker = ParseTreeWalker() + # walker.walk(listener, tree) + walker.walk(symbol_listener, tree) + if if_listener: + walker.walk(if_listener, tree) + + +if __name__ == "__main__": + project_path = "F:/benchmarks-proposal/ant" + symbol_listener = SymbolTableListener() + # Build symbol table + for file in get_all_java_files(project_path): + parse_java_file(file, symbol_listener) + # Detect type checks + if_listener = MissingHierarchyListener(symbol_listener.abstract_classes, + symbol_listener.subclasses, + symbol_listener.abstract_methods) + for file in get_all_java_files(project_path): + parse_java_file(file, symbol_listener, if_listener) + + print("\n=== Type Checking Smells Detected in ===") + + smells = sorted( + if_listener.type_checking_smells, + key=lambda s: (s.package, s.class_name, s.method, s.line) + ) + print(f"Total smells detected: {len(smells)}\n") + for smell in smells: + print( + f"package: {smell.package}\n" + f"class: {smell.class_name}\n" + f"method: {smell.method}\n" + f"(line {smell.line}) --> [{smell.kind}]" + ) diff --git a/codart/refactorings/extract_hierarchy_execution_pipeline.py b/codart/refactorings/extract_hierarchy_execution_pipeline.py new file mode 100644 index 00000000..f37e71fb --- /dev/null +++ b/codart/refactorings/extract_hierarchy_execution_pipeline.py @@ -0,0 +1,171 @@ +import os +import argparse +from time import time +from antlr4 import * +from collections import defaultdict +from codart.gen.JavaLexer import JavaLexer +from codart.gen.JavaParserLabeled import JavaParserLabeled +from codart.refactorings.extract_hierarchy_detection import SymbolTableListener, MissingHierarchyListener, TypeCheckingSmell +from codart.refactorings.encapsulate_field_for_extract_hierarchy import apply_encapsulation +from codart.refactorings.strategy_pattern_refactoring_for_switch_case import StrategyPatternRefactoringListenerForSwitch +from codart.refactorings.strategy_pattern_refactoring_if import StrategyPatternRefactoringListenerForIfElse +from codart.refactorings.replace_conditional_with_polymorphism2 import refactor_project + + +def get_all_java_files(directory): + java_files = [] + for root, _, files in os.walk(directory): + for file in files: + if file.endswith(".java"): + java_files.append(os.path.join(root, file)) + return java_files + + +def parse_java_file(file_path, listeners): + """ Walk a parse tree with one or multiple listeners """ + input_stream = FileStream(file_path, encoding="utf-8", errors="ignore") + lexer = JavaLexer(input_stream) + token_stream = CommonTokenStream(lexer) + parser = JavaParserLabeled(token_stream) + tree = parser.compilationUnit() + + walker = ParseTreeWalker() + for listener in listeners: + walker.walk(listener, tree) + + return token_stream, tree + + +def build_typecheck_index(smells: set[TypeCheckingSmell]): + by_class = defaultdict(list) + by_class_method = defaultdict(list) + + for s in smells: + by_class[s.class_name].append(s) + by_class_method[(s.class_name, s.method)].append(s) + + return by_class, by_class_method + + +def main(project_path): + begin_time = time() + + if not os.path.isdir(project_path): + print("Error: Provided path is not a directory") + return + + java_files = get_all_java_files(project_path) + print(f"Found {len(java_files)} Java files") + + # ===== Step 1: Build symbol table ===== + symbol_listener = SymbolTableListener() + for file_path in java_files: + parse_java_file(file_path, [symbol_listener]) + + # ===== Step 2: Detect type-checking smells ===== + if_listener = MissingHierarchyListener( + symbol_listener.abstract_classes, + symbol_listener.subclasses, + symbol_listener.abstract_methods + ) + for file_path in java_files: + parse_java_file(file_path, [if_listener]) + + # === INDEX SMELLS === + smells_by_class, smells_by_class_method = build_typecheck_index(if_listener.type_checking_smells) + + # Map class_name -> file_path for easy refactoring + class_to_file = {os.path.splitext(os.path.basename(f))[0]: f for f in java_files} + + # Report detected smells + print("\n=== Type Checking Smells Detected ===") + smells = sorted( + if_listener.type_checking_smells, + key=lambda s: (s.package, s.class_name, s.method, s.line) + ) + print(f"Total smells detected: {len(smells)}\n") + for smell in smells: + print( + f"package: {smell.package}\n" + f"class: {smell.class_name}\n" + f"method: {smell.method}\n" + f"(line {smell.line}) --> [{smell.kind}]" + ) + + # ===== Step 3: Apply encapsulation/refactoring for smelly files if needed ===== + for file_path in java_files: + class_name = os.path.splitext(os.path.basename(file_path))[0] + smells_for_class = smells_by_class.get(class_name, []) + print(f"Encapsulation: Processing {file_path} smells={len(smells_for_class)}") + new_source = apply_encapsulation(file_path, smells_for_class) + with open(file_path, "w", encoding="utf-8") as f: + f.write(new_source) + + # ===== Step 4: Apply strategy pattern refactoring ===== + for smell in if_listener.type_checking_smells: + target_file = class_to_file.get(smell.class_name) + if not target_file or not os.path.exists(target_file): + continue + + print(f"Strategy Refactoring: {smell.class_name}.{smell.method} in {target_file}") + + # Parse the file again for refactoring + input_stream = FileStream(target_file, encoding="utf-8") + lexer = JavaLexer(input_stream) + token_stream = CommonTokenStream(lexer) + parser = JavaParserLabeled(token_stream) + tree = parser.compilationUnit() + + # Choose listener based on smell kind + if "switch" in smell.kind.lower(): + listener = StrategyPatternRefactoringListenerForSwitch( + common_token_stream=token_stream, + method_identifier=smell.method + ) + elif smell.kind == "IF_ELSE_TYPE_CHECK": # if-else + listener = StrategyPatternRefactoringListenerForIfElse( + common_token_stream=token_stream, + method_identifier=smell.method, + smell_line=smell.line + ) + elif smell.kind == "RTTI": + print(f"RTTI Refactoring: {smell.class_name}.{smell.method}") + + # IMPORTANT: You need a .udb file path + udb_path = os.path.join(project_path, f"{os.path.basename(project_path)}.udb") + + refactor_project( + project_root=project_path, + output_root=project_path, # in-place refactor + udb_path=udb_path, + target_method=smell.method + ) + + continue + + # Set base_dir dynamically based on the file location + listener.base_dir = os.path.dirname(target_file) + + walker = ParseTreeWalker() + walker.walk(listener, tree) + + # Write back the refactored code + refactored_code = listener.token_stream_rewriter.getDefaultText() + with open(target_file, "w", encoding="utf-8") as f: + f.write(refactored_code) + + end_time = time() + print("Execution completed") + print("Total files processed:", len(java_files)) + print("Total execution time:", end_time - begin_time) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument( + '-p', '--path', + default=r"F:\my-refactoring\1", + help='Path to project folder containing Java files' + ) + args = parser.parse_args() + main(args.path) diff --git a/codart/refactorings/replace_conditional_with_polymorphism2.py b/codart/refactorings/replace_conditional_with_polymorphism2.py new file mode 100644 index 00000000..c66210f0 --- /dev/null +++ b/codart/refactorings/replace_conditional_with_polymorphism2.py @@ -0,0 +1,413 @@ +import os +import re +import sys +from codart.gen.JavaParserLabeled import JavaParserLabeled +from codart.gen.JavaLexer import JavaLexer +from antlr4 import * +from antlr4.TokenStreamRewriter import TokenStreamRewriter +from codart.gen.JavaParserLabeledListener import JavaParserLabeledListener + +os.add_dll_directory("C:\\Program Files\\SciTools\\bin\\pc-win64\\") +sys.path.append("C:\\Program Files\\SciTools\\bin\\pc-win64\\python") +import understand + + +GLOBAL_INJECTIONS = { + "abstract": {}, # baseType -> abstractMethodText + "concrete": {}, # classType -> [methodText, ...] +} + + +def analyze_project_with_understand(udb_path, target_method): + db = understand.open(udb_path) + + owners = set() + fields_map = {} # owner_class -> set of fields + getters_map = {} # owner_class -> {getter_name: field_name} + + for m in db.ents("Method"): + if m.name() != target_method: + continue + + # Get defining class + parent_class = None + for ref in m.refs("Definein"): + if ref.ent().kindname().endswith("Class"): + parent_class = ref.ent() + break + if not parent_class: + parent_class = m.parent() + if not parent_class: + continue + + owner_class = parent_class.name() + owners.add(owner_class) + + # Get fields of the owner class + if owner_class not in fields_map: + fields_map[owner_class] = { + f.name(): f.type() + for f in parent_class.ents("Define", "Java Variable") + } + + # Get getters mapping + if owner_class not in getters_map: + getters = {} + for method in parent_class.ents("Define", "Java Method"): + if not (method.name().startswith("get") or method.name().startswith("is")): + continue + # Try to find which field it returns + for ref in method.refs("Return", "Java Variable"): + ret_ent = ref.ent() + if ret_ent and ret_ent.name() in fields_map[owner_class]: + getters[method.name()] = ret_ent.name() + getters_map[owner_class] = getters + print(getters_map) + + return list(owners), fields_map, getters_map + + +class ClassCollector(JavaParserLabeledListener): + def __init__(self, class_ctx): + self.class_ctx = class_ctx + + def enterClassDeclaration(self, ctx): + name = ctx.IDENTIFIER().getText() + self.class_ctx[name] = ctx + + +class ReplaceConditionalWithPolymorphismRTTIListener(JavaParserLabeledListener): + """ + Generic Replace Conditional with Polymorphism (RTTI). + Works for any Java input satisfying preconditions. + """ + + def __init__(self, tokens: CommonTokenStream, target_method: str, global_class_ctx, method_owners, fields_map, getters_map): + self.tokens = tokens + self.rewriter = TokenStreamRewriter(tokens) + self.target_method = target_method + + self.current_class = None + self.in_target_method = False + + self.param_name = None + self.param_type = None + + self.if_ctx = None + self.current_subtype = None + + self.cases = [] # [{type, body}] + self.default_body = None + + self.class_ctx = global_class_ctx + self.method_owners = set(method_owners) + self.class_fields = set() + self.getters = {} # method_name -> field_name + self.fields_map = fields_map # owner_class -> {field: type} + self.getters_map = getters_map # owner_class -> {getter_name: field_name} + + self.capture_body_for = None # class name for current if/else-if + self.pending_cases = [] # [{type, body}] + self.is_else_block = False + + def enterClassDeclaration(self, ctx): + name = ctx.IDENTIFIER().getText() + self.current_class = name + self.class_ctx[name] = ctx + + def enterMethodDeclaration(self, ctx): + + if ctx.IDENTIFIER().getText() != self.target_method: + return + + self.in_target_method = True + + params = ctx.formalParameters().formalParameterList() + if not params or len(params.formalParameter()) != 1: + self.in_target_method = False + return + + p = params.formalParameter()[0] + self.param_name = p.variableDeclaratorId().getText() + self.param_type = p.typeType().getText() + + def exitMethodDeclaration(self, ctx): + if not self.in_target_method: + return + self._inject_abstract_method(ctx) + self._inject_concrete_methods(ctx) + self._replace_if_with_dispatch() + self._reset() + + # RTTI detection + def enterStatement2(self, ctx): + if not self.in_target_method or not ctx.parExpression(): + return + + expr = ctx.parExpression().expression() + if not expr.INSTANCEOF(): + return + + lhs = expr.expression().getText() + if lhs != self.param_name: + return + + self.capture_body_for = expr.typeType().getText().split(".")[-1] + self.if_ctx = ctx.parentCtx + + # Capture bodies + def enterStatement0(self, ctx: JavaParserLabeled.StatementContext): + if not self.if_ctx: + return + + start = ctx.start.tokenIndex + stop = ctx.stop.tokenIndex + + body_text = self.rewriter.getText("", start, stop).strip() + + # Remove surrounding braces + if body_text.startswith("{") and body_text.endswith("}"): + body_text = body_text[1:-1].strip() + + # Decide target class + if self.capture_body_for: + class_label = self.capture_body_for + self.capture_body_for = None + else: + # ELSE branch → base type + # ELSE branch → infer from cast + inferred = self._extract_cast_type(body_text) + if not inferred: + print("ELSE branch without cast — skipping") + return + + class_label = inferred + self.is_else_block = True + + body_text = self._transform_body(body_text) + + self.pending_cases.append({ + "type": class_label, + "body": body_text + }) + + print(f"Captured body for {class_label}:\n{body_text}\n") + + # Refactoring + def _transform_body(self, body): + # Take first owner from the set (or list) + owner_class = next(iter(self.method_owners)) # e.g., "InstructionFactory" + param_name = owner_class[0].lower() + owner_class[1:] # camelCase + + # Replace field access with getter call + if owner_class in self.getters_map: + for getter, field in self.getters_map[owner_class].items(): + # Replace all occurrences of the field in the body + body = re.sub(rf"\b{field}\b", f"{param_name}.{getter}()", body) + + # Replace RTTI parameter name with "this" + # Step 1: Replace any cast around the RTTI parameter with "( this )" + if self.param_name: + # Matches: (Type) param_name, possibly with spaces + body = re.sub( + rf"\(\s*[\w<>]+\s*\)\s*{self.param_name}\b", + "(this)", + body + ) + return body + + def _inject_abstract_method(self, method_ctx): + base = self.class_ctx.get(self.param_type) + if not base: + return + + ret_type = method_ctx.typeTypeOrVoid().getText() + + abstract_method = ( + f"\n\tpublic abstract {ret_type} " + f"{self.target_method}(" + f"{self._caller_signature(method_ctx)});\n" + ) + base = self.param_type + GLOBAL_INJECTIONS["abstract"][base] = abstract_method + + print(f"Registered abstract method for {base}") + + def _inject_concrete_methods(self, method_ctx): + ret_type = method_ctx.typeTypeOrVoid().getText() + + for case in self.pending_cases: + ctx = self.class_ctx.get(case["type"]) + if not ctx: + continue + + method = ( + "\n\t@Override\n" + f"\tpublic {ret_type} {self._method_signature()} {{\n" + f"{case['body']}\n\t}}\n" + ) + print(f"Injecting concrete method into {case['type']}") + + cls = case["type"] + GLOBAL_INJECTIONS.setdefault("concrete", {}).setdefault(cls, []).append(method) + + print(f"Registered concrete method for {cls}") + + def _replace_if_with_dispatch(self): + if not self.if_ctx: + return + + call = f"{self.param_name}.{self.target_method}({self._caller_args()})" + replacement = f"return {call};" if self._returns() else f"{call};" + + self.rewriter.replaceRange( + self.if_ctx.start.tokenIndex, + self.if_ctx.stop.tokenIndex, + replacement + ) + + def _get_text(self, ctx): + return self.rewriter.getText( + "", + ctx.start.tokenIndex, + ctx.stop.tokenIndex + ) + + def _method_signature(self): + return f"{self.target_method}({self._caller_signature(None)})" + + def _caller_signature(self, ctx): + # Take first owner from the set (or list) + owner_class = next(iter(self.method_owners)) # e.g., "InstructionFactory" + + # Generate parameter name in camelCase: instructionFactory + param_name = owner_class[0].lower() + owner_class[1:] + + return f"{owner_class} {param_name}" + + def _caller_args(self): + return "this" + + def _returns(self): + return True + + def _reset(self): + self.in_target_method = False + self.if_ctx = None + self.cases.clear() + self.default_body = None + + def _extract_cast_type(self, body_text): + """ + Extracts cast type from expressions like: + (ObjectType) t + (pkg.ObjectType) t + """ + m = re.search(r"\(\s*([\w\.<>]+)\s*\)\s*" + re.escape(self.param_name), body_text) + if m: + return m.group(1).split(".")[-1] + return None + + +def find_java_files(root_dir): + java_files = [] + for root, _, files in os.walk(root_dir): + for f in files: + if f.endswith(".java"): + java_files.append(os.path.join(root, f)) + return java_files + + +def apply_global_injections(project_root): + for f in find_java_files(project_root): + tokens, tree = parse_file(f) + + class_ctx = {} + collector = ClassCollector(class_ctx) + ParseTreeWalker().walk(collector, tree) + + rewriter = TokenStreamRewriter(tokens) + + for cls, ctx in class_ctx.items(): + + # abstract methods + if cls in GLOBAL_INJECTIONS["abstract"]: + rewriter.insertBefore( + rewriter.DEFAULT_PROGRAM_NAME, + ctx.classBody().stop.tokenIndex, + GLOBAL_INJECTIONS["abstract"][cls] + ) + print(f"Injected abstract method into {cls}") + + # concrete methods + for m in GLOBAL_INJECTIONS["concrete"].get(cls, []): + rewriter.insertBefore( + rewriter.DEFAULT_PROGRAM_NAME, + ctx.classBody().stop.tokenIndex, + m + ) + print(f"Injected concrete method into {cls}") + + new_code = rewriter.getDefaultText() + open(f, "w", encoding="utf-8").write(new_code) + + +def parse_file(java_path): + with open(java_path, "r", encoding="utf-8") as f: + code = f.read() + + lexer = JavaLexer(InputStream(code)) + tokens = CommonTokenStream(lexer) + parser = JavaParserLabeled(tokens) + tree = parser.compilationUnit() + + return tokens, tree + + +def refactor_project(project_root, output_root, udb_path, target_method): + + # semantic analysis + owners, fields_map, getters_map = analyze_project_with_understand(udb_path, target_method) + + # collect classes + class_ctx = {} + for f in find_java_files(project_root): + _, tree = parse_file(f) + collector = ClassCollector(class_ctx) + ParseTreeWalker().walk(collector, tree) + class_ctx.update(collector.class_ctx) + + # rewrite + for f in find_java_files(project_root): + tokens, tree = parse_file(f) + + listener = ReplaceConditionalWithPolymorphismRTTIListener( + tokens, + target_method, + class_ctx, + owners, + fields_map, + getters_map + ) + + ParseTreeWalker().walk(listener, tree) + # write replaced if-dispatch only + out = os.path.join(output_root, os.path.relpath(f, project_root)) + os.makedirs(os.path.dirname(out), exist_ok=True) + open(out, "w", encoding="utf-8").write( + listener.rewriter.getDefaultText() + ) + + # inject everything + apply_global_injections(output_root) + + +if __name__ == "__main__": + refactor_project( + project_root="C:/Users/98910/Desktop/type_checking_examples_before_refactor/rtti", + output_root="C:/Users/98910/Desktop/type_checking_examples_before_refactor/rtti", + udb_path="C:/Users/98910/Desktop/type_checking_examples_before_refactor/rtti/rtti.udb", + target_method="createCheckCast" + ) + + print("RTTI refactoring complete.") diff --git a/codart/refactorings/strategy_pattern_refactoring_for_switch_case.py b/codart/refactorings/strategy_pattern_refactoring_for_switch_case.py new file mode 100644 index 00000000..8d3fe4b3 --- /dev/null +++ b/codart/refactorings/strategy_pattern_refactoring_for_switch_case.py @@ -0,0 +1,778 @@ +import argparse +import re +from collections import defaultdict +from time import time +import os +from antlr4 import * +from antlr4.TokenStreamRewriter import TokenStreamRewriter +from codart.gen.JavaLexer import JavaLexer +from codart.gen.JavaParserLabeled import JavaParserLabeled +from codart.gen.JavaParserLabeledListener import JavaParserLabeledListener + + +class StrategyPatternRefactoringListenerForSwitch(JavaParserLabeledListener): + """ + Implementing State/Strategy pattern to refactor type checking + """ + + def __init__(self, common_token_stream: CommonTokenStream = None, method_identifier: str = None, base_dir= None): + """ + :param common_token_stream: + """ + self.base_dir = base_dir + self.package_name = "" + self.enter_class = False + self.class_identifier = "" + self.class_fields = set() + self.field_contexts = {} + self.current_method = None + self.uses_instance_fields = False + self.instance_fields = set() + self.current_case_instance_fields = set() + self.method_identifier = method_identifier + self.method_contexts = {} + self.methods = {} + self.enter_method = False + self.method_selected = False + self.selected_method_ctx = {} + self.switch_method_name = "" + self.method_returns = defaultdict(list) + self.local_vars = [] + self.old_method_declaration = "" + self.modified_old_method_declaration = "" + self.method_parameters = set() + self.local_variables = set() + self.switch_stmt_ctx = {} + self.switch_condition_name = "" + self.switch = False + self.switch_condition = None + self.body = "" + self.case_has_return = False + self.old_method_dec_with_body_per_case = {} + self.params_injected = False + self.token_stream = common_token_stream + self.newClasses = "" + self.abstractClass = "" + self.switch_label = "" + self.new_abstract_method_declaration = "" + self.method_name = "" + self.method_return = "" + self.new_abstract_method = "" + self.private_fields = {} + self.pending_cases = [] + self.generated_subclasses = [] + self.final_method_return = None + self.label_text = "" + self.static_fields = set() + self.null_object_body = None + self.current_labels = [] + self.getters = {} + self.switch_field_name = "" + self.switch_name = "" + self.extra_params = [] + + # Move all the tokens in the source code in a buffer, token_stream_rewriter. + if common_token_stream is not None: + self.token_stream_rewriter = TokenStreamRewriter(common_token_stream) + else: + raise TypeError('common_token_stream is None') + + def enterCompilationUnit(self, ctx: JavaParserLabeled.CompilationUnitContext): + if ctx.packageDeclaration(): + self.package_name = ctx.packageDeclaration().getText().replace(";", "").strip() + + def enterClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + self.enter_class = True + self.class_identifier = ctx.IDENTIFIER().getText() + + def enterFieldDeclaration(self, ctx: JavaParserLabeled.FieldDeclarationContext): + # Save class field names + for declarator in ctx.variableDeclarators().variableDeclarator(): + field_name = declarator.variableDeclaratorId().getText() + self.class_fields.add(field_name) + # Save context for later type replacement + if not hasattr(self, "field_contexts"): + self.field_contexts = {} + self.field_contexts[field_name] = ctx + + # Go up to classBodyDeclaration + class_body_dec = ctx.parentCtx.parentCtx + modifiers = [] + if hasattr(class_body_dec, "modifier"): + modifiers = [m.getText() for m in class_body_dec.modifier()] + if "private" in modifiers: + field_type = ctx.typeType().getText() + var_dec = ctx.variableDeclarators().variableDeclarator(0) + field_name = var_dec.variableDeclaratorId().getText() + self.private_fields[field_name] = field_type + # print(f"Collected private field: {field_name} ({field_type})") + if "static" in modifiers: + var_dec = ctx.variableDeclarators().variableDeclarator(0) + field_name = var_dec.variableDeclaratorId().getText() + self.static_fields.add(field_name) + + def enterMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + method_name = ctx.IDENTIFIER().getText() + if method_name.startswith("get") or method_name.startswith("is"): + method_body = ctx.methodBody().getText() + for field in self.class_fields: + if f"return{field}" in method_body or f"return this.{field}" in method_body: + self.getters[method_name] = field + + self.method_contexts[method_name] = ctx + grand_parent_ctx = ctx.parentCtx.parentCtx + method_dec = self.token_stream_rewriter.getText("", grand_parent_ctx.modifier(0). + start.tokenIndex, + ctx.formalParameters().stop.tokenIndex) + self.methods[method_name] = method_dec + if self.method_identifier is None or method_name == self.method_identifier: + self.current_method = method_name + self.uses_instance_fields = False # reset for this method + if method_name == self.method_identifier: + self.enter_method = True + self.method_selected = True + self.selected_method_ctx = ctx + self.switch_method_name = method_name + self.current_method = method_name + self.method_returns[self.current_method] = [] + self.local_vars = [] + + self.old_method_declaration = method_dec + self.modified_old_method_declaration = method_dec + print("Old declaration of method:", self.old_method_declaration) + + def enterFormalParameter(self, ctx): + param_name = ctx.variableDeclaratorId().getText() + self.method_parameters.add(param_name) + + def enterLocalVariableDeclaration(self, ctx: JavaParserLabeled.LocalVariableDeclarationContext): + for dec in ctx.variableDeclarators().variableDeclarator(): + var_name = dec.variableDeclaratorId().getText() + self.local_variables.add(var_name) + + if not self.current_method: + return + + var_type = ctx.typeType().getText() + declarator = ctx.variableDeclarators().variableDeclarator(0) + var_name = declarator.variableDeclaratorId().getText() + # Collect locals + if (var_type, var_name) not in self.local_vars: + self.local_vars.append((var_type, var_name)) + + def enterExpression0(self, ctx: JavaParserLabeled.Expression0Context): + if not self.current_method: + return + name = ctx.getText() + # skip qualified access + if "." in name: + return + # skip locals, parameters, static fields + if name in self.local_variables or name in self.method_parameters or name in self.static_fields: + return + # check exact match against instance fields + if name in self.private_fields: + self.uses_instance_fields = True + self.instance_fields.add(name) + self.current_case_instance_fields.add(name) + + def enterStatement8(self, ctx: JavaParserLabeled.Statement8Context): + if not self.method_selected: + return + if ctx: + self.switch_stmt_ctx = ctx + self.switch_condition_name = ctx.parExpression().expression().getText() + self.switch_name = ctx.parExpression().expression().getText().replace("()", "") + if self.switch_name in self.getters: + self.switch_field_name = self.getters[self.switch_name] + if not self.method_selected: + return + self.switch = True + expr = ctx.parExpression().expression() + + # Case 1: switch(getSomething()) + if hasattr(expr, "methodCall") and expr.methodCall(): + self.method_name = expr.methodCall().IDENTIFIER().getText() + if self.method_name.startswith("get") and len(self.method_name) > 3: + self.switch_condition = self.method_name[3:] + else: + self.switch_condition = self.method_name + # Case 2: switch(type) + else: + self.switch_condition = expr.getText() + s = self.switch_condition[0].upper() + self.switch_condition[1:] + # self.method_name = f"get{self.switch_condition.capitalize()}" + self.method_name = f"get{s}" + + def enterSwitchBlockStatementGroup(self, ctx: JavaParserLabeled.SwitchBlockStatementGroupContext): + if not self.method_selected: + return + block_statements = ctx.blockStatement() + if not block_statements: + return + + start = block_statements[0].start.tokenIndex + stop = block_statements[-1].stop.tokenIndex + body_text = self.token_stream_rewriter.getText("", start, stop).strip() + + # Remove trailing break; + if body_text.endswith("break;"): + body_text = body_text[:-6].strip() + + # Detect return inside case + if re.search(r"\breturn\b", body_text): + self.case_has_return = True + + # Detect DEFAULT label + is_default = any(label.DEFAULT() for label in ctx.switchLabel()) + + if is_default: + # capture Null Object behavior + self.null_object_body = body_text + return + + self.body = body_text + + def enterSwitchLabel(self, ctx: JavaParserLabeled.SwitchLabelContext): + if not self.method_selected: + return + self.switch = True + + # Detect label + if ctx.expression(): + self.label_text = ctx.expression().getText() + + # Normalize label (ENGINEER → Employee.ENGINEER) + if "." not in self.label_text: + self.label_text = f"{self.class_identifier}.{self.label_text}" + + # Extract subclass name (ENGINEER) + self.switch_label = self.to_pascal_case(self.label_text.split(".")[-1]) + self.current_labels.append((self.switch_label, self.label_text)) + + # Normalize switch condition (getType → EmployeeType) + self.switch_condition = self.to_pascal_case(self.switch_condition) + + # Rewrite concrete method body + if self.method_name in self.methods: + old_dec = self.methods[self.method_name] + return_stmt = f"return {self.label_text};" + new_body = "{ " + return_stmt + " }" + self.old_method_dec_with_body_per_case[self.label_text] = old_dec + new_body + + # Handle case body + self.case_has_return = "return" in self.body + if self.body and self.body[0] != "{": + self.body = "{\n\t" + self.body + "\n\t}" + + def exitSwitchBlockStatementGroup(self, ctx: JavaParserLabeled.SwitchBlockStatementGroupContext): + if not self.method_selected: + return + + for switch_label, label_text in self.current_labels: + case_info = { + "label": switch_label, + "switch_label": label_text, + "body": self.body, + "case_has_return": "return" in self.body, + "instance_fields": set(self.current_case_instance_fields) + } + self.pending_cases.append(case_info) + self.current_case_instance_fields.clear() + + # Clear labels for next group + self.current_labels.clear() + + def enterStatement10(self, ctx: JavaParserLabeled.StatementContext): + if not self.current_method: + return + start = ctx.start.tokenIndex + stop = ctx.stop.tokenIndex + text = self.token_stream_rewriter.getText("", start, stop).strip() + # Capture ONLY method-level return + if text.startswith("return"): + self.final_method_return = text + + def exitMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + # Create abstract method + if self.method_name in self.methods: + abstract_dec = self.methods[self.method_name] + if "abstract" not in abstract_dec: + # insert after visibility modifier if present, otherwise at the beginning + if abstract_dec.startswith(("public ", "protected ", "private ")): + parts = abstract_dec.split(" ", 1) + # abstract_dec = parts[0] + " abstract " + parts[1] + abstract_dec = parts[1] + self.new_abstract_method = abstract_dec + ";" + else: + # abstract_dec = "abstract " + self.modified_old_method_declaration + abstract_dec = self.modified_old_method_declaration + self.new_abstract_method = abstract_dec + ";" + print("Modified getter declaration (abstract modifier):", self.new_abstract_method) + + if not self.current_method: + return + + if getattr(self, "params_injected", False): + return + + self.extra_params = [] + receiver_name = None + + # Case 1: locals exist + if self.local_vars: + self.extra_params = [f"{t} {n}" for t, n in self.local_vars] + + # Case 2: no locals, but instance fields used + elif self.uses_instance_fields: + receiver_type = self.class_identifier + receiver_name = receiver_type[0].lower() + receiver_type[1:] + self.extra_params = [f"{receiver_type} {receiver_name}"] + + # Inject into method signature + paren_index = self.old_method_declaration.find("(") + if paren_index == -1: + return # still invalid method signature, keep this check + + before = self.old_method_declaration[:paren_index + 1] + after = self.old_method_declaration[paren_index + 1:] + + if after.strip().startswith(")"): + self.modified_old_method_declaration = before + if self.extra_params: + self.modified_old_method_declaration += ", ".join(self.extra_params) + self.modified_old_method_declaration += after + + self.new_abstract_method_declaration = self.modified_old_method_declaration + else: + self.modified_old_method_declaration = before + if self.extra_params: + self.modified_old_method_declaration += ", ".join(self.extra_params) + ", " + self.modified_old_method_declaration += after + + self.new_abstract_method_declaration = self.modified_old_method_declaration + + self.params_injected = bool(self.extra_params) + print("Modified declaration (method signature) :", self.modified_old_method_declaration) + + if "abstract" not in self.modified_old_method_declaration: + # insert after visibility modifier if present, otherwise at the beginning + if self.modified_old_method_declaration.startswith(("public ", "protected ", "private ")): + parts = self.modified_old_method_declaration.split(" ", 1) + self.new_abstract_method_declaration = parts[1] + else: + self.new_abstract_method_declaration = self.modified_old_method_declaration + print("Modified declaration (abstract modifier):", self.new_abstract_method_declaration) + + for case in self.pending_cases: + body = case["body"] + if case["instance_fields"]: + body = self.rewrite_body_with_receiver( + body, + receiver_name + ) + + if body and body[0] != "{": + body = "{\n\t" + body + "\n\t}" + + case_label = case["switch_label"] + label = case["label"] + new_sub_class = ( + "\n public class " + label + + " implements " + self.switch_condition + "{\n\t" + + "@Override" + "\n\t" + + self.old_method_dec_with_body_per_case[case_label] + + "\n\t" + "@Override" + "\n\t" + + self.modified_old_method_declaration + ) + + self.generated_subclasses.append({ + "label": label, + "header": new_sub_class, + "body": body, + "case_has_return": case["case_has_return"] + }) + + def exitClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + # print("....", self.method_selected, self.switch) + if self.method_selected and self.switch: + for case in self.generated_subclasses: + body = case["body"] + + # Append method-level return ONLY if case has no return + if not case.get("case_has_return", False) and getattr(self, "final_method_return", None): + # Ensure proper indentation + body = body.rstrip("}") + f"\t\t{self.final_method_return}\n}}" + case["body"] = body + + self.newClasses += case["header"] + body + "\n}" + self._write_single_strategy_file(case) + + # ---- Generate Null Object ---- + if self.null_object_body: + null_class_name = f"Null{self.switch_condition}" + + null_method_body = self.null_object_body + if null_method_body and null_method_body[0] != "{": + null_method_body = "{\n\t" + null_method_body + "\n\t}" + + null_class = ( + f"\npublic class {null_class_name} implements {self.switch_condition} {{\n\t" + f"@Override \n\t {self.modified_old_method_declaration}\n\t" + f"{null_method_body}\n}}" + ) + + self.newClasses += null_class + self._write_null_object_file() + + self.generated_subclasses.clear() + + # Build abstract class + self.abstractClass = ( + f"\npublic interface {self.switch_condition} {{\n\t" + f"{self.new_abstract_method}\n\t" + f"{self.new_abstract_method_declaration};\n}}" + ) + self._write_abstract_strategy_file(self.base_dir) + + print("New subclasses:", self.newClasses) + print("Interface:", self.abstractClass) + + self.enter_method = False + self.method_selected = False + self.enter_class = False + + # Step 1: Replace int type with Type + # Find token indices for 'private int type;' + self.replace_type_variable() + + # Step 2: replace switch with delegation + if self.switch: + # Assuming last Statement8Context is the switch + self.replace_switch_with_delegation() + + # Reset + self.pending_cases.clear() + self.switch = False + + def to_pascal_case(self, label: str) -> str: + parts = label.lower().split('_') + return ''.join(word.capitalize() for word in parts) + + def rewrite_body_with_receiver(self, body, receiver_name): + for field in self.instance_fields: + body = re.sub( + rf"\b{field}\b", + f"{receiver_name}.{field}", + body + ) + return body + + def replace_type_variable(self): + """ + Replace any int/enum field used as a type in a switch with the new abstract class type. + This modifies: + 1. Field declaration type + 2. Getter return value + Handles both void and non-void methods and works for any field used as switch condition. + """ + abstract_type = self.switch_condition # the new abstract class type + + for field_name, field_type in self.private_fields.items(): + # Only replace if this field was used in the switch + # if field_name.lower() == self.switch_condition_name.lower(): + if field_name == self.switch_field_name: + + # Replace field declaration type + field_ctx = self.field_contexts.get(field_name) + if field_ctx: + start = field_ctx.typeType().start.tokenIndex + stop = field_ctx.typeType().stop.tokenIndex + self.token_stream_rewriter.replaceRange(start, stop, abstract_type) + print(f"Replaced field '{field_name}' type '{field_type}' with '{abstract_type}'") + + # Replace getter return type + # getter_name = f"get{field_name[0].upper() + field_name[1:]}" + # getter_ctx = self.method_contexts.get(getter_name) + getter_ctx = self.method_contexts.get(self.switch_name) + if getter_ctx: + body = getter_ctx.methodBody() + if body and body.block(): + for stmt in body.block().blockStatement(): + statement = stmt.statement() + if statement and statement.RETURN(): + expr = statement.expression() + if expr: + start = expr.start.tokenIndex + stop = expr.stop.tokenIndex + new_expr = f"{field_name}.{self.switch_condition_name}" + self.token_stream_rewriter.replaceRange(start, stop, new_expr) + print("Updated getter return delegation") + + # Initialize with Null Object + field_ctx = self.field_contexts.get(field_name) + if field_ctx and self.null_object_body: + parent = field_ctx.parentCtx.parentCtx + insert_index = parent.stop.tokenIndex - 1 + self.token_stream_rewriter.insertAfter( + insert_index, + f"\n\t{field_name} = new Null{abstract_type}()" + ) + + def _write_single_strategy_file(self, case): + package_path = "" + if self.package_name: + package_path = ( + self.package_name + .replace("package ", "") + .replace(";", "") + .replace(".", os.sep) + ) + output_dir = os.path.join(self.base_dir, package_path) + else: + # No package -> put next to original class file + output_dir = "C:/Users/98910/Desktop/all_example_for_type_checking/Video_Store_7/Video_Store_7" + + # Make sure directory exists + os.makedirs(output_dir, exist_ok=True) + + class_name = case["label"] + file_path = os.path.join(output_dir, f"{class_name}.java") + + if os.path.exists(file_path): + # File exists: append before last closing brace + with open(file_path, "r", encoding="utf8") as f: + content = f.read() + + # Find last closing brace + last_brace_index = content.rfind("}") + if last_brace_index == -1: + # No closing brace found, append at end + new_content = content + "\n" + self.modified_old_method_declaration + "\n}\n" + else: + new_content = content[:last_brace_index] + "\n" + self.modified_old_method_declaration + "\n}\n" + + with open(file_path, "w", encoding="utf8") as f: + f.write(new_content) + + print(f"Appended to {file_path}") + + else: + print("uuuu") + # File does not exist: create normally + with open(file_path, "w", encoding="utf8") as f: + if self.package_name: + f.write(self.package_name + ";\n\n") + + f.write(case["header"]) + f.write(case["body"]) + f.write("\n}\n") + + print(f"Created {file_path}") + + def _write_abstract_strategy_file(self, base_dir: str): + + if not self.abstractClass: + return + + package_path = "" + if self.package_name: + package_path = ( + self.package_name + .replace("package ", "") + .replace(";", "") + .replace(".", os.sep) + ) + + output_dir = os.path.join(base_dir, package_path) + else: + output_dir = "C:/Users/98910/Desktop/all_example_for_type_checking/Video_Store_7/Video_Store_7" + + os.makedirs(output_dir, exist_ok=True) + + class_name = self.switch_condition + file_path = os.path.join(output_dir, f"{class_name}.java") + + # CASE 1: File does not exist → create it + if not os.path.exists(file_path): + + with open(file_path, "w", encoding="utf8") as f: + if self.package_name: + f.write(self.package_name + ";\n\n") + + f.write(self.abstractClass) + + print(f"Created abstract strategy: {file_path}") + return + + # CASE 2: File exists → append safely + with open(file_path, "r", encoding="utf8") as f: + content = f.read() + + last_brace_index = content.rfind("}") + if last_brace_index == -1: + # corrupted file fallback + new_content = content + "\n" + self.new_abstract_method_declaration + "\n}\n" + else: + new_content = ( + content[:last_brace_index] + + "\n" + + self.new_abstract_method_declaration + + "\n}\n" + ) + + with open(file_path, "w", encoding="utf8") as f: + f.write(new_content) + + print(f"Updated abstract strategy: {file_path}") + + def replace_switch_with_delegation(self): + """ + Replace switch with unconditional delegation. + Default behavior is moved into a Null Object. + """ + + if not self.switch_stmt_ctx or not self.selected_method_ctx: + return + + # type_var = self.switch_condition_name # e.g., type + type_var = self.switch_field_name + method_decl = self.new_abstract_method_declaration + match = re.search(r"\((.*)\)", method_decl) + param_str = "" + param_names = [] + if match: + param_str = match.group(1) + # Extract just variable names + for p in param_str.split(","): + p = p.strip() # remove spaces + if p: + name = p.split()[-1] # last token is the param name + param_names.append(name) + params_str = ", ".join(param_names) + method_name = f"{self.switch_method_name}({params_str})" + # method_name = self.switch_method_name # e.g., m + + default_body = "" + + for block in self.switch_stmt_ctx.switchBlockStatementGroup(): + for label in block.switchLabel(): + if label.DEFAULT(): + stmts = block.blockStatement() + default_body = "\n".join(stmt.getText() for stmt in stmts) + + # TRUE Null Object delegation (no condition!) + delegation_code = f"{type_var}.{method_name};" + + start = self.switch_stmt_ctx.start.tokenIndex + stop = self.switch_stmt_ctx.stop.tokenIndex + + self.token_stream_rewriter.replaceRange(start, stop, delegation_code) + + def write_generated_strategy_files(self, base_dir: str): + """ + Write all generated subclasses into the **same package** as the base class. + """ + # Compute folder based on package + package_path = "" + if self.package_name: + package_path = self.package_name.replace("package ", "").replace(";", "").replace(".", os.sep) + output_dir = os.path.join(base_dir, package_path) + os.makedirs(output_dir, exist_ok=True) + + # Write each generated subclass + for case in self.generated_subclasses: + class_name = case["label"] + file_path = os.path.join(output_dir, f"{class_name}.java") + with open(file_path, "w", encoding="utf8") as f: + if self.package_name: + f.write(self.package_name + ";\n\n") + f.write(case["header"]) + f.write(case["body"]) + print(f"Wrote {file_path}") + + def _write_null_object_file(self): + if not self.null_object_body: + return + + package_path = "" + if self.package_name: + package_path = ( + self.package_name + .replace("package ", "") + .replace(";", "") + .replace(".", os.sep) + ) + + output_dir = os.path.join(self.base_dir, package_path) + os.makedirs(output_dir, exist_ok=True) + + class_name = f"Null{self.switch_condition}" + file_path = os.path.join(output_dir, f"{class_name}.java") + + body = self.null_object_body + if body and not body.startswith("{"): + body = "{\n\t" + body + "\n\t}" + + with open(file_path, "w", encoding="utf8") as f: + if self.package_name: + f.write(self.package_name + ";\n\n") + + f.write( + f"public class {class_name} implements {self.switch_condition} {{\n\t" + f"{self.modified_old_method_declaration}\n\t" + f"{body}\n}}\n" + ) + + print(f"Wrote {file_path}") + + +def main(args): + begin_time = time() + + # Step 1: Load input source + stream = FileStream(args.file, encoding='utf8', errors='ignore') + + # Step 2: Lexer + lexer = JavaLexer(stream) + + # Step 3: Token stream + token_stream = CommonTokenStream(lexer) + + # Step 4: Parser + parser = JavaParserLabeled(token_stream) + + # Step 5: Parse tree + parse_tree = parser.compilationUnit() + + # Step 6: Listener + my_listener = StrategyPatternRefactoringListenerForSwitch( + common_token_stream=token_stream, + method_identifier='getCharge' + ) + + # Step 7: Walk tree (THIS populates abstractClass & newClasses) + walker = ParseTreeWalker() + walker.walk(t=parse_tree, listener=my_listener) + + # Step 9: Write refactored base class + refactored_code = my_listener.token_stream_rewriter.getDefaultText() + + print('Compiler result:') + print(refactored_code) + + with open(args.file, "w", encoding="utf8") as f: + f.write(refactored_code) + + end_time = time() + print("execute time : ", end_time - begin_time) + + +if __name__ == '__main__': + argparser = argparse.ArgumentParser() + argparser.add_argument( + '-n', '--file', + help='Input source', + default=r'C:\Users\98910\Desktop\all_example_for_type_checking\Video_Store_7\Video_Store_7\Movie.java') + args_ = argparser.parse_args() + main(args_) diff --git a/codart/refactorings/strategy_pattern_refactoring_if.py b/codart/refactorings/strategy_pattern_refactoring_if.py new file mode 100644 index 00000000..d92d19bf --- /dev/null +++ b/codart/refactorings/strategy_pattern_refactoring_if.py @@ -0,0 +1,840 @@ +import argparse +import re +import os +from collections import defaultdict +from time import time + +from antlr4 import * +from antlr4.TokenStreamRewriter import TokenStreamRewriter + +from codart.gen.JavaLexer import JavaLexer +from codart.gen.JavaParserLabeled import JavaParserLabeled +from codart.gen.JavaParserLabeledListener import JavaParserLabeledListener + + +class StrategyPatternRefactoringListenerForIfElse(JavaParserLabeledListener): + """ + Implementing State/Strategy pattern to refactor type checking + """ + + def __init__(self, common_token_stream: CommonTokenStream = None, method_identifier: str = None, smell_line=None): + """ + :param common_token_stream: + """ + self.smell_line = smell_line + self.capture_body_for = None + self.is_else_block = False + self.base_dir = "C:/Users/98910/Desktop/type_checking_examples_before_refactor/if" + self.getter_signature = None + self.package_name = None + self.condition_field = "" + self.method_identifier = method_identifier + self.class_identifier = "" + self.class_fields = set() + self.private_fields = {} + self.local_vars = [] + self.old_method_Declaration = "" + self.new_abstract_method_Declaration = "" + self.modified_old_method_Declaration = "" + self.new_abstract_method = "" + self.method_selected = False + self.switch = False + self.switch_condition = "" + self.method_name = "" + self.class_methods = {} + self.getter_declaration = "" + self.current_method = None + self.method_returns = {} + self.method_returns = defaultdict(list) + self.enter_method = False + self.methods = {} + self.methods2 = {} + self.if_else = False + self.para = "" + self.newClasses = "" + self.body = "" + self.switch_label = "" + self.old_method_Declaration2 = "" + self.new_method_Declaration = "" + self.inputPara = False + self.newPara = [] + self.oldPara = [] + self.typePara = [] + self.i = 0 + self.token_stream = common_token_stream + self.method_identifie = "" + self.enter_class = False + self.interface = "" + self.abstractClass = "" + self.if_has_return = False + self.getter_Declaration = [] + self.new_sub_class = "" + self.getters = {} + self.filtered_methods = "" + self.method_return = "" + self.pending_cases = [] + self.final_method_return = None + self.method_return = None + self.pending_cases = [] + self.new_abstract_method_declaration = "" + self.package_name = "" + self.enter_class = False + self.class_identifier = "" + self.class_fields = set() + self.field_contexts = {} + self.current_method = None + self.uses_instance_fields = False + self.instance_fields = set() + self.current_case_instance_fields = set() + self.method_identifier = method_identifier + self.method_contexts = {} + self.methods = {} + self.enter_method = False + self.method_selected = False + self.selected_method_ctx = {} + self.switch_method_name = "" + self.method_returns = defaultdict(list) + self.local_vars = [] + self.old_method_declaration = "" + self.modified_old_method_declaration = "" + self.method_parameters = set() + self.local_variables = set() + self.if_stmt_ctx = {} + self.switch_condition_name = "" + self.switch = False + self.switch_condition = None + self.body = "" + self.case_has_return = False + self.old_method_dec_with_body_per_case = {} + self.params_injected = False + self.token_stream = common_token_stream + self.newClasses = "" + self.abstractClass = "" + self.switch_label = "" + self.new_abstract_method_declaration = "" + self.method_name = "" + self.method_return = "" + self.new_abstract_method = "" + self.private_fields = {} + self.class_f = {} + self.pending_cases = [] + self.generated_subclasses = [] + self.final_method_return = None + self.label_text = "" + self.static_fields = set() + self.if_expression_text = "" + self.current_if_label = "" + self.null_object_body = None + self.flag = False + self.constructor_contexts = [] + self.strategy_required_fields = set() + # Move all the tokens in the source code in a buffer, token_stream_rewriter. + if common_token_stream is not None: + self.token_stream_rewriter = TokenStreamRewriter(common_token_stream) + else: + raise TypeError('common_token_stream is None') + + def enterPackageDeclaration(self, ctx: JavaParserLabeled.PackageDeclarationContext): + self.package_name = ctx.qualifiedName().getText() + + def enterCompilationUnit(self, ctx: JavaParserLabeled.CompilationUnitContext): + self.package_name = "" + + def enterClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + self.enter_class = True + self.class_identifier = ctx.IDENTIFIER().getText() + + def enterConstructorDeclaration(self, ctx): + if not hasattr(self, "constructor_contexts"): + self.constructor_contexts = [] + self.constructor_contexts.append(ctx) + + def enterFieldDeclaration(self, ctx: JavaParserLabeled.FieldDeclarationContext): + # Save class field names + for declarator in ctx.variableDeclarators().variableDeclarator(): + field_name = declarator.variableDeclaratorId().getText() + field_type = ctx.typeType().getText() + self.class_f[field_name] = field_type + self.class_fields.add(field_name) + # Save context for later type replacement + if not hasattr(self, "field_contexts"): + self.field_contexts = {} + self.field_contexts[field_name] = ctx + + # Go up to classBodyDeclaration + class_body_dec = ctx.parentCtx.parentCtx + modifiers = [] + if hasattr(class_body_dec, "modifier"): + modifiers = [m.getText() for m in class_body_dec.modifier()] + if "private" in modifiers: + field_type = ctx.typeType().getText() + var_dec = ctx.variableDeclarators().variableDeclarator(0) + field_name = var_dec.variableDeclaratorId().getText() + self.private_fields[field_name] = field_type + # print(f"Collected private field: {field_name} ({field_type})") + if "static" in modifiers: + var_dec = ctx.variableDeclarators().variableDeclarator(0) + field_name = var_dec.variableDeclaratorId().getText() + self.static_fields.add(field_name) + + def enterMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + method_name = ctx.IDENTIFIER().getText() + self.method_contexts[method_name] = ctx + member_ctx = ctx.parentCtx # MemberDeclaration* + class_body_ctx = member_ctx.parentCtx # ClassBodyDeclaration + + # extract modifiers if present + modifiers = [] + if hasattr(class_body_ctx, "modifier"): + modifiers = class_body_ctx.modifier() + + if modifiers: + start_index = modifiers[0].start.tokenIndex + else: + # fallback: start at method identifier + start_index = ctx.IDENTIFIER().symbol.tokenIndex + + end_index = ctx.formalParameters().stop.tokenIndex + + method_dec = self.token_stream_rewriter.getText( + "", + start_index, + end_index + ) + + self.methods[method_name] = method_dec + if self.method_identifier is None or method_name == self.method_identifier: + self.current_method = method_name + self.uses_instance_fields = False # reset for this method + if method_name == self.method_identifier: + self.enter_method = True + self.method_selected = True + self.selected_method_ctx = ctx + self.switch_method_name = method_name + self.current_method = method_name + self.method_returns[self.current_method] = [] + self.local_vars = [] + + self.old_method_declaration = method_dec + self.modified_old_method_declaration = method_dec + print("Old declaration of method:", self.old_method_declaration) + + def enterFormalParameter(self, ctx): + if self.method_selected: + self.inputPara = True + self.oldPara.append(ctx.typeType().getText() + " " + ctx.variableDeclaratorId().getText()) + self.typePara.append(ctx.typeType().getText()) + param_name = ctx.variableDeclaratorId().getText() + self.method_parameters.add(param_name) + + def enterLocalVariableDeclaration(self, ctx: JavaParserLabeled.LocalVariableDeclarationContext): + for dec in ctx.variableDeclarators().variableDeclarator(): + var_name = dec.variableDeclaratorId().getText() + self.local_variables.add(var_name) + + if not self.current_method: + return + + var_type = ctx.typeType().getText() + declarator = ctx.variableDeclarators().variableDeclarator(0) + var_name = declarator.variableDeclaratorId().getText() + # Collect locals + if (var_type, var_name) not in self.local_vars: + self.local_vars.append((var_type, var_name)) + + def enterExpression0(self, ctx: JavaParserLabeled.Expression0Context): + if not self.current_method: + return + name = ctx.getText() + # skip qualified access + if "." in name: + return + # skip locals, parameters, static fields + if name in self.local_variables or name in self.method_parameters or name in self.static_fields: + return + # check exact match against instance fields + if name in self.private_fields: + self.uses_instance_fields = True + self.instance_fields.add(name) + self.current_case_instance_fields.add(name) + + # def enterMethodBody(self, ctx:JavaParserLabeled.MethodBodyContext): + # if self.method_selected: + # self.if_stmt_ctx = ctx.block() + + def enterStatement2(self, ctx: JavaParserLabeled.Statement2Context): + if not ctx or not self.method_selected: + return + + if self.smell_line and ctx.start.line != self.smell_line: + return + self.if_stmt_ctx = ctx + + # Mark that we are inside an if-based type check + self.switch = True # reuse same pipeline + self.if_else = True + + expr = ctx.parExpression().expression() + self.if_expression_text = expr.getText() + + # Expected patterns: + # if (type == VALUE) + # if (getType() == VALUE) + + # Case 1: getType() == VALUE + if hasattr(expr, "binaryOp") or expr.getChildCount() >= 3: + left = expr.getChild(0).getText() + right = expr.getChild(2).getText() + + # Normalize condition source + if left.startswith("get"): + self.method_name = left + self.switch_condition = left[3:] + else: + self.switch_condition = left + self.method_name = f"get{left.capitalize()}" + + + # Normalize label (ENGINEER → Employee.ENGINEER) + if "." not in right: + right = f"{self.class_identifier}.{right}" + + self.label_text = right + self.switch_label = self.to_pascal_case(right.split(".")[-1]) + + self.switch_condition_name = self.switch_condition + # Normalize switch condition (type → Type) + self.switch_condition = self.to_pascal_case(self.switch_condition) + + # --- Capture IF body --- + # Mark that we need to capture the body for this if block + self.capture_body_for = self.switch_label + + # ----- Rewrite concrete getter if needed ----- + if self.method_name in self.methods: + old_dec = self.methods[self.method_name] + return_stmt = f"return {right};" + new_body = "{ " + return_stmt + " }" + # Store mapping using switch_label + self.old_method_dec_with_body_per_case[self.switch_label] = old_dec + " " + new_body + # self.old_method_dec_with_body_per_case[right] = old_dec + new_body + + # Track that this is not an else block + self.is_else_block = False + + def enterStatement0(self, ctx: JavaParserLabeled.StatementContext): + if not ctx or not self.method_selected: + return + + if self.smell_line and ctx.start.line != self.smell_line: + return + start = ctx.start.tokenIndex + stop = ctx.stop.tokenIndex + + # Get the full text of the block + body_text = self.token_stream_rewriter.getText("", start, stop).strip() + + # Remove surrounding braces if any + if body_text.startswith("{") and body_text.endswith("}"): + body_text = body_text[1:-1].strip() + + # Decide class label + if getattr(self, "capture_body_for", None): + # Regular if/else if block + class_label = self.capture_body_for + body = body_text + self.capture_body_for = None + else: + # Else block → NullType + class_label = f"Null{self.switch_condition}" + body = body_text + self.is_else_block = True + + # Track return statements + case_has_return = bool(re.search(r"\breturn\b", body)) + + # Prepare case info + case_info = { + "label": class_label, + "switch_label": class_label, + "body": body, + "case_has_return": case_has_return, + "instance_fields": set(self.current_case_instance_fields) + } + # Track globally required fields for abstract class + self.strategy_required_fields.update(self.current_case_instance_fields) + + # Avoid duplicates + if class_label not in [c["switch_label"] for c in self.pending_cases]: + self.pending_cases.append(case_info) + + # Clear temporary data + self.current_case_instance_fields.clear() + + # Optional: print captured body + print(f"Captured {class_label} body:\n{body}\n") + + def enterStatement10(self, ctx: JavaParserLabeled.StatementContext): + if not self.current_method: + return + start = ctx.start.tokenIndex + stop = ctx.stop.tokenIndex + text = self.token_stream_rewriter.getText("", start, stop).strip() + # Capture ONLY method-level return + if text.startswith("return"): + self.final_method_return = text + + def exitMethodDeclaration(self, ctx: JavaParserLabeled.MethodDeclarationContext): + # 1. Create ABSTRACT getter / discriminator method + if self.method_name in self.methods: + abstract_dec = self.methods[self.method_name] + + if "abstract" not in abstract_dec: + if abstract_dec.startswith(("public ", "protected ", "private ")): + vis, rest = abstract_dec.split(" ", 1) + abstract_dec = f"{rest}" + else: + abstract_dec = f"{abstract_dec}" + + self.new_abstract_method = abstract_dec + ";" + print("Modified getter declaration (abstract):", self.new_abstract_method) + + # 2. Sanity checks + if not self.current_method: + return + + if getattr(self, "params_injected", False): + return + + # 3. Decide extra parameters + extra_params = [] + receiver_name = None + + # Case 1: local variables used + if self.local_vars: + extra_params = [f"{t} {n}" for t, n in self.local_vars] + + # Case 2: instance fields used + elif self.uses_instance_fields: + receiver_type = self.class_identifier + receiver_name = receiver_type[0].lower() + receiver_type[1:] + extra_params = [f"{receiver_type} {receiver_name}"] + + # 4. Inject parameters into method signature + old_decl = self.old_method_declaration + paren_index = old_decl.find("(") + if paren_index == -1: + return # invalid signature, stay safe + + before = old_decl[:paren_index + 1] + after = old_decl[paren_index + 1:] + + self.modified_old_method_declaration = before + + if after.strip().startswith(")"): + if extra_params: + self.modified_old_method_declaration += ", ".join(extra_params) + self.modified_old_method_declaration += after + else: + if extra_params: + self.modified_old_method_declaration += ", ".join(extra_params) + ", " + self.modified_old_method_declaration += after + + self.params_injected = bool(extra_params) + + print("Modified declaration (signature):", + self.modified_old_method_declaration) + + # 5. Make the method ABSTRACT + if "abstract" not in self.modified_old_method_declaration: + if self.modified_old_method_declaration.startswith( + ("public ", "protected ", "private ") + ): + vis, rest = self.modified_old_method_declaration.split(" ", 1) + self.new_abstract_method_declaration = f"{rest}" + else: + self.new_abstract_method_declaration = ( + f"{self.modified_old_method_declaration}" + ) + else: + self.new_abstract_method_declaration = ( + self.modified_old_method_declaration + ) + + print("Modified declaration (abstract):", + self.new_abstract_method_declaration) + + # 6. Generate subclasses (works for SWITCH and IF) + for case in self.pending_cases: + body = case["body"] + if case["instance_fields"] and receiver_name: + body = self.rewrite_body_with_receiver(body, receiver_name) + if body and not body.startswith("{"): + body = "{\n\t" + body + "\n\t}" + case_label = case["switch_label"] + subclass_name = case["label"] + + abstract_name = self.switch_condition.split('.')[0] + abstract_name = abstract_name[0].upper() + abstract_name[1:] + + new_sub_class = ( + "\npublic class " + subclass_name + + " implements " + abstract_name + " {\n\t" + + "@Override" + "\n\t" + + self.old_method_dec_with_body_per_case.get(case_label, "") + + "\n\t" + "@Override" + "\n\t" + self.modified_old_method_declaration + ) + + self.generated_subclasses.append({ + "label": subclass_name, + "header": new_sub_class, + "body": body, + "case_has_return": case["case_has_return"] + }) + + def exitClassDeclaration(self, ctx: JavaParserLabeled.ClassDeclarationContext): + if not self.method_selected or not self.switch: + return + + # --- Write all subclasses --- + for case in self.generated_subclasses: + body = case["body"] + if not case.get("case_has_return", False) and getattr(self, "final_method_return", None): + body = body.rstrip("}") + f"\n\t\t{self.final_method_return}\n\t}}" + case["body"] = body + + self._write_single_strategy_file(case) + self.newClasses += case["header"] + body + "\n}" + + # --- Null Object --- + if getattr(self, "null_object_body", None): + null_class_name = f"Null{self.switch_condition}" + if any(c['label'] == null_class_name for c in self.generated_subclasses): + null_class_name += "Default" + null_method_body = self.null_object_body + if null_method_body and not null_method_body.startswith("{"): + null_method_body = "{\n\t" + null_method_body + "\n\t}" + null_class = ( + f"\npublic class {null_class_name} implements {self.switch_condition} {{\n\t" + f"{self.modified_old_method_declaration}\n\t" + f"{null_method_body}\n}}" + ) + self.newClasses += null_class + self._write_null_object_file() + + # --- Abstract base class --- + abstract_name = self.switch_condition.split('.')[0] + abstract_name = abstract_name[0].upper() + abstract_name[1:] + self.abstractClass = ( + f"\npublic interface {abstract_name} {{\n\t" + f"{self.new_abstract_method}\n\t" + f"{self.new_abstract_method_declaration};\n}}" + ) + + self._write_abstract_strategy_file(self.base_dir) + + print("Generated subclasses:", self.newClasses) + print("Generated interface:", self.abstractClass) + + # --- Reset listener state --- + self.enter_method = False + self.method_selected = False + self.enter_class = False + self.switch = False + + # --- Replace int type with Type field --- + self.replace_type_variable() + self.replace_constructor_parameter_type() + if self.if_else: + self.replace_if_else_with_delegation() + self.if_else = False + + def to_pascal_case(self, label: str) -> str: + parts = label.lower().split('_') + return ''.join(word.capitalize() for word in parts) + + def make_getter(self, field_name, field_type): + getter_name = "get" + field_name[0].upper() + field_name[1:] + return f"{field_type} {getter_name}();" + + def rewrite_body_with_receiver(self, body, receiver_name): + for field in self.instance_fields: + body = re.sub( + rf"\b{field}\b", + f"{receiver_name}.{field}", + body + ) + return body + + def replace_type_variable(self): + """ + Replace any int/enum field used as a type in a switch with the new abstract class type. + This modifies: + 1. Field declaration type + 2. Getter return type + 3. Setter parameter type + Handles both void and non-void methods and works for any field used as switch condition. + """ + # Replace field type, getter, setter, and initialize with Null Object + abstract_type = self.switch_condition # e.g., "Type" + + for field_name, field_type in self.class_f.items(): + if field_name.lower() != self.switch_condition.lower(): + continue + + field_ctx = self.field_contexts.get(field_name) + if field_ctx: + # Replace field type + start = field_ctx.typeType().start.tokenIndex + stop = field_ctx.typeType().stop.tokenIndex + self.token_stream_rewriter.replaceRange(start, stop, abstract_type) + print(f"Replaced field '{field_name}' type with '{abstract_type}'") + + # Initialize field with Null Object + stop_token = field_ctx.stop + insert_index = stop_token.tokenIndex + 1 # insert AFTER semicolon + + # Prepare the assignment code + # code = f"\n\t{field_name} = new Null{abstract_type}();\n" + # Pass required fields into Null object constructor + args = ", ".join(self.strategy_required_fields) + + code = f"\n\t{field_name} = new Null{abstract_type}({args});\n" + + # Insert it using the rewriter + self.token_stream_rewriter.insertBefore( + program_name=self.token_stream_rewriter.DEFAULT_PROGRAM_NAME, + index=insert_index, + text=code + ) + print(f"Inserted Null Object assignment for field '{field_name}' after declaration") + + # Replace getter return type + getter_name = f"get{field_name[0].upper() + field_name[1:]}" + getter_ctx = self.method_contexts.get(getter_name) + if getter_ctx: + result_ctx = getter_ctx.typeTypeOrVoid() # call the method + if result_ctx.typeType: + start = result_ctx.start.tokenIndex + stop = result_ctx.stop.tokenIndex + self.token_stream_rewriter.replaceRange(start, stop, abstract_type) + print(f"Replaced getter '{getter_name}' return type with '{abstract_type}'") + + # Replace setter parameter type + setter_name = f"set{field_name[0].upper() + field_name[1:]}" + setter_ctx = self.method_contexts.get(setter_name) + if setter_ctx: + param_list_ctx = setter_ctx.formalParameters().formalParameterList() + if param_list_ctx: + for param in param_list_ctx.formalParameter(): + param_name = param.variableDeclaratorId().IDENTIFIER().getText() + if param_name.lower() == field_name.lower(): + start = param.typeType().start.tokenIndex + stop = param.typeType().stop.tokenIndex + self.token_stream_rewriter.replaceRange(start, stop, abstract_type) + print(f"Replaced setter '{setter_name}' parameter type with '{abstract_type}'") + + def _write_single_strategy_file(self, case): + package_path = "" + if self.package_name: + package_path = ( + self.package_name + .replace("package ", "") + .replace(";", "") + .replace(".", os.sep) + ) + + output_dir = os.path.join(self.base_dir, package_path) + os.makedirs(output_dir, exist_ok=True) + + class_name = case["label"] + file_path = os.path.join(output_dir, f"{class_name}.java") + + with open(file_path, "w", encoding="utf8") as f: + if self.package_name: + f.write("package " + self.package_name + ";\n\n") + + f.write(case["header"]) + f.write(case["body"]) + + # CLOSE THE CLASS + f.write("\n}\n") + + print(f"Wrote {file_path}") + + def _write_abstract_strategy_file(self, base_dir: str): + """ + Write the abstract base strategy class into the same package. + """ + if not self.abstractClass: + return + + # Compute package folder + package_path = "" + if self.package_name: + package_path = ( + self.package_name + .replace("package ", "") + .replace(";", "") + .replace(".", os.sep) + ) + + output_dir = os.path.join(base_dir, package_path) + os.makedirs(output_dir, exist_ok=True) + + abstract_name = self.switch_condition.split('.')[0] + class_name = abstract_name[0].upper() + abstract_name[1:] + # class_name = self.switch_condition + file_path = os.path.join(output_dir, f"{class_name}.java") + + with open(file_path, "w", encoding="utf8") as f: + if self.package_name: + f.write("package " + self.package_name + ";\n\n") + + f.write(self.abstractClass) + + print(f"Wrote abstract strategy: {file_path}") + + def replace_if_else_with_delegation(self): + """ + Replace the entire if/else chain in a method with a delegation to the type object. + The original method body is fully replaced. + """ + + if not self.if_else or not self.switch_method_name: + return + # abstract_name = self.switch_condition.split('.')[0] + # type_var = abstract_name[0].upper() + abstract_name[1:] + type_var = self.switch_condition_name # e.g., "type" + method_name = self.switch_method_name # e.g., "m" + + # Delegation code: simple method call to the polymorphic object + # delegation_code = f"{{\n\t{type_var}.{method_name}();\n}}" + args = ", ".join(self.strategy_required_fields) + + delegation_code = f"{{\n\t{type_var}.{method_name}({args});\n}}" + + start = self.if_stmt_ctx.start.tokenIndex + stop = self.if_stmt_ctx.stop.tokenIndex + + # Replace the full method body with delegation + self.token_stream_rewriter.replaceRange(start, stop, delegation_code) + + + def _write_null_object_file(self): + if not self.null_object_body: + return + + package_path = "" + if self.package_name: + package_path = ( + self.package_name + .replace("package ", "") + .replace(";", "") + .replace(".", os.sep) + ) + + output_dir = os.path.join(self.base_dir, package_path) + os.makedirs(output_dir, exist_ok=True) + + class_name = f"Null{self.switch_condition}" + file_path = os.path.join(output_dir, f"{class_name}.java") + + body = self.null_object_body + if body and not body.startswith("{"): + body = "{\n\t" + body + "\n\t}" + + with open(file_path, "w", encoding="utf8") as f: + if self.package_name: + f.write(self.package_name + ";\n\n") + + f.write( + f"public class {class_name} implements {self.switch_condition} {{\n\t" + f"{self.modified_old_method_declaration}\n\t" + f"{body}\n}}\n" + ) + + print(f"Wrote {file_path}") + + def replace_constructor_parameter_type(self): + + if not hasattr(self, "constructor_contexts"): + return + + abstract_type = self.switch_condition + + for ctx in self.constructor_contexts: + + param_list_ctx = ctx.formalParameters().formalParameterList() + + if not param_list_ctx: + continue + + for param in param_list_ctx.formalParameter(): + + param_name = param.variableDeclaratorId().getText() + + if param_name.lower() == self.switch_condition_name.lower(): + start = param.typeType().start.tokenIndex + stop = param.typeType().stop.tokenIndex + + self.token_stream_rewriter.replaceRange( + start, + stop, + abstract_type + ) + + print("Constructor parameter type replaced") + + def detect_used_fields(self, ctx): + used_fields = set() + + tokens = ctx.getText() + + for field_name in self.class_fields: + if field_name in tokens: + used_fields.add(field_name) + + return used_fields + + +def main(args): + # Step 1: Load input source into stream + begin_time = time() + stream = FileStream(args.file, encoding='utf8', errors='ignore') + # input_stream = StdinStream() + # Step 2: Create an instance of AssignmentStLexer + lexer = JavaLexer(stream) + # Step 3: Convert the input source into a list of tokens + token_stream = CommonTokenStream(lexer) + # Step 4: Create an instance of the AssignmentStParser + parser = JavaParserLabeled(token_stream) + parser.getTokenStream() + # Step 5: Create parse tree + parse_tree = parser.compilationUnit() + # Step 6: Create an instance of the refactoringListener, and send as a parameter the list of tokens to the class + my_listener = StrategyPatternRefactoringListenerForIfElse(common_token_stream=token_stream, + method_identifier='m') + # method_identifier='read') + # method_identifier='write') + + walker = ParseTreeWalker() + walker.walk(t=parse_tree, listener=my_listener) + refactored_code = my_listener.token_stream_rewriter.getDefaultText() + print('Compiler result:') + print(refactored_code) + with open(args.file, "w", encoding="utf8") as f: + f.write(refactored_code) + end_time = time() + print("execute time : ", end_time - begin_time) + + +# Test driver +if __name__ == '__main__': + argparser = argparse.ArgumentParser() + argparser.add_argument( + '-n', '--file', + help='Input source', + default=r'C:\Users\98910\Desktop\type_checking_examples_before_refactor\if\ContextIfStatement.java') + args_ = argparser.parse_args() + main(args_)