feat: init异常检测

罗广辉

10 hours ago 2e68b1338ea04125ce4360e6d2ddf03f54df86ac

feat: init异常检测

 PROJECT_CONTEXT.md

@@ -1,6 +1,6 @@
# GeoAI Workbench Current Context

Last updated: 2026-08-18
Last updated: 2026-08-19

This file is the current project snapshot for new Codex tasks. Keep it concise and replace stale facts instead of appending a conversation diary.

@@ -40,6 +40,22 @@
| ultralytics | 8.4.118 |

`pip check` passes in this environment.

Dedicated anomaly-detection environment:

`D:\lgh\code\geoai-workbench\.venvs\09-anomaly-detection`

| Component | Version |
| --- | --- |
| Python | 3.12.10 |
| geoai-py | 0.42.0 |
| scikit-learn | 1.9.0 |
| SciPy | 1.18.0 |
| rasterio | 1.5.1 |
| geopandas | 1.1.4 |
| OpenCV | 4.14.0 |

The environment is CPU-only, contains no model weights, and `pip check` passes.

Dedicated semantic-mapping environment:

@@ -112,7 +128,7 @@
- Start command: `py -3.12 .\scripts\serve_workbench_console.py`
- URL: `http://127.0.0.1:6173` (built console); Vite development is `http://127.0.0.1:6174/apps/workbench-console/`. Only `6xxx` ports are accepted.
- Scope: independent local experiment workbench for this repository. It has no code, account, or product API link to the two drone-product repositories. The map client directly requests public ArcGIS tiles and, only when configured, TianDiTu tiles.
- Current workflow: `00-change-detection`, `01-object-detection`, `02-semantic-mapping`, `04-spatial-measurement`, and `15-trajectory-analysis` support new local runs, searchable case libraries, and visual result workspaces. Change detection shows two-date imagery, a ChangeStar raster overlay, pixel-coordinate GeoJSON, per-feature probability stats and downloads. Spatial measurement shows raster/vector results, per-object metrics and downloads. Change detection now uploads large TIFFs as raw binary streams and copies bytes into a new run directory before invoking the fixed capability script; the original file is not rewritten. The server creates a new run ID for every submission, only accepts allowlisted input types and enabled tasks, and calls fixed capability scripts in fixed virtual environments.
- Current workflow: `00-change-detection`, `01-object-detection`, `02-semantic-mapping`, `04-spatial-measurement`, `09-anomaly-detection`, and `15-trajectory-analysis` support new local runs, searchable case libraries, and visual result workspaces. Anomaly detection accepts separate normal-reference and inspection groups through bounded binary uploads, runs asynchronously, and shows references, input/output comparison, heatmaps, candidates and downloads. Change detection shows two-date imagery, a ChangeStar raster overlay, pixel-coordinate GeoJSON, per-feature probability stats and downloads. Spatial measurement shows raster/vector results, per-object metrics and downloads. The server creates a new run ID for every submission, only accepts allowlisted input types and enabled tasks, and calls fixed capability scripts in fixed virtual environments.
- File exposure: the static handler permits only console assets, `shared/outputs`, and source images required for result comparison. Upload APIs write new raw and processed run directories but do not expose the rest of the repository.
- Styling: `apps/workbench-console/src/styles.css` is organized into design variables, application shell, shared workbench components, capability workspaces, maps, and responsive rules. New capability pages must use a scoped workspace class and shared spacing variables instead of global override patches.

@@ -124,11 +140,25 @@
| `01-object-detection` | In progress; runnable people and vehicle experiments | Tiled YOLO helps people; GeoAI NWPU substantially improves top-down vehicles; tree detection is not implemented. |
| `02-semantic-mapping` | Runnable CPU baseline verified | B capability: deterministic RGB/HSV segmentation plus `geoai.masks_to_vector`; raster and vector outputs are inspectable, but this is not a trained GeoAI model. |
| `04-spatial-measurement` | Runnable CPU Demo verified | B capability: `geoai.masks_to_vector` plus ecosystem measurement; counts labelled raster objects and reports area/perimeter in projected or explicitly non-metric pixel/coordinate units. |
| `09-anomaly-detection` | Runnable CPU CLI Demo verified | B capability: interpretable image features plus robust rules and Isolation Forest score visual outliers; `geoai-py` supplies `masks_to_vector`. Results are review candidates, not business-event conclusions. |
| `15-trajectory-analysis` | Runnable CPU Demo verified | C capability consuming timestamped tracks: spatial metrics, DBSCAN and explicit behavior rules; not a built-in `geoai-py` function. |
| `03`, `05` through `14`, `16` through `18` | Directory and initial README only | No verified local Demo yet. Start each one through `$geoai-capability-builder`. |
| `03`, `05` through `08`, `10` through `14`, `16` through `18` | Directory and initial README only | No verified local Demo yet. Start each one through `$geoai-capability-builder`. |

## Semantic Mapping Snapshot

## Anomaly Detection Snapshot

- Script: `capabilities/09-anomaly-detection/run_anomaly_detection.py`; reproducible input generator: `generate_validation_inputs.py` in the same directory.
- Boundary: B. Seventeen interpretable RGB/HSV/texture features feed a robust MAD rule baseline and a 200-tree Isolation Forest; fixed-view inputs can use corresponding-location robust rules after image-wide appearance normalization. `geoai-py 0.42.0` only supplies `masks_to_vector`. There are no model weights.
- Contract: image or directory input; minimum 50 reference windows; outputs are two heatmaps, comparison overlay, encoded mask GeoTIFF, candidate GeoJSON, tile CSV and run metadata. Ordinary images use pixel coordinates; valid GeoTIFF CRS/transform is preserved.
- Validation: controlled color, blur and dark anomalies were injected into a copy of `07-16.jpg`; `07-19.jpg` was used unchanged as a difficult real sample. Source-copy SHA256 values match the originals.
- Measured result: 588 reference windows and three inputs completed in 41.051 seconds CPU. The rule method hit 3/3 controlled regions with precision 0.583, recall 0.672 and IoU 0.454; Isolation Forest hit 2/3 with precision 0.831, recall 0.500 and IoU 0.454. Normal-image coverage was 0.33% and 0.00%, respectively. Every image completed within 15 seconds.
- Visual limitation: the normal image has a small rule false positive at a vegetation/rock transition. Difficult-image candidates occur around concrete structure, a dark opening and a top dark region, consistent with viewpoint/material/lighting shift; no real anomaly or business accuracy is claimed.
- Console: `GET/POST /api/anomaly-detection/runs`, binary `PUT /api/anomaly-detection/uploads/<upload-id>` and asynchronous `GET /api/anomaly-detection/jobs/<job-id>`. A real one-reference/one-input upload produced run `anomaly-20260818-102845-61f55b` with six review candidates. Desktop and 390 px layouts were visually checked; tables use internal horizontal scrolling and the collapsed menu trigger remains inside the viewport.
- Synthetic fixed-view validation: five ImageGen normal references cover overcast, sunny, after-rain, morning-haze and late-afternoon conditions; one test image adds a localized branch/leaf/mud/rock blockage. The original pooled-feature run returned zero candidates and is retained as `ImageGen 五参考图堵塞复核（漏检案例）`.
- Fixed-view optimization: `--spatial-mode auto` now selects same-position robust rules when 3+ references and all target images share dimensions, after normalizing each image's global feature profile; otherwise it falls back to the original global rule. At the unchanged 0.995 quantile, the synthetic blockage produced one correctly located rule candidate with 9.375% coverage. Five leave-one-normal-out checks produced 7.292%, 0%, 0%, 0%, and 0% coverage. The successful output is retained as `ImageGen 五参考图堵塞复核（同位置优化）`; Isolation Forest still missed this sample.
- Next decision: add image registration and/or a channel ROI or specialist blockage detector before accepting camera motion or claiming blockage semantics. Obtain real normal/labelled anomaly pairs; the synthetic result is only a pipeline check.

## Change Detection Snapshot

- Script: `capabilities/00-change-detection/run_change_detection.py`.

 apps/workbench-console/README.md

@@ -57,7 +57,7 @@
- `scripts/serve_workbench_console.py`：本地控制台服务和受限运行 API；仅暴露构建后的控制台、结果工件与必要原图，且只调用固定虚拟环境和能力脚本。
- `shared/outputs/`：能力原始输出；控制台不复制、不修改这些文件。

当前已接入 `00-change-detection`、`01-object-detection`、`02-semantic-mapping`、`04-spatial-measurement` 和 `15-trajectory-analysis`。变化检测案例显示两期原图、变化叠加、像素坐标 GeoJSON 图斑和概率/配准指标；上传接口只接受一对 JPG/JPEG/PNG/TIF/TIFF，服务端固定调用 `.venvs/00-change-detection`。田墩实飞案例会显示
当前已接入 `00-change-detection`、`01-object-detection`、`02-semantic-mapping`、`04-spatial-measurement`、`09-anomaly-detection` 和 `15-trajectory-analysis`。变化检测案例显示两期原图、变化叠加、像素坐标 GeoJSON 图斑和概率/配准指标；上传接口只接受一对 JPG/JPEG/PNG/TIF/TIFF，服务端固定调用 `.venvs/00-change-detection`。异常检测分别上传正常参考影像和待检测影像，展示原图、方法对比、热力图、候选表和下载工件。田墩实飞案例会显示
实飞轨迹、计划航线、禁飞区与适飞区，并提供轨迹和区域范围的聚焦按钮。其他能力
保留目录和边界状态，待其首个 Demo 产出可检查工件后再接入。
## 本地运行工作台
@@ -69,5 +69,6 @@
- 语义分割上传最多 6 张 `JPG/JPEG/PNG/TIF/TIFF`，任务预设从能力目录动态读取；当前只有通用颜色流程可运行，排洪沟、边坡、竖井和尾矿库任务在模型与真值齐备前保持禁用。结果展示原图、栅格叠加、GeoJSON 矢量预览、类别比例和可下载工件。
- 空间测量上传最多 4 份 `PNG/TIF/TIFF` 单波段标签栅格，展示彩色栅格、GeoAI 矢量对象、计数、面积、周长和对象明细；只有有效投影 CRS 才使用地图单位。
- 变化检测上传一对 `JPG/JPEG/PNG/TIF/TIFF`，每次运行可选择自动识别、普通图片或 GeoTIFF 地理参考模式；阈值支持 `0.01～0.99`（默认 `0.5`），分辨率支持 GeoTIFF 原始分辨率/自动以及 `1024/1536/2048/3072` 档位。GeoTIFF 模式保留 CRS 和地图坐标，普通图片模式输出像素坐标。单文件最大 `1024 MB`，单次 Base64 JSON 请求最大 `3072 MB`。
- 异常检测分别上传 1～6 张正常参考影像和 1～6 张待检测影像，支持 `JPG/JPEG/PNG/TIF/TIFF`。3 张以上固定机位且全部尺寸一致时自动启用同位置规则，否则回退为全局规则；结果页显示实际选择的模式。大文件使用二进制流，后台任务完成后进入案例库；结果展示参考输入、待检测原图、规则/Isolation Forest 对比、热力图、候选区字段与下载工件。
- 本地 API 还包括 `GET/POST /api/change-detection/runs` 和 `PUT /api/change-detection/uploads/<upload-id>`；变化检测前端使用二进制流上传，服务端按原字节暂存并复制到独立 raw、processed 和 output 目录，不修改原始 TIFF，也不覆盖既有运行。
- 本地 API 为 `GET/POST /api/trajectory/runs`、`GET/POST /api/object-detection/runs`、`GET/POST /api/semantic-mapping/runs`、`GET/POST /api/spatial-measurement/runs` 和只读的 `GET /api/semantic-mapping/tasks`，仅监听回环地址，端口必须是 `6xxx`；服务只调用固定虚拟环境和能力脚本，不接受任意命令或任意路径。
- 本地 API 为 `GET/POST /api/trajectory/runs`、`GET/POST /api/object-detection/runs`、`GET/POST /api/semantic-mapping/runs`、`GET/POST /api/spatial-measurement/runs`、`GET/POST /api/anomaly-detection/runs` 和只读的 `GET /api/semantic-mapping/tasks`；异常检测另有二进制上传和后台任务状态端点。服务仅监听回环地址，端口必须是 `6xxx`，只调用固定虚拟环境和能力脚本，不接受任意命令或任意路径。

 apps/workbench-console/src/api/artifacts.ts

@@ -9,6 +9,10 @@
export interface MeasurementImage { file: string; width: number; height: number; raster_file: string; preview_file: string; vector_file: string; csv_file: string; object_count: number; class_counts: Record<string, number>; total_area: number; total_perimeter: number; area_unit: string; length_unit: string; measurement_basis: string; crs: string | null; georeferenced: boolean; vectorizer: string; elapsed_seconds: number; }
export interface MeasurementRun { capability: string; classification: string; created_at?: string; geoai_version: string; method: string; model: string; device: string; input_count: number; processed_images: number; elapsed_seconds: number; images: MeasurementImage[]; limitations: string[]; input_dir?: string; raw_input_dir?: string; }
export interface MeasurementCase { id: string; label: string; note: string; artifactRoot: string; createdAt: string; run: MeasurementRun; }
export interface AnomalyCandidate { type: "Feature"; properties: { feature_id?: number; mask_code?: number; method?: string; rule_flag?: boolean; isolation_flag?: boolean; agreement?: boolean; max_rule_score?: number; max_isolation_score?: number; reason_feature?: string; area_pixels?: number }; geometry: { type: string; coordinates: unknown }; }
export interface AnomalyImage { file: string; width: number; height: number; tile_count: number; rule_heatmap_file: string; isolation_heatmap_file: string; overlay_file: string; mask_file: string; vector_file: string; tiles_file: string; candidate_count: number; rule_anomaly_pixels: number; isolation_anomaly_pixels: number; agreement_pixels: number; rule_anomaly_coverage: number; isolation_anomaly_coverage: number; agreement_coverage: number; georeferenced: boolean; crs: string | null; coordinate_basis: string; vectorizer: string; elapsed_seconds: number; }
export interface AnomalyRun { capability: string; classification: string; created_at?: string; versions: Record<string, string>; method: string; model: string; device: string; reference_count: number; reference_tile_count: number; input_count: number; reference_images: Array<{ file: string; tile_count: number }>; feature_names: string[]; parameters: { tile_size: number; stride: number; threshold_quantile: number; random_state: number; spatial_mode_requested?: "auto" | "global" | "aligned"; rule_mode_selected?: "global" | "aligned" }; thresholds: { rule_score: number; isolation_score: number }; images: AnomalyImage[]; elapsed_seconds: number; limitations: string[]; }
export interface AnomalyCase { id: string; label: string; note: string; artifactRoot: string; inputRoot: string; referenceRoot: string; createdAt: string; run: AnomalyRun; candidates: Record<string, AnomalyCandidate[]>; }
export interface ChangeFeature { type: "Feature"; properties: { feature_id?: number; area_pixels?: number; mean_probability?: number; max_probability?: number; bounds_pixel?: number[]; confidence?: number; class?: number }; geometry: { type: string; coordinates: unknown }; }
export interface ChangeRun { schema_version: number; capability: string; classification: string; geoai_version: string; method: string; model: string; device: string; processing_mode?: "image" | "geotiff"; requested_processing_mode?: "auto" | "image" | "geotiff"; thresholds: Record<string, number>; tile_size: number; overlap: number; max_dimension?: number; effective_max_dimension?: number; input_files: string[]; input_shape: number[]; processed_shape: number[]; registration: { method: string; matches: number; inliers: number; inlier_ratio: number; valid_ratio: number }; valid_pixel_ratio: number; raw_changed_pixels: number; changed_pixels: number; changed_pixel_ratio: number; vector_feature_count: number; georeferenced: boolean; coordinate_basis: string; crs: string | null; elapsed_seconds: number; limitations: string[]; artifacts: { probability_raster: string; raw_mask_raster: string; mask_raster: string; overlay: string; vector: string; features: string }; created_at?: string; }
export interface ChangeCase { id: string; label: string; note: string; artifactRoot: string; beforeImage: string; afterImage: string; createdAt: string; run: ChangeRun; features: ChangeFeature[]; }
@@ -22,8 +26,11 @@
interface SemanticDefinition extends CaseDefinition { inputRoot: string; rawInputRoot: string; }
interface MeasurementDefinition extends CaseDefinition {}
interface ChangeDefinition extends CaseDefinition { beforeImage: string; afterImage: string; }
interface AnomalyDefinition extends CaseDefinition { inputRoot: string; referenceRoot: string; }
export interface UploadFilePayload { name: string; content: string; }
export interface ChangeUploadRef { uploadId: string; role: "before" | "after"; name: string; size: number; }
export interface AnomalyUploadRef { uploadId: string; role: "reference" | "input"; name: string; size: number; sha256: string; }
export interface AnomalyJob { id: string; runId: string; status: "queued" | "running" | "complete" | "failed"; createdAt: string; finishedAt?: string; error?: string; run?: AnomalyDefinition; }

export const artifactUrl = (path: string) => `/${path.replace(/\\/g, "/").split("/").map(encodeURIComponent).join("/")}`;

@@ -80,6 +87,19 @@
  return Object.fromEntries(cases.map((item) => [item.id, item]));
}

export async function loadAnomalyArtifacts(): Promise<Record<string, AnomalyCase>> {
  const { runs } = await getJson<{ runs: AnomalyDefinition[] }>("api/anomaly-detection/runs");
  const cases = await Promise.all(runs.map(async (definition) => {
    const run = await getJson<AnomalyRun>(`${definition.artifactRoot}/run_metadata.json`);
    const entries = await Promise.all(run.images.map(async (image) => {
      const payload = await getJson<{ features?: AnomalyCandidate[] }>(`${definition.artifactRoot}/${image.vector_file}`);
      return [image.file, payload.features ?? []] as const;
    }));
    return { ...definition, run, candidates: Object.fromEntries(entries) } satisfies AnomalyCase;
  }));
  return Object.fromEntries(cases.map((item) => [item.id, item]));
}

async function postRun<T>(path: string, body: object): Promise<T> { return responseJson<T>(await fetch(artifactUrl(path), { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(body) })); }
export async function createTrajectoryRun(files: { flight: UploadFilePayload; route: UploadFilePayload; restricted: UploadFilePayload; flyable?: UploadFilePayload }) { return postRun<{ run: TrajectoryDefinition }>("api/trajectory/runs", { files }); }
export async function createDetectionRun(images: UploadFilePayload[]) { return postRun<{ run: DetectionDefinition }>("api/object-detection/runs", { images }); }
@@ -89,8 +109,16 @@
export async function uploadChangeFile(file: File, role: "before" | "after"): Promise<ChangeUploadRef> {
  const uploadId = globalThis.crypto.randomUUID().replaceAll("-", "");
  const query = new URLSearchParams({ role });
  const response = await fetch(`/api/change-detection/uploads/${uploadId}?${query.toString()}`, { method: "PUT", headers: { "Content-Type": "application/octet-stream", "X-Upload-Name": file.name }, body: file });
  const response = await fetch(`/api/change-detection/uploads/${uploadId}?${query.toString()}`, { method: "PUT", headers: { "Content-Type": "application/octet-stream", "X-Upload-Name": encodeURIComponent(file.name) }, body: file });
  return responseJson<ChangeUploadRef>(response);
}
export async function createChangeRunFromUploads(uploads: { before: ChangeUploadRef; after: ChangeUploadRef }, threshold = 0.5, maxDimension = 0, processingMode: "auto" | "image" | "geotiff" = "auto") { return postRun<{ run: ChangeDefinition }>("api/change-detection/runs", { uploads, threshold, maxDimension, processingMode }); }
export async function uploadAnomalyFile(file: File, role: "reference" | "input"): Promise<AnomalyUploadRef> {
  const uploadId = globalThis.crypto.randomUUID().replaceAll("-", "");
  const query = new URLSearchParams({ role });
  const response = await fetch(`/api/anomaly-detection/uploads/${uploadId}?${query.toString()}`, { method: "PUT", headers: { "Content-Type": "application/octet-stream", "X-Upload-Name": encodeURIComponent(file.name) }, body: file });
  return responseJson<AnomalyUploadRef>(response);
}
export async function createAnomalyRun(uploads: { reference: AnomalyUploadRef[]; input: AnomalyUploadRef[] }, parameters: { tileSize: number; stride: number; thresholdQuantile: number; randomState: number }) { return postRun<{ job: AnomalyJob }>("api/anomaly-detection/runs", { uploads, ...parameters }); }
export async function loadAnomalyJob(jobId: string) { return (await getJson<{ job: AnomalyJob }>(`api/anomaly-detection/jobs/${jobId}`)).job; }
export function readFileAsPayload(file: File): Promise<UploadFilePayload> { return new Promise((resolve, reject) => { const reader = new FileReader(); reader.onerror = () => reject(new Error(`无法读取 ${file.name}`)); reader.onload = () => { const value = String(reader.result ?? ""); resolve({ name: file.name, content: value.slice(value.indexOf(",") + 1) }); }; reader.readAsDataURL(file); }); }

 apps/workbench-console/src/components/AnomalyDetectionPanel.vue

New file
@@ -0,0 +1,163 @@
<script setup lang="ts">
import { computed, onMounted, ref } from "vue";
import { DownloadOutlined, FileImageOutlined, FileOutlined, PlayCircleOutlined, UploadOutlined } from "@ant-design/icons-vue";

import { artifactUrl, createAnomalyRun, loadAnomalyJob, uploadAnomalyFile, type AnomalyCase } from "@/api/artifacts";
import ArtifactState from "@/components/ArtifactState.vue";
import { useArtifactStore } from "@/stores/artifacts";

const store = useArtifactStore();
const caseId = ref("");
const selectedName = ref("");
const referenceFiles = ref<File[]>([]);
const inputFiles = ref<File[]>([]);
const tileSize = ref(256);
const stride = ref(128);
const thresholdQuantile = ref(0.995);
const randomState = ref(42);
const showRunForm = ref(false);
const running = ref(false);
const runStage = ref("");
const runError = ref<string | null>(null);
const searchText = ref("");

const currentCase = computed<AnomalyCase | undefined>(() => store.anomalyCases[caseId.value] ?? Object.values(store.anomalyCases)[0]);
const selectedImage = computed(() => currentCase.value?.run.images.find((item) => item.file === selectedName.value) ?? currentCase.value?.run.images[0]);
const candidates = computed(() => selectedImage.value && currentCase.value ? currentCase.value.candidates[selectedImage.value.file] ?? [] : []);
const candidateRows = computed(() => candidates.value.map((item) => item.properties));
const caseOptions = computed(() => Object.values(store.anomalyCases).map((item) => ({ value: item.id, label: item.label })));
const filteredCaseOptions = computed(() => caseOptions.value.filter((item) => item.label.toLowerCase().includes(searchText.value.trim().toLowerCase())));
const methodLabel: Record<string, string> = { rule_only: "仅规则", isolation_only: "仅 Isolation Forest", agreement: "两方法一致" };
const featureLabel: Record<string, string> = {
  rgb_mean_r: "红色均值", rgb_mean_g: "绿色均值", rgb_mean_b: "蓝色均值",
  rgb_std_r: "红色波动", rgb_std_g: "绿色波动", rgb_std_b: "蓝色波动",
  hsv_mean_h: "色相均值", hsv_mean_s: "饱和度均值", hsv_mean_v: "亮度均值",
  hsv_std_h: "色相波动", hsv_std_s: "饱和度波动", hsv_std_v: "亮度波动",
  gray_entropy: "灰度熵", edge_density: "边缘密度", laplacian_variance: "清晰度", dark_ratio: "暗像素比例", bright_ratio: "亮像素比例"
};

function addFile(target: "reference" | "input", file: File) {
  const list = target === "reference" ? referenceFiles : inputFiles;
  if (!list.value.some((item) => item.name.toLowerCase() === file.name.toLowerCase())) list.value = [...list.value, file];
  return false;
}
function removeFile(target: "reference" | "input", name: string) {
  const list = target === "reference" ? referenceFiles : inputFiles;
  list.value = list.value.filter((item) => item.name !== name);
}
function syncSelection() { selectedName.value = currentCase.value?.run.images[0]?.file ?? ""; }
const delay = (milliseconds: number) => new Promise((resolve) => globalThis.setTimeout(resolve, milliseconds));

async function submitRun() {
  if (!referenceFiles.value.length || !inputFiles.value.length) { runError.value = "请至少选择一张正常参考影像和一张待检测影像。"; return; }
  if (stride.value > tileSize.value) { runError.value = "步长不能大于窗口尺寸。"; return; }
  running.value = true; runError.value = null;
  try {
    runStage.value = "正在上传原始影像…";
    const references = [];
    const inputs = [];
    for (const file of referenceFiles.value) references.push(await uploadAnomalyFile(file, "reference"));
    for (const file of inputFiles.value) inputs.push(await uploadAnomalyFile(file, "input"));
    runStage.value = "已提交，正在本机 CPU 后台检测…";
    const { job } = await createAnomalyRun(
      { reference: references, input: inputs },
      { tileSize: tileSize.value, stride: stride.value, thresholdQuantile: thresholdQuantile.value, randomState: randomState.value }
    );
    let state = job;
    while (state.status === "queued" || state.status === "running") { await delay(1200); state = await loadAnomalyJob(job.id); }
    if (state.status === "failed") throw new Error(state.error || "异常检测后台任务失败");
    runStage.value = "正在加载结果…";
    await store.loadAnomaly(true);
    caseId.value = state.runId;
    syncSelection();
    referenceFiles.value = []; inputFiles.value = []; showRunForm.value = false;
  } catch (error) { runError.value = error instanceof Error ? error.message : "异常检测运行失败"; }
  finally { running.value = false; runStage.value = ""; }
}

onMounted(async () => { await store.loadAnomaly(); caseId.value = Object.keys(store.anomalyCases)[0] ?? ""; syncSelection(); });
</script>

<template>
  <ArtifactState :loading="store.loading" :error="store.error" />
  <section class="workspace-command anomaly-command">
    <div><h2>新建异常检测运行</h2><p>固定机位且尺寸一致时自动进行同位置比较，再与 Isolation Forest 结果对照。</p></div>
    <a-button type="primary" @click="showRunForm = !showRunForm"><PlayCircleOutlined />{{ showRunForm ? "收起运行表单" : "上传并运行" }}</a-button>
  </section>

  <section v-if="showRunForm" class="surface-section anomaly-run-form">
    <a-alert type="warning" show-icon message="建议上传 3～6 张固定机位、相同尺寸的正常图；系统会自动启用同位置规则。机位或尺寸不一致时回退为全局比较；候选不是业务告警。" />
    <div class="anomaly-upload-grid">
      <div class="upload-group">
        <div><strong>正常参考影像</strong><span>1～6 张；合计至少产生 50 个窗口</span></div>
        <a-upload multiple accept=".jpg,.jpeg,.png,.tif,.tiff" :show-upload-list="false" :before-upload="(file: File) => addFile('reference', file)"><a-button><UploadOutlined />选择参考影像</a-button></a-upload>
        <div class="file-tags"><a-tag v-for="file in referenceFiles" :key="file.name" closable @close="removeFile('reference', file.name)">{{ file.name }}</a-tag><span v-if="!referenceFiles.length">未选择</span></div>
      </div>
      <div class="upload-group">
        <div><strong>待检测影像</strong><span>1～6 张；每张分别生成结果</span></div>
        <a-upload multiple accept=".jpg,.jpeg,.png,.tif,.tiff" :show-upload-list="false" :before-upload="(file: File) => addFile('input', file)"><a-button><UploadOutlined />选择待检测影像</a-button></a-upload>
        <div class="file-tags"><a-tag v-for="file in inputFiles" :key="file.name" closable @close="removeFile('input', file.name)">{{ file.name }}</a-tag><span v-if="!inputFiles.length">未选择</span></div>
      </div>
    </div>
    <div class="anomaly-parameters">
      <label>窗口尺寸<a-input-number v-model:value="tileSize" :min="128" :max="1024" :step="64" /></label>
      <label>步长<a-input-number v-model:value="stride" :min="32" :max="tileSize" :step="32" /></label>
      <label>阈值分位数<a-input-number v-model:value="thresholdQuantile" :min="0.9" :max="0.9999" :step="0.001" :precision="4" /></label>
      <label>随机种子<a-input-number v-model:value="randomState" :min="0" :max="2147483647" /></label>
    </div>
    <a-alert v-if="runError" type="error" show-icon :message="runError" />
    <a-button type="primary" :loading="running" :disabled="!referenceFiles.length || !inputFiles.length" @click="submitRun"><PlayCircleOutlined />开始检测</a-button>
    <span v-if="running" class="run-stage">{{ runStage }}</span>
  </section>

  <template v-if="currentCase && selectedImage">
    <a-row :gutter="[18, 18]" class="anomaly-workspace">
      <a-col :xs="24" :xl="5"><section class="surface-section run-library"><h2>案例库</h2><a-input-search v-model:value="searchText" placeholder="搜索运行" allow-clear /><a-list size="small" :data-source="filteredCaseOptions"><template #renderItem="{ item }"><a-list-item class="run-item" :class="{ active: item.value === currentCase.id }" @click="caseId = item.value; syncSelection()">{{ item.label }}</a-list-item></template></a-list></section></a-col>
      <a-col :xs="24" :xl="19"><section class="surface-section"><div class="section-toolbar"><a-select v-model:value="selectedName" :options="currentCase.run.images.map((item) => ({ value: item.file, label: item.file }))" /><span>{{ selectedImage.width }} × {{ selectedImage.height }} · {{ selectedImage.tile_count }} 窗口</span></div><div class="comparison-grid anomaly-primary"><figure><figcaption>待检测原图</figcaption><a-image :src="artifactUrl(`${currentCase.inputRoot}/${selectedImage.file}`)" /></figure><figure><figcaption>两方法对比结果</figcaption><a-image :src="artifactUrl(`${currentCase.artifactRoot}/${selectedImage.overlay_file}`)" /></figure></div><div class="anomaly-legend"><span class="rule">仅规则</span><span class="isolation">仅 Isolation Forest</span><span class="agreement">两方法一致</span></div></section></a-col>
    </a-row>

    <section class="surface-section result-band"><div class="section-heading"><div><h2>正常参考输入</h2><p>模型只从这些影像的窗口学习“正常”视觉分布。</p></div><a-tag>{{ currentCase.run.reference_tile_count }} 个参考窗口</a-tag></div><div class="reference-grid"><figure v-for="reference in currentCase.run.reference_images" :key="reference.file"><a-image :src="artifactUrl(`${currentCase.referenceRoot}/${reference.file}`)" /><figcaption>{{ reference.file }} · {{ reference.tile_count }} 窗口</figcaption></figure></div></section>

    <a-row :gutter="[18, 18]" class="metric-row anomaly-metrics"><a-col :xs="12" :lg="6"><a-statistic title="候选区" :value="selectedImage.candidate_count" /></a-col><a-col :xs="12" :lg="6"><a-statistic title="规则覆盖率" :value="(selectedImage.rule_anomaly_coverage * 100).toFixed(2)" suffix="%" /></a-col><a-col :xs="12" :lg="6"><a-statistic title="Isolation 覆盖率" :value="(selectedImage.isolation_anomaly_coverage * 100).toFixed(2)" suffix="%" /></a-col><a-col :xs="12" :lg="6"><a-statistic title="单图耗时" :value="selectedImage.elapsed_seconds" suffix="秒" /></a-col></a-row>

    <a-row :gutter="[18, 18]" class="result-band anomaly-secondary"><a-col :xs="24" :xl="12"><section class="surface-section"><h2>规则异常热力图</h2><a-image :src="artifactUrl(`${currentCase.artifactRoot}/${selectedImage.rule_heatmap_file}`)" /></section></a-col><a-col :xs="24" :xl="12"><section class="surface-section"><h2>Isolation Forest 热力图</h2><a-image :src="artifactUrl(`${currentCase.artifactRoot}/${selectedImage.isolation_heatmap_file}`)" /></section></a-col></a-row>

    <section class="surface-section result-band"><div class="section-heading"><div><h2>候选区明细</h2><p>原因特征表示规则评分贡献最大项，不是业务原因诊断。</p></div><a-tag>{{ selectedImage.georeferenced ? selectedImage.crs : "像素坐标" }}</a-tag></div><a-table :data-source="candidateRows" :pagination="{ pageSize: 8 }" row-key="feature_id" size="small" :scroll="{ x: 760 }"><a-table-column title="编号" data-index="feature_id" key="feature_id" /><a-table-column title="方法" data-index="method" key="method"><template #default="{ text }">{{ methodLabel[text] || text }}</template></a-table-column><a-table-column title="主要特征" data-index="reason_feature" key="reason_feature"><template #default="{ text }">{{ featureLabel[text] || text || '—' }}</template></a-table-column><a-table-column title="规则分数" data-index="max_rule_score" key="max_rule_score" align="right" /><a-table-column title="Isolation 分数" data-index="max_isolation_score" key="max_isolation_score" align="right" /><a-table-column title="面积 (px)" data-index="area_pixels" key="area_pixels" align="right" /></a-table></section>

    <section class="surface-section result-band anomaly-files"><div class="section-heading"><div><h2>运行与下载</h2><p>{{ currentCase.note }}</p></div></div><a-descriptions size="small" :column="{ xs: 1, sm: 2, lg: 6 }"><a-descriptions-item label="分类边界">{{ currentCase.run.classification }}</a-descriptions-item><a-descriptions-item label="规则模式">{{ currentCase.run.parameters.rule_mode_selected === 'aligned' ? '同位置' : '全局' }}</a-descriptions-item><a-descriptions-item label="设备">{{ currentCase.run.device }}</a-descriptions-item><a-descriptions-item label="参考影像">{{ currentCase.run.reference_count }}</a-descriptions-item><a-descriptions-item label="参考窗口">{{ currentCase.run.reference_tile_count }}</a-descriptions-item><a-descriptions-item label="总耗时">{{ currentCase.run.elapsed_seconds }} 秒</a-descriptions-item></a-descriptions><a-space wrap><a-button :href="artifactUrl(`${currentCase.artifactRoot}/${selectedImage.mask_file}`)" download><FileImageOutlined />编码掩膜 GeoTIFF</a-button><a-button :href="artifactUrl(`${currentCase.artifactRoot}/${selectedImage.vector_file}`)" download><FileOutlined />候选 GeoJSON</a-button><a-button :href="artifactUrl(`${currentCase.artifactRoot}/${selectedImage.tiles_file}`)" download><FileOutlined />窗口特征 CSV</a-button><a-button :href="artifactUrl(`${currentCase.artifactRoot}/run_metadata.json`)" download><DownloadOutlined />运行元数据</a-button></a-space></section>
    <a-alert class="anomaly-limit" type="warning" show-icon message="结果表示相对正常参考影像的视觉离群，只是人工复核候选，不能直接形成堵塞、损坏、渗漏或生产告警结论。" />
  </template>
  <a-empty v-else-if="!store.loading" description="尚无异常检测案例"><template #description><p>点击“上传并运行”，选择正常参考影像和待检测影像。</p></template></a-empty>
</template>

<style scoped>
.anomaly-run-form { display: grid; gap: 16px; margin-bottom: 24px; }
.anomaly-upload-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 16px; }
.upload-group { display: grid; align-content: start; gap: 12px; padding: 16px; border: 1px solid var(--wb-border); background: #f7f9f7; }
.upload-group > div:first-child { display: grid; gap: 4px; }
.upload-group span, .run-stage { color: var(--wb-muted); font-size: 12px; }
.file-tags { min-height: 30px; }
.file-tags :deep(.ant-tag) { margin-bottom: 6px; }
.anomaly-parameters { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 12px; }
.anomaly-parameters label { display: grid; gap: 7px; color: var(--wb-muted); font-size: 12px; font-weight: 700; }
.anomaly-parameters :deep(.ant-input-number) { width: 100%; }
.anomaly-workspace, .result-band { margin-bottom: 24px; }
.anomaly-workspace > :deep(.ant-col), .anomaly-secondary > :deep(.ant-col) { display: flex; }
.anomaly-workspace .surface-section, .anomaly-secondary .surface-section { width: 100%; }
.anomaly-workspace .run-library { align-self: flex-start; max-height: clamp(430px, calc(100vh - 260px), 660px); }
.anomaly-primary :deep(.ant-image), .anomaly-primary :deep(img) { width: 100%; }
.anomaly-primary :deep(img) { height: 390px; object-fit: contain; background: #18221d; }
.anomaly-secondary :deep(.ant-image), .anomaly-secondary :deep(img) { width: 100%; }
.anomaly-secondary :deep(img) { height: 330px; object-fit: contain; background: #18221d; }
.reference-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(180px, 1fr)); gap: 12px; }
.reference-grid figure { min-width: 0; margin: 0; }
.reference-grid :deep(.ant-image), .reference-grid :deep(img) { width: 100%; }
.reference-grid :deep(img) { height: 150px; object-fit: cover; background: #18221d; }
.reference-grid figcaption { margin-top: 6px; overflow: hidden; color: var(--wb-muted); font-size: 12px; text-overflow: ellipsis; white-space: nowrap; }
.anomaly-legend { display: flex; flex-wrap: wrap; gap: 18px; margin-top: 12px; color: var(--wb-muted); font-size: 12px; }
.anomaly-legend span::before { display: inline-block; width: 10px; height: 10px; margin-right: 6px; content: ""; }
.anomaly-legend .rule::before { background: #ff9f1c; }.anomaly-legend .isolation::before { background: #2196f3; }.anomaly-legend .agreement::before { background: #e53935; }
.anomaly-limit { margin-bottom: 24px; }
@media (max-width: 1199px) { .anomaly-workspace > :deep(.ant-col), .anomaly-secondary > :deep(.ant-col) { display: block; } .anomaly-workspace .run-library { max-height: none; } .anomaly-parameters { grid-template-columns: repeat(2, minmax(0, 1fr)); } }
@media (max-width: 760px) { .anomaly-upload-grid, .anomaly-primary, .anomaly-parameters, .reference-grid { grid-template-columns: minmax(0, 1fr); } .section-heading { flex-wrap: wrap; } .anomaly-primary :deep(img), .anomaly-secondary :deep(img) { height: 270px; } }
</style>

 apps/workbench-console/src/data/capabilities.ts

@@ -18,7 +18,7 @@
  { id: "06-spatial-reasoning", title: "空间推理", level: "C", status: "planned", note: "等待首个本地 Demo。" },
  { id: "07-risk-rule-engine", title: "风险规则引擎", level: "C", status: "planned", note: "等待首个本地 Demo。" },
  { id: "08-spatiotemporal-forecasting", title: "时空预测", level: "B", status: "planned", note: "等待首个本地 Demo。" },
  { id: "09-anomaly-detection", title: "异常检测", level: "B", status: "planned", note: "等待首个本地 Demo。" },
  { id: "09-anomaly-detection", title: "异常检测", level: "B", status: "verified", note: "CPU 可解释特征与 Isolation Forest 对比，输出热力图、候选区和结构化结果。" },
  { id: "10-smart-route-planning", title: "智能航线规划", level: "C", status: "planned", note: "等待首个本地 Demo。" },
  { id: "11-fleet-scheduling", title: "集群调度", level: "C", status: "planned", note: "等待首个本地 Demo。" },
  { id: "12-quality-control-refly", title: "质量检查与补飞", level: "B", status: "planned", note: "等待首个本地 Demo。" },

 apps/workbench-console/src/stores/artifacts.ts

@@ -6,11 +6,13 @@
  loadSemanticArtifacts,
  loadMeasurementArtifacts,
  loadTrajectoryArtifacts,
  loadAnomalyArtifacts,
  type DetectionCase,
  type ChangeCase,
  type SemanticCase,
  type MeasurementCase,
  type TrajectoryCase
  type TrajectoryCase,
  type AnomalyCase
} from "@/api/artifacts";

interface ArtifactState {
@@ -19,6 +21,7 @@
  semanticCases: Record<string, SemanticCase>;
  measurementCases: Record<string, MeasurementCase>;
  changeCases: Record<string, ChangeCase>;
  anomalyCases: Record<string, AnomalyCase>;
  loading: boolean;
  error: string | null;
}
@@ -28,7 +31,7 @@
    detectionRun: (state) => Object.values(state.detectionCases)[0]?.run ?? null,
    detectionImages: (state) => Object.values(state.detectionCases)[0]?.images ?? []
  },
  state: (): ArtifactState => ({ detectionCases: {}, trajectoryCases: {}, semanticCases: {}, measurementCases: {}, changeCases: {}, loading: false, error: null }),
  state: (): ArtifactState => ({ detectionCases: {}, trajectoryCases: {}, semanticCases: {}, measurementCases: {}, changeCases: {}, anomalyCases: {}, loading: false, error: null }),
  actions: {
    async loadChange(force = false) {
      if (!force && Object.keys(this.changeCases).length) return;
@@ -64,6 +67,13 @@
      try { this.measurementCases = await loadMeasurementArtifacts(); }
      catch (error) { this.error = error instanceof Error ? error.message : "空间测量结果读取失败"; }
      finally { this.loading = false; }
    },
    async loadAnomaly(force = false) {
      if (!force && Object.keys(this.anomalyCases).length) return;
      this.loading = true; this.error = null;
      try { this.anomalyCases = await loadAnomalyArtifacts(); }
      catch (error) { this.error = error instanceof Error ? error.message : "异常检测结果读取失败"; }
      finally { this.loading = false; }
    }
  }
});

 apps/workbench-console/src/styles.css

@@ -156,6 +156,7 @@

@media (max-width: 991px) {
  .application-sider { position: static !important; height: auto; }
  .application-sider .ant-layout-sider-zero-width-trigger { position: fixed; right: 0; left: auto; }
  .navigation-menu { height: auto; max-height: 290px; }
  .sider-status { display: none; }
}

 apps/workbench-console/src/views/CapabilityView.vue

@@ -8,6 +8,7 @@
import SemanticMappingPanel from "@/components/SemanticMappingPanel.vue";
import SpatialMeasurementPanel from "@/components/SpatialMeasurementPanel.vue";
import ChangeDetectionPanel from "@/components/ChangeDetectionPanel.vue";
import AnomalyDetectionPanel from "@/components/AnomalyDetectionPanel.vue";
import { capabilityById } from "@/data/capabilities";

const route = useRoute();
@@ -22,6 +23,7 @@
    <TrajectoryAnalysisPanel v-else-if="capability.id === '15-trajectory-analysis'" />
    <SemanticMappingPanel v-else-if="capability.id === '02-semantic-mapping'" />
    <SpatialMeasurementPanel v-else-if="capability.id === '04-spatial-measurement'" />
    <AnomalyDetectionPanel v-else-if="capability.id === '09-anomaly-detection'" />
    <a-empty v-else description="等待第一个可验证 Demo"><template #description><h2>尚无本地结果工件</h2><p>该能力已登记在目录中，但尚未产生可视化结果、结构化输出或运行元数据。</p></template></a-empty>
  </div>
  <div v-else class="view-container"><a-result status="404" title="未找到能力页面" sub-title="请从左侧目录选择已登记的能力。" /></div>

 capabilities/09-anomaly-detection/README.md

@@ -1,6 +1,148 @@
# Anomaly Detection
# 09 Anomaly Detection

- 输入：影像特征、轨迹特征或业务指标。
- 输出：异常分数、异常清单和原因特征。
- 首个 Demo：规则基线与 Isolation Forest 对比。
## 状态与能力边界

- 状态：可运行 CPU CLI Demo，已于 2026-08-18 完成受控样本和困难真实样本验证。
- 分类：B 类能力。RGB/HSV 等可解释影像特征、稳健规则评分和 Isolation Forest 来自生态库；`geoai-py 0.42.x` 仅用于调用 `masks_to_vector` 将异常掩膜转成候选多边形。
- 目标：寻找与正常参考影像分布明显不同的视觉区域，供人工复核。
- 非目标：不能把候选直接解释成堵塞、损坏、渗漏或生产告警，也不代表真实矿山场景精度。

首版没有引入深度嵌入权重。`geoai-py` 当前的 Embeddings API 主要服务于预计算地球嵌入数据集，不是本 Demo 的本地通用异常模型。

## 输入约束

- 参考和待检测输入可以是单张影像或目录，支持 JPG、JPEG、PNG、TIF、TIFF。
- JPG/PNG 转为 RGB 并使用像素坐标；GeoTIFF 默认读取前三个波段，只有有效 CRS 和仿射变换才会写入空间参考。
- 默认窗口 `256 x 256`、步长 `128`，边缘窗口锚定到影像边界。
- 参考集至少需要 50 个窗口，影像尺寸不能小于窗口，输出目录必须不存在或为空。

## 方法

每个窗口提取 17 个特征：RGB/HSV 均值与标准差、灰度熵、边缘密度、Laplacian 清晰度以及暗/亮像素比例。

- 规则基线：`auto` 模式下，3 张以上同尺寸固定机位参考图会先消除每张图的整体亮度/色彩漂移，再将测试窗口只与参考图的相同位置比较；条件不满足时回退到全局中位数/MAD。两种模式都记录贡献最大的原因特征。
- Isolation Forest：先对参考特征拟合 `StandardScaler`，再训练 `IsolationForest(n_estimators=200, random_state=42)`。
- 两种方法分别用参考分数的默认 `99.5%` 分位数定阈值，不通过固定污染率强制标记正常窗口。
- 重叠窗口分数平均回像素网格，掩膜编码为 `0=正常、1=仅规则、2=仅 Isolation Forest、3=两者一致`。

## 环境与运行

环境固定在仓库内的 `.venvs/09-anomaly-detection`，使用 Python 3.12，不安装到系统 Python：

```powershell
.\scripts\setup.ps1 -Capability 09-anomaly-detection
.\.venvs\09-anomaly-detection\Scripts\python.exe -m pip check
```

运行 CLI：

```powershell
.\.venvs\09-anomaly-detection\Scripts\python.exe `
  .\capabilities\09-anomaly-detection\run_anomaly_detection.py `
  --reference <正常参考影像或目录> `
  --input <待检测影像或目录> `
  --output <新的空目录> `
  --tile-size 256 `
  --stride 128 `
  --threshold-quantile 0.995 `
  --random-state 42 `
  --spatial-mode auto `
  --truth-manifest <可选验证真值JSON>
```

生成可重复的验证输入（复制源图时校验 SHA256，绝不覆盖 `baseData` 原图）：

```powershell
.\.venvs\09-anomaly-detection\Scripts\python.exe `
  .\capabilities\09-anomaly-detection\generate_validation_inputs.py `
  --source-dir .\baseData `
  --raw-output .\shared\data\raw\09-anomaly-detection\validation-20260818-v2 `
  --processed-output .\shared\data\processed\09-anomaly-detection\validation-20260818-v2
```

## 输出接口

每张输入影像生成：

- `*.rule.heatmap.png` 和 `*.isolation.heatmap.png`
- `*.comparison.overlay.png`
- `*.anomaly-mask.tif`
- `*.anomalies.geojson`
- `*.tiles.csv`

运行目录还会生成 `run_metadata.json`，记录环境版本、阈值、特征、参考/输入数量、耗时、空间参考、逐图统计、真值指标、矢量化路径和限制。GeoJSON 包含方法标记、最大分数、主要原因特征和像素面积。

## 验证结果

验证输出位于 `shared/outputs/09-anomaly-detection/validation-20260818-v3/`。参考图产生 588 个窗口，共处理 3 张目标图，CPU 总耗时 41.051 秒。

| 样本 | 规则覆盖率 | Isolation Forest 覆盖率 | 单图耗时 |
| --- | ---: | ---: | ---: |
| 注入异常 `07-16-injected.png` | 7.16% | 3.74% | 14.741 秒 |
| 未修改正常图 `07-16-normal.jpg` | 0.33% | 0.00% | 9.211 秒 |
| 困难真实图 `07-19-difficult.jpg` | 1.46% | 0.56% | 10.341 秒 |

受控注入包含颜色块、局部模糊和暗色遮挡。规则方法命中 3/3，precision 0.583、recall 0.672、IoU 0.454；Isolation Forest 命中 2/3，precision 0.831、recall 0.500、IoU 0.454。这些指标只验证受控链路，不代表真实业务准确率。

人工查看结果图后确认：颜色块和暗色遮挡与候选明显对齐；模糊区域主要由规则方法检出。正常图在左上植被/岩体过渡处有少量规则误报；困难图的候选集中在混凝土结构、门洞和顶部暗区，可能来自视角、材质和光照差异，不能称为真实异常。

## 测试

```powershell
.\.venvs\09-anomaly-detection\Scripts\python.exe -m unittest discover `
  -s .\capabilities\09-anomaly-detection\tests -v
```

测试覆盖边缘切片、特征响应、随机种子复现、完整 CLI 输出、非空输出拒绝、参考窗口下限、目录输入、无支持格式，以及带投影三波段 GeoTIFF 的 CRS/仿射变换保留。

## 许可证与限制

| 项目 | 许可证/状态 |
| --- | --- |
| `geoai-py` | MIT |
| scikit-learn、SciPy、Rasterio、GeoPandas | BSD 系列；Rasterio 依赖的 GDAL 需独立核查 |
| OpenCV | Apache-2.0 |
| Pillow | HPND |
| 模型权重 | 无 |
| 验证影像 | 用户已有样本；本 Demo 不扩展其授权范围 |

当前只处理 RGB 可视影像，不支持热红外、SAR 或任意多光谱组合。下一步应先取得多场景正常参考图、真实异常人工标注和地理配准影像，测量跨视角/光照的误报与漏报，再决定是否接入控制台或产品告警。

## 本地实验控制台

控制台已接入输入、后台运行、案例库和结果工作区：

```text
http://127.0.0.1:6173/apps/workbench-console/#/capability/09-anomaly-detection
```

- 输入：分别多选 1～6 张正常参考影像和 1～6 张待检测影像，可调整窗口、步长、阈值分位数和随机种子。
- 上传：JPG/PNG/GeoTIFF 都通过受限二进制流传输，保存原始字节、大小和 SHA256；每次创建新的 raw、processed 和 output 运行目录。
- 运行：提交后进入本机后台任务，页面轮询状态，不维持长时间上传请求。
- 输出：展示参考输入、待检测原图、对比叠加、两张热力图、覆盖率、候选表格，并下载掩膜 GeoTIFF、GeoJSON、窗口 CSV 和元数据。
- API：`GET/POST /api/anomaly-detection/runs`、`PUT /api/anomaly-detection/uploads/<upload-id>`、`GET /api/anomaly-detection/jobs/<job-id>`。

2026-08-18 使用一张 3664 × 2748 参考图和一张困难影像完成控制台真实上传验证，后台任务成功产生 6 个候选区；原图与结果工件均通过本地 HTTP 读取。控制台仍只用于实验复核，不连接产品告警。

## 五参考图合成场景复核（2026-08-19）

使用内置 ImageGen 生成同一固定机位的排水沟场景：5 张正常参考图分别覆盖阴天、晴天、雨后、晨雾和傍晚，异常图仅在沟内加入树枝、落叶、泥土和石块形成的堵塞。生成信息位于 `shared/data/raw/09-anomaly-detection/validation-20260819-imagegen-v1/generation_metadata.json`。

- 5 张参考图产生 385 个参考窗口；默认参数运行耗时 5.861 秒。
- 多参考图结果为 0 个候选，规则法和 Isolation Forest 均漏掉堵塞。
- 只使用最接近的阴天基准图时产生 4 个候选；两方法一致区域覆盖堵塞位置，但规则覆盖率达到 41.67%，画面两侧和远端出现大面积误报。
- 原因：当前特征模型汇总所有位置的窗口分布，不理解“岩石在坡面正常、类似材质进入排水沟才异常”的空间语义；天气参考和生成编辑的全局纹理变化又扩大了正常范围。

该运行已登记到控制台案例库，名称为 `ImageGen 五参考图堵塞复核（漏检案例）`，保留 0 候选的真实输出以及五张参考图、测试原图、热力图和下载工件。

该结果否定了“多放几张正常图就能可靠识别排水沟堵塞”的假设。为此新增了固定机位同位置规则：`--spatial-mode auto` 在至少 3 张参考图且所有图像尺寸一致时自动启用，否则保留原全局规则；也可显式选择 `aligned` 或 `global`。Isolation Forest 仍作为全局外观基线保留。

优化后的新运行位于 `shared/outputs/09-anomaly-detection/validation-20260819-imagegen-aligned-v2/`，没有覆盖原漏检结果：

- 默认 `99.5%` 参考分位阈值为 13.6367，没有为命中样本降低阈值。
- 堵塞图产生 1 个规则候选，覆盖率 9.375%；人工查看确认候选覆盖沟内树枝、落叶、泥石，没有扩散到两侧山坡和道路。Isolation Forest 在该样本仍为 0 个候选。
- 将 5 张正常图逐张留出、其余 4 张作参考时，规则覆盖率依次为 7.292%、0%、0%、0%、0%，均不高于 10%。
- 新结果已登记为 `ImageGen 五参考图堵塞复核（同位置优化）`，与 `漏检案例` 并列用于前后对照。

同位置模式解决的是固定机位、相同构图中的局部画面变化，不具备“堵塞”语义。明显机位偏移仍会造成误报；真实部署还应增加图像配准、排水沟 ROI 或专门的检测/分割模型，并取得真实标注数据。


 capabilities/09-anomaly-detection/generate_validation_inputs.py

New file
@@ -0,0 +1,123 @@
"""Prepare unchanged real samples plus controlled anomaly validation inputs."""

from __future__ import annotations

import argparse
import hashlib
import json
import shutil
from pathlib import Path

import cv2
import numpy as np
from PIL import Image


def _require_empty(path: Path) -> None:
    if path.exists() and any(path.iterdir()):
        raise ValueError(f"Output directory is not empty: {path}")
    path.mkdir(parents=True, exist_ok=True)


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _bbox(width: int, height: int, fractions: tuple[float, float, float, float]) -> list[int]:
    x0, y0, x1, y1 = fractions
    return [int(width * x0), int(height * y0), int(width * x1), int(height * y1)]


def generate(normal_source: Path, difficult_source: Path, raw_output: Path, processed_output: Path) -> dict:
    _require_empty(raw_output)
    _require_empty(processed_output)
    reference_dir = raw_output / "reference"
    raw_inputs = raw_output / "inputs"
    processed_inputs = processed_output / "inputs"
    truth_dir = processed_output / "truth"
    for directory in (reference_dir, raw_inputs, processed_inputs, truth_dir):
        directory.mkdir(parents=True, exist_ok=True)

    reference_path = reference_dir / "07-16-reference.jpg"
    normal_path = raw_inputs / "07-16-normal.jpg"
    difficult_path = raw_inputs / "07-19-difficult.jpg"
    shutil.copy2(normal_source, reference_path)
    shutil.copy2(normal_source, normal_path)
    shutil.copy2(difficult_source, difficult_path)
    shutil.copy2(normal_path, processed_inputs / normal_path.name)
    shutil.copy2(difficult_path, processed_inputs / difficult_path.name)

    with Image.open(normal_source) as image:
        rgb = np.asarray(image.convert("RGB")).copy()
    height, width = rgb.shape[:2]
    regions = [
        {"name": "high_saturation_colour", "bbox": _bbox(width, height, (0.10, 0.16, 0.21, 0.29))},
        {"name": "local_blur", "bbox": _bbox(width, height, (0.43, 0.12, 0.58, 0.30))},
        {"name": "dark_occlusion", "bbox": _bbox(width, height, (0.68, 0.56, 0.81, 0.72))},
    ]
    truth = np.zeros((height, width), dtype=np.uint8)
    x0, y0, x1, y1 = regions[0]["bbox"]
    rgb[y0:y1, x0:x1] = np.array([230, 20, 210], dtype=np.uint8)
    truth[y0:y1, x0:x1] = 255
    x0, y0, x1, y1 = regions[1]["bbox"]
    rgb[y0:y1, x0:x1] = cv2.GaussianBlur(rgb[y0:y1, x0:x1], (151, 151), 0)
    truth[y0:y1, x0:x1] = 255
    x0, y0, x1, y1 = regions[2]["bbox"]
    rgb[y0:y1, x0:x1] = np.array([8, 8, 8], dtype=np.uint8)
    truth[y0:y1, x0:x1] = 255

    injected_name = "07-16-injected.png"
    Image.fromarray(rgb).save(processed_inputs / injected_name)
    truth_name = "07-16-injected.truth.png"
    Image.fromarray(truth).save(truth_dir / truth_name)
    manifest = {
        "images": {
            injected_name: {
                "mask": f"truth/{truth_name}",
                "regions": regions,
            }
        }
    }
    manifest_path = processed_output / "truth-manifest.json"
    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    metadata = {
        "normal_source": normal_source.as_posix(),
        "difficult_source": difficult_source.as_posix(),
        "normal_source_sha256": _sha256(normal_source),
        "reference_copy_sha256": _sha256(reference_path),
        "difficult_source_sha256": _sha256(difficult_source),
        "difficult_copy_sha256": _sha256(difficult_path),
        "width": width,
        "height": height,
        "injected_regions": regions,
        "truth_manifest": manifest_path.as_posix(),
        "limitations": ["注入异常仅用于验证检测链路，不代表真实业务异常。"],
    }
    (processed_output / "generation_metadata.json").write_text(
        json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    return metadata


def main() -> int:
    parser = argparse.ArgumentParser(description="Generate anomaly-detection validation inputs.")
    parser.add_argument("--normal-source", type=Path, required=True)
    parser.add_argument("--difficult-source", type=Path, required=True)
    parser.add_argument("--raw-output", type=Path, required=True)
    parser.add_argument("--processed-output", type=Path, required=True)
    args = parser.parse_args()
    try:
        result = generate(args.normal_source, args.difficult_source, args.raw_output, args.processed_output)
    except (OSError, ValueError) as error:
        print(f"ERROR: {error}", file=__import__("sys").stderr)
        return 2
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

 capabilities/09-anomaly-detection/requirements.txt

@@ -1,4 +1,8 @@
-r ../../requirements/base.txt
scikit-learn>=1.5,<2
scipy>=1.13,<2
# GeoAI vectorization and georeferenced raster outputs.
geoai-py>=0.42,<0.43
rasterio>=1.4,<2
geopandas>=1,<2


 capabilities/09-anomaly-detection/run_anomaly_detection.py

New file
@@ -0,0 +1,835 @@
"""Run an explainable CPU image-region anomaly baseline.

The anomaly scores come from lightweight image statistics and Isolation Forest.
GeoAI is used for anomaly-mask vectorization, so this remains a B capability:
GeoAI supplies the geospatial output workflow while ecosystem code supplies the
anomaly model.  A high score is only a review candidate, not a business event.
"""

from __future__ import annotations

import argparse
import json
import math
import time
from dataclasses import dataclass
from datetime import UTC, datetime
from importlib import metadata as importlib_metadata
from pathlib import Path
from typing import Any, Iterable

import cv2
import geopandas as gpd
import numpy as np
import pandas as pd
import rasterio
from PIL import Image
from rasterio.features import geometry_mask, shapes
from rasterio.transform import Affine
from shapely.geometry import Point, box, mapping, shape
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler


SUPPORTED_SUFFIXES = {".jpg", ".jpeg", ".png", ".tif", ".tiff"}
MASK_LABELS = {1: "rule_only", 2: "isolation_only", 3: "agreement"}
MASK_COLORS = {
    0: np.array([0, 0, 0], dtype=np.uint8),
    1: np.array([255, 158, 44], dtype=np.uint8),
    2: np.array([47, 128, 237], dtype=np.uint8),
    3: np.array([224, 49, 49], dtype=np.uint8),
}
FEATURE_NAMES = [
    "rgb_mean_r",
    "rgb_mean_g",
    "rgb_mean_b",
    "rgb_std_r",
    "rgb_std_g",
    "rgb_std_b",
    "hsv_mean_h",
    "hsv_mean_s",
    "hsv_mean_v",
    "hsv_std_h",
    "hsv_std_s",
    "hsv_std_v",
    "gray_entropy",
    "edge_density",
    "laplacian_variance",
    "dark_ratio",
    "bright_ratio",
]


@dataclass(frozen=True)
class RasterInfo:
    transform: Affine
    crs: str | None
    georeferenced: bool
    source_bands: list[int]
    normalization: str


def axis_positions(length: int, tile_size: int, stride: int) -> list[int]:
    """Return deterministic starts and always anchor the final tile to the edge."""
    if length < tile_size:
        raise ValueError(f"Image dimension {length} is smaller than tile size {tile_size}.")
    positions = list(range(0, length - tile_size + 1, stride))
    last = length - tile_size
    if positions[-1] != last:
        positions.append(last)
    return positions


def tile_windows(width: int, height: int, tile_size: int, stride: int) -> list[tuple[int, int, int, int]]:
    if tile_size <= 0 or stride <= 0 or stride > tile_size:
        raise ValueError("tile_size must be positive and stride must be in 1..tile_size.")
    return [
        (x, y, x + tile_size, y + tile_size)
        for y in axis_positions(height, tile_size, stride)
        for x in axis_positions(width, tile_size, stride)
    ]


def _normalize_band(band: np.ndarray) -> np.ndarray:
    values = band.astype(np.float32)
    finite = values[np.isfinite(values)]
    if finite.size == 0:
        return np.zeros(band.shape, dtype=np.uint8)
    low, high = np.percentile(finite, [2.0, 98.0])
    if high <= low:
        high = low + 1.0
    return np.clip((values - low) * 255.0 / (high - low), 0, 255).astype(np.uint8)


def read_rgb(path: Path) -> tuple[np.ndarray, RasterInfo]:
    """Read common imagery and retain spatial metadata only when it is real."""
    if path.suffix.lower() in {".tif", ".tiff"}:
        with rasterio.open(path) as src:
            if src.count < 3:
                raise ValueError(f"GeoTIFF requires at least three bands: {path.name}")
            raw = src.read([1, 2, 3])
            if raw.dtype == np.uint8:
                rgb = np.moveaxis(raw, 0, 2)
                normalization = "uint8_identity"
            else:
                rgb = np.stack([_normalize_band(raw[index]) for index in range(3)], axis=2)
                normalization = "per_band_percentile_2_98"
            transform = src.transform
            crs = src.crs.to_string() if src.crs else None
            georeferenced = bool(src.crs and src.transform != Affine.identity())
            return rgb, RasterInfo(transform, crs, georeferenced, [1, 2, 3], normalization)
    with Image.open(path) as image:
        rgb = np.asarray(image.convert("RGB"))
    return rgb, RasterInfo(Affine.identity(), None, False, [1, 2, 3], "pillow_rgb")


def extract_features(tile: np.ndarray) -> np.ndarray:
    """Extract explainable colour, texture, edge, focus, and exposure features."""
    if tile.ndim != 3 or tile.shape[2] != 3:
        raise ValueError("Expected an RGB tile with shape (height, width, 3).")
    rgb = tile.astype(np.float32)
    hsv = cv2.cvtColor(tile, cv2.COLOR_RGB2HSV).astype(np.float32)
    gray = cv2.cvtColor(tile, cv2.COLOR_RGB2GRAY)
    histogram = cv2.calcHist([gray], [0], None, [32], [0, 256]).ravel().astype(np.float64)
    probabilities = histogram / max(float(histogram.sum()), 1.0)
    probabilities = probabilities[probabilities > 0]
    entropy = float(-(probabilities * np.log2(probabilities)).sum())
    edges = cv2.Canny(gray, 80, 160)
    laplacian_variance = float(cv2.Laplacian(gray, cv2.CV_64F).var())
    values = [
        *rgb.mean(axis=(0, 1)).tolist(),
        *rgb.std(axis=(0, 1)).tolist(),
        *hsv.mean(axis=(0, 1)).tolist(),
        *hsv.std(axis=(0, 1)).tolist(),
        entropy,
        float(np.mean(edges > 0)),
        laplacian_variance,
        float(np.mean(gray < 32)),
        float(np.mean(gray > 223)),
    ]
    return np.asarray(values, dtype=np.float64)


def image_features(
    rgb: np.ndarray, tile_size: int, stride: int
) -> tuple[np.ndarray, list[tuple[int, int, int, int]]]:
    height, width = rgb.shape[:2]
    windows = tile_windows(width, height, tile_size, stride)
    features = np.vstack([extract_features(rgb[y0:y1, x0:x1]) for x0, y0, x1, y1 in windows])
    return features, windows


def robust_rule_scores(
    features: np.ndarray,
    center: np.ndarray,
    scale: np.ndarray,
    feature_cutoffs: np.ndarray | None = None,
) -> tuple[np.ndarray, np.ndarray]:
    z_scores = np.abs((features - center) / scale)
    normalized = z_scores if feature_cutoffs is None else z_scores / feature_cutoffs
    return normalized.max(axis=1), normalized.argmax(axis=1)


def _spatial_profile(features: np.ndarray) -> np.ndarray:
    """Remove image-wide colour/lighting shifts while retaining local spatial patterns."""
    center = np.median(features, axis=0)
    mad = 1.4826 * np.median(np.abs(features - center), axis=0)
    fallback = features.std(axis=0)
    scale = np.where(mad > 1e-9, mad, np.where(fallback > 1e-9, fallback, 1.0))
    return (features - center) / scale


def aligned_rule_scores(
    features: np.ndarray, center: np.ndarray, scale: np.ndarray
) -> tuple[np.ndarray, np.ndarray]:
    """Compare every target tile only with the same tile position in the references."""
    profile = _spatial_profile(features)
    z_scores = np.abs((profile - center) / scale)
    top_count = min(3, z_scores.shape[1])
    scores = np.sort(z_scores, axis=1)[:, -top_count:].mean(axis=1)
    return scores, z_scores.argmax(axis=1)


def fit_aligned_rule_model(
    reference_feature_sets: list[np.ndarray], threshold_quantile: float
) -> dict[str, Any]:
    """Fit a fixed-camera rule model and calibrate it by leave-one-reference-out scores."""
    if len(reference_feature_sets) < 3:
        raise ValueError("Aligned spatial mode requires at least three reference images.")
    if len({values.shape for values in reference_feature_sets}) != 1:
        raise ValueError("Aligned spatial mode requires identical reference image dimensions.")
    profiles = np.stack([_spatial_profile(values) for values in reference_feature_sets])
    feature_floor = np.maximum(
        profiles.reshape(-1, profiles.shape[-1]).std(axis=0) * 0.15, 0.1
    )

    def spatial_stats(values: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
        center = np.median(values, axis=0)
        mad = 1.4826 * np.median(np.abs(values - center), axis=0)
        return center, np.maximum(mad, feature_floor)

    calibration_scores: list[np.ndarray] = []
    for index in range(profiles.shape[0]):
        center, scale = spatial_stats(np.delete(profiles, index, axis=0))
        z_scores = np.abs((profiles[index] - center) / scale)
        top_count = min(3, z_scores.shape[1])
        calibration_scores.append(np.sort(z_scores, axis=1)[:, -top_count:].mean(axis=1))
    center, scale = spatial_stats(profiles)
    reference_scores = np.concatenate(calibration_scores)
    return {
        "rule_mode": "aligned",
        "aligned_center": center,
        "aligned_scale": scale,
        "rule_threshold": float(np.quantile(reference_scores, threshold_quantile)),
        "reference_rule_scores": reference_scores,
    }


def fit_models(
    reference_features: np.ndarray,
    threshold_quantile: float,
    random_state: int,
    aligned_reference_features: list[np.ndarray] | None = None,
) -> dict[str, Any]:
    if reference_features.shape[0] < 50:
        raise ValueError(
            f"At least 50 reference tiles are required; found {reference_features.shape[0]}."
        )
    if not 0.9 <= threshold_quantile < 1.0:
        raise ValueError("threshold_quantile must be in [0.9, 1.0).")
    center = np.median(reference_features, axis=0)
    mad = np.median(np.abs(reference_features - center), axis=0)
    fallback = reference_features.std(axis=0)
    scale = np.where(mad > 1e-9, 1.4826 * mad, np.where(fallback > 1e-9, fallback, 1.0))
    reference_z = np.abs((reference_features - center) / scale)
    feature_cutoffs = np.quantile(reference_z, threshold_quantile, axis=0)
    feature_cutoffs = np.maximum(feature_cutoffs, 1.0)
    rule_scores, _ = robust_rule_scores(reference_features, center, scale, feature_cutoffs)
    scaler = StandardScaler().fit(reference_features)
    standardized = scaler.transform(reference_features)
    isolation = IsolationForest(
        n_estimators=200,
        contamination="auto",
        random_state=random_state,
        n_jobs=-1,
    ).fit(standardized)
    isolation_scores = -isolation.score_samples(standardized)
    result = {
        "rule_mode": "global",
        "center": center,
        "scale": scale,
        "feature_cutoffs": feature_cutoffs,
        "scaler": scaler,
        "isolation": isolation,
        "rule_threshold": float(np.quantile(rule_scores, threshold_quantile)),
        "isolation_threshold": float(np.quantile(isolation_scores, threshold_quantile)),
        "reference_rule_scores": rule_scores,
        "reference_isolation_scores": isolation_scores,
    }
    if aligned_reference_features is not None:
        result.update(fit_aligned_rule_model(aligned_reference_features, threshold_quantile))
    return result


def score_features(features: np.ndarray, models: dict[str, Any]) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    if models.get("rule_mode") == "aligned":
        if features.shape[0] != models["aligned_center"].shape[0]:
            raise ValueError(
                "Aligned spatial mode requires target images with the same dimensions as references."
            )
        rule_scores, reason_indices = aligned_rule_scores(
            features, models["aligned_center"], models["aligned_scale"]
        )
    else:
        rule_scores, reason_indices = robust_rule_scores(
            features, models["center"], models["scale"], models["feature_cutoffs"]
        )
    isolation_scores = -models["isolation"].score_samples(models["scaler"].transform(features))
    return rule_scores, isolation_scores, reason_indices


def scores_to_raster(
    scores: np.ndarray,
    windows: list[tuple[int, int, int, int]],
    shape_: tuple[int, int],
) -> np.ndarray:
    sums = np.zeros(shape_, dtype=np.float32)
    counts = np.zeros(shape_, dtype=np.uint16)
    for score, (x0, y0, x1, y1) in zip(scores, windows, strict=True):
        sums[y0:y1, x0:x1] += float(score)
        counts[y0:y1, x0:x1] += 1
    return sums / np.maximum(counts, 1)


def _heatmap(score_map: np.ndarray, threshold: float) -> np.ndarray:
    scaled = np.clip(score_map / max(threshold, 1e-9), 0.0, 2.0) * 127.5
    colored_bgr = cv2.applyColorMap(scaled.astype(np.uint8), cv2.COLORMAP_TURBO)
    return cv2.cvtColor(colored_bgr, cv2.COLOR_BGR2RGB)


def _write_mask_raster(path: Path, mask: np.ndarray, info: RasterInfo) -> None:
    with rasterio.open(
        path,
        "w",
        driver="GTiff",
        height=mask.shape[0],
        width=mask.shape[1],
        count=1,
        dtype="uint8",
        transform=info.transform,
        crs=info.crs,
        nodata=0,
        compress="deflate",
    ) as dst:
        dst.write(mask, 1)


def _fallback_geometries(binary: np.ndarray, transform: Affine, crs: str | None) -> gpd.GeoDataFrame:
    records: list[dict[str, Any]] = []
    for geometry, value in shapes(binary.astype(np.uint8), mask=binary, transform=transform):
        if value != 1:
            continue
        polygon = shape(geometry)
        if polygon.is_empty:
            continue
        records.append({"geometry": polygon})
    return gpd.GeoDataFrame(records, geometry="geometry", crs=crs)


def _polygon_score_stats(
    geometry: Any,
    transform: Affine,
    shape_: tuple[int, int],
    rule_map: np.ndarray,
    isolation_map: np.ndarray,
) -> tuple[float, float, int]:
    inverse = ~transform
    corners = [
        inverse * (geometry.bounds[0], geometry.bounds[1]),
        inverse * (geometry.bounds[0], geometry.bounds[3]),
        inverse * (geometry.bounds[2], geometry.bounds[1]),
        inverse * (geometry.bounds[2], geometry.bounds[3]),
    ]
    xs = [point[0] for point in corners]
    ys = [point[1] for point in corners]
    x0 = max(0, int(math.floor(min(xs))))
    x1 = min(shape_[1], int(math.ceil(max(xs))) + 1)
    y0 = max(0, int(math.floor(min(ys))))
    y1 = min(shape_[0], int(math.ceil(max(ys))) + 1)
    if x1 <= x0 or y1 <= y0:
        return 0.0, 0.0, 0
    local_transform = transform * Affine.translation(x0, y0)
    inside = geometry_mask(
        [mapping(geometry)],
        out_shape=(y1 - y0, x1 - x0),
        transform=local_transform,
        invert=True,
    )
    if not inside.any():
        return 0.0, 0.0, 0
    return (
        float(rule_map[y0:y1, x0:x1][inside].max()),
        float(isolation_map[y0:y1, x0:x1][inside].max()),
        int(inside.sum()),
    )


def vectorize_candidates(
    mask: np.ndarray,
    rule_map: np.ndarray,
    isolation_map: np.ndarray,
    transform: Affine,
    crs: str | None,
    tile_rows: list[dict[str, Any]],
    output_path: Path,
    minimum_area: int,
    source_file: str,
) -> tuple[int, str]:
    frames: list[gpd.GeoDataFrame] = []
    vectorizers: list[str] = []
    for code in (1, 2, 3):
        binary = mask == code
        if not binary.any():
            continue
        reference = _fallback_geometries(binary, transform, crs)
        temporary = output_path.with_name(f".{output_path.stem}-{code}.tif")
        _write_mask_raster(
            temporary,
            np.where(binary, 255, 0).astype(np.uint8),
            RasterInfo(transform, crs, bool(crs and transform != Affine.identity()), [1], "binary"),
        )
        selected = reference
        source = "rasterio.features.shapes fallback"
        try:
            from geoai import masks_to_vector

            candidate = masks_to_vector(
                str(temporary), min_object_area=minimum_area, simplify_tolerance=2.0
            )
            pixel_area = abs(transform.a * transform.e - transform.b * transform.d) or 1.0
            reference = reference[
                reference.geometry.area / pixel_area >= minimum_area
            ].copy()
            reference_area = float(reference.geometry.area.sum())
            candidate_area = float(candidate.geometry.area.sum()) if not candidate.empty else 0.0
            ratio = candidate_area / reference_area if reference_area else 0.0
            if len(candidate) == len(reference) and (not reference_area or 0.9 <= ratio <= 1.1):
                selected = candidate[["geometry"]].copy()
                source = "geoai.masks_to_vector"
            else:
                selected = reference
                source = "geoai.masks_to_vector + rasterio completeness repair"
        except Exception:
            pixel_area = abs(transform.a * transform.e - transform.b * transform.d) or 1.0
            selected = reference[reference.geometry.area / pixel_area >= minimum_area].copy()
        finally:
            temporary.unlink(missing_ok=True)
        vectorizers.append(source)
        rows: list[dict[str, Any]] = []
        for _, record in selected.iterrows():
            geometry = record.geometry
            max_rule, max_isolation, area_pixels = _polygon_score_stats(
                geometry, transform, mask.shape, rule_map, isolation_map
            )
            reason = ""
            best_rule = -math.inf
            for tile in tile_rows:
                center_x = (tile["x0"] + tile["x1"]) / 2.0
                center_y = (tile["y0"] + tile["y1"]) / 2.0
                coordinate = transform * (center_x, center_y)
                if geometry.covers(Point(coordinate)) and tile["rule_score"] > best_rule:
                    best_rule = float(tile["rule_score"])
                    reason = str(tile["reason_feature"])
            rows.append(
                {
                    "geometry": geometry,
                    "source_file": source_file,
                    "mask_code": code,
                    "method": MASK_LABELS[code],
                    "rule_flag": code in (1, 3),
                    "isolation_flag": code in (2, 3),
                    "agreement": code == 3,
                    "max_rule_score": round(max_rule, 6),
                    "max_isolation_score": round(max_isolation, 6),
                    "reason_feature": reason,
                    "area_pixels": area_pixels,
                }
            )
        if rows:
            frames.append(gpd.GeoDataFrame(rows, geometry="geometry", crs=crs))
    if frames:
        merged = gpd.GeoDataFrame(pd.concat(frames, ignore_index=True), geometry="geometry", crs=crs)
        merged.insert(0, "feature_id", np.arange(1, len(merged) + 1))
    else:
        merged = gpd.GeoDataFrame(
            {
                "feature_id": pd.Series(dtype="int64"),
                "mask_code": pd.Series(dtype="int64"),
                "method": pd.Series(dtype="str"),
                "geometry": gpd.GeoSeries([], crs=crs),
            },
            geometry="geometry",
            crs=crs,
        )
    merged.to_file(output_path, driver="GeoJSON")
    unique_vectorizers = set(vectorizers)
    if "geoai.masks_to_vector + rasterio completeness repair" in unique_vectorizers:
        vectorizer = "geoai.masks_to_vector + rasterio completeness repair"
    elif "geoai.masks_to_vector" in unique_vectorizers:
        vectorizer = "geoai.masks_to_vector"
    elif unique_vectorizers:
        vectorizer = "rasterio.features.shapes fallback"
    else:
        vectorizer = "no candidates"
    return len(merged), vectorizer


def _truth_entry(manifest: dict[str, Any] | None, filename: str) -> dict[str, Any] | None:
    if not manifest:
        return None
    images = manifest.get("images", {})
    entry = images.get(filename)
    return entry if isinstance(entry, dict) else None


def _binary_metrics(predicted: np.ndarray, truth: np.ndarray) -> dict[str, float | int]:
    predicted = predicted.astype(bool)
    truth = truth.astype(bool)
    true_positive = int(np.count_nonzero(predicted & truth))
    false_positive = int(np.count_nonzero(predicted & ~truth))
    false_negative = int(np.count_nonzero(~predicted & truth))
    precision = true_positive / (true_positive + false_positive) if true_positive + false_positive else 0.0
    recall = true_positive / (true_positive + false_negative) if true_positive + false_negative else 0.0
    union = true_positive + false_positive + false_negative
    return {
        "true_positive_pixels": true_positive,
        "false_positive_pixels": false_positive,
        "false_negative_pixels": false_negative,
        "precision": round(precision, 6),
        "recall": round(recall, 6),
        "iou": round(true_positive / union if union else 0.0, 6),
    }


def evaluate_truth(
    mask: np.ndarray,
    entry: dict[str, Any],
    manifest_dir: Path,
    expected_shape: tuple[int, int],
) -> dict[str, Any]:
    truth_path = (manifest_dir / str(entry["mask"])).resolve()
    with Image.open(truth_path) as image:
        truth = np.asarray(image.convert("L")) > 0
    if truth.shape != expected_shape:
        raise ValueError(f"Truth mask shape {truth.shape} does not match image {expected_shape}.")
    methods = {
        "rule": np.isin(mask, [1, 3]),
        "isolation": np.isin(mask, [2, 3]),
        "agreement": mask == 3,
    }
    result: dict[str, Any] = {name: _binary_metrics(values, truth) for name, values in methods.items()}
    region_results: dict[str, list[dict[str, Any]]] = {}
    for name, values in methods.items():
        regions: list[dict[str, Any]] = []
        for region in entry.get("regions", []):
            x0, y0, x1, y1 = map(int, region["bbox"])
            region_truth = truth[y0:y1, x0:x1]
            detected = values[y0:y1, x0:x1]
            overlap = int(np.count_nonzero(region_truth & detected))
            truth_pixels = int(np.count_nonzero(region_truth))
            regions.append(
                {
                    "name": region["name"],
                    "overlap_pixels": overlap,
                    "truth_pixels": truth_pixels,
                    "overlap_ratio": round(overlap / truth_pixels if truth_pixels else 0.0, 6),
                    "hit": overlap > 0,
                }
            )
        region_results[name] = regions
    result["regions"] = region_results
    return result


def process_image(
    input_path: Path,
    output_dir: Path,
    models: dict[str, Any],
    tile_size: int,
    stride: int,
    truth_manifest: dict[str, Any] | None,
    truth_manifest_dir: Path | None,
) -> dict[str, Any]:
    started = time.perf_counter()
    rgb, raster_info = read_rgb(input_path)
    features, windows = image_features(rgb, tile_size, stride)
    rule_scores, isolation_scores, reason_indices = score_features(features, models)
    rule_map = scores_to_raster(rule_scores, windows, rgb.shape[:2])
    isolation_map = scores_to_raster(isolation_scores, windows, rgb.shape[:2])
    rule_flag = rule_map > models["rule_threshold"]
    isolation_flag = isolation_map > models["isolation_threshold"]
    encoded = rule_flag.astype(np.uint8) + isolation_flag.astype(np.uint8) * 2

    stem = input_path.stem
    rule_heatmap_path = output_dir / f"{stem}.rule.heatmap.png"
    isolation_heatmap_path = output_dir / f"{stem}.isolation.heatmap.png"
    overlay_path = output_dir / f"{stem}.comparison.overlay.png"
    mask_path = output_dir / f"{stem}.anomaly-mask.tif"
    vector_path = output_dir / f"{stem}.anomalies.geojson"
    csv_path = output_dir / f"{stem}.tiles.csv"

    Image.fromarray(_heatmap(rule_map, models["rule_threshold"])).save(rule_heatmap_path)
    Image.fromarray(_heatmap(isolation_map, models["isolation_threshold"])).save(
        isolation_heatmap_path
    )
    color_mask = np.zeros_like(rgb)
    for code, color in MASK_COLORS.items():
        color_mask[encoded == code] = color
    overlay = rgb.copy()
    candidates = encoded > 0
    overlay[candidates] = (
        rgb[candidates].astype(np.float32) * 0.42
        + color_mask[candidates].astype(np.float32) * 0.58
    ).clip(0, 255).astype(np.uint8)
    Image.fromarray(overlay).save(overlay_path, quality=92)
    _write_mask_raster(mask_path, encoded, raster_info)

    tile_rows: list[dict[str, Any]] = []
    for index, ((x0, y0, x1, y1), feature_values) in enumerate(
        zip(windows, features, strict=True)
    ):
        row: dict[str, Any] = {
            "tile_id": index + 1,
            "x0": x0,
            "y0": y0,
            "x1": x1,
            "y1": y1,
            "rule_score": float(rule_scores[index]),
            "isolation_score": float(isolation_scores[index]),
            "rule_anomaly": bool(rule_scores[index] > models["rule_threshold"]),
            "isolation_anomaly": bool(isolation_scores[index] > models["isolation_threshold"]),
            "reason_feature": FEATURE_NAMES[int(reason_indices[index])],
        }
        row.update({name: float(value) for name, value in zip(FEATURE_NAMES, feature_values, strict=True)})
        tile_rows.append(row)
    pd.DataFrame(tile_rows).to_csv(csv_path, index=False, encoding="utf-8-sig")
    candidate_count, vectorizer = vectorize_candidates(
        encoded,
        rule_map,
        isolation_map,
        raster_info.transform,
        raster_info.crs,
        tile_rows,
        vector_path,
        minimum_area=max(64, tile_size * tile_size // 16),
        source_file=input_path.name,
    )

    truth_metrics = None
    entry = _truth_entry(truth_manifest, input_path.name)
    if entry and truth_manifest_dir:
        truth_metrics = evaluate_truth(encoded, entry, truth_manifest_dir, rgb.shape[:2])
    pixel_count = int(encoded.size)
    return {
        "file": input_path.name,
        "width": int(rgb.shape[1]),
        "height": int(rgb.shape[0]),
        "tile_count": len(windows),
        "rule_heatmap_file": rule_heatmap_path.name,
        "isolation_heatmap_file": isolation_heatmap_path.name,
        "overlay_file": overlay_path.name,
        "mask_file": mask_path.name,
        "vector_file": vector_path.name,
        "tiles_file": csv_path.name,
        "candidate_count": candidate_count,
        "rule_anomaly_pixels": int(np.count_nonzero(rule_flag)),
        "isolation_anomaly_pixels": int(np.count_nonzero(isolation_flag)),
        "agreement_pixels": int(np.count_nonzero(encoded == 3)),
        "rule_anomaly_coverage": round(float(np.count_nonzero(rule_flag) / pixel_count), 6),
        "isolation_anomaly_coverage": round(
            float(np.count_nonzero(isolation_flag) / pixel_count), 6
        ),
        "agreement_coverage": round(float(np.count_nonzero(encoded == 3) / pixel_count), 6),
        "georeferenced": raster_info.georeferenced,
        "crs": raster_info.crs,
        "source_bands": raster_info.source_bands,
        "normalization": raster_info.normalization,
        "coordinate_basis": "map_coordinates" if raster_info.georeferenced else "pixel_coordinates",
        "vectorizer": vectorizer,
        "truth_metrics": truth_metrics,
        "elapsed_seconds": round(time.perf_counter() - started, 3),
    }


def collect_inputs(path: Path) -> list[Path]:
    if path.is_file():
        candidates: Iterable[Path] = [path]
    elif path.is_dir():
        candidates = sorted(item for item in path.iterdir() if item.is_file())
    else:
        raise ValueError(f"Input path does not exist: {path}")
    inputs = [item for item in candidates if item.suffix.lower() in SUPPORTED_SUFFIXES]
    if not inputs:
        raise ValueError(f"No JPG, PNG, or GeoTIFF inputs were found in: {path}")
    return inputs


def load_truth_manifest(path: Path | None) -> tuple[dict[str, Any] | None, Path | None]:
    if path is None:
        return None, None
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload.get("images"), dict):
        raise ValueError("Truth manifest must contain an 'images' object.")
    return payload, path.parent


def run(args: argparse.Namespace) -> dict[str, Any]:
    if args.output.exists() and any(args.output.iterdir()):
        raise ValueError(f"Output directory is not empty: {args.output}. Use a new run directory.")
    args.output.mkdir(parents=True, exist_ok=True)
    reference_paths = collect_inputs(args.reference)
    input_paths = collect_inputs(args.input)
    started = time.perf_counter()
    reference_feature_sets: list[np.ndarray] = []
    reference_images: list[dict[str, Any]] = []
    reference_shapes: list[tuple[int, int]] = []
    for path in reference_paths:
        rgb, _ = read_rgb(path)
        features, windows = image_features(rgb, args.tile_size, args.stride)
        reference_feature_sets.append(features)
        reference_shapes.append(rgb.shape[:2])
        reference_images.append(
            {
                "file": path.name,
                "tile_count": len(windows),
                "width": rgb.shape[1],
                "height": rgb.shape[0],
            }
        )
    reference_features = np.vstack(reference_feature_sets)
    requested_spatial_mode = getattr(args, "spatial_mode", "auto")
    if requested_spatial_mode not in {"auto", "global", "aligned"}:
        raise ValueError("spatial_mode must be auto, global, or aligned.")
    input_shapes = [read_rgb(path)[0].shape[:2] for path in input_paths]
    aligned_compatible = (
        len(reference_paths) >= 3
        and len(set(reference_shapes)) == 1
        and all(shape_ == reference_shapes[0] for shape_ in input_shapes)
    )
    if requested_spatial_mode == "aligned" and not aligned_compatible:
        raise ValueError(
            "Aligned spatial mode requires at least three references and all reference/target images "
            "to have identical dimensions."
        )
    selected_rule_mode = (
        "aligned"
        if requested_spatial_mode == "aligned"
        or (requested_spatial_mode == "auto" and aligned_compatible)
        else "global"
    )
    models = fit_models(
        reference_features,
        args.threshold_quantile,
        args.random_state,
        reference_feature_sets if selected_rule_mode == "aligned" else None,
    )
    truth_manifest, truth_manifest_dir = load_truth_manifest(args.truth_manifest)
    images = [
        process_image(
            path,
            args.output,
            models,
            args.tile_size,
            args.stride,
            truth_manifest,
            truth_manifest_dir,
        )
        for path in input_paths
    ]
    metadata = {
        "capability": "09-anomaly-detection",
        "classification": "B",
        "created_at": datetime.now(UTC).isoformat(),
        "versions": {
            "python": __import__("platform").python_version(),
            "geoai-py": importlib_metadata.version("geoai-py"),
            "scikit-learn": importlib_metadata.version("scikit-learn"),
            "rasterio": importlib_metadata.version("rasterio"),
            "opencv-python-headless": importlib_metadata.version("opencv-python-headless"),
        },
        "method": (
            "position-aligned robust feature rules compared with global Isolation Forest"
            if models["rule_mode"] == "aligned"
            else "global robust feature rules compared with Isolation Forest"
        ),
        "model": "IsolationForest(n_estimators=200, contamination='auto')",
        "device": "CPU",
        "reference_path": args.reference.as_posix(),
        "input_path": args.input.as_posix(),
        "reference_count": len(reference_paths),
        "reference_tile_count": int(reference_features.shape[0]),
        "input_count": len(input_paths),
        "feature_names": FEATURE_NAMES,
        "parameters": {
            "tile_size": args.tile_size,
            "stride": args.stride,
            "threshold_quantile": args.threshold_quantile,
            "random_state": args.random_state,
            "minimum_reference_tiles": 50,
            "spatial_mode_requested": requested_spatial_mode,
            "rule_mode_selected": models["rule_mode"],
        },
        "thresholds": {
            "rule_score": round(float(models["rule_threshold"]), 8),
            "isolation_score": round(float(models["isolation_threshold"]), 8),
        },
        "reference_images": reference_images,
        "images": images,
        "truth_manifest": args.truth_manifest.as_posix() if args.truth_manifest else None,
        "elapsed_seconds": round(time.perf_counter() - started, 3),
        "limitations": [
            "异常表示相对正常参考影像的统计离群，只是人工复核候选，不是堵塞、损坏、渗漏或告警结论。",
            "受控注入异常只能验证检测链路，不能代表真实矿山场景精度。",
            "自然材质稀有区域、视角、光照、季节和清晰度变化都可能产生误报。",
            "同位置规则模式要求固定机位和相同像素尺寸；auto 条件不满足时自动回退到全局规则。",
            "普通 JPG/PNG 使用像素坐标；只有含有效 CRS 和仿射变换的 GeoTIFF 才保留地图坐标。",
        ],
    }
    (args.output / "run_metadata.json").write_text(
        json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    return metadata


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Run the CPU image-region anomaly baseline.")
    parser.add_argument("--reference", type=Path, required=True, help="Normal reference image or directory.")
    parser.add_argument("--input", type=Path, required=True, help="Target image or directory.")
    parser.add_argument("--output", type=Path, required=True, help="A new, empty output directory.")
    parser.add_argument("--tile-size", type=int, default=256)
    parser.add_argument("--stride", type=int, default=128)
    parser.add_argument("--threshold-quantile", type=float, default=0.995)
    parser.add_argument("--random-state", type=int, default=42)
    parser.add_argument(
        "--spatial-mode",
        choices=("auto", "global", "aligned"),
        default="auto",
        help="auto uses same-position rules for 3+ aligned references, otherwise global rules.",
    )
    parser.add_argument("--truth-manifest", type=Path)
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        metadata = run(args)
    except (OSError, ValueError, json.JSONDecodeError) as error:
        print(f"ERROR: {error}", file=__import__("sys").stderr)
        return 2
    print(json.dumps(metadata, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

 capabilities/09-anomaly-detection/tests/test_anomaly_detection.py

New file
@@ -0,0 +1,205 @@
import argparse
import json
import sys
import tempfile
import unittest
from pathlib import Path

import cv2
import numpy as np
import rasterio
from PIL import Image
from rasterio.transform import from_origin


CAPABILITY = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(CAPABILITY))

from generate_validation_inputs import generate  # noqa: E402
from run_anomaly_detection import (  # noqa: E402
    FEATURE_NAMES,
    aligned_rule_scores,
    axis_positions,
    extract_features,
    fit_aligned_rule_model,
    fit_models,
    run,
)


def patterned_image(width: int = 192, height: int = 160) -> np.ndarray:
    yy, xx = np.mgrid[:height, :width]
    return np.stack(
        [
            (xx * 3 + yy) % 256,
            (xx + yy * 2 + 40) % 256,
            (xx * 2 + yy * 3 + 80) % 256,
        ],
        axis=2,
    ).astype(np.uint8)


class AnomalyDetectionTests(unittest.TestCase):
    def test_axis_positions_anchor_final_tile(self):
        self.assertEqual(axis_positions(100, 32, 20), [0, 20, 40, 60, 68])
        with self.assertRaises(ValueError):
            axis_positions(20, 32, 16)

    def test_features_explain_blur_and_colour(self):
        source = patterned_image(64, 64)
        blurred = cv2.GaussianBlur(source, (31, 31), 0)
        magenta = np.full((64, 64, 3), (230, 20, 210), dtype=np.uint8)
        source_features = extract_features(source)
        blurred_features = extract_features(blurred)
        magenta_features = extract_features(magenta)
        laplacian_index = FEATURE_NAMES.index("laplacian_variance")
        saturation_index = FEATURE_NAMES.index("hsv_mean_s")
        self.assertLess(blurred_features[laplacian_index], source_features[laplacian_index])
        self.assertGreater(magenta_features[saturation_index], source_features[saturation_index])

    def test_isolation_model_is_repeatable(self):
        rng = np.random.default_rng(7)
        features = rng.normal(size=(80, len(FEATURE_NAMES)))
        first = fit_models(features, 0.995, 42)
        second = fit_models(features, 0.995, 42)
        self.assertAlmostEqual(first["isolation_threshold"], second["isolation_threshold"], places=12)
        np.testing.assert_allclose(
            first["reference_isolation_scores"], second["reference_isolation_scores"]
        )

    def test_aligned_rules_detect_local_change_despite_global_weather_shift(self):
        rng = np.random.default_rng(12)
        base = rng.normal(size=(60, len(FEATURE_NAMES)))
        references = [
            base + rng.normal(scale=0.08, size=base.shape) + offset
            for offset in (-0.3, -0.1, 0.0, 0.15, 0.35)
        ]
        model = fit_aligned_rule_model(references, 0.995)
        target = base + 0.22
        target[25:29, FEATURE_NAMES.index("dark_ratio")] += 8.0
        scores, reasons = aligned_rule_scores(
            target, model["aligned_center"], model["aligned_scale"]
        )
        flagged = set(np.flatnonzero(scores > model["rule_threshold"]))
        self.assertTrue(set(range(25, 29)).issubset(flagged))
        self.assertEqual(FEATURE_NAMES[int(reasons[26])], "dark_ratio")

    def test_cli_contract_produces_visual_and_structured_outputs(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            reference = root / "reference"
            inputs = root / "inputs"
            reference.mkdir()
            inputs.mkdir()
            normal = patterned_image()
            Image.fromarray(normal).save(reference / "normal.png")
            Image.fromarray(normal).save(inputs / "normal.png")
            anomalous = normal.copy()
            anomalous[48:96, 64:112] = (5, 5, 5)
            Image.fromarray(anomalous).save(inputs / "anomalous.png")
            output = root / "output"
            metadata = run(
                argparse.Namespace(
                    reference=reference,
                    input=inputs,
                    output=output,
                    tile_size=32,
                    stride=16,
                    threshold_quantile=0.99,
                    random_state=42,
                    truth_manifest=None,
                )
            )
            self.assertEqual(metadata["input_count"], 2)
            self.assertGreaterEqual(metadata["reference_tile_count"], 50)
            for record in metadata["images"]:
                for key in (
                    "rule_heatmap_file",
                    "isolation_heatmap_file",
                    "overlay_file",
                    "mask_file",
                    "vector_file",
                    "tiles_file",
                ):
                    self.assertTrue((output / record[key]).is_file(), key)
            stored = json.loads((output / "run_metadata.json").read_text(encoding="utf-8"))
            self.assertEqual(stored["classification"], "B")
            with self.assertRaises(ValueError):
                run(
                    argparse.Namespace(
                        reference=reference,
                        input=inputs,
                        output=output,
                        tile_size=32,
                        stride=16,
                        threshold_quantile=0.99,
                        random_state=42,
                        truth_manifest=None,
                    )
                )

    def test_projected_geotiff_preserves_crs(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            source = patterned_image(96, 96)
            reference = root / "reference.tif"
            target = root / "target.tif"
            transform = from_origin(500000, 4000000, 2, 2)
            for path, values in ((reference, source), (target, source.copy())):
                if path == target:
                    values[32:64, 32:64] = (250, 10, 230)
                with rasterio.open(
                    path,
                    "w",
                    driver="GTiff",
                    width=96,
                    height=96,
                    count=3,
                    dtype="uint8",
                    transform=transform,
                    crs="EPSG:3857",
                ) as dst:
                    dst.write(np.moveaxis(values, 2, 0))
            output = root / "out"
            metadata = run(
                argparse.Namespace(
                    reference=reference,
                    input=target,
                    output=output,
                    tile_size=16,
                    stride=8,
                    threshold_quantile=0.99,
                    random_state=42,
                    truth_manifest=None,
                )
            )
            record = metadata["images"][0]
            self.assertTrue(record["georeferenced"])
            self.assertEqual(record["crs"], "EPSG:3857")
            with rasterio.open(output / record["mask_file"]) as src:
                self.assertEqual(src.crs.to_string(), "EPSG:3857")
                self.assertEqual(src.transform, transform)

    def test_too_few_reference_tiles_fails(self):
        rng = np.random.default_rng(1)
        with self.assertRaisesRegex(ValueError, "At least 50"):
            fit_models(rng.normal(size=(49, len(FEATURE_NAMES))), 0.995, 42)

    def test_validation_generator_preserves_source_bytes(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            normal = root / "normal.jpg"
            difficult = root / "difficult.jpg"
            Image.fromarray(patterned_image()).save(normal)
            Image.fromarray(np.flipud(patterned_image())).save(difficult)
            result = generate(normal, difficult, root / "raw", root / "processed")
            self.assertEqual(result["normal_source_sha256"], result["reference_copy_sha256"])
            self.assertEqual(result["difficult_source_sha256"], result["difficult_copy_sha256"])
            manifest = json.loads(
                (root / "processed" / "truth-manifest.json").read_text(encoding="utf-8")
            )
            self.assertEqual(len(manifest["images"]["07-16-injected.png"]["regions"]), 3)


if __name__ == "__main__":
    unittest.main()

 scripts/serve_workbench_console.py

@@ -5,6 +5,7 @@
import argparse
import base64
import binascii
import hashlib
import json
import os
import re
@@ -29,6 +30,7 @@
MAX_IMAGES_PER_RUN = 12
MAX_SEGMENTATION_IMAGES_PER_RUN = 6
MAX_MEASUREMENT_RASTERS_PER_RUN = 4
MAX_ANOMALY_IMAGES_PER_ROLE = 6
CHANGE_THRESHOLD_DEFAULT = 0.5
CHANGE_THRESHOLD_MIN = 0.01
CHANGE_THRESHOLD_MAX = 0.99
@@ -44,10 +46,13 @@
    "shared/data/raw/00-change-detection",
    "shared/data/raw/01-object-detection",
    "shared/data/raw/02-semantic-mapping",
    "shared/data/raw/09-anomaly-detection",
)
SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9._-]+")
SAFE_FILE_NAME = re.compile(r"[^\w.-]+", re.UNICODE)
SAFE_UPLOAD_ID = re.compile(r"^[0-9a-f]{32}$")
RUN_LOCK = threading.Lock()
ANOMALY_JOB_LOCK = threading.Lock()
ANOMALY_JOBS: dict[str, dict[str, Any]] = {}


class ApiError(ValueError):
@@ -92,6 +97,14 @@
    except (OSError, json.JSONDecodeError):
        return {}
    return payload if isinstance(payload, dict) else {}


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def trajectory_runs(root: Path) -> list[dict[str, Any]]:
@@ -233,6 +246,120 @@
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def anomaly_runs(root: Path) -> list[dict[str, Any]]:
    output_root = root / "shared" / "outputs" / "09-anomaly-detection"
    allowed_raw = (root / "shared" / "data" / "raw" / "09-anomaly-detection").resolve()
    records: list[dict[str, Any]] = []
    for metadata_path in output_root.rglob("run_metadata.json"):
        artifact = metadata_path.parent
        metadata = load_json(metadata_path)
        images = metadata.get("images")
        if metadata.get("capability") != "09-anomaly-detection" or not isinstance(images, list):
            continue
        raw_input_value = str(metadata.get("raw_input_dir") or "")
        raw_reference_value = str(metadata.get("raw_reference_dir") or "")
        if not raw_input_value or not raw_reference_value:
            continue
        try:
            raw_input = (root / raw_input_value).resolve()
            raw_reference = (root / raw_reference_value).resolve()
            raw_input.relative_to(allowed_raw)
            raw_reference.relative_to(allowed_raw)
        except ValueError:
            continue
        if not raw_input.is_dir() or not raw_reference.is_dir():
            continue
        if any(not (artifact / str(item.get("overlay_file") or "")).is_file() for item in images if isinstance(item, dict)):
            continue
        run_id = artifact.name
        records.append(
            {
                "id": run_id,
                "label": str(metadata.get("display_name") or run_id),
                "note": str(metadata.get("case_note") or "规则基线与 Isolation Forest 的视觉离群候选，只供人工复核。"),
                "artifactRoot": relative_path(root, artifact),
                "inputRoot": relative_path(root, raw_input),
                "referenceRoot": relative_path(root, raw_reference),
                "createdAt": str(metadata.get("created_at") or ""),
            }
        )
    return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)


def anomaly_job(job_id: str) -> dict[str, Any] | None:
    with ANOMALY_JOB_LOCK:
        value = ANOMALY_JOBS.get(job_id)
        return dict(value) if value else None


def validate_anomaly_parameters(payload: dict[str, Any]) -> tuple[int, int, float, int]:
    tile_size = payload.get("tileSize", 256)
    stride = payload.get("stride", 128)
    threshold_quantile = payload.get("thresholdQuantile", 0.995)
    random_state = payload.get("randomState", 42)
    if isinstance(tile_size, bool) or not isinstance(tile_size, int) or not 128 <= tile_size <= 1024:
        raise ApiError("Tile size must be an integer between 128 and 1024.")
    if isinstance(stride, bool) or not isinstance(stride, int) or not 32 <= stride <= tile_size:
        raise ApiError("Stride must be an integer between 32 and tile size.")
    if isinstance(threshold_quantile, bool) or not isinstance(threshold_quantile, (int, float)) or not 0.9 <= float(threshold_quantile) <= 0.9999:
        raise ApiError("Threshold quantile must be between 0.9 and 0.9999.")
    if isinstance(random_state, bool) or not isinstance(random_state, int) or not 0 <= random_state <= 2_147_483_647:
        raise ApiError("Random state must be a non-negative integer.")
    return tile_size, stride, float(threshold_quantile), random_state


def execute_anomaly_job(
    root: Path,
    job_id: str,
    run_id: str,
    raw_reference: Path,
    raw_input: Path,
    processed_reference: Path,
    processed_input: Path,
    output: Path,
    tile_size: int,
    stride: int,
    threshold_quantile: float,
    random_state: int,
) -> None:
    with ANOMALY_JOB_LOCK:
        ANOMALY_JOBS[job_id]["status"] = "running"
    python = root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe"
    command = [
        str(python),
        str(root / "capabilities" / "09-anomaly-detection" / "run_anomaly_detection.py"),
        "--reference", str(processed_reference),
        "--input", str(processed_input),
        "--output", str(output),
        "--tile-size", str(tile_size),
        "--stride", str(stride),
        "--threshold-quantile", f"{threshold_quantile:.6f}",
        "--random-state", str(random_state),
        "--spatial-mode", "auto",
    ]
    try:
        with RUN_LOCK:
            completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=1800, check=False)
        if completed.returncode:
            message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
            raise ApiError(f"Processing failed: {message[:600]}")
        metadata_path = output / "run_metadata.json"
        if not metadata_path.is_file():
            raise ApiError("Anomaly-detection script finished without the expected result metadata.")
        metadata = load_json(metadata_path)
        metadata["raw_input_dir"] = relative_path(root, raw_input)
        metadata["raw_reference_dir"] = relative_path(root, raw_reference)
        metadata["processed_input_dir"] = relative_path(root, processed_input)
        metadata["processed_reference_dir"] = relative_path(root, processed_reference)
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        definition = next(item for item in anomaly_runs(root) if item["id"] == run_id)
        with ANOMALY_JOB_LOCK:
            ANOMALY_JOBS[job_id].update({"status": "complete", "run": definition, "finishedAt": datetime.now(UTC).isoformat()})
    except Exception as exc:  # pragma: no cover - background boundary
        with ANOMALY_JOB_LOCK:
            ANOMALY_JOBS[job_id].update({"status": "failed", "error": str(exc), "finishedAt": datetime.now(UTC).isoformat()})


def semantic_tasks(root: Path) -> list[dict[str, Any]]:
    catalog = load_json(root / "capabilities" / "02-semantic-mapping" / "configs" / "task-catalog.json")
    tasks = catalog.get("tasks")
@@ -270,6 +397,14 @@
        if path == "/api/spatial-measurement/runs":
            self.send_json(HTTPStatus.OK, {"runs": measurement_runs(self.root)})
            return
        if path == "/api/anomaly-detection/runs":
            self.send_json(HTTPStatus.OK, {"runs": anomaly_runs(self.root)})
            return
        if path.startswith("/api/anomaly-detection/jobs/"):
            job_id = path.rstrip("/").rsplit("/", 1)[-1]
            job = anomaly_job(job_id)
            self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown anomaly-detection job."})
            return
        if path == "/":
            self.send_response(HTTPStatus.FOUND)
            self.send_header("Location", "/apps/workbench-console/")
@@ -296,6 +431,9 @@
            if path == "/api/spatial-measurement/runs":
                self.send_json(HTTPStatus.CREATED, {"run": self.create_measurement_run(payload)})
                return
            if path == "/api/anomaly-detection/runs":
                self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_anomaly_run(payload)})
                return
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
        except ApiError as exc:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
@@ -307,11 +445,15 @@

    def do_PUT(self) -> None:  # noqa: N802 - binary upload endpoint
        path = urlsplit(self.path).path
        if not path.startswith("/api/change-detection/uploads/"):
        if not path.startswith("/api/change-detection/uploads/") and not path.startswith("/api/anomaly-detection/uploads/"):
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
            return
        try:
            self.send_json(HTTPStatus.CREATED, self.receive_change_upload(path))
            if path.startswith("/api/anomaly-detection/uploads/"):
                result = self.receive_anomaly_upload(path)
            else:
                result = self.receive_change_upload(path)
            self.send_json(HTTPStatus.CREATED, result)
        except ApiError as exc:
            self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
        except Exception as exc:  # pragma: no cover - defensive server boundary
@@ -320,7 +462,7 @@

    def do_OPTIONS(self) -> None:  # noqa: N802
        self.send_response(HTTPStatus.NO_CONTENT)
        self.send_header("Allow", "GET, POST, OPTIONS")
        self.send_header("Allow", "GET, POST, PUT, OPTIONS")
        self.end_headers()

    def read_json_body(self) -> dict[str, Any]:
@@ -347,14 +489,32 @@
            raise ApiError(f"Processing failed: {message[:600]}")

    def receive_change_upload(self, path: str) -> dict[str, Any]:
        return self.receive_binary_upload(path, "00-change-detection", {"before", "after"}, "change-detection")

    def receive_anomaly_upload(self, path: str) -> dict[str, Any]:
        return self.receive_binary_upload(path, "09-anomaly-detection", {"reference", "input"}, "anomaly-detection")

    def receive_binary_upload(
        self,
        path: str,
        capability: str,
        allowed_roles: set[str],
        label: str,
    ) -> dict[str, Any]:
        upload_id = path.rstrip("/").rsplit("/", 1)[-1]
        if not SAFE_UPLOAD_ID.fullmatch(upload_id):
            raise ApiError("Invalid change-detection upload id.")
            raise ApiError(f"Invalid {label} upload id.")
        query = parse_qs(urlsplit(self.path).query)
        role = query.get("role", [""])[0]
        if role not in {"before", "after"}:
            raise ApiError("Change-detection upload role must be before or after.")
        name = self.headers.get("X-Upload-Name", "")
        if role not in allowed_roles:
            raise ApiError(f"Invalid {label} upload role.")
        encoded_name = self.headers.get("X-Upload-Name", "")
        if len(encoded_name) > 2048:
            raise ApiError("Encoded upload name is too long.")
        try:
            name = unquote(encoded_name, encoding="utf-8", errors="strict")
        except UnicodeError as exc:
            raise ApiError("Upload name is not valid UTF-8 percent encoding.") from exc
        safe_name = safe_file_name(name, {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
        content_length = self.headers.get("Content-Length")
        if content_length is None or not content_length.isdigit():
@@ -362,11 +522,12 @@
        size = int(content_length)
        if size <= 0 or size > MAX_FILE_BYTES:
            raise ApiError(f"Uploaded file must be between 1 byte and {MAX_FILE_BYTES // (1024 * 1024)} MB: {safe_name}.")
        staging = self.root / "shared" / "data" / "raw" / "00-change-detection" / "uploads" / upload_id
        staging = self.root / "shared" / "data" / "raw" / capability / "uploads" / upload_id
        staging.mkdir(parents=True, exist_ok=False)
        part = staging / f"{role}.part"
        target = staging / f"{role}{Path(safe_name).suffix.lower()}"
        remaining = size
        digest = hashlib.sha256()
        try:
            with part.open("wb") as stream:
                while remaining:
@@ -374,27 +535,36 @@
                    if not chunk:
                        raise ApiError("Binary upload ended before Content-Length was reached.")
                    stream.write(chunk)
                    digest.update(chunk)
                    remaining -= len(chunk)
            part.replace(target)
            (staging / f"{role}.json").write_text(json.dumps({"role": role, "name": safe_name, "size": size}), encoding="utf-8")
            (staging / f"{role}.json").write_text(json.dumps({"role": role, "name": safe_name, "size": size, "sha256": digest.hexdigest()}), encoding="utf-8")
        except Exception:
            part.unlink(missing_ok=True)
            target.unlink(missing_ok=True)
            raise
        return {"uploadId": upload_id, "role": role, "name": safe_name, "size": size}
        return {"uploadId": upload_id, "role": role, "name": safe_name, "size": size, "sha256": digest.hexdigest()}

    def resolve_change_upload(self, payload: Any, role: str) -> tuple[str, Path]:
        return self.resolve_binary_upload(payload, role, "00-change-detection", "change-detection")

    def resolve_anomaly_upload(self, payload: Any, role: str) -> tuple[str, Path, str]:
        name, path = self.resolve_binary_upload(payload, role, "09-anomaly-detection", "anomaly-detection")
        manifest = load_json(path.parent / f"{role}.json")
        return name, path, str(manifest.get("sha256") or "")

    def resolve_binary_upload(self, payload: Any, role: str, capability: str, label: str) -> tuple[str, Path]:
        if not isinstance(payload, dict) or not isinstance(payload.get("uploadId"), str):
            raise ApiError(f"Change-detection uploads must include a {role} uploadId.")
            raise ApiError(f"{label} uploads must include a {role} uploadId.")
        upload_id = payload["uploadId"]
        if not SAFE_UPLOAD_ID.fullmatch(upload_id):
            raise ApiError("Invalid change-detection upload id.")
        staging = self.root / "shared" / "data" / "raw" / "00-change-detection" / "uploads" / upload_id
            raise ApiError(f"Invalid {label} upload id.")
        staging = self.root / "shared" / "data" / "raw" / capability / "uploads" / upload_id
        manifest = load_json(staging / f"{role}.json")
        name = str(manifest.get("name") or "")
        path = staging / f"{role}{Path(name).suffix.lower()}"
        if manifest.get("role") != role or not name or not path.is_file():
            raise ApiError(f"The staged {role} TIFF upload is unavailable or incomplete.")
            raise ApiError(f"The staged {role} upload is unavailable or incomplete.")
        return name, path

    def create_trajectory_run(self, payload: dict[str, Any]) -> dict[str, Any]:
@@ -568,6 +738,64 @@
        metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
        return next(item for item in semantic_runs(self.root) if item["id"] == run_id)

    def create_anomaly_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        tile_size, stride, threshold_quantile, random_state = validate_anomaly_parameters(payload)
        uploads = payload.get("uploads")
        if not isinstance(uploads, dict):
            raise ApiError("Anomaly-detection request must contain reference and input uploads.")
        reference_values = uploads.get("reference")
        input_values = uploads.get("input")
        if not isinstance(reference_values, list) or not reference_values:
            raise ApiError("Select at least one normal reference image.")
        if not isinstance(input_values, list) or not input_values:
            raise ApiError("Select at least one image to inspect.")
        if len(reference_values) > MAX_ANOMALY_IMAGES_PER_ROLE or len(input_values) > MAX_ANOMALY_IMAGES_PER_ROLE:
            raise ApiError(f"An anomaly-detection run accepts at most {MAX_ANOMALY_IMAGES_PER_ROLE} images in each group.")
        references = [self.resolve_anomaly_upload(value, "reference") for value in reference_values]
        inputs = [self.resolve_anomaly_upload(value, "input") for value in input_values]
        if len({name.casefold() for name, _, _ in references}) != len(references):
            raise ApiError("Normal reference image names must be unique within one run.")
        if len({name.casefold() for name, _, _ in inputs}) != len(inputs):
            raise ApiError("Input image names must be unique within one run.")

        python = self.root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe"
        if not python.is_file():
            raise ApiError("Anomaly-detection virtual environment is unavailable. Run the capability setup first.")
        run_id = make_run_id("anomaly")
        job_id = uuid4().hex
        raw_root = self.root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / run_id
        raw_reference = raw_root / "reference"
        raw_input = raw_root / "input"
        processed_root = self.root / "shared" / "data" / "processed" / "09-anomaly-detection" / run_id
        processed_reference = processed_root / "reference"
        processed_input = processed_root / "input"
        output = self.root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / run_id
        for directory in (raw_reference, raw_input, processed_reference, processed_input):
            directory.mkdir(parents=True, exist_ok=False)
        for group, raw_dir, processed_dir in ((references, raw_reference, processed_reference), (inputs, raw_input, processed_input)):
            for name, staged_path, expected_sha256 in group:
                raw_path = raw_dir / name
                processed_path = processed_dir / name
                shutil.copyfile(staged_path, raw_path)
                if expected_sha256 and file_sha256(raw_path) != expected_sha256:
                    raise ApiError(f"Uploaded file checksum changed while staging: {name}.")
                shutil.copyfile(raw_path, processed_path)
        for _, staged_path, _ in references + inputs:
            shutil.rmtree(staged_path.parent)

        created_at = datetime.now(UTC).isoformat()
        job = {"id": job_id, "runId": run_id, "status": "queued", "createdAt": created_at}
        with ANOMALY_JOB_LOCK:
            ANOMALY_JOBS[job_id] = job
        thread = threading.Thread(
            target=execute_anomaly_job,
            args=(self.root, job_id, run_id, raw_reference, raw_input, processed_reference, processed_input, output, tile_size, stride, threshold_quantile, random_state),
            daemon=True,
            name=f"anomaly-{run_id}",
        )
        thread.start()
        return dict(job)

    def create_measurement_run(self, payload: dict[str, Any]) -> dict[str, Any]:
        uploads = payload.get("rasters")
        if not isinstance(uploads, list) or not uploads:

 tests/test_serve_workbench_console.py

@@ -5,6 +5,8 @@
import sys
import tempfile
import unittest
import json
from urllib.parse import quote
from pathlib import Path


@@ -44,9 +46,18 @@
            Path(handler.translate_path("/shared/data/raw/00-change-detection/sample.jpg")),
            ROOT / "shared" / "data" / "raw" / "00-change-detection" / "sample.jpg",
        )
        self.assertEqual(
            Path(handler.translate_path("/shared/data/raw/09-anomaly-detection/sample.tif")),
            ROOT / "shared" / "data" / "raw" / "09-anomaly-detection" / "sample.tif",
        )
        self.assertEqual(
            Path(handler.translate_path("/shared/data/processed/09-anomaly-detection/sample.tif")),
            ROOT / ".console-forbidden",
        )

    def test_upload_name_is_sanitized_and_extension_is_allowlisted(self) -> None:
        self.assertEqual(MODULE.safe_file_name("../../unsafe name.JPG", {".jpg"}), "unsafe_name.jpg")
        self.assertEqual(MODULE.safe_file_name("../正常参考图.JPG", {".jpg"}), "正常参考图.jpg")
        with self.assertRaises(MODULE.ApiError):
            MODULE.safe_file_name("image.exe", {".jpg", ".png"})

@@ -73,6 +84,66 @@
            self.assertEqual(staged.read_bytes(), b"raw-tif-bytes")
            self.assertEqual(result["name"], "1.tif")

    def test_binary_anomaly_upload_preserves_bytes_and_checksum(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef?role=reference"
            handler.headers = {"Content-Length": "15", "X-Upload-Name": "../normal.JPG"}
            handler.rfile = io.BytesIO(b"reference-bytes")
            result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "reference.jpg"
            self.assertEqual(staged.read_bytes(), b"reference-bytes")
            self.assertEqual(result["sha256"], MODULE.file_sha256(staged))

    def test_binary_anomaly_upload_decodes_chinese_file_name(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef?role=input"
            handler.headers = {"Content-Length": "11", "X-Upload-Name": quote("异常测试图.JPG", safe="")}
            handler.rfile = io.BytesIO(b"image-bytes")
            result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "input.jpg"
            self.assertEqual(result["name"], "异常测试图.jpg")
            self.assertEqual(staged.read_bytes(), b"image-bytes")

    def test_interrupted_anomaly_upload_removes_partial_file(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef?role=input"
            handler.headers = {"Content-Length": "20", "X-Upload-Name": "target.tif"}
            handler.rfile = io.BytesIO(b"short")
            with self.assertRaisesRegex(MODULE.ApiError, "ended before"):
                handler.receive_anomaly_upload("/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef")
            staging = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / "2123456789abcdef0123456789abcdef"
            self.assertFalse(any(staging.glob("*.part")))

    def test_anomaly_parameters_are_bounded(self) -> None:
        self.assertEqual(MODULE.validate_anomaly_parameters({}), (256, 128, 0.995, 42))
        with self.assertRaisesRegex(MODULE.ApiError, "Stride"):
            MODULE.validate_anomaly_parameters({"tileSize": 128, "stride": 256})
        with self.assertRaisesRegex(MODULE.ApiError, "quantile"):
            MODULE.validate_anomaly_parameters({"thresholdQuantile": 1.0})

    def test_anomaly_run_discovery_requires_exposed_raw_sources(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            artifact = root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / "anomaly-test"
            raw_input = root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / "anomaly-test" / "input"
            raw_reference = raw_input.parent / "reference"
            artifact.mkdir(parents=True)
            raw_input.mkdir(parents=True)
            raw_reference.mkdir(parents=True)
            (artifact / "target.comparison.overlay.png").write_bytes(b"png")
            metadata = {"capability": "09-anomaly-detection", "created_at": "2026-08-18", "display_name": "五参考图边界案例", "case_note": "保留真实漏检结果。", "raw_input_dir": raw_input.relative_to(root).as_posix(), "raw_reference_dir": raw_reference.relative_to(root).as_posix(), "images": [{"overlay_file": "target.comparison.overlay.png"}]}
            (artifact / "run_metadata.json").write_text(json.dumps(metadata), encoding="utf-8")
            runs = MODULE.anomaly_runs(root)
            self.assertEqual(runs[0]["id"], "anomaly-test")
            self.assertEqual(runs[0]["label"], "五参考图边界案例")
            self.assertEqual(runs[0]["note"], "保留真实漏检结果。")

    def test_script_failure_becomes_a_useful_api_error(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "Processing failed"):