fix codestyle

Author: fangfangssj

graph_net/analysis_util.py

@@ -1,8 +1,7 @@
 import os
 import re
-import numpy as np
+import sys
 from scipy.stats import gmean
-from collections import OrderedDict, defaultdict
 from graph_net.config.datatype_tolerance_config import get_precision
 
 
@@ -265,7 +264,7 @@ def scan_all_folders(benchmark_path: str) -> dict:
         print(f"Detected log file: '{benchmark_path}'")
         samples = parse_logs_to_data(benchmark_path)
         if not samples:
-            print(f"  - No valid data found in log file.")
+            print("  - No valid data found in log file.")
             return {}
 
         folder_name = (

graph_net/constraint_util.py

@@ -90,12 +90,12 @@ def __call__(self, model_path):
 
         tensor_metas = self._get_tensor_metas(model_path)
         tensor_meta_attrs_list = [asdict(tensor_meta) for tensor_meta in tensor_metas]
-        logging.warning(f"before create_inputs_by_metas")
+        logging.warning("before create_inputs_by_metas")
         inputs = self.get_dimension_generalizer().create_inputs_by_metas(
             module=self.get_model(model_path),
             tensor_meta_attrs_list=tensor_meta_attrs_list,
         )
-        logging.warning(f"after create_inputs_by_metas")
+        logging.warning("after create_inputs_by_metas")
         dyn_dim_cstr = make_dyn_dim_cstr_from_tensor_metas(tensor_metas)
 
         def data_input_predicator(input_var_name):
@@ -157,23 +157,23 @@ def get_model(self, model_path):
 
     @contextmanager
     def _try_dimension_generalization(self, dim_axes_pairs, model_path, inputs):
-        logging.warning(f"enter _try_dimension_generalization")
+        logging.warning("enter _try_dimension_generalization")
         if self.config["dimension_generalizer_filepath"] is None:
             yield model_path, ()
             return
         model = self.get_model(model_path)
         dim_generalizer = self.get_dimension_generalizer()
         dim_gen_pass = dim_generalizer(model, dim_axes_pairs)
-        logging.warning(f"before need_rewrite")
+        logging.warning("before need_rewrite")
         need_rewrite = dim_gen_pass.need_rewrite(inputs)
-        logging.warning(f"after need_rewrite")
+        logging.warning("after need_rewrite")
         if not need_rewrite:
             yield model_path, ()
             return
 
-        logging.warning(f"before rewrite")
+        logging.warning("before rewrite")
         graph_module = dim_gen_pass.rewrite(inputs)
-        logging.warning(f"after rewrite")
+        logging.warning("after rewrite")
         with tempfile.TemporaryDirectory() as tmp_dir:
             shutil.copytree(Path(model_path), Path(tmp_dir), dirs_exist_ok=True)
             dim_gen_pass.save_graph_module(graph_module, tmp_dir)
@@ -344,17 +344,17 @@ def filter_fn(input_name, input_idx, axis, dim):
             (dim, axes) for dim in unqiue_dims[: i + 1] for axes in [dim2axes[dim]]
         )
         ctx_mgr = dyn_dim_cstr_feasibility_ctx_mgr
-        logging.warning(f"before dyn_dim_cstr_feasibility_ctx_mgr")
+        logging.warning("before dyn_dim_cstr_feasibility_ctx_mgr")
         with ctx_mgr(dim_axes_pairs) as dyn_dim_cstr_feasibility:
-            logging.warning(f"enter dyn_dim_cstr_feasibility_ctx_mgr")
+            logging.warning("enter dyn_dim_cstr_feasibility_ctx_mgr")
             tmp_dyn_dim_cstr = copy.deepcopy(cur_dyn_dim_cstr)
             tmp_dyn_dim_cstr.update_symbol2example_value(sym2example_value)
-            logging.warning(f"before dyn_dim_cstr_feasibility")
+            logging.warning("before dyn_dim_cstr_feasibility")
             is_dyn_dim_cstr_feasible = dyn_dim_cstr_feasibility(tmp_dyn_dim_cstr)
-            logging.warning(f"after dyn_dim_cstr_feasibility")
+            logging.warning("after dyn_dim_cstr_feasibility")
             if not is_dyn_dim_cstr_feasible:
                 continue
             dyn_dim_cstr = cur_dyn_dim_cstr
             append_dim_gen_pass_names(dyn_dim_cstr_feasibility.dim_gen_pass_names)
-            logging.warning(f"leave dyn_dim_cstr_feasibility_ctx_mgr")
+            logging.warning("leave dyn_dim_cstr_feasibility_ctx_mgr")
     return dyn_dim_cstr, total_dim_gen_pass_names

graph_net/dynamic_dim_constraints.py

@@ -1,8 +1,6 @@
-import sys
 import sympy
 import importlib.util as imp
 from dataclasses import dataclass
-import copy
 from typing import Callable
 from collections import namedtuple
 
@@ -22,7 +20,7 @@ class DynamicDimConstraints:
     kRelations = "dynamic_dim_constraint_relations"
 
     # len(input_shapes) equals number of Model.forward arguments
-    input_shapes: list[(tuple[sympy.Expr | int], "var-name")]
+    input_shapes: list[(tuple[sympy.Expr | int], str)]
     kInputShapes = "dynamic_dim_constraint_input_shapes"
 
     @classmethod
@@ -44,7 +42,6 @@ def symbolize(
         ]
         Returns created symbol.
         """
-        import logging
 
         InputDim = namedtuple("InputDim", ["input_idx", "axis", "dim"])
         input_dims = [

graph_net/model_path_handler.py

@@ -1,16 +1,10 @@
 import traceback
 import argparse
-import importlib.util
 from graph_net.imp_util import load_module
-import inspect
 import logging
-from pathlib import Path
-from typing import Type, Any
 import sys
 import json
 import base64
-from contextlib import contextmanager
-import logging
 
 logging.basicConfig(
     level=logging.WARNING, format="%(asctime)s [%(levelname)s] %(message)s"

graph_net/plot_ESt.py

@@ -88,7 +88,7 @@ def print_verification_result(
         )
     else:
         print(
-            f"t={tolerance:3d}: MISMATCH - Microscopic: {microscopic_es:.6f}, Aggregated: {aggregated_es:.6f}, Diff: {diff:.2e} ({relative_diff*100:.4f}%)"
+            f"t={tolerance:3d}: MISMATCH - Microscopic: {microscopic_es:.6f}, Aggregated: {aggregated_es:.6f}, Diff: {diff:.2e} ({relative_diff * 100:.4f}%)"
         )
 
 
@@ -110,9 +110,9 @@ def get_verified_aggregated_es_values(es_scores: dict, folder_name: str) -> dict
     verified_es_values = {}
     mismatches = []
 
-    print(f"\n{'='*80}")
+    print(f"\n{'=' * 80}")
     print(f"Verifying Aggregated/Microscopic Consistency for '{folder_name}'")
-    print(f"{'='*80}")
+    print(f"{'=' * 80}")
 
     for tolerance, microscopic_es in es_scores.items():
         aggregated_es = aggregated_results.get(tolerance)
@@ -136,29 +136,29 @@ def get_verified_aggregated_es_values(es_scores: dict, folder_name: str) -> dict
         elif not is_matched:
             mismatches.append(
                 f"t={tolerance}: Mismatch - Microscopic={microscopic_es:.6f}, "
-                f"Aggregated={aggregated_es:.6f}, Diff={diff:.2e} ({relative_diff*100:.4f}%)"
+                f"Aggregated={aggregated_es:.6f}, Diff={diff:.2e} ({relative_diff * 100:.4f}%)"
             )
         else:
             verified_es_values[tolerance] = microscopic_es
 
     if mismatches:
         error_msg = (
-            f"\n{'='*80}\n"
+            f"\n{'=' * 80}\n"
             f"ERROR: Aggregated and microscopic results do not match for '{folder_name}'!\n"
-            f"{'='*80}\n"
+            f"{'=' * 80}\n"
             f"Mismatches:\n"
             + "\n".join(f"  - {mismatch}" for mismatch in mismatches)
             + f"\n\nCalculation validation failed. Please verify the calculation logic "
             f"using verify_aggregated_params.py\n"
-            f"{'='*80}\n"
+            f"{'=' * 80}\n"
         )
         print(error_msg)
         raise AssertionError(error_msg)
 
     print(
         f"\nSUCCESS: All aggregated and microscopic results match for '{folder_name}'."
     )
-    print(f"{'='*80}\n")
+    print(f"{'=' * 80}\n")
     return verified_es_values
 
 

graph_net/test_compiler_util.py

@@ -56,7 +56,7 @@ def get_device_utilization(device_id, device_count, synchronizer_func):
                 cmd = [
                     "nvidia-smi",
                     f"--id={selected_gpu_id}",
-                    f"--query-gpu=index,gpu_uuid,utilization.gpu,memory.used,memory.total",
+                    "--query-gpu=index,gpu_uuid,utilization.gpu,memory.used,memory.total",
                     "--format=csv,noheader,nounits",
                 ]
                 output = subprocess.check_output(cmd).decode().strip()
@@ -78,7 +78,7 @@ def get_device_utilization(device_id, device_count, synchronizer_func):
             cmd = [
                 "nvidia-smi",
                 f"--id={selected_gpu_id}",
-                f"--query-compute-apps=gpu_uuid,pid,used_memory",
+                "--query-compute-apps=gpu_uuid,pid,used_memory",
                 "--format=csv,noheader,nounits",
             ]
             output = subprocess.check_output(cmd).decode().strip()
@@ -126,14 +126,14 @@ def get_model_name(model_path):
 
     if model_name is None:
         fields = model_path.split(os.sep)
-        pattern = rf"^subgraph(_\d+)?$"
+        pattern = r"^subgraph(_\d+)?$"
         model_name = fields[-2] if re.match(pattern, fields[-1]) else fields[-1]
     return model_name
 
 
 def get_subgraph_tag(model_path):
     fields = model_path.split(os.sep)
-    pattern = rf"^subgraph(_\d+)?$"
+    pattern = r"^subgraph(_\d+)?$"
     return fields[-1] if re.match(pattern, fields[-1]) else ""
 
 

graph_net/verify_aggregated_params.py

@@ -1,12 +1,9 @@
-import os
 import argparse
-import numpy as np
 from collections import OrderedDict, Counter
 from graph_net import analysis_util
 from graph_net import samples_statistics
 from graph_net.samples_statistics import (
     get_errno_from_error_type,
-    get_error_type_from_errno,
 )
 
 
@@ -298,9 +295,9 @@ def verify_es_constructor_params_across_tolerances(
     """
     total_samples = len(samples)
 
-    print(f"\n{'='*80}")
+    print(f"\n{'=' * 80}")
     print(f"Verifying Aggregated Parameters for '{folder_name}'")
-    print(f"{'='*80}")
+    print(f"{'=' * 80}")
 
     tolerances = determine_tolerances(samples)
     builder = ToleranceReportBuilder(
@@ -314,9 +311,9 @@ def verify_es_constructor_params_across_tolerances(
         (tolerance, builder.build_report(tolerance)) for tolerance in tolerances
     )
 
-    print(f"\n{'='*80}")
-    print(f"Aggregated Parameter Verification Complete")
-    print(f"{'='*80}\n")
+    print(f"\n{'=' * 80}")
+    print("Aggregated Parameter Verification Complete")
+    print(f"{'=' * 80}\n")
 
     return results
 
@@ -355,12 +352,12 @@ def main():
 
     # Calculate and print aggregated parameters for each curve
     for folder_name, samples in all_results.items():
-        aggregated_results = verify_es_constructor_params_across_tolerances(
+        _ = verify_es_constructor_params_across_tolerances(
             samples,
             folder_name,
             negative_speedup_penalty=args.negative_speedup_penalty,
             fpdb=args.fpdb,
-        )
+        )  # noqa: F841
 
 
 if __name__ == "__main__":

tools/update_model_configs.py

@@ -77,7 +77,7 @@ def process_single_model(model_dir_path: str, root_dir: str, failures: list):
     relative_path = os.path.relpath(model_dir_path, root_dir)
     potential_model_id = relative_path.replace(os.path.sep, "/")
 
-    print(f"\n{'='*20}\nProcessing: {potential_model_id} (at {model_dir_path})")
+    print(f"\n{'=' * 20}\nProcessing: {potential_model_id} (at {model_dir_path})")
 
     try:
         with open(json_path, "r", encoding="utf-8") as f:
@@ -90,9 +90,9 @@ def process_single_model(model_dir_path: str, root_dir: str, failures: list):
         print(f"  [Phase 1a] Trying precise lookup for '{potential_model_id}'...")
         try:
             best_match_info = model_info(potential_model_id)
-            print(f"  [Phase 1a] Success! Found exact match from path.")
+            print("  [Phase 1a] Success! Found exact match from path.")
         except RepositoryNotFoundError:
-            print(f"  [Phase 1a] Precise lookup failed.")
+            print("  [Phase 1a] Precise lookup failed.")
             # Phase 1b: If that fails, and it's a single-level dir, try replacing the first '_' with '/'
             if "/" not in potential_model_id and "_" in potential_model_id:
                 hypothetical_id = potential_model_id.replace("_", "/", 1)
@@ -102,10 +102,10 @@ def process_single_model(model_dir_path: str, root_dir: str, failures: list):
                 try:
                     best_match_info = model_info(hypothetical_id)
                     print(
-                        f"  [Phase 1b] Success! Found exact match by replacing underscore."
+                        "  [Phase 1b] Success! Found exact match by replacing underscore."
                     )
                 except RepositoryNotFoundError:
-                    print(f"  [Phase 1b] Alternative lookup failed.")
+                    print("  [Phase 1b] Alternative lookup failed.")
 
         # --- Stage 2: Advanced Fuzzy Search (Fallback) ---
         if not best_match_info:
@@ -174,7 +174,7 @@ def process_model_directories(root_dir: str):
     """
     if not root_dir or not os.path.isdir(root_dir):
         print(
-            f"❌ ERROR: Root directory not provided or not found. Please use the --directory argument."
+            "❌ ERROR: Root directory not provided or not found. Please use the --directory argument."
         )
         return
 
@@ -199,7 +199,7 @@ def process_model_directories(root_dir: str):
     for model_path in model_paths_to_process:
         process_single_model(model_path, root_dir, failures)
 
-    print(f"\n{'='*40}\n🎉 All directories processed!")
+    print(f"\n{'=' * 40}\n🎉 All directories processed!")
 
     if failures:
         log_and_print_failures(failures)