3d40e88e35
modified: evaluation_studio.py modified: scripts/dataset.py new file: scripts/inference_utils.py modified: scripts/predict_single.py modified: scripts_2/__pycache__/task3_identify.cpython-314.pyc modified: scripts_2/task3_identify.py
736 lines
30 KiB
Python
736 lines
30 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
from datetime import datetime
|
||
import math
|
||
import pickle
|
||
from pathlib import Path
|
||
import shutil
|
||
import threading
|
||
import traceback
|
||
import tkinter as tk
|
||
from tkinter import filedialog, messagebox, ttk
|
||
from tkinter.scrolledtext import ScrolledText
|
||
|
||
import matplotlib
|
||
|
||
matplotlib.use("Agg")
|
||
|
||
import matplotlib.pyplot as plt
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent
|
||
CHECKPOINT_DEFAULT = PROJECT_ROOT / "checkpoints_final" / "task1_final" / "task1_final_model.pkl"
|
||
ADAPTER_DEFAULT = PROJECT_ROOT / "checkpoints_final" / "task1_final" / "task1_tmd_adapter.pkl"
|
||
APP_OUTPUT_DIR = PROJECT_ROOT / "evaluation_outputs" / "studio_outputs"
|
||
APP_CACHE_DIR = PROJECT_ROOT / ".studio_cache"
|
||
|
||
TASK1_LABEL = "任务1：顶部RMS预测（Non_TMD）"
|
||
TASK2_LABEL = "任务2：自由衰减识别自振频率和阻尼比"
|
||
TASK3_LABEL = "任务3：顶部响应反推激振频率"
|
||
TASK4_LABEL = "挑战任务：TMD RMS预测（基础模型+适配器）"
|
||
|
||
TASK_CHOICES = [TASK1_LABEL, TASK2_LABEL, TASK3_LABEL, TASK4_LABEL]
|
||
|
||
HAS_DND = False
|
||
TkRootClass: type[tk.Tk] = tk.Tk
|
||
try:
|
||
from tkinterdnd2 import DND_FILES, TkinterDnD # type: ignore
|
||
|
||
HAS_DND = True
|
||
TkRootClass = TkinterDnD.Tk
|
||
except Exception:
|
||
DND_FILES = "DND_Files" # type: ignore
|
||
|
||
|
||
def pretty_json_like_lines(payload: dict) -> str:
|
||
lines: list[str] = []
|
||
for key in sorted(payload.keys()):
|
||
value = payload[key]
|
||
if isinstance(value, float):
|
||
lines.append(f"- {key}: {value:.6f}")
|
||
else:
|
||
lines.append(f"- {key}: {value}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def ensure_long_format_csv(input_path: Path, force_convert: bool, tz_offset_hours: float) -> tuple[Path, bool]:
|
||
from convert_teacher_csv import convert_dataframe
|
||
|
||
df_head = pd.read_csv(input_path, nrows=5)
|
||
lower_columns = {column.lower() for column in df_head.columns}
|
||
long_required = {"code", "type", "time", "value1", "value2", "value3"}
|
||
is_long = long_required.issubset(lower_columns)
|
||
if is_long and not force_convert:
|
||
return input_path, False
|
||
|
||
APP_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
full_df = pd.read_csv(input_path)
|
||
converted_df = convert_dataframe(
|
||
input_df=full_df,
|
||
time_column="time",
|
||
timezone_offset_hours=tz_offset_hours,
|
||
)
|
||
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
output_path = APP_CACHE_DIR / f"{input_path.stem}_converted_{stamp}.csv"
|
||
converted_df.to_csv(output_path, index=False)
|
||
return output_path, True
|
||
|
||
|
||
def _load_sensor_frame(df: pd.DataFrame, sensor_code: str, value_col: str, time_col: str = "time") -> pd.DataFrame:
|
||
sensor_df = df.loc[df["code"] == sensor_code, [time_col, value_col]].copy()
|
||
if sensor_df.empty:
|
||
return sensor_df
|
||
sensor_df = sensor_df.sort_values(time_col)
|
||
sensor_df = sensor_df.drop_duplicates(subset=time_col, keep="first")
|
||
sensor_df[time_col] = pd.to_numeric(sensor_df[time_col], errors="coerce")
|
||
sensor_df[value_col] = pd.to_numeric(sensor_df[value_col], errors="coerce")
|
||
sensor_df = sensor_df[sensor_df[time_col].notna() & sensor_df[value_col].notna()]
|
||
return sensor_df
|
||
|
||
|
||
def build_prediction_input(csv_path: Path, for_tmd: bool) -> dict:
|
||
from scripts.config import make_experiment_config
|
||
from scripts.dataset import (
|
||
calculate_rms,
|
||
estimate_sampling_rate,
|
||
extract_core_features,
|
||
get_dominant_frequency,
|
||
get_middle_segment,
|
||
)
|
||
|
||
config = make_experiment_config().data
|
||
if for_tmd:
|
||
config.data_dir = "downloads/TMD"
|
||
config.__post_init__()
|
||
|
||
df = pd.read_csv(csv_path)
|
||
base_df = _load_sensor_frame(df, sensor_code=config.base_sensor_code, value_col=config.base_axis, time_col=config.time_column)
|
||
if base_df.empty:
|
||
raise ValueError(
|
||
f"缺少基底传感器 {config.base_sensor_code}/{config.base_axis}，"
|
||
"无法执行RMS预测任务。"
|
||
)
|
||
|
||
response_df = _load_sensor_frame(
|
||
df,
|
||
sensor_code=config.response_sensor_code,
|
||
value_col=config.response_axis,
|
||
time_col=config.time_column,
|
||
)
|
||
|
||
if response_df.empty:
|
||
time_values = base_df[config.time_column].to_numpy(dtype=np.float64)
|
||
x_values = base_df[config.base_axis].to_numpy(dtype=np.float32)
|
||
time_middle, x_middle, _ = get_middle_segment(time_values, x_values, x_values, config)
|
||
y_middle = None
|
||
has_truth = False
|
||
else:
|
||
aligned = base_df.rename(columns={config.base_axis: "base_signal"}).merge(
|
||
response_df.rename(columns={config.response_axis: "response_signal"}),
|
||
on=config.time_column,
|
||
how="left",
|
||
)
|
||
aligned["response_signal"] = aligned["response_signal"].interpolate(method=config.interpolation_method, limit_direction="both")
|
||
aligned["response_signal"] = aligned["response_signal"].ffill().bfill()
|
||
time_values = aligned[config.time_column].to_numpy(dtype=np.float64)
|
||
x_values = aligned["base_signal"].to_numpy(dtype=np.float32)
|
||
y_values = aligned["response_signal"].to_numpy(dtype=np.float32)
|
||
time_middle, x_middle, y_middle = get_middle_segment(time_values, x_values, y_values, config)
|
||
has_truth = True
|
||
|
||
sampling_rate = estimate_sampling_rate(np.asarray(time_middle, dtype=np.float64))
|
||
frequency_hz = get_dominant_frequency(np.asarray(x_middle, dtype=np.float32), sampling_rate)
|
||
features = extract_core_features(np.asarray(x_middle, dtype=np.float32), sampling_rate, config, known_frequency_hz=frequency_hz)
|
||
x_rms = float(calculate_rms(np.asarray(x_middle, dtype=np.float32)))
|
||
|
||
true_y_rms = None
|
||
if has_truth and y_middle is not None:
|
||
true_y_rms = float(calculate_rms(np.asarray(y_middle, dtype=np.float32)))
|
||
|
||
return {
|
||
"config": config,
|
||
"time_middle": np.asarray(time_middle, dtype=np.float64),
|
||
"x_middle": np.asarray(x_middle, dtype=np.float32),
|
||
"y_middle": None if y_middle is None else np.asarray(y_middle, dtype=np.float32),
|
||
"sampling_rate": float(sampling_rate),
|
||
"frequency_hz": float(frequency_hz),
|
||
"features": np.asarray(features, dtype=np.float32),
|
||
"x_rms": x_rms,
|
||
"true_y_rms": true_y_rms,
|
||
"has_truth": has_truth,
|
||
}
|
||
|
||
|
||
def save_prediction_overview_figure(
|
||
*,
|
||
file_path: Path,
|
||
output_dir: Path,
|
||
title_prefix: str,
|
||
time_middle: np.ndarray,
|
||
x_middle: np.ndarray,
|
||
sampling_rate: float,
|
||
frequency_hz: float,
|
||
pred_rms: float,
|
||
true_rms: float | None,
|
||
) -> Path:
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
fft_values = np.fft.rfft(x_middle - np.mean(x_middle))
|
||
freqs = np.fft.rfftfreq(x_middle.size, d=1.0 / sampling_rate)
|
||
amplitudes = np.abs(fft_values)
|
||
|
||
fig, axes = plt.subplots(3, 1, figsize=(10, 8))
|
||
fig.suptitle(f"{title_prefix} | {file_path.name}")
|
||
|
||
axes[0].plot(time_middle, x_middle, color="tab:blue")
|
||
axes[0].set_title("Input Base Signal (Middle Segment)")
|
||
axes[0].set_xlabel("Time")
|
||
axes[0].set_ylabel("Acceleration")
|
||
axes[0].grid(True, alpha=0.25)
|
||
|
||
axes[1].plot(freqs, amplitudes, color="tab:purple")
|
||
axes[1].axvline(frequency_hz, color="tab:red", linestyle="--", linewidth=1.2, label=f"f={frequency_hz:.4f}Hz")
|
||
axes[1].set_xlim(0.0, 5.0)
|
||
axes[1].set_title("Input Spectrum")
|
||
axes[1].set_xlabel("Frequency (Hz)")
|
||
axes[1].set_ylabel("Amplitude")
|
||
axes[1].grid(True, alpha=0.25)
|
||
axes[1].legend()
|
||
|
||
if true_rms is None:
|
||
axes[2].bar(["Pred RMS"], [pred_rms], color=["tab:orange"])
|
||
axes[2].set_title(f"Pred RMS = {pred_rms:.6f} (truth sensor not provided)")
|
||
else:
|
||
axes[2].bar(["True RMS", "Pred RMS"], [true_rms, pred_rms], color=["tab:green", "tab:orange"])
|
||
err_pct = abs(pred_rms - true_rms) / max(abs(true_rms), 1e-12) * 100.0
|
||
axes[2].set_title(f"True={true_rms:.6f} | Pred={pred_rms:.6f} | Error={err_pct:.2f}%")
|
||
axes[2].set_ylabel("RMS")
|
||
axes[2].grid(True, axis="y", alpha=0.25)
|
||
|
||
plt.tight_layout()
|
||
figure_path = output_dir / f"{file_path.stem}_{title_prefix.replace(' ', '_').lower()}.png"
|
||
plt.savefig(figure_path, dpi=170, bbox_inches="tight")
|
||
plt.close(fig)
|
||
return figure_path
|
||
|
||
|
||
def run_task1(csv_path: Path) -> tuple[dict, Path]:
|
||
from scripts.config import CORE_FEATURE_NAMES
|
||
from scripts.inference_utils import predict_task1_tr
|
||
|
||
with CHECKPOINT_DEFAULT.open("rb") as handle:
|
||
payload = pickle.load(handle)
|
||
model = payload["model"]
|
||
pred_input = build_prediction_input(csv_path=csv_path, for_tmd=False)
|
||
pred_tr, tr_source = predict_task1_tr(
|
||
model,
|
||
feature_vector=pred_input["features"],
|
||
frequency_hz=float(pred_input["frequency_hz"]),
|
||
normalization_eps=float(pred_input["config"].normalization_eps),
|
||
project_root=PROJECT_ROOT,
|
||
prefer_curve=True,
|
||
)
|
||
pred_rms = pred_tr * float(pred_input["x_rms"])
|
||
true_rms = pred_input["true_y_rms"]
|
||
rel_err_pct = None if true_rms is None else abs(pred_rms - true_rms) / max(abs(true_rms), 1e-12) * 100.0
|
||
|
||
out_dir = APP_OUTPUT_DIR / "task1"
|
||
fig_path = save_prediction_overview_figure(
|
||
file_path=csv_path,
|
||
output_dir=out_dir,
|
||
title_prefix="Task1 Prediction",
|
||
time_middle=pred_input["time_middle"],
|
||
x_middle=pred_input["x_middle"],
|
||
sampling_rate=float(pred_input["sampling_rate"]),
|
||
frequency_hz=float(pred_input["frequency_hz"]),
|
||
pred_rms=float(pred_rms),
|
||
true_rms=true_rms,
|
||
)
|
||
result = {
|
||
"task": "task1_rms_prediction",
|
||
"file_path": str(csv_path),
|
||
"frequency_hz": float(pred_input["frequency_hz"]),
|
||
"x_rms": float(pred_input["x_rms"]),
|
||
"true_y_rms": None if true_rms is None else float(true_rms),
|
||
"pred_tr": float(pred_tr),
|
||
"pred_tr_source": str(tr_source),
|
||
"pred_y_rms": float(pred_rms),
|
||
"relative_error_percent": None if rel_err_pct is None else float(rel_err_pct),
|
||
"truth_available": bool(pred_input["has_truth"]),
|
||
"model_name": str(payload.get("model_name", "unknown")),
|
||
"feature_names": ",".join(CORE_FEATURE_NAMES),
|
||
}
|
||
return result, Path(fig_path)
|
||
|
||
|
||
def run_task2(csv_path: Path) -> tuple[dict, Path]:
|
||
from scripts_2.task3_identify import analyze_free_vibration
|
||
|
||
real_csv = csv_path
|
||
if "free_vib" not in {part.lower() for part in csv_path.parts}:
|
||
temp_dir = APP_CACHE_DIR / "free_vib"
|
||
temp_dir.mkdir(parents=True, exist_ok=True)
|
||
real_csv = temp_dir / csv_path.name
|
||
shutil.copyfile(csv_path, real_csv)
|
||
|
||
output_dir = APP_OUTPUT_DIR / "task2"
|
||
result = analyze_free_vibration(
|
||
file_path=real_csv,
|
||
output_dir=output_dir,
|
||
bandwidth_hz=0.3,
|
||
show=False,
|
||
)
|
||
fig_path = Path(str(result["figure_path"]))
|
||
return result, fig_path
|
||
|
||
|
||
def run_task3(csv_path: Path) -> tuple[dict, Path]:
|
||
from scripts.config import make_experiment_config
|
||
from scripts_2.signal_utils import (
|
||
TOP_RESPONSE_AXIS,
|
||
TOP_RESPONSE_SENSOR,
|
||
calculate_rms,
|
||
get_middle_segment,
|
||
read_sensor_signal,
|
||
)
|
||
from scripts_3.task4_predict_freq import analyze_forced_vibration
|
||
|
||
output_dir = APP_OUTPUT_DIR / "task3"
|
||
result = analyze_forced_vibration(
|
||
file_path=csv_path,
|
||
output_dir=output_dir,
|
||
show=False,
|
||
)
|
||
|
||
# Optional inverse estimation:
|
||
# infer base RMS from top-response RMS with fixed-point iteration:
|
||
# x = y_rms / TR_model(f, x)
|
||
try:
|
||
config = make_experiment_config()
|
||
with CHECKPOINT_DEFAULT.open("rb") as handle:
|
||
payload = pickle.load(handle)
|
||
model = payload["model"]
|
||
|
||
time_values, raw_signal = read_sensor_signal(
|
||
csv_path,
|
||
sensor_code=TOP_RESPONSE_SENSOR,
|
||
value_column=TOP_RESPONSE_AXIS,
|
||
)
|
||
_, signal_segment = get_middle_segment(time_values, raw_signal)
|
||
top_rms = float(calculate_rms(signal_segment))
|
||
freq_hz = float(result["predicted_frequency_hz"])
|
||
|
||
min_x = config.data.normalization_eps
|
||
x_est = max(top_rms, min_x)
|
||
converged = False
|
||
rel_change = float("inf")
|
||
iter_count = 0
|
||
|
||
for i in range(1, 41):
|
||
feat = np.asarray(
|
||
[
|
||
[
|
||
freq_hz,
|
||
freq_hz * freq_hz,
|
||
x_est,
|
||
(2.0 * np.pi * freq_hz) ** 2 * config.data.harmonic_amplitude_m,
|
||
]
|
||
],
|
||
dtype=np.float64,
|
||
)
|
||
tr_est = max(float(model.predict(feat)[0]), config.data.normalization_eps)
|
||
x_new = max(top_rms / max(tr_est, config.data.normalization_eps), min_x)
|
||
if not np.isfinite(x_new):
|
||
break
|
||
rel_change = abs(x_new - x_est) / max(abs(x_est), min_x)
|
||
# Damping update to improve stability in fixed-point iteration.
|
||
x_est = 0.6 * x_est + 0.4 * x_new
|
||
iter_count = i
|
||
if rel_change < 1e-4:
|
||
converged = True
|
||
break
|
||
|
||
feat_final = np.asarray(
|
||
[
|
||
[
|
||
freq_hz,
|
||
freq_hz * freq_hz,
|
||
x_est,
|
||
(2.0 * np.pi * freq_hz) ** 2 * config.data.harmonic_amplitude_m,
|
||
]
|
||
],
|
||
dtype=np.float64,
|
||
)
|
||
tr_final = max(float(model.predict(feat_final)[0]), config.data.normalization_eps)
|
||
y_reconstructed = tr_final * x_est
|
||
residual_pct = abs(y_reconstructed - top_rms) / max(abs(top_rms), min_x) * 100.0
|
||
|
||
result["top_response_rms"] = float(top_rms)
|
||
result["estimated_base_rms"] = float(x_est)
|
||
result["estimated_tr_from_model"] = float(tr_final)
|
||
result["inversion_converged"] = bool(converged)
|
||
result["inversion_iterations"] = int(iter_count)
|
||
result["inversion_relative_change"] = float(rel_change)
|
||
result["inversion_residual_percent"] = float(residual_pct)
|
||
result["inversion_note"] = "Estimated by fixed-point inversion using task1 TR model; use as an estimate only."
|
||
except Exception as error:
|
||
result["inversion_note"] = f"Base RMS inversion unavailable: {error}"
|
||
|
||
fig_path = Path(str(result["figure_path"]))
|
||
return result, fig_path
|
||
|
||
|
||
def run_task4_tmd(csv_path: Path) -> tuple[dict, Path]:
|
||
from scripts_4.adapter import load_adapter
|
||
from scripts.inference_utils import predict_task1_tr
|
||
|
||
with CHECKPOINT_DEFAULT.open("rb") as handle:
|
||
payload = pickle.load(handle)
|
||
model = payload["model"]
|
||
adapter, adapter_extra = load_adapter(ADAPTER_DEFAULT)
|
||
pred_input = build_prediction_input(csv_path=csv_path, for_tmd=True)
|
||
pred_tr, tr_source = predict_task1_tr(
|
||
model,
|
||
feature_vector=pred_input["features"],
|
||
frequency_hz=float(pred_input["frequency_hz"]),
|
||
normalization_eps=float(pred_input["config"].normalization_eps),
|
||
project_root=PROJECT_ROOT,
|
||
prefer_curve=True,
|
||
)
|
||
pred_base = pred_tr * float(pred_input["x_rms"])
|
||
pred_tmd = adapter.predict(pred_base, float(pred_input["frequency_hz"]))
|
||
true_rms = pred_input["true_y_rms"]
|
||
rel_err_pct = None if true_rms is None else abs(pred_tmd - true_rms) / max(abs(true_rms), 1e-12) * 100.0
|
||
|
||
out_dir = APP_OUTPUT_DIR / "task4"
|
||
fig_path = save_prediction_overview_figure(
|
||
file_path=csv_path,
|
||
output_dir=out_dir,
|
||
title_prefix="Task4 TMD Prediction",
|
||
time_middle=pred_input["time_middle"],
|
||
x_middle=pred_input["x_middle"],
|
||
sampling_rate=float(pred_input["sampling_rate"]),
|
||
frequency_hz=float(pred_input["frequency_hz"]),
|
||
pred_rms=float(pred_tmd),
|
||
true_rms=true_rms,
|
||
)
|
||
result = {
|
||
"task": "task4_tmd_rms_prediction",
|
||
"file_path": str(csv_path),
|
||
"frequency_hz": float(pred_input["frequency_hz"]),
|
||
"x_rms": float(pred_input["x_rms"]),
|
||
"true_y_rms": None if true_rms is None else float(true_rms),
|
||
"pred_base_y_rms": float(pred_base),
|
||
"pred_tr_source": str(tr_source),
|
||
"pred_tmd_y_rms": float(pred_tmd),
|
||
"adapter_scale_k": float(adapter.scale_at(float(pred_input["frequency_hz"]))),
|
||
"relative_error_percent": None if rel_err_pct is None else float(rel_err_pct),
|
||
"truth_available": bool(pred_input["has_truth"]),
|
||
"model_name": str(payload.get("model_name", "unknown")),
|
||
"adapter_extra": str(adapter_extra),
|
||
}
|
||
return result, Path(fig_path)
|
||
|
||
|
||
def build_key_metrics(task_label: str, result: dict) -> str:
|
||
if task_label == TASK1_LABEL:
|
||
lines = [
|
||
f"预测RMS: {result.get('pred_y_rms')}",
|
||
f"预测TR: {result.get('pred_tr')}",
|
||
f"TR来源: {result.get('pred_tr_source')}",
|
||
f"激振频率(Hz): {result.get('frequency_hz')}",
|
||
f"输入RMS(x): {result.get('x_rms')}",
|
||
f"是否有真值: {'是' if result.get('truth_available') else '否'}",
|
||
]
|
||
if result.get("truth_available"):
|
||
lines.append(f"真值RMS(y): {result.get('true_y_rms')}")
|
||
lines.append(f"误差(%): {result.get('relative_error_percent')}")
|
||
return "\n".join(lines)
|
||
if task_label == TASK2_LABEL:
|
||
lines = [
|
||
f"自振频率f_n(Hz): {result.get('natural_frequency_hz')}",
|
||
f"局部阻尼比zeta: {result.get('damping_ratio')}",
|
||
f"全局阻尼比zeta: {result.get('global_damping_ratio')}",
|
||
f"最高峰后衰减阻尼比zeta: {result.get('max_tail_damping_ratio')}",
|
||
f"局部精确阻尼比: {result.get('damping_ratio_exact')}",
|
||
f"全局精确阻尼比: {result.get('global_damping_ratio_exact')}",
|
||
f"最高峰后衰减精确阻尼比: {result.get('max_tail_damping_ratio_exact')}",
|
||
f"局部包络R²: {result.get('envelope_r2')}",
|
||
f"全局包络R²: {result.get('global_envelope_r2')}",
|
||
f"最高峰后衰减包络R²: {result.get('max_tail_envelope_r2')}",
|
||
f"采样率(Hz): {result.get('sampling_rate_hz')}",
|
||
]
|
||
return "\n".join(lines)
|
||
if task_label == TASK3_LABEL:
|
||
lines = [
|
||
f"预测激振频率(Hz): {result.get('predicted_frequency_hz')}",
|
||
f"稳态采样点数: {result.get('segment_sample_count')}",
|
||
f"采样率(Hz): {result.get('sampling_rate_hz')}",
|
||
]
|
||
if result.get("top_response_rms") is not None:
|
||
lines.append(f"顶部RMS(y): {result.get('top_response_rms')}")
|
||
if result.get("estimated_base_rms") is not None:
|
||
lines.append(f"反演底部RMS(x): {result.get('estimated_base_rms')}")
|
||
lines.append(f"反演TR: {result.get('estimated_tr_from_model')}")
|
||
lines.append(f"反演残差(%): {result.get('inversion_residual_percent')}")
|
||
lines.append(f"反演收敛: {result.get('inversion_converged')}")
|
||
if result.get("true_frequency_hz") is not None:
|
||
lines.append(f"文件名真值(Hz): {result.get('true_frequency_hz')}")
|
||
lines.append(f"绝对误差(Hz): {result.get('absolute_error_hz')}")
|
||
return "\n".join(lines)
|
||
if task_label == TASK4_LABEL:
|
||
lines = [
|
||
f"挑战任务预测RMS: {result.get('pred_tmd_y_rms')}",
|
||
f"基础模型RMS: {result.get('pred_base_y_rms')}",
|
||
f"TR来源: {result.get('pred_tr_source')}",
|
||
f"适配系数k(f): {result.get('adapter_scale_k')}",
|
||
f"激振频率(Hz): {result.get('frequency_hz')}",
|
||
f"是否有真值: {'是' if result.get('truth_available') else '否'}",
|
||
]
|
||
if result.get("truth_available"):
|
||
lines.append(f"真值RMS(y): {result.get('true_y_rms')}")
|
||
lines.append(f"误差(%): {result.get('relative_error_percent')}")
|
||
return "\n".join(lines)
|
||
return pretty_json_like_lines(result)
|
||
|
||
|
||
class EvaluationStudio:
|
||
def __init__(self) -> None:
|
||
self.root = TkRootClass()
|
||
self.root.title("建筑评测一体式测试系统")
|
||
self.root.geometry("1180x760")
|
||
self.root.minsize(980, 620)
|
||
|
||
self.task_var = tk.StringVar(value=TASK1_LABEL)
|
||
self.file_var = tk.StringVar(value="")
|
||
self.force_convert_var = tk.BooleanVar(value=False)
|
||
self.timezone_var = tk.StringVar(value="8")
|
||
self.status_var = tk.StringVar(value="就绪")
|
||
self.preview_image: tk.PhotoImage | None = None
|
||
self.original_image: tk.PhotoImage | None = None
|
||
self.current_image_path: Path | None = None
|
||
|
||
self._build_ui()
|
||
|
||
def _build_ui(self) -> None:
|
||
main = ttk.Frame(self.root, padding=10)
|
||
main.pack(fill="both", expand=True)
|
||
main.columnconfigure(1, weight=1)
|
||
main.rowconfigure(4, weight=1)
|
||
|
||
ttk.Label(main, text="评测任务").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=5)
|
||
task_box = ttk.Combobox(main, textvariable=self.task_var, values=TASK_CHOICES, state="readonly")
|
||
task_box.grid(row=0, column=1, sticky="ew", pady=5)
|
||
|
||
ttk.Label(main, text="CSV 文件").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=5)
|
||
file_entry = ttk.Entry(main, textvariable=self.file_var)
|
||
file_entry.grid(row=1, column=1, sticky="ew", pady=5)
|
||
ttk.Button(main, text="选择文件", command=self._choose_file).grid(row=1, column=2, padx=(8, 0), pady=5)
|
||
|
||
self.drop_label = ttk.Label(
|
||
main,
|
||
text="将 CSV 拖到这里（若无拖拽支持，可用“选择文件”）",
|
||
relief="groove",
|
||
anchor="center",
|
||
)
|
||
self.drop_label.grid(row=2, column=0, columnspan=3, sticky="ew", pady=(4, 8), ipady=12)
|
||
if HAS_DND:
|
||
self.drop_label.drop_target_register(DND_FILES)
|
||
self.drop_label.dnd_bind("<<Drop>>", self._on_drop)
|
||
|
||
config_line = ttk.Frame(main)
|
||
config_line.grid(row=3, column=0, columnspan=3, sticky="ew", pady=5)
|
||
config_line.columnconfigure(6, weight=1)
|
||
ttk.Checkbutton(
|
||
config_line,
|
||
text="强制按老师宽表格式转换",
|
||
variable=self.force_convert_var,
|
||
).grid(row=0, column=0, sticky="w")
|
||
ttk.Label(config_line, text="字符串时间时区偏移(小时)").grid(row=0, column=1, sticky="w", padx=(12, 4))
|
||
ttk.Entry(config_line, textvariable=self.timezone_var, width=7).grid(row=0, column=2, sticky="w")
|
||
ttk.Button(config_line, text="开始评测", command=self._run_clicked).grid(row=0, column=3, padx=(12, 0))
|
||
ttk.Label(config_line, textvariable=self.status_var).grid(row=0, column=4, sticky="w", padx=(12, 0))
|
||
if not HAS_DND:
|
||
ttk.Label(
|
||
config_line,
|
||
text="提示：未检测到 tkinterdnd2，拖拽不可用；可先 pip install tkinterdnd2",
|
||
foreground="#aa5500",
|
||
).grid(row=0, column=5, sticky="w", padx=(12, 0))
|
||
|
||
paned = ttk.Panedwindow(main, orient="horizontal")
|
||
paned.grid(row=4, column=0, columnspan=3, sticky="nsew", pady=(6, 0))
|
||
|
||
left = ttk.Frame(paned, padding=8)
|
||
right = ttk.Frame(paned, padding=8)
|
||
paned.add(left, weight=3)
|
||
paned.add(right, weight=2)
|
||
|
||
ttk.Label(left, text="关键结果").pack(anchor="w")
|
||
self.key_text = ScrolledText(left, wrap="word", height=8)
|
||
self.key_text.pack(fill="x", expand=False, pady=(0, 6))
|
||
|
||
ttk.Label(left, text="详细结果").pack(anchor="w")
|
||
self.result_text = ScrolledText(left, wrap="word", height=16)
|
||
self.result_text.pack(fill="both", expand=True)
|
||
|
||
ttk.Label(right, text="图像预览（最新结果）").pack(anchor="w")
|
||
self.image_label = ttk.Label(right, relief="sunken", anchor="center")
|
||
self.image_label.pack(fill="both", expand=True)
|
||
self.image_label.bind("<Configure>", self._on_image_area_resize)
|
||
|
||
def _choose_file(self) -> None:
|
||
file_path = filedialog.askopenfilename(
|
||
title="选择CSV文件",
|
||
filetypes=[("CSV Files", "*.csv"), ("All Files", "*.*")],
|
||
initialdir=str(PROJECT_ROOT),
|
||
)
|
||
if file_path:
|
||
self.file_var.set(file_path)
|
||
|
||
def _on_drop(self, event) -> None: # noqa: ANN001
|
||
raw = event.data
|
||
if not raw:
|
||
return
|
||
paths = self.root.tk.splitlist(raw)
|
||
if not paths:
|
||
return
|
||
self.file_var.set(str(paths[0]).strip("{}"))
|
||
|
||
def _append_result(self, text: str) -> None:
|
||
self.result_text.insert("end", text + "\n")
|
||
self.result_text.see("end")
|
||
|
||
def _on_image_area_resize(self, _event) -> None: # noqa: ANN001
|
||
self._refresh_image_fit()
|
||
|
||
def _refresh_image_fit(self) -> None:
|
||
if self.original_image is None:
|
||
return
|
||
width = max(self.image_label.winfo_width(), 1)
|
||
height = max(self.image_label.winfo_height(), 1)
|
||
orig_w = max(self.original_image.width(), 1)
|
||
orig_h = max(self.original_image.height(), 1)
|
||
ratio_w = math.ceil(orig_w / width)
|
||
ratio_h = math.ceil(orig_h / height)
|
||
ratio = max(1, ratio_w, ratio_h)
|
||
scaled = self.original_image.subsample(ratio, ratio)
|
||
self.preview_image = scaled
|
||
self.image_label.configure(image=scaled, text="")
|
||
|
||
def _set_image(self, image_path: Path | None) -> None:
|
||
if image_path is None or not image_path.exists():
|
||
self.image_label.configure(image="", text="暂无图像")
|
||
self.preview_image = None
|
||
self.original_image = None
|
||
self.current_image_path = None
|
||
return
|
||
try:
|
||
self.original_image = tk.PhotoImage(file=str(image_path))
|
||
self.current_image_path = image_path
|
||
self._refresh_image_fit()
|
||
except Exception:
|
||
self.image_label.configure(text=f"图像已生成：\n{image_path}", image="")
|
||
self.preview_image = None
|
||
self.original_image = None
|
||
self.current_image_path = image_path
|
||
|
||
def _run_clicked(self) -> None:
|
||
csv_text = self.file_var.get().strip()
|
||
if not csv_text:
|
||
messagebox.showerror("缺少输入", "请先选择或拖入CSV文件。")
|
||
return
|
||
csv_path = Path(csv_text).expanduser().resolve()
|
||
if not csv_path.exists():
|
||
messagebox.showerror("文件不存在", f"找不到文件：\n{csv_path}")
|
||
return
|
||
|
||
try:
|
||
tz_offset = float(self.timezone_var.get().strip())
|
||
except ValueError:
|
||
messagebox.showerror("参数错误", "时区偏移必须是数字。")
|
||
return
|
||
|
||
self.status_var.set("运行中...")
|
||
self.key_text.delete("1.0", "end")
|
||
self.result_text.delete("1.0", "end")
|
||
self._set_image(None)
|
||
|
||
thread = threading.Thread(
|
||
target=self._run_pipeline,
|
||
args=(csv_path, self.task_var.get(), self.force_convert_var.get(), tz_offset),
|
||
daemon=True,
|
||
)
|
||
thread.start()
|
||
|
||
def _run_pipeline(self, csv_path: Path, task_label: str, force_convert: bool, tz_offset: float) -> None:
|
||
try:
|
||
APP_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
use_csv, converted = ensure_long_format_csv(csv_path, force_convert=force_convert, tz_offset_hours=tz_offset)
|
||
notes = [
|
||
f"输入文件: {csv_path}",
|
||
f"实际评测文件: {use_csv}",
|
||
f"是否执行宽转长: {'是' if converted else '否'}",
|
||
f"任务: {task_label}",
|
||
"",
|
||
]
|
||
|
||
if task_label == TASK1_LABEL:
|
||
result, fig_path = run_task1(use_csv)
|
||
elif task_label == TASK2_LABEL:
|
||
result, fig_path = run_task2(use_csv)
|
||
elif task_label == TASK3_LABEL:
|
||
result, fig_path = run_task3(use_csv)
|
||
elif task_label == TASK4_LABEL:
|
||
result, fig_path = run_task4_tmd(use_csv)
|
||
else:
|
||
raise ValueError(f"不支持的任务：{task_label}")
|
||
|
||
output_text = "\n".join(notes) + pretty_json_like_lines(result)
|
||
key_text = build_key_metrics(task_label, result)
|
||
self.root.after(0, lambda: self.key_text.insert("end", key_text + "\n"))
|
||
self.root.after(0, lambda: self.result_text.insert("end", output_text + "\n"))
|
||
self.root.after(0, lambda: self._set_image(fig_path))
|
||
self.root.after(0, lambda: self.status_var.set("完成"))
|
||
except Exception as error:
|
||
stack = traceback.format_exc()
|
||
message = f"运行失败：{error}\n\n{stack}"
|
||
self.root.after(0, lambda: self.key_text.insert("end", "运行失败，请查看详细错误。\n"))
|
||
self.root.after(0, lambda: self.result_text.insert("end", message + "\n"))
|
||
self.root.after(0, lambda: self.status_var.set("失败"))
|
||
|
||
def run(self) -> None:
|
||
self.root.mainloop()
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="Integrated visual evaluation studio for tasks 1~4.")
|
||
parser.add_argument("--task", type=str, default=None, choices=TASK_CHOICES)
|
||
parser.add_argument("--file", type=str, default=None, help="Optional CSV file path to preload.")
|
||
parser.add_argument(
|
||
"--force-convert",
|
||
action="store_true",
|
||
help="Force teacher wide-format conversion before running tasks.",
|
||
)
|
||
parser.add_argument("--timezone-offset-hours", type=float, default=8.0)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
app = EvaluationStudio()
|
||
if args.task:
|
||
app.task_var.set(args.task)
|
||
if args.file:
|
||
app.file_var.set(str(Path(args.file).expanduser().resolve()))
|
||
if args.force_convert:
|
||
app.force_convert_var.set(True)
|
||
app.timezone_var.set(str(args.timezone_offset_hours))
|
||
app.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|