"""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())
|
