"""可视化模块:双栏对比图(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