181 lines
7.0 KiB
Python
181 lines
7.0 KiB
Python
"""
|
||
Smoke test for Step 10 散点 CSV 模式 (WaterIndexCsvProcessor)
|
||
|
||
模拟 Step 4 输出格式 (sampling_spectra.csv):
|
||
x_coord, y_coord, pixel_x, pixel_y, "400.000000", "401.000000", ...
|
||
|
||
验证 WaterIndexCsvProcessor.compute_indices_from_csv:
|
||
1. 正确读取 x_coord/y_coord → 重命名为 longitude/latitude
|
||
2. 正确识别数字列名 = 光谱列
|
||
3. 复用 WaterQualityIndexCalculator 逐行计算
|
||
4. 输出每个公式一个 CSV,三列严格为 longitude, latitude, <formula_name>
|
||
5. 公式值数量级合理 (非 NaN,非 inf)
|
||
"""
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import shutil
|
||
from pathlib import Path
|
||
|
||
# 让脚本能找到项目根
|
||
PROJECT_ROOT = Path(__file__).parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
|
||
def create_synthetic_sampling_csv(path: str, n_points: int = 5):
|
||
"""模拟 Step 4 输出: x_coord, y_coord, pixel_x, pixel_y, 数字列名光谱"""
|
||
import csv
|
||
# 选一组关键波段（确保 waterindex.csv 中的 BGA_Am09KBBI 等公式都能找到）
|
||
wavelengths = [400.0, 443.0, 458.0, 486.0, 500.0, 510.0, 531.0, 547.0, 555.0,
|
||
615.0, 622.0, 629.0, 644.0, 658.0, 665.0, 672.0, 681.0, 686.0,
|
||
700.0, 709.0, 714.0, 715.0, 753.0, 857.0, 900.0]
|
||
fieldnames = ['x_coord', 'y_coord', 'pixel_x', 'pixel_y'] + [f'{w:.6f}' for w in wavelengths]
|
||
|
||
with open(path, 'w', newline='', encoding='utf-8-sig') as f:
|
||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||
writer.writeheader()
|
||
for i in range(n_points):
|
||
# 模拟水体光谱（典型内陆湖泊反射率 0.005-0.05）
|
||
row = {
|
||
'x_coord': 100.0 + i * 10,
|
||
'y_coord': 30.0 + i * 5,
|
||
'pixel_x': 100 + i,
|
||
'pixel_y': 30 + i,
|
||
}
|
||
for w in wavelengths:
|
||
# 简单合成光谱: 蓝光 < 红光 + 一点叶绿素峰
|
||
base = 0.01 + 0.0001 * (w - 400)
|
||
chl_peak = 0.005 * (1 - abs(w - 560) / 200) if abs(w - 560) < 200 else 0
|
||
row[f'{w:.6f}'] = round(base + chl_peak, 6)
|
||
writer.writerow(row)
|
||
|
||
|
||
def run_smoke():
|
||
print("=" * 70)
|
||
print("Step 10 散点 CSV 模式 Smoke Test")
|
||
print("=" * 70)
|
||
|
||
tmpdir = tempfile.mkdtemp(prefix="step10_smoke_")
|
||
print(f"Tempdir: {tmpdir}")
|
||
|
||
sampling_csv = os.path.join(tmpdir, "sampling_spectra.csv")
|
||
output_dir = os.path.join(tmpdir, "10_WaterIndex_CSV")
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# 1) 创建合成 sampling CSV
|
||
create_synthetic_sampling_csv(sampling_csv, n_points=5)
|
||
print(f"Created sampling CSV: {sampling_csv}")
|
||
with open(sampling_csv, encoding='utf-8-sig') as f:
|
||
header_line = f.readline().strip()
|
||
print(f" header: {header_line[:120]}...")
|
||
|
||
# 2) 找到项目自带的 waterindex.csv
|
||
waterindex_csv = PROJECT_ROOT / "src" / "gui" / "model" / "waterindex.csv"
|
||
print(f"Using waterindex.csv: {waterindex_csv}")
|
||
assert waterindex_csv.is_file(), "waterindex.csv not found!"
|
||
|
||
# 3) 调 WaterIndexCsvProcessor
|
||
# 注意：父包 __init__.py 顶部有 `from osgeo import gdal, osr`，
|
||
# 在没装 gdal 的 venv 里任何 from ...waterindex_inversion import ... 都会炸。
|
||
# 这里用 importlib 按文件路径直接加载 csv_processor.py 子模块，
|
||
# 完全绕开 __init__.py 的 osgeo 加载。
|
||
import importlib.util as _ilu
|
||
_csv_proc_path = (
|
||
PROJECT_ROOT / "src" / "core" / "algorithms" / "waterindex_inversion"
|
||
/ "csv_processor.py"
|
||
)
|
||
_spec = _ilu.spec_from_file_location("waterindex_csv_processor", _csv_proc_path)
|
||
_mod = _ilu.module_from_spec(_spec)
|
||
sys.modules["waterindex_csv_processor"] = _mod
|
||
_spec.loader.exec_module(_mod)
|
||
WaterIndexCsvProcessor = _mod.WaterIndexCsvProcessor
|
||
|
||
progress_log = []
|
||
def progress_cb(msg, pct):
|
||
progress_log.append((msg, pct))
|
||
|
||
proc = WaterIndexCsvProcessor(str(waterindex_csv))
|
||
print(f"\n[Step] compute_indices_from_csv...")
|
||
out_files = proc.compute_indices_from_csv(
|
||
sampling_csv_path=sampling_csv,
|
||
output_dir=output_dir,
|
||
selected_formulas=["BGA_Am09KBBI", "BGA_Da052BDA", "BGA_Be16NDPhyI"],
|
||
progress_callback=progress_cb,
|
||
)
|
||
|
||
print(f"\n[Result] Generated {len(out_files)} CSV files:")
|
||
for name, path in out_files.items():
|
||
size = os.path.getsize(path)
|
||
print(f" {name:30s} -> {os.path.basename(path)} ({size} bytes)")
|
||
|
||
# 4) 验证每个输出 CSV 的列结构
|
||
print(f"\n[Verify] Column structure check:")
|
||
all_pass = True
|
||
import pandas as pd
|
||
for name, path in out_files.items():
|
||
df = pd.read_csv(path, encoding='utf-8-sig')
|
||
cols = list(df.columns)
|
||
expected = ['longitude', 'latitude', name]
|
||
ok = (cols == expected) and (len(df) == 5)
|
||
flag = "✓" if ok else "✗"
|
||
if not ok:
|
||
all_pass = False
|
||
print(f" {flag} {name:30s} cols={cols} rows={len(df)}")
|
||
|
||
# 5) 验证坐标重命名
|
||
print(f"\n[Verify] Coordinate rename (x_coord→longitude, y_coord→latitude):")
|
||
sample = pd.read_csv(out_files[list(out_files.keys())[0]], encoding='utf-8-sig')
|
||
print(f" longitude values: {sample['longitude'].tolist()}")
|
||
print(f" latitude values: {sample['latitude'].tolist()}")
|
||
coord_ok = (sample['longitude'].iloc[0] == 100.0 and
|
||
sample['latitude'].iloc[0] == 30.0)
|
||
if not coord_ok:
|
||
all_pass = False
|
||
print(f" {'✓' if coord_ok else '✗'} coordinate rename correct")
|
||
|
||
# 6) 验证公式值非 NaN
|
||
print(f"\n[Verify] Formula values (no NaN):")
|
||
for name, path in out_files.items():
|
||
df = pd.read_csv(path, encoding='utf-8-sig')
|
||
col = df[name]
|
||
n_nan = col.isna().sum()
|
||
n_inf = ((col == float('inf')) | (col == float('-inf'))).sum()
|
||
all_nan = col.dropna().empty
|
||
if n_nan > 0 or n_inf > 0 or all_nan:
|
||
print(f" ✗ {name:30s}: NaN={n_nan} Inf={n_inf} empty={all_nan}")
|
||
print(f" values: {col.tolist()}")
|
||
all_pass = False
|
||
else:
|
||
mn, mx = col.min(), col.max()
|
||
print(f" ✓ {name:30s}: range=[{mn:.4f}, {mx:.4f}]")
|
||
|
||
# 7) 进度回调检查
|
||
print(f"\n[Verify] Progress callback:")
|
||
print(f" Total progress events: {len(progress_log)}")
|
||
if progress_log:
|
||
first_msg, first_pct = progress_log[0]
|
||
last_msg, last_pct = progress_log[-1]
|
||
print(f" First: ({first_pct:.1f}%) {first_msg}")
|
||
print(f" Last : ({last_pct:.1f}%) {last_msg}")
|
||
progress_ok = (last_pct == 100.0)
|
||
if not progress_ok:
|
||
all_pass = False
|
||
print(f" {'✓' if progress_ok else '✗'} last progress = 100%")
|
||
|
||
# 8) 总结
|
||
print(f"\n{'=' * 70}")
|
||
if all_pass:
|
||
print(f"✓ ALL CHECKS PASSED")
|
||
else:
|
||
print(f"✗ SOME CHECKS FAILED — inspect output above")
|
||
print(f"{'=' * 70}")
|
||
|
||
# 清理
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
return all_pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
success = run_smoke()
|
||
sys.exit(0 if success else 1)
|