ADAS/HSAP
2
2
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add support for ADAS project in labeling and export processes

Browse Source 
- Implemented class name loading for ADAS campaigns in `_class_names_for_campaign`.
- Enhanced annotation parsing to support cuboid and 2D rectangle labels in `_parse_ls_annotations`.
- Updated catalog building to include ADAS packs and batches in `build_catalog_signature`.
- Added ADAS cuboid promotion logic in `adas_cuboid.py`, supporting both quaternion_json and YOLO HBB formats.
- Introduced validation for ADAS batches to check for YOLO HBB labels in `adas_cuboid.py`.
- Modified delivery scanning to include ADAS project deliveries in `scan.py`.
- Extended job execution to handle ADAS exports for YOLO format in `runner.py`.
- Updated labeling export logic to accommodate ADAS project in `export_cuboid_batch.py`.
- Enhanced format conversion to support rectangle labels in `format_converter.py`.
- Improved CVAT task handling in `service.py` to ensure task ID updates on each labeling job.
- Updated web API endpoints to reflect ADAS project changes in `hsap-api.ts`.
- Added ADAS catalog types and UI components for displaying ADAS packs and batches in `dmsCatalog.ts`, `CatalogPage.tsx`, and `ExportPage.tsx`.
- Adjusted workflow registry to align with new ADAS project structure.
This commit is contained in:
jiacheng.lin
2026-07-16 16:03:04 +08:00
parent be8a885b07
commit 1bde0fe430
23 changed files with 472 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
platform/as_platform/data/catalog_cache.py
View File

@@ -83,7 +83,7 @@ def build_catalog_signature(wf: dict, proj_root_fn) -> dict[str, Any]:
files.append({"path": str(p), "missing": True})
dirs: list[dict[str, Any]] = []
for pname in ("dms", "lane"):
for pname in ("dms", "lane", "adas"):
root = proj_root_fn(wf, pname)
dirs.append(_dir_fingerprint(root, scan_children=False))
dirs.append(_dir_fingerprint(root / "inbox"))

73
platform/as_platform/data/core.py
View File

@@ -11,7 +11,7 @@ from typing import Any
import yaml
from as_platform.config import WORKSPACE, WORKSPACE_ROOT, LANE_DATA_VIZ_ENABLED
from as_platform.data.batch import META_FILENAME, dms_has_images, enrich_batch, write_meta
from as_platform.data.batch import META_FILENAME, dms_has_images, enrich_batch, read_meta, write_meta
from as_platform.data.catalog_cache import (
build_catalog_signature,
get_cached_catalog,
@@ -811,7 +811,7 @@ def _normalize_catalog_dms(dms: dict[str, Any], reg: dict) -> dict[str, Any]:
def _build_catalog(wf: dict, *, prefer_reports: bool = True) -> tuple[dict[str, Any], str]:
out: dict[str, Any] = {"workspace": str(WORKSPACE), "dms": {}, "lane": {}}
out: dict[str, Any] = {"workspace": str(WORKSPACE), "dms": {}, "lane": {}, "adas": {}}
build_source = "scan"
reports = load_dms_reports()
@@ -966,6 +966,71 @@ def _build_catalog(wf: dict, *, prefer_reports: bool = True) -> tuple[dict[str,
"add_template": "python as.py add lane --src <archive> --engineer <name> --date YYYYMMDD",
"quality": lane_quality,
}
# ── ADAS catalog ──
adas_root = proj_root(wf, "adas")
out["adas"] = {}
adas_packs_dir = adas_root / "packs"
if adas_packs_dir.is_dir():
for pack_dir in sorted(adas_packs_dir.iterdir()):
if pack_dir.name.startswith(".") or not pack_dir.is_dir():
continue
pack_name = pack_dir.name
sources_dir = pack_dir / "sources"
pack_entry: dict[str, Any] = {
"name": pack_name,
"path": str(pack_dir.relative_to(adas_root)),
"enabled": pack_name in wf["projects"]["adas"].get("active_packs", []),
"batches": [],
"total_images": 0,
"total_labels": 0,
"class_counts": {},
}
if sources_dir.is_dir():
for batch_dir in sorted(sources_dir.iterdir()):
if batch_dir.name.startswith(".") or not batch_dir.is_dir():
continue
batch_name = batch_dir.name
meta = read_meta(batch_dir) or {}
stage = meta.get("stage", "unknown")
task = meta.get("task", "")
counts = meta.get("counts", {})
img_count = counts.get("images", 0)
# Count quaternion_json detections
qdir = batch_dir / "labels" / "quaternion_json"
label_count = 0
class_counts: dict[str, int] = {}
if qdir.is_dir():
for qf in sorted(qdir.glob("*.json")):
try:
qdata = json.loads(qf.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
continue
for det in qdata.get("detections") or []:
label_count += 1
cls = str(det.get("class_name") or det.get("class_id", "unknown"))
class_counts[cls] = class_counts.get(cls, 0) + 1
# Count images from directory if meta is missing
if img_count == 0:
imgs_dir = batch_dir / "images"
if imgs_dir.is_dir():
img_count = sum(1 for f in imgs_dir.rglob("*") if f.suffix.lower() in IMG_EXTS)
batch_entry = {
"batch": batch_name,
"task": task,
"stage": stage,
"images": img_count,
"labels": label_count,
"class_counts": class_counts,
"ingested_at": meta.get("ingested_at"),
}
pack_entry["batches"].append(batch_entry)
pack_entry["total_images"] += img_count
pack_entry["total_labels"] += label_count
for cls, n in class_counts.items():
pack_entry["class_counts"][cls] = pack_entry["class_counts"].get(cls, 0) + n
out["adas"][pack_name] = pack_entry
return out, build_source
@@ -996,7 +1061,9 @@ def get_catalog(
result = {"task": task_or_pack, **(full_catalog.get("dms", {}).get(task_or_pack, {}))}
elif project == "lane" and task_or_pack:
result = {"pack": task_or_pack, **(full_catalog.get("lane", {}).get(task_or_pack, {}))}
elif project in ("dms", "lane"):
elif project == "adas" and task_or_pack:
result = {"pack": task_or_pack, **(full_catalog.get("adas", {}).get(task_or_pack, {}))}
elif project in ("dms", "lane", "adas"):
result = {"workspace": full_catalog.get("workspace", str(WORKSPACE)), project: full_catalog.get(project, {})}
else:
result = dict(full_catalog)

15
platform/as_platform/data/promote/adas_cuboid.py
View File

@@ -86,14 +86,20 @@ class AdasCuboidPromoteAdapter(PackPromoteAdapter):
def promote(self, ctx: PromoteContext) -> PromoteResult:
warnings = list(ctx.extra.get("validate_warnings") or [])
qdir = ctx.batch_dir / "labels" / "quaternion_json"
if not qdir.is_dir() or not any(qdir.glob("*.json")):
has_quat = qdir.is_dir() and any(qdir.glob("*.json"))
# YOLO HBB 回退
labels_dir = ctx.batch_dir / "labels"
has_yolo = (labels_dir / "yolo-hbb").is_dir() and any((labels_dir / "yolo-hbb").glob("*.txt"))
if not has_quat and not has_yolo:
return PromoteResult(
ok=False,
project=ctx.project,
task=ctx.task,
batch=ctx.batch,
pack=ctx.pack,
warnings=["missing quaternion_json export"],
warnings=["missing quaternion_json or YOLO HBB export"],
)
pack_dir = ctx.project_root / "packs" / ctx.pack
@@ -119,7 +125,9 @@ class AdasCuboidPromoteAdapter(PackPromoteAdapter):
if src_sub.is_dir():
copied += _sync_tree(src_sub, dest / sub)
normalized = _normalize_quaternion_json(dest)
normalized = _normalize_quaternion_json(dest) if has_quat else 0
if has_yolo and not has_quat:
warnings.append("using YOLO HBB labels (no quaternion_json)")
meta = read_meta(ctx.batch_dir) or {}
meta.update({
@@ -130,6 +138,7 @@ class AdasCuboidPromoteAdapter(PackPromoteAdapter):
"pack": ctx.pack,
"ingested_at": datetime.now(timezone.utc).isoformat(),
"pipeline_version": 2,
"label_format": "quaternion_json" if has_quat else "yolo_hbb",
})
write_meta(dest, meta)
write_meta(ctx.batch_dir, meta)

73
platform/as_platform/data/promote/validate/adas_cuboid.py
View File

@@ -1,4 +1,4 @@
"""ADAS cuboid batch validation before promote."""
"""ADAS batch validation before promote（支持 quaternion_json 和 YOLO HBB）。"""
from __future__ import annotations
import json
@@ -7,20 +7,87 @@ from pathlib import Path
from as_platform.labeling.class_map import load_adas_class_names
def _has_yolo_hbb_labels(batch_dir: Path) -> bool:
"""检查是否有 YOLO HBB txt 标注文件。"""
yolo_dir = batch_dir / "labels" / "yolo-hbb"
return yolo_dir.is_dir() and any(yolo_dir.glob("*.txt"))
def _validate_yolo_hbb(batch_dir: Path, expected_names: list[str]) -> tuple[list[str], list[str], dict]:
"""校验 YOLO HBB 格式标注。"""
errors: list[str] = []
warnings: list[str] = []
n_classes = len(expected_names)
labels_dir = batch_dir / "labels" / "yolo-hbb"
txt_files = sorted(labels_dir.glob("*.txt")) if labels_dir.is_dir() else []
total_boxes = 0
files_with_boxes = 0
for p in txt_files:
try:
lines = p.read_text(encoding="utf-8").strip().splitlines()
except OSError as e:
errors.append(f"{p.name}: read error ({e})")
continue
if not lines:
continue
has_valid = False
for line in lines:
line = line.strip()
if not line:
continue
parts = line.split()
if len(parts) < 5:
warnings.append(f"{p.name}: invalid YOLO line '{line[:40]}...'")
continue
try:
cid = int(parts[0])
if cid < 0 or cid >= n_classes:
errors.append(f"{p.name}: class_id {cid} out of range (0-{n_classes - 1})")
continue
# 验证 bbox 值在 [0,1]
for v in parts[1:5]:
fv = float(v)
if fv < 0 or fv > 1:
warnings.append(f"{p.name}: bbox value {fv} out of [0,1]")
except ValueError:
warnings.append(f"{p.name}: non-numeric value in '{line[:40]}...'")
continue
total_boxes += 1
has_valid = True
if has_valid:
files_with_boxes += 1
if files_with_boxes == 0:
errors.append("no valid YOLO HBB label files")
stats = {
"format": "yolo_hbb",
"label_files": len(txt_files),
"files_with_boxes": files_with_boxes,
"total_boxes": total_boxes,
}
return errors, warnings, stats
def validate_adas_cuboid_batch(
batch_dir: Path,
*,
allow_partial_3d: bool = False,
min_fit_ratio: float = 0.8,
) -> tuple[list[str], list[str], dict]:
"""Return (errors, warnings, stats)."""
"""Return (errors, warnings, stats). 优先 quaternion_json，其次 YOLO HBB。"""
errors: list[str] = []
warnings: list[str] = []
qdir = batch_dir / "labels" / "quaternion_json"
expected_names = load_adas_class_names()
if not qdir.is_dir():
errors.append(f"missing labels/quaternion_json under {batch_dir}")
# 回退到 YOLO HBB 格式
if _has_yolo_hbb_labels(batch_dir):
return _validate_yolo_hbb(batch_dir, expected_names)
errors.append(f"missing labels/quaternion_json or YOLO HBB labels under {batch_dir}")
return errors, warnings, {}
files = sorted(qdir.glob("*.json"))