feat: 波长偏移修正 + 项目架构文档

- BandMathCalculator 支持 wavelength_offset 参数,公式波长统一加减偏移后匹配传感器波段
- WaterQualityIndexCalculator 全链传递偏移量 (band_math → calculate_one → calculate_many)
- WaterIndexCsvProcessor / Step7Handler / DataPreparationStep 传播偏移参数
- Step7/Step10 面板新增 QDoubleSpinBox 波长偏移控件 (±200nm, 默认0)
- 偏移控件去除单位后缀,避免编辑时需手动移动光标
- 新增 ARCHITECTURE.md 完整项目架构文档
This commit is contained in:
duxin
2026-06-30 13:13:38 +08:00
parent 8ff5b08190
commit 9e433395f4
8 changed files with 537 additions and 15 deletions

View File

@ -4,13 +4,16 @@ import re
class BandMathCalculator:
def __init__(self, csv_file):
def __init__(self, csv_file, wavelength_offset=0.0):
"""
初始化计算器
csv_file: 包含光谱反射率的CSV文件路径
wavelength_offset: 波长偏移修正量(nm),公式中的目标波长会统一加上此偏移后再匹配最近的传感器波段。
例如 offset=100 意味着公式中的 w450 实际会去找传感器波段 ~550nm。
"""
self.df = pd.read_csv(csv_file)
self.wavelengths = self._extract_wavelengths()
self.wavelength_offset = float(wavelength_offset)
def _extract_wavelengths(self):
"""从列名中提取波长信息"""
@ -25,19 +28,25 @@ class BandMathCalculator:
return wavelengths
def _find_closest_wavelength(self, target_wavelength):
"""找到最接近目标波长的列索引"""
"""找到最接近目标波长的列索引(自动应用波长偏移修正)"""
# 应用波长偏移修正
adjusted_target = target_wavelength + self.wavelength_offset
valid_indices = [i for i, wl in enumerate(self.wavelengths) if wl is not None]
if not valid_indices:
raise ValueError("未找到有效的波长列")
# 计算与目标波长的差值
differences = [abs(self.wavelengths[i] - target_wavelength) for i in valid_indices]
differences = [abs(self.wavelengths[i] - adjusted_target) for i in valid_indices]
min_diff_index = np.argmin(differences)
closest_index = valid_indices[min_diff_index]
closest_wavelength = self.wavelengths[closest_index]
print(
f"目标波长 {target_wavelength}nm -> 最接近波长 {closest_wavelength}nm (列: {self.df.columns[closest_index]})")
if abs(self.wavelength_offset) > 0.01:
print(
f"公式波长 {target_wavelength}nm + 偏移 {self.wavelength_offset}nm → 目标 {adjusted_target}nm → 最接近波段 {closest_wavelength}nm (列: {self.df.columns[closest_index]})")
else:
print(
f"目标波长 {target_wavelength}nm -> 最接近波长 {closest_wavelength}nm (列: {self.df.columns[closest_index]})")
return closest_index
def _parse_expression(self, expression):

View File

@ -90,13 +90,14 @@ class WaterQualityIndexCalculator:
parts = [float(x.strip()) for x in s.split(",")]
return np.array(parts)
def _band_math_all_rows(self, df: pd.DataFrame, expression: str) -> pd.Series:
def _band_math_all_rows(self, df: pd.DataFrame, expression: str, wavelength_offset: float = 0.0) -> pd.Series:
"""
使用 BandMathCalculator 的公式计算引擎,在整个 DataFrame 上批量求值。
Args:
df: 输入光谱数据(列名为 wNNN 格式)
expression: 波段计算表达式,如 "(w715 - w686) / (w715 + w686)"
wavelength_offset: 波长偏移修正量(nm)
Returns:
pd.Series,与 df 等长的计算结果
@ -104,6 +105,7 @@ class WaterQualityIndexCalculator:
calc = BandMathCalculator.__new__(BandMathCalculator)
calc.df = df.copy()
calc.wavelengths = calc._extract_wavelengths()
calc.wavelength_offset = float(wavelength_offset)
variables = calc._parse_expression(expression)
results = []
@ -126,13 +128,14 @@ class WaterQualityIndexCalculator:
return pd.Series(results, index=df.index, name=expression)
def calculate_one(self, name: str, df: pd.DataFrame) -> pd.Series:
def calculate_one(self, name: str, df: pd.DataFrame, wavelength_offset: float = 0.0) -> pd.Series:
"""
计算单个水质指数。
Args:
name: 公式名称(对应 Formula_Name)
df: 光谱反射率 DataFrame
wavelength_offset: 波长偏移修正量(nm)
Returns:
pd.Series,计算结果
@ -145,7 +148,7 @@ class WaterQualityIndexCalculator:
ftype = cfg["type"]
coeff_str = cfg["coeff"]
raw = self._band_math_all_rows(df, expr)
raw = self._band_math_all_rows(df, expr, wavelength_offset=wavelength_offset)
if ftype == "concentration":
coeff = self._parse_coeff(coeff_str)
@ -155,13 +158,14 @@ class WaterQualityIndexCalculator:
raw.name = name
return raw
def calculate_many(self, names: List[str], df: pd.DataFrame) -> pd.DataFrame:
def calculate_many(self, names: List[str], df: pd.DataFrame, wavelength_offset: float = 0.0) -> pd.DataFrame:
"""
批量计算多个水质指数。
Args:
names: 公式名称列表
df: 光谱反射率 DataFrame
wavelength_offset: 波长偏移修正量(nm)
Returns:
pd.DataFrame,每列对应一个公式的计算结果
@ -169,7 +173,7 @@ class WaterQualityIndexCalculator:
results = {}
for name in names:
try:
results[name] = self.calculate_one(name, df)
results[name] = self.calculate_one(name, df, wavelength_offset=wavelength_offset)
except Exception as e:
print(f"⚠️ 计算 {name} 失败: {e}")
results[name] = pd.Series(np.nan, index=df.index, name=name)