Spaces:
Sleeping
Sleeping
File size: 7,760 Bytes
67fb03c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 |
#!/usr/bin/env python3
"""
Simple test function to run main.py with cadillac_testcase config
and validate that specific geometries have R² scores greater than 0.7.
"""
########################################## global variables #########################################################
config_path = "/raid/ansysai/udbhav/alphaLPFM/testcases/config_eval.yaml"
main_script_path = "/raid/ansysai/udbhav/alphaLPFM/main.py"
project_root = "/raid/ansysai/udbhav/alphaLPFM"
test_geometries = ['cadillac_19', 'cadillac_73', 'cadillac_104']
thresholds = {
'cadillac_19': 0.895,
'cadillac_73': 0.855,
'cadillac_104': 0.769,
}
###### import libraries #######
import pytest
import os
import sys
import subprocess
import tempfile
import shutil
from pathlib import Path
from omegaconf import OmegaConf
class TestMainPipeline:
"""Test class for running main.py and validating R² scores."""
@pytest.fixture(scope="class")
def config_path(self):
"""Path to the cadillac_testcase config file."""
# return "/raid/ansysai/udbhav/alphaLPFM/configs/testcase/cadillac_testcase/config.yaml"
return config_path
@pytest.fixture(scope="class")
def main_script_path(self):
"""Path to the main.py script."""
return main_script_path
@pytest.fixture(scope="class")
def project_root(self):
"""Project root directory."""
return project_root
def test_main_pipeline_r2_scores(self, config_path, main_script_path, project_root):
"""
Test that main.py runs successfully and top 5 R² scores are > test_r2_threshold.
This test:
1. Runs main.py with cadillac_testcase config
2. Loads the R² scores from the results
3. Validates that top 5 scores are > test_r2_threshold
"""
# Verify config file exists
assert os.path.exists(config_path), f"Config file not found: {config_path}"
assert os.path.exists(main_script_path), f"Main script not found: {main_script_path}"
# Load config to get test_name for path construction
cfg = OmegaConf.load(config_path)
test_name = cfg.test_name
project_name = cfg.project_name
model_name = cfg.model
# Construct expected output path
expected_output_dir = os.path.join(
project_root,
"metrics",
project_name,
f"{model_name}_{test_name}"
)
print(f"Running main.py with config: {config_path}")
print(f"Expected output directory: {expected_output_dir}")
# Change to project root directory
original_cwd = os.getcwd()
os.chdir(project_root)
try:
# Run main.py with the config
cmd = [sys.executable, "main.py", "--config_path", config_path]
print(f"Executing command: {' '.join(cmd)}")
# Run the command
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=1800 # 30 minutes timeout
)
# Check if command succeeded
if result.returncode != 0:
print(f"STDOUT: {result.stdout}")
print(f"STDERR: {result.stderr}")
pytest.fail(f"main.py failed with return code {result.returncode}")
print("✅ main.py executed successfully")
print(f"STDOUT: {result.stdout}")
# Verify output directory was created
assert os.path.exists(expected_output_dir), f"Output directory not created: {expected_output_dir}"
# Check for best_case directory
best_case_dir = os.path.join(expected_output_dir, "best_case")
assert os.path.exists(best_case_dir), f"Best case directory not found: {best_case_dir}"
# Check for test results directory
test_dir = os.path.join(best_case_dir, "test")
assert os.path.exists(test_dir), f"Test directory not found: {test_dir}"
# Check for R² scores file
r_squared_file = os.path.join(test_dir, "r_squared.txt")
assert os.path.exists(r_squared_file), f"R² scores file not found: {r_squared_file}"
# Load and validate R² scores
r2_scores_dict = self._load_r2_scores_with_ids(r_squared_file)
print(f"Loaded {len(r2_scores_dict)} R² scores")
# Specific geometries to test with required thresholds
print(f"Testing specific geometries: {test_geometries} with thresholds: {thresholds}")
# Validate that specific geometries have R² > 0.7
failed_geometries = []
passed_geometries = []
for geometry_id in test_geometries:
if geometry_id in r2_scores_dict:
r2_score = r2_scores_dict[geometry_id]
required = thresholds[geometry_id]
if r2_score > 0.99*required and r2_score < 1.01*required:
passed_geometries.append((geometry_id, r2_score))
print(f"✅ {geometry_id}: R² = {r2_score:.6f} > {0.99*required} and {r2_score:.6f} < {1.01*required}")
else:
failed_geometries.append((geometry_id, r2_score))
print(f"❌ {geometry_id}: R² = {r2_score:.6f} <= {0.99*required} or {r2_score:.6f} >= {1.01*required}")
else:
failed_geometries.append((geometry_id, "NOT_FOUND"))
print(f"❌ {geometry_id}: Not found in results")
if failed_geometries:
failed_msg = "Specific geometries validation failed:\n"
for geometry_id, score in failed_geometries:
failed_msg += f" - {geometry_id}: {score}\n"
failed_msg += f"\nPassed geometries: {passed_geometries}"
pytest.fail(failed_msg)
print(f"✅ All specific geometries validation passed!")
print(f"Passed geometries: {passed_geometries}")
except subprocess.TimeoutExpired:
pytest.fail("main.py execution timed out after 30 minutes")
except Exception as e:
pytest.fail(f"Unexpected error during main.py execution: {str(e)}")
finally:
# Restore original working directory
os.chdir(original_cwd)
def _load_r2_scores_with_ids(self, r_squared_file):
"""
Load R² scores from the results file with geometry IDs.
Expected format: "geometry_id r2_score"
Returns: dict with geometry_id as key and r2_score as value
"""
r2_scores_dict = {}
with open(r_squared_file, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#'):
parts = line.split()
if len(parts) >= 2:
try:
geometry_id = parts[0]
r2_score = float(parts[1])
r2_scores_dict[geometry_id] = r2_score
except ValueError:
print(f"Warning: Could not parse R² score from line: {line}")
continue
return r2_scores_dict
if __name__ == "__main__":
# Run the tests
pytest.main([__file__, "-v", "--tb=short"])
|