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, extract_frequency_hz, 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 = extract_frequency_hz(csv_path.name) if frequency_hz is None: 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 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 = max(float(model.predict([pred_input["features"].tolist()])[0]), pred_input["config"].normalization_eps) 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_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 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 = max(float(model.predict([pred_input["features"].tolist()])[0]), pred_input["config"].normalization_eps) 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_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"激振频率(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"精确阻尼比: {result.get('damping_ratio_exact')}", f"包络R²: {result.get('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"适配系数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("<>", 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("", 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()