Hsyshasuzh/longmemory-geo / project /evaluation /render_detection_error_sheets.py
Hsyshasuzh's picture
download
raw
4.08 kB
"""Render representative development error panels for manual review."""
import argparse
from collections import defaultdict
import json
from pathlib import Path
from PIL import Image, ImageDraw
from load_dataset import dataset, load_sample
from training.model_factory import ROOT
from training.soda_a import inventory, materialize, tile_descriptors
COLORS = {"target": "#25d366", "prediction": "#f44336", "highlight": "#ffd600"}
def draw_polygon(draw, points, color, width=4):
vertices = list(zip(points[::2], points[1::2]))
if vertices:
draw.line(vertices + vertices[:1], fill=color, width=width)
def panel(image, targets, predictions, highlighted, title):
canvas = Image.fromarray(image).convert("RGB")
draw = ImageDraw.Draw(canvas)
for item in targets:
draw_polygon(draw, item["obb_polygon"], COLORS["target"], 3)
for item in predictions:
draw_polygon(draw, item["obb_polygon"], COLORS["prediction"], 2)
draw_polygon(draw, highlighted["obb_polygon"], COLORS["highlight"], 5)
canvas.thumbnail((384, 384))
result = Image.new("RGB", (400, 424), "black")
result.paste(canvas, (8, 32))
ImageDraw.Draw(result).text((8, 8), title, fill="white")
return result
def records_by_id(records):
return ({row["sample_id"]: row.get("annotations", []) for row in records["targets"]},
{row["sample_id"]: row.get("annotations", []) for row in records["predictions"]})
def sheet(items, path):
columns = 3
rows = (len(items) + columns - 1) // columns
output = Image.new("RGB", (columns * 400, rows * 424), "black")
for index, item in enumerate(items):
output.paste(item, ((index % columns) * 400, (index // columns) * 424))
path.parent.mkdir(parents=True, exist_ok=True)
output.save(path)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("locations", type=Path)
parser.add_argument("soda_records", type=Path)
parser.add_argument("dota_records", type=Path)
parser.add_argument("output_dir", type=Path)
args = parser.parse_args()
if args.output_dir.exists():
raise FileExistsError("output directory already exists")
locations = json.loads(args.locations.read_text())
soda_records = json.loads(args.soda_records.read_text())
dota_records = json.loads(args.dota_records.read_text())["candidate"]
soda_targets, soda_predictions = records_by_id(soda_records)
dota_targets, dota_predictions = records_by_id(dota_records)
soda_descriptors = {item["id"]: item for row in inventory(splits=frozenset({"val"})) if row["split"] == "val"
for item in tile_descriptors(row)}
dota_data = dataset(ROOT / "datasets/corpus", "detection", "val", license_scope="all")
dota_samples = {row["id"]: row for row in dota_data.samples}
for source, target_rows, prediction_rows in (("soda_a_validation", soda_targets, soda_predictions),
("dota_selection", dota_targets, dota_predictions)):
panels = []
for class_id, errors in locations[source].items():
for kind in ("false_positives", "missed_targets"):
if not errors[kind]:
continue
item = errors[kind][0]
sample_id = item["sample_id"]
if source == "soda_a_validation":
image = materialize(soda_descriptors[sample_id])["image"]
else:
image = load_sample(dota_data.root, dota_samples[sample_id], dota_data.expected_hashes)["image"]
confidence = item.get("confidence")
suffix = f" conf={confidence:.3f}" if confidence is not None else ""
title = f"{errors['class']} {kind[:-1]}{suffix}\n{sample_id}"
panels.append(panel(image, target_rows[sample_id], prediction_rows[sample_id], item, title))
sheet(panels, args.output_dir / f"{source}.jpg")
print(json.dumps({"output_dir": str(args.output_dir)}))
if __name__ == "__main__":
main()

Xet Storage Details

Size:
4.08 kB
·
Xet hash:
2a31dd29753b5cda51cb7a39fa4f421ec5ae7f42e061c07ff672c18664bb1b9f

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.