Code Fix 2

按照comment更新代码

Author: 3619117923

graph_net/analysis_util.py

@@ -3,7 +3,7 @@
 import sys
 from scipy.stats import gmean
 from graph_net.config.datatype_tolerance_config import get_precision
-from graph_net.positive_tolerance_interpretation_manager import get_positive_tolerance_interpretation
+from graph_net.positive_tolerance_interpretation import PositiveToleranceInterpretation
 from graph_net.verify_aggregated_params import determine_tolerances
 
 
@@ -294,25 +294,22 @@ def get_correctness(dtype: str, t: int, correctness_data: dict, index: int) -> b
         return bool(result[index])
     return False
 
-def fake_perf_degrad(tolerance, error_code, type="default") -> str:
+def fake_perf_degrad(tolerance, error_code, positive_tolerance_interpretation: PositiveToleranceInterpretation,) -> str:
     """
     Judge current correctness based on tolerance t and status.
     Refactored to delegate logic to PositiveToleranceInterpretation classes.
     """
-
-    strategy = get_positive_tolerance_interpretation(type)
-
-    if strategy.is_error_tolerated(tolerance, error_code):
+    if positive_tolerance_interpretation.is_error_tolerated(tolerance, error_code):
         return "correct"
 
     return error_code
 
 def calculate_scores(
     samples: list,
+    positive_tolerance_interpretation: PositiveToleranceInterpretation,
     p: float = 0,
     b: float = 0.1,
     type: str = "ESt",
-    interpretation_type: str = "default",
 ) -> tuple:
     """
     Use a standard tolerance to evaluate all samples and calculate S(t) and ES(t) scores for each tolerance level.
@@ -324,8 +321,8 @@ def calculate_scores(
 
     scores = {}
 
-    strategy = get_positive_tolerance_interpretation(interpretation_type)
-    tolerances = determine_tolerances(samples,interpretation_type)
+    strategy = positive_tolerance_interpretation
+    tolerances = determine_tolerances(samples,positive_tolerance_interpretation)
 
     for tolerance in tolerances:
         rectified_speedups = []

graph_net/plot_ESt.py

@@ -4,8 +4,9 @@
 import matplotlib.pyplot as plt
 from graph_net import analysis_util
 from graph_net import verify_aggregated_params
-from graph_net.positive_tolerance_interpretation_manager import g_type_name2_positive_tolerance_interpretation_cls
-
+from graph_net.positive_tolerance_interpretation_manager import (
+    get_supported_positive_tolerance_interpretation_types, get_positive_tolerance_interpretation
+)
 
 class ESScoresWrapper:
     """Wrapper for es_scores dict to allow attribute assignment."""
@@ -263,6 +264,7 @@ def main(args):
     # 2. Calculate scores for each curve and verify aggregated/microscopic consistency
     all_es_scores = {}
     all_aggregated_results = {}
+    positive_tolerance_interpretation = get_positive_tolerance_interpretation(args.interpretation_type)
 
     for folder_name, samples in all_results.items():
         print(f"\nCalculating ESt scores for '{folder_name}'...")
@@ -272,7 +274,7 @@ def main(args):
             p=args.negative_speedup_penalty,
             b=args.fpdb,
             type="ESt",
-            interpretation_type=args.interpretation_type,
+            positive_tolerance_interpretation=positive_tolerance_interpretation,
         )
 
         # Keep original behavior: assign es_scores directly
@@ -287,7 +289,7 @@ def main(args):
                     folder_name,
                     negative_speedup_penalty=args.negative_speedup_penalty,
                     fpdb=args.fpdb,
-                    interpretation_type=args.interpretation_type
+                    positive_tolerance_interpretation=positive_tolerance_interpretation,
                 )
             )
             # Store aggregated results for plotting
@@ -435,7 +437,7 @@ def main(args):
     parser.add_argument(
         "--positive-tolerance-interpretation",
         dest="interpretation_type",
-        choices=list(g_type_name2_positive_tolerance_interpretation_cls.keys()),
+        choices=get_supported_positive_tolerance_interpretation_types(),
         default="default",
         help="Select how positive tolerance values are interpreted into error types."
     )

graph_net/positive_tolerance_interpretation_manager.py

@@ -1,16 +1,9 @@
-from typing import Type
+from typing import Type, List
 from graph_net.positive_tolerance_interpretation import PositiveToleranceInterpretation
 from graph_net.default_positive_tolerance_interpretation import DefaultPositiveToleranceInterpretation
 from graph_net.mismatch_extended_positive_tolerance_interpretation import \
     MismatchExtendedPositiveToleranceInterpretation
 
-# Registry of available classes
-g_type_name2_positive_tolerance_interpretation_cls: dict[str, Type[PositiveToleranceInterpretation]] = {
-    'default': DefaultPositiveToleranceInterpretation,
-    'mismatch_extended': MismatchExtendedPositiveToleranceInterpretation
-}
-
-
 def get_positive_tolerance_interpretation(type_name: str) -> PositiveToleranceInterpretation:
     """
     Factory function to retrieve an instance of the requested interpretation strategy.
@@ -24,11 +17,20 @@ def get_positive_tolerance_interpretation(type_name: str) -> PositiveToleranceIn
     Raises:
         ValueError: If type_name is not registered.
     """
-    if type_name not in g_type_name2_positive_tolerance_interpretation_cls:
-        supported = list(g_type_name2_positive_tolerance_interpretation_cls.keys())
+    if type_name not in _g_type_name2_positive_tolerance_interpretation_cls:
+        supported = list(_g_type_name2_positive_tolerance_interpretation_cls.keys())
         raise ValueError(f"Unknown positive tolerance interpretation: '{type_name}'. Supported: {supported}")
 
     # Instantiate and return.
     # If stateful caching is needed, this logic can be modified to return singletons.
-    cls = g_type_name2_positive_tolerance_interpretation_cls[type_name]
-    return cls()
+    cls = _g_type_name2_positive_tolerance_interpretation_cls[type_name]
+    return cls()
+
+def get_supported_positive_tolerance_interpretation_types() -> List[str]:
+    return list(_g_type_name2_positive_tolerance_interpretation_cls.keys())
+
+# Registry of available classes
+_g_type_name2_positive_tolerance_interpretation_cls: dict[str, Type[PositiveToleranceInterpretation]] = {
+    'default': DefaultPositiveToleranceInterpretation,
+    'mismatch_extended': MismatchExtendedPositiveToleranceInterpretation
+}

graph_net/samples_statistics.py

@@ -9,25 +9,24 @@
 import scipy
 from scipy.stats import gmean
 from collections.abc import Callable
-from abc import ABC, abstractmethod
+from graph_net.positive_tolerance_interpretation import PositiveToleranceInterpretation
 
-from graph_net.positive_tolerance_interpretation_manager import get_positive_tolerance_interpretation
-
-def get_errno_from_error_type(error_type: str, interpretation_type: str = "default") -> int:
+def get_errno_from_error_type(error_type: str,
+                              positive_tolerance_interpretation: PositiveToleranceInterpretation) -> int:
     """
         Map error type string to errno (error number) using the appropriate strategy.
 
         Args:
             error_type: Error type string (e.g., "accuracy", "runtime_fail")
-            interpretation_type: Evaluation mode ("default" or "mismatch_extended")
+            positive_tolerance_interpretation: Evaluation mode ("default" or "mismatch_extended")
 
         Returns:
-            int: Errno based on the selected interpretation_type's logic.
+            int: Errno based on the selected positive_tolerance_interpretation's logic.
         """
-    interpreter = get_positive_tolerance_interpretation(interpretation_type)
-    return interpreter.get_errno(error_type)
+    return positive_tolerance_interpretation.get_errno(error_type)
 
-def get_errno_tolerance_mapping(custom_mapping, interpretation_type: str = "default"):
+def get_errno_tolerance_mapping(custom_mapping,
+                                positive_tolerance_interpretation: PositiveToleranceInterpretation):
     """
         Map errno (error number) back to error type string.
 
@@ -36,14 +35,13 @@ def get_errno_tolerance_mapping(custom_mapping, interpretation_type: str = "defa
 
         Args:
             errno: Error number
-            interpretation_type: Evaluation mode ("default" or "mismatch_extended")
+            positive_tolerance_interpretation: Evaluation mode ("default" or "mismatch_extended")
 
         Returns:
             Representative error type string (e.g., "accuracy", "compile_fail")
         """
     if custom_mapping: return custom_mapping
-    interpreter = get_positive_tolerance_interpretation(interpretation_type)
-    return interpreter.get_tolerance_mapping()
+    return positive_tolerance_interpretation.get_tolerance_mapping()
 
 def calculate_alpha(correct_speedups: list[float]) -> float:
     """
@@ -116,8 +114,8 @@ def calculate_eta(correct_speedups: list[float]) -> float:
 
 
 def calculate_pi(
-    errno2count: dict[int, int], total_samples: int, correct_speedups: list[float]
-) -> dict[int, float]:
+    errno2count: dict[Union[int, str], int], total_samples: int, correct_speedups: list[float]
+) -> dict[Union[int, str], float]:
     """
     Calculate pi: error type proportions for t > 0.
 
@@ -285,26 +283,22 @@ def calculate_es_t_from_aggregated(
 def calculate_es_components_values(
     total_samples: int,
     correct_speedups: list[float],
-    errno2count: dict[Union[int, str], int],  # 更新类型注解支持 str
+    errno2count: dict[Union[int, str], int],  # support str
     tolerance: int,
+    positive_tolerance_interpretation: PositiveToleranceInterpretation,
     negative_speedup_penalty: float = 0.0,
     b: float = 0.1,
     pi: Optional[dict[Union[int, str], float]] = None,
     errno_to_tolerance: Optional[dict[Union[int, str], int]] = None,
-    interpretation_type: str = "default"
 ) -> dict:
     """
     Calculate aggregated parameters for a given tolerance level.
-
-    Args:
-        ...
-        interpretation_type: "default" (int error codes) or "mismatch_extended" (str error codes).
     """
 
     if pi is None:
         pi = calculate_pi(errno2count, total_samples, correct_speedups)
 
-    errno_to_tolerance = get_errno_tolerance_mapping(errno_to_tolerance, interpretation_type)
+    errno_to_tolerance = get_errno_tolerance_mapping(errno_to_tolerance, positive_tolerance_interpretation)
 
     errno2tolerance = resolve_errno_tolerance(errno2count, errno_to_tolerance)
 

graph_net/verify_aggregated_params.py

@@ -2,23 +2,26 @@
 from collections import OrderedDict, Counter
 from graph_net import analysis_util
 from graph_net import samples_statistics
+from graph_net.positive_tolerance_interpretation import PositiveToleranceInterpretation
 from graph_net.samples_statistics import (
     get_errno_from_error_type,
 )
 
 
-def determine_tolerances(samples: list,interpretation_type = "default") -> range:
+def determine_tolerances(samples: list,
+                         positive_tolerance_interpretation: PositiveToleranceInterpretation,) -> range:
     """Determine tolerance range based on observed errno categories."""
     # Currently errno categories are 1=accuracy, 2=runtime, 3=compile.
     # Keep logic data-driven for future extension.
-    if interpretation_type == "default":
-        default_errnos = {1, 2, 3}
-    elif interpretation_type == "mismatch_extended":
-        default_errnos = {1, 2, 3, 4}
-    return range(-10, len(default_errnos) + 2)
+    mapping = positive_tolerance_interpretation.get_tolerance_mapping()
 
+    if not mapping:
+        max_errno = 3
+    else:
+        max_errno = max(mapping.keys())
+    return range(-10, max_errno + 2)
 
-def extract_statistics_at_tolerance(samples: list, tolerance: int,interpretation_type:str = "default") -> dict:
+def extract_statistics_at_tolerance(samples: list, tolerance: int,positive_tolerance_interpretation: PositiveToleranceInterpretation) -> dict:
     """Extract statistics for a given tolerance level."""
     sample_data = [
         (
@@ -42,7 +45,7 @@ def extract_statistics_at_tolerance(samples: list, tolerance: int,interpretation
 
     errno2count = dict(
         Counter(
-            get_errno_from_error_type(fail_type,interpretation_type)
+            get_errno_from_error_type(fail_type,positive_tolerance_interpretation)
             for _, _, _, _, fail_type in sample_data
             if fail_type is not None
         )
@@ -105,7 +108,7 @@ def calculate_es_constructor_params_for_tolerance(
     pi: dict,
     negative_speedup_penalty: float,
     fpdb: float,
-    interpretation_type: str = "default",
+    positive_tolerance_interpretation: PositiveToleranceInterpretation
 ) -> dict:
     """Calculate ES(t) constructor parameters (alpha, beta, gamma, lambda, eta) and final scores for a tolerance level."""
     aggregated_params = samples_statistics.calculate_es_components_values(
@@ -116,7 +119,7 @@ def calculate_es_constructor_params_for_tolerance(
         negative_speedup_penalty=negative_speedup_penalty,
         b=fpdb,
         pi=pi,
-        interpretation_type=interpretation_type,
+        positive_tolerance_interpretation=positive_tolerance_interpretation,
     )
 
     alpha = aggregated_params["alpha"]
@@ -205,7 +208,7 @@ def __init__(
         total_samples: int,
         negative_speedup_penalty: float,
         fpdb: float,
-        interpretation_type: str = "default",
+        positive_tolerance_interpretation: PositiveToleranceInterpretation,
     ):
         self.samples = samples
         self.total_samples = total_samples
@@ -218,10 +221,10 @@ def __init__(
             "slowdown_speedups": [],
             "errno2count": {},
         }
-        self.interpretation_type = interpretation_type
+        self.positive_tolerance_interpretation = positive_tolerance_interpretation
 
     def build_report(self, tolerance: int) -> dict:
-        current_stats = extract_statistics_at_tolerance(self.samples, tolerance,self.interpretation_type)
+        current_stats = extract_statistics_at_tolerance(self.samples, tolerance,self.positive_tolerance_interpretation)
 
         if tolerance == 1:
             self.pi = _freeze_statistics_at_tolerance(
@@ -247,7 +250,7 @@ def build_report(self, tolerance: int) -> dict:
             pi_for_calc,
             self.negative_speedup_penalty,
             self.fpdb,
-            self.interpretation_type,
+            self.positive_tolerance_interpretation,
         )
         # Use calculated pi from es_constructor_params for display and return
         calculated_pi = es_constructor_params.get("pi", self.pi)
@@ -289,9 +292,9 @@ def _empty_report(self, tolerance: int) -> dict:
 def verify_es_constructor_params_across_tolerances(
     samples: list,
     folder_name: str,
+    positive_tolerance_interpretation: PositiveToleranceInterpretation,
     negative_speedup_penalty: float = 0,
     fpdb: float = 0.1,
-    interpretation_type: str = "default",
 ) -> dict:
     """
     Verify and print ES constructor parameters (alpha, beta, gamma, lambda, eta, pi) for each
@@ -307,13 +310,13 @@ def verify_es_constructor_params_across_tolerances(
     print(f"Verifying Aggregated Parameters for '{folder_name}'")
     print(f"{'=' * 80}")
 
-    tolerances = determine_tolerances(samples,interpretation_type)
+    tolerances = determine_tolerances(samples,positive_tolerance_interpretation)
     builder = ToleranceReportBuilder(
         samples=samples,
         total_samples=total_samples,
         negative_speedup_penalty=negative_speedup_penalty,
         fpdb=fpdb,
-        interpretation_type=interpretation_type,
+        positive_tolerance_interpretation=positive_tolerance_interpretation,
     )
 
     results = OrderedDict(