From ab31b11fb5f14269d7bdde8aa3d932011bc33c57 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 21:19:09 +0800 Subject: [PATCH] feat: add render-algo subcommand for single-algorithm GT+detection PNG --- tests/test_render_algo.py | 37 +++++++++ ts_anomaly_td/cli.py | 30 +++++++ ts_anomaly_td/visualize.py | 159 +++++++++++++++++++++++++++++++++++++ 3 files changed, 226 insertions(+) create mode 100644 tests/test_render_algo.py diff --git a/tests/test_render_algo.py b/tests/test_render_algo.py new file mode 100644 index 0000000..b183e44 --- /dev/null +++ b/tests/test_render_algo.py @@ -0,0 +1,37 @@ +"""render_algo 单元测试。""" + +import json +import tempfile +from pathlib import Path + +from ts_anomaly_td.visualize import render_algo + + +def _make_ts_vals_labels(n=200): + """生成测试用时间序列。""" + ts = [i * 3600_000 for i in range(n)] # 每小时 + vals = [float(i % 20) for i in range(n)] + labels = [1 if 50 <= i <= 60 else 0 for i in range(n)] + return ts, vals, labels + + +def test_render_algo_basic(): + """render_algo 应生成有效 PNG。""" + ts, vals, labels = _make_ts_vals_labels() + windows = [[ts[50], ts[60]]] + + with tempfile.TemporaryDirectory() as tmpdir: + out = Path(tmpdir) / "test_algo.png" + result = render_algo(ts, vals, labels, windows, "ksigma", str(out), title="test") + assert Path(result).exists() + assert Path(result).stat().st_size > 5000 + + +def test_render_algo_empty_windows(): + """空检测窗口应正常渲染。""" + ts, vals, labels = _make_ts_vals_labels(50) + + with tempfile.TemporaryDirectory() as tmpdir: + out = Path(tmpdir) / "empty.png" + result = render_algo(ts, vals, labels, [], "iqr", str(out)) + assert Path(result).exists() diff --git a/ts_anomaly_td/cli.py b/ts_anomaly_td/cli.py index 47cdbe5..b4828fc 100644 --- a/ts_anomaly_td/cli.py +++ b/ts_anomaly_td/cli.py @@ -209,9 +209,38 @@ def build_parser() -> argparse.ArgumentParser: p_watch.add_argument("--window-sec", type=int, default=30, help="检测间隔秒数 (default: 30)") p_watch.add_argument("--url", default="ws://root:taosdata@localhost:6041", help="TDengine WebSocket URL") + # render-algo + p_ra = sub.add_parser("render-algo", help="单算法 GT+检测渲染 PNG") + p_ra.add_argument("--gt-csv", required=True, help="GT CSV 文件路径") + p_ra.add_argument("--det-json", required=True, help="detect-batch --json 输出的 JSON 文件") + p_ra.add_argument("--algo", required=True, help="算法名称") + p_ra.add_argument("--output", required=True, help="PNG 输出路径") + p_ra.add_argument("--value-col", default="value", help="数值列名 (default: value)") + p_ra.add_argument("--title", default="", help="图表标题") + return parser +def cmd_render_algo(args): + """render-algo 子命令：单算法 GT+检测渲染 PNG。""" + logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") + + import json + from ts_anomaly_td.io_csv import read_csv + from ts_anomaly_td.visualize import render_algo + + ts, vals, labels = read_csv(args.gt_csv, value_col=args.value_col) + + with open(args.det_json) as f: + det_data = json.load(f) + + # det_json 格式: {"algo": "ksigma", "windows": [[start, end], ...], "error": null} + windows = det_data.get("windows", []) + + title = args.title or f"{Path(args.gt_csv).stem} / {args.algo}" + render_algo(ts, vals, labels, windows, args.algo, args.output, title=title) + + def main(): parser = build_parser() args = parser.parse_args() @@ -223,6 +252,7 @@ def main(): "e2e": cmd_e2e, "visualize": cmd_visualize, "watch": cmd_watch, + "render-algo": cmd_render_algo, } fn = dispatch.get(args.command) diff --git a/ts_anomaly_td/visualize.py b/ts_anomaly_td/visualize.py index 979e32c..938ed6a 100644 --- a/ts_anomaly_td/visualize.py +++ b/ts_anomaly_td/visualize.py @@ -153,3 +153,162 @@ def render_contrast( plt.close(fig) logger.info("chart saved: %s", output_path) return output_path + + +def render_algo( + timestamps: list[int], + values: list[float], + gt_labels: list[int], + det_windows: list[list[int]], + algo: str, + output_path: str, + title: str = "", +) -> str: + """单算法渲染：GT（红）+ 该算法检测窗口（算法色）→ PNG。 + + Args: + timestamps: 毫秒时间戳列表 + values: 数值列表 + gt_labels: GT 标签（0/1） + det_windows: [[start_ts, end_ts], ...] 该算法检测窗口 + algo: 算法名称 + 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 = plt.subplots(1, 1, figsize=(14, 5)) + + t0 = timestamps[0] + hours = [(t - t0) / 3_600_000 for t in timestamps] + + # 原始序列 + ax.plot(hours, values, color="black", linewidth=0.6, alpha=0.8) + + # GT 红色 axvspan + 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.axvspan(anom_start, hours[i], alpha=0.15, color="red", label="GT" if len(gt_windows) == 0 else None) + gt_windows.append((anom_start, hours[i])) + if in_anomaly: + ax.axvspan(anom_start, hours[-1], alpha=0.15, color="red", label="GT" if len(gt_windows) == 0 else None) + gt_windows.append((anom_start, hours[-1])) + + # 该算法检测窗口 + color = ALGO_COLORS.get(algo, "#888888") + det_hours = [] + for i, (wstart, wend) in enumerate(det_windows): + ws_h = (wstart - t0) / 3_600_000 + we_h = (wend - t0) / 3_600_000 + ax.axvspan(ws_h, we_h, alpha=0.2, color=color, label=algo if i == 0 else None) + det_hours.append((ws_h, we_h)) + + # IoU + 统计 + iou = _compute_iou(gt_windows, det_hours) + stats_text = f"GT: {len(gt_windows)} Det: {len(det_windows)} IoU: {iou:.3f}" + ax.set_title(f"{title} — {stats_text}" if title else stats_text) + ax.set_ylabel("Value") + ax.set_xlabel("Time (hours)") + + if gt_windows or det_windows: + ax.legend(loc="upper right", fontsize=7) + + 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("render_algo saved: %s", output_path) + return output_path + + +def montage_pngs( + image_paths: list[str], + output_path: str, + cols: int = 2, + title: str = "", +) -> str: + """将多张等宽 PNG 按网格拼成一张对比图。 + + Args: + image_paths: PNG 文件路径列表 + output_path: 拼图输出路径 + cols: 列数 + title: 总标题 + + Returns: + output_path + """ + from PIL import Image, ImageDraw, ImageFont + + if not image_paths: + raise ValueError("image_paths 为空") + + images = [] + for p in image_paths: + img = Image.open(p) + images.append(img) + + # 统一宽度为第一张图的宽度 + target_w = images[0].width + resized = [] + for img in images: + if img.width != target_w: + ratio = target_w / img.width + new_h = int(img.height * ratio) + img = img.resize((target_w, new_h), Image.LANCZOS) + resized.append(img) + + n = len(resized) + rows_count = (n + cols - 1) // cols + + # 计算每行高度 + row_heights = [] + for r in range(rows_count): + max_h = max(resized[r * cols + c].height for c in range(cols) if r * cols + c < n) + row_heights.append(max_h) + + padding = 10 + title_h = 40 if title else 0 + total_w = target_w * cols + padding * (cols + 1) + total_h = sum(row_heights) + padding * (rows_count + 1) + title_h + + canvas = Image.new("RGB", (total_w, total_h), "white") + draw = ImageDraw.Draw(canvas) + + if title: + try: + font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 24) + except Exception: + font = ImageFont.load_default() + draw.text((total_w // 2, 15), title, fill="black", anchor="mm", font=font) + + y_offset = title_h + padding + for r in range(rows_count): + x_offset = padding + for c in range(cols): + idx = r * cols + c + if idx >= n: + break + img = resized[idx] + # 垂直居中 + y_pad = (row_heights[r] - img.height) // 2 + canvas.paste(img, (x_offset, y_offset + y_pad)) + x_offset += target_w + padding + y_offset += row_heights[r] + padding + + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + canvas.save(output_path) + logger.info("montage saved: %s (%d images, %dx%d)", output_path, n, cols, rows_count) + return output_path