import os import sys import shutil import subprocess import numpy as np from generate_synthetic_data import generate_synthetic_grid def run_synthetic_regression_test(): # Resolve paths dynamically relative to this script's location code_gen_dir = os.path.dirname(os.path.abspath(__file__)) code_dir = os.path.dirname(code_gen_dir) workspace_dir = os.path.dirname(code_dir) test_run_dir = os.path.join(workspace_dir, "scratch", "synthetic_test_run") # 1. Create a clean testing directory in workspace if os.path.exists(test_run_dir): shutil.rmtree(test_run_dir) os.makedirs(test_run_dir, exist_ok=True) # 2. Generate the synthetic grid file (fort.1000) inside the testing directory synthetic_grid_path = os.path.join(test_run_dir, "fort.1000") nx, ny, nz = 32, 32, 32 generate_synthetic_grid(synthetic_grid_path, nx, ny, nz) # 3. Create config files post-edge-cold-bc-hybrid-intro and empty otape intro_content = f""" y_leng 2.0 ! Length of y-dir. domain. [PI] startnum 1000 endnum 1000 skipnum 1 shiftnum 999 vis 0.02 sc 0.75 le 1. min_wr 0.0 prof_wr 1.0 min_fsd 0.0 min_c 1.e-12 pre 2.10e+4 ac 26.7 bc 3.0 c_cut 0.001 c_ref 0.01 syp 1 eyp {ny} SL_u 0.21 omitnum 0 """ with open(os.path.join(test_run_dir, "post-edge-cold-bc-hybrid-intro"), "w") as f: f.write(intro_content) with open(os.path.join(test_run_dir, "otape"), "w") as f: f.write("") # 4. Compile the Fortran code using the test termspec print("Compiling Fortran modules with test_terms.input...") try: # Run make clean to remove old objects subprocess.run(["make", "cleanAll"], cwd=code_dir, check=True) # Compile with specific test termspec subprocess.run( ["make", "TERMSPEC=code_gen/test_terms.input"], cwd=code_dir, check=True ) print("Compilation successful!") except subprocess.CalledProcessError as e: print("Fortran compilation failed!") sys.exit(1) # 5. Copy compiled binary to test run directory binary_name = "x-edge-cold-bc-uPrime-hybrid" shutil.copy(os.path.join(code_dir, binary_name), test_run_dir) # 6. Execute binary inside test run directory # Using 4 ranks for fast execution print("Running compiled MPI binary on synthetic grid...") run_cmd = [f"./{binary_name}"] try: subprocess.run( run_cmd, cwd=test_run_dir, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True ) print("MPI Binary execution finished successfully!") except subprocess.CalledProcessError as e: print("MPI execution failed!") sys.exit(1) # 7. Load results and mathematically verify accuracy output_dat_path = os.path.join(test_run_dir, "qEdge_X.dat") if not os.path.exists(output_dat_path): print("Error: Output qEdge_X.dat was not generated!") sys.exit(1) # Parse headers and load numeric data with open(output_dat_path, 'r') as f: headers = f.readline().strip().split() data = np.loadtxt(output_dat_path, skiprows=1) # Set up analytical solution mesh lx, ly, lz = 2.0 * np.pi, 2.0 * np.pi, 2.0 * np.pi hx, hy, hz = lx / nx, ly / ny, lz / nz x = np.arange(1, nx + 1) * hx y = np.arange(1, ny + 1) * hy z = np.arange(1, nz + 1) * hz # Pre-calculate discrete averages over y and z dimensions for exact matching cos_y_avg = np.mean(np.cos(y)) cos_z_avg = np.mean(np.cos(z)) sin_y_avg = np.mean(np.sin(y)) sin_z_avg = np.mean(np.sin(z)) # Expected discrete X-profiles expected = {} # avg_y(i) expected["avg_y"] = (np.cos(x) * cos_y_avg * cos_z_avg) + 2.0 # dy_dx = -sin(x)*cos(y)*cos(z) => avg_dy_dx(i) = -sin(x_i) * * expected["avg_dy_dx"] = -np.sin(x) * cos_y_avg * cos_z_avg # d2y_dx = -cos(x)*cos(y)*cos(z) => avg_d2y_dx(i) = -cos(x_i) * * expected["avg_d2y_dx"] = -np.cos(x) * cos_y_avg * cos_z_avg # Fluctuation averages should be exactly 0 (u' = u - ) expected["avg_u_prime"] = np.zeros(nx) expected["avg_v_prime"] = np.zeros(nx) expected["avg_y_prime"] = np.zeros(nx) # Weighted averages: y_avg_u = / # u = sin(x)cos(y)sin(z), y = cos(x)cos(y)cos(z) + 2.0 # u * y = sin(x)cos(x) * cos^2(y) * sin(z)cos(z) + 2.0 * sin(x) * cos(y) * sin(z) u_y_term1 = np.sin(x)*np.cos(x) * np.mean(np.cos(y)**2) * np.mean(np.sin(z)*np.cos(z)) u_y_term2 = 2.0 * np.sin(x) * np.mean(np.cos(y)) * np.mean(np.sin(z)) expected["y_avg_u"] = (u_y_term1 + u_y_term2) / expected["avg_y"] # Numeric column comparison failed = False tolerance = 1e-10 # Floating-point tolerance for averages diff_tolerance = 1e-4 # Slightly higher tolerance for compact scheme derivatives print("\n=== Analytical Grid Verification Report ===") print(f"{'Variable Name':<20} | {'Max Absolute Error':<20} | {'Status':<10}") print("-" * 60) for col_idx, col_name in enumerate(headers): if col_name == 'x': continue col_data = data[:, col_idx] if col_name in expected: max_err = np.max(np.abs(col_data - expected[col_name])) limit = diff_tolerance if "dy" in col_name or "d2" in col_name else tolerance status = "PASSED" if max_err > limit: status = "FAILED" failed = True print(f"{col_name:<20} | {max_err:<20.4e} | {status:<10}") # Clean up build objects to avoid polluting git tree subprocess.run(["make", "cleanAll"], cwd=code_dir, stdout=subprocess.DEVNULL) if failed: print("\nVerification FAILED: Some outputs deviated from exact analytical formulas!") sys.exit(1) else: print("\nVerification SUCCESSFUL! All compiler stages (Jinja2, SymPy Optimization, Liveness, Array Pooling) verified.") # Cleanup temporary run directory shutil.rmtree(test_run_dir) sys.exit(0) if __name__ == '__main__': run_synthetic_regression_test()