From dd532059219792b7d725532b1bb94d45debdb6ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E5=AE=97=E5=B9=B3?= Date: Thu, 11 Jun 2026 15:45:57 +0800 Subject: [PATCH] feat: add visualize + e2e orchestrator + run_e2e.sh --- scripts/run_e2e.sh | 48 ++++++++++++ ts_anomaly_td/e2e.py | 120 +++++++++++++++++++++++++++- ts_anomaly_td/visualize.py | 155 +++++++++++++++++++++++++++++++++++++ 3 files changed, 321 insertions(+), 2 deletions(-) create mode 100755 scripts/run_e2e.sh create mode 100644 ts_anomaly_td/visualize.py diff --git a/scripts/run_e2e.sh b/scripts/run_e2e.sh new file mode 100755 index 0000000..b4db834 --- /dev/null +++ b/scripts/run_e2e.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +PROJECT_DIR="$(dirname "$SCRIPT_DIR")" +cd "$PROJECT_DIR" + +echo "=== ts-anomaly-td E2E ===" +echo "" + +echo "[1/5] Starting TDengine container..." +docker compose up -d + +echo "[2/5] Waiting for TDengine to be healthy..." +for i in $(seq 1 60); do + if docker compose exec -T tdengine taos -s "SELECT SERVER_VERSION()" > /dev/null 2>&1; then + echo "TDengine ready (attempt $i)" + break + fi + sleep 2 +done + +mkdir -p render logs + +echo "[3/5] Running E2E pipeline..." +uv run python -m ts_anomaly_td e2e --data-dir data --output-dir render --log-dir logs + +echo "[4/5] Checking results..." +for png in render/*.png; do + if [ -f "$png" ]; then + sz=$(stat -c%s "$png" 2>/dev/null || stat -f%z "$png" 2>/dev/null || echo 0) + echo " $png: ${sz} bytes" + if [ "$sz" -lt 10240 ]; then + echo " WARNING: PNG too small (< 10KB)" + fi + fi +done + +for log in logs/e2e_*.json; do + if [ -f "$log" ]; then + echo " $log: $(wc -c < "$log") bytes" + fi +done + +echo "[5/5] Done." +echo "" +echo "To clean up: docker compose down" +echo "To keep containers for debug: leave them running" diff --git a/ts_anomaly_td/e2e.py b/ts_anomaly_td/e2e.py index b534e7e..95799bf 100644 --- a/ts_anomaly_td/e2e.py +++ b/ts_anomaly_td/e2e.py @@ -1,3 +1,119 @@ -"""E2E 编排器（占位 stub）""" +"""E2E 编排器：inject → 6 算法 → FORECAST → 可视化 → JSON 日志。""" + +import json +import logging +import time +from pathlib import Path +from datetime import datetime + +from ts_anomaly_td.connector import TDConnection +from ts_anomaly_td.schema import setup_schema +from ts_anomaly_td.io_csv import read_csv +from ts_anomaly_td.detection import detect_all_algos, ALL_ALGOS +from ts_anomaly_td.forecast import forecast_anomaly +from ts_anomaly_td.visualize import render_contrast + +logger = logging.getLogger(__name__) + + def run_e2e(args): - raise NotImplementedError("e2e 将在 Task 7 实现") + """执行 E2E 全流程。 + + Args: + args: argparse namespace（含 data_dir, output_dir, log_dir, url） + """ + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + ) + + data_dir = Path(args.data_dir) + output_dir = Path(args.output_dir) + log_dir = Path(args.log_dir) + output_dir.mkdir(parents=True, exist_ok=True) + log_dir.mkdir(parents=True, exist_ok=True) + + csv_files = sorted(data_dir.glob("*.csv")) + if not csv_files: + logger.error("no CSV files found in %s", data_dir) + return 1 + + conn = TDConnection(args.url) + summary = {"run_at": datetime.now().isoformat(), "datasets": {}, "exit_code": 0} + last_log_path = None + + try: + for csv_path in csv_files: + stable = csv_path.stem + logger.info("=== E2E %s ===", stable) + + ts, vals, labels = read_csv(str(csv_path)) + rows = list(zip(ts, vals, labels)) + + t0 = time.monotonic() + setup_schema(conn, stable) + conn.batch_insert(stable, rows) + inject_time = time.monotonic() - t0 + logger.info("inject: %d rows in %.1fs", len(rows), inject_time) + + t0 = time.monotonic() + det_results = detect_all_algos(conn, stable) + detect_time = time.monotonic() - t0 + for algo, r in det_results.items(): + if r["error"]: + logger.info(" %s: ERROR %s", algo, r["error"]) + else: + logger.info(" %s: %d windows", algo, len(r["windows"])) + + t0 = time.monotonic() + try: + fc_points = forecast_anomaly(conn, stable) + fc_time = time.monotonic() - t0 + fc_anom_count = sum(1 for p in fc_points if p["is_anomaly"]) + logger.info("FORECAST: %d points, %d anomalies in %.1fs", len(fc_points), fc_anom_count, fc_time) + except Exception as e: + logger.warning("FORECAST failed (non-fatal): %s", e) + fc_points = [] + fc_time = time.monotonic() - t0 + fc_anom_count = 0 + + t0 = time.monotonic() + png_path = output_dir / f"{stable}_gt_vs_tdengine.png" + render_contrast(ts, vals, labels, det_results, str(png_path), title=stable) + viz_time = time.monotonic() - t0 + + ds_summary = { + "rows": len(rows), + "inject_time_s": round(inject_time, 1), + "detect_time_s": round(detect_time, 1), + "forecast_time_s": round(fc_time, 1), + "viz_time_s": round(viz_time, 1), + "algorithms": {}, + "forecast": { + "point_count": len(fc_points), + "anomaly_count": fc_anom_count, + }, + "png": str(png_path), + } + for algo, r in det_results.items(): + ds_summary["algorithms"][algo] = { + "window_count": len(r["windows"]), + "error": r["error"], + } + + summary["datasets"][stable] = ds_summary + + last_log_path = log_dir / f"e2e_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + last_log_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False)) + logger.info("log saved: %s", last_log_path) + + except Exception as e: + logger.exception("E2E failed: %s", e) + summary["exit_code"] = 1 + summary["error"] = str(e) + last_log_path = log_dir / f"e2e_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + last_log_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False)) + finally: + conn.close() + + return summary["exit_code"] diff --git a/ts_anomaly_td/visualize.py b/ts_anomaly_td/visualize.py new file mode 100644 index 0000000..979e32c --- /dev/null +++ b/ts_anomaly_td/visualize.py @@ -0,0 +1,155 @@ +"""可视化模块：双栏对比图（GT 上、TDengine 检测下）。""" + +import logging +from pathlib import Path +import matplotlib +matplotlib.use("Agg") +import matplotlib.pyplot as plt +import matplotlib.font_manager as fm + +logger = logging.getLogger(__name__) + +ALGO_COLORS = { + "ksigma": "#1f77b4", + "iqr": "#2ca02c", + "grubbs": "#9467bd", + "shesd": "#ff7f0e", + "lof": "#17becf", + "sample_ad_model": "#e377c2", +} + + +def _find_cjk_font(): + """查找可用 CJK 字体。""" + candidates = ["Noto Serif CJK SC", "Noto Sans CJK SC", "WenQuanYi Micro Hei", + "DejaVu Sans", "sans-serif"] + available = {f.name for f in fm.fontManager.ttflist} + for name in candidates: + if name in available: + return name + return "sans-serif" + + +def _compute_iou(gt_windows, det_windows): + """计算 GT 与检测窗口的 IoU。 + + 窗口格式: [(start_ts, end_ts), ...] + IoU = 交集时长 / 并集时长 + """ + if not gt_windows or not det_windows: + return 0.0 + + def overlap(a_start, a_end, b_start, b_end): + o_start = max(a_start, b_start) + o_end = min(a_end, b_end) + return max(0, o_end - o_start) + + intersection = 0.0 + union = 0.0 + + for gw in gt_windows: + union += gw[1] - gw[0] + for dw in det_windows: + intersection += overlap(gw[0], gw[1], dw[0], dw[1]) + + for dw in det_windows: + union += dw[1] - dw[0] + + if union == 0: + return 0.0 + return min(intersection / union, 1.0) + + +def render_contrast( + timestamps: list[int], + values: list[float], + gt_labels: list[int], + detection_results: dict, + output_path: str, + title: str = "", +) -> str: + """渲染双栏对比图。 + + Args: + timestamps: 毫秒时间戳列表 + values: 数值列表 + gt_labels: GT 标签（0/1） + detection_results: {algo: {"windows": [...]}} 来自 detect_all_algos + output_path: PNG 输出路径 + title: 图表标题 + + Returns: + output_path + """ + font_name = _find_cjk_font() + plt.rcParams["font.family"] = font_name + plt.rcParams["font.size"] = 9 + + fig, (ax_gt, ax_det) = plt.subplots(2, 1, sharex=True, figsize=(14, 8)) + + t0 = timestamps[0] + hours = [(t - t0) / 3_600_000 for t in timestamps] + + ax_gt.plot(hours, values, color="black", linewidth=0.6, alpha=0.8) + ax_gt.set_ylabel("Value") + ax_gt.set_title(f"Ground Truth — {title}" if title else "Ground Truth") + + in_anomaly = False + anom_start = None + gt_windows = [] + for i, label in enumerate(gt_labels): + if label == 1 and not in_anomaly: + in_anomaly = True + anom_start = hours[i] + elif label == 0 and in_anomaly: + in_anomaly = False + ax_gt.axvspan(anom_start, hours[i], alpha=0.25, color="red") + gt_windows.append((anom_start, hours[i])) + + if in_anomaly: + ax_gt.axvspan(anom_start, hours[-1], alpha=0.25, color="red") + gt_windows.append((anom_start, hours[-1])) + + ax_det.plot(hours, values, color="black", linewidth=0.6, alpha=0.3) + ax_det.set_ylabel("Value") + ax_det.set_xlabel("Time (hours)") + + all_det_windows = [] + for algo, result in detection_results.items(): + if result.get("error"): + continue + color = ALGO_COLORS.get(algo, "#888888") + for wstart, wend in result.get("windows", []): + wstart_h = (wstart - timestamps[0]) / 3_600_000 if wstart > timestamps[0] else 0 + wend_h = (wend - timestamps[0]) / 3_600_000 + ax_det.axvspan(wstart_h, wend_h, alpha=0.2, color=color) + all_det_windows.append((wstart_h, wend_h)) + + legend_handles = [] + for algo in detection_results: + if algo in ALGO_COLORS and not detection_results[algo].get("error"): + from matplotlib.patches import Patch + legend_handles.append(Patch(color=ALGO_COLORS[algo], alpha=0.5, label=algo)) + if legend_handles: + ax_det.legend(handles=legend_handles, loc="upper right", fontsize=7) + + ax_det.set_title(f"TDengine Detection — {title}" if title else "TDengine Detection (6 algos)") + + iou = _compute_iou(gt_windows, all_det_windows) + total_det = len(all_det_windows) + stats_text = f"GT windows: {len(gt_windows)}\nDetected: {total_det}\nIoU: {iou:.3f}" + ax_det.text( + 0.98, 0.97, stats_text, + transform=ax_det.transAxes, + verticalalignment="top", + horizontalalignment="right", + bbox=dict(boxstyle="round", facecolor="white", alpha=0.8), + fontsize=8, + ) + + plt.tight_layout() + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + fig.savefig(output_path, dpi=150, bbox_inches="tight") + plt.close(fig) + logger.info("chart saved: %s", output_path) + return output_path