Buckets:
| """cwm3/baselines_dms/smoke.py — reproducible smoke tests of every installed DMS component: shapes, dims and field names only. | |
| python -m baselines_dms.smoke [--only openvino,mediapipe,l2cs,videomaev2,openface3] [--device cuda:0] | |
| Inputs are synthetic noise images plus the PUBLIC MediaPipe sample portrait (registry id `sample_portrait`, fetched on demand) — no BATON frame | |
| is read here. Every component is resolved through registry.resolve (sha256 asserted). Writes baselines_dms/weights/smoke_latest.json | |
| (never results/). Failures are recorded per section instead of aborting the run; exit code 1 if any section failed. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import time | |
| import traceback | |
| import numpy as np | |
| from baselines_dms import registry as R | |
| OUT = R.WEIGHTS / "smoke_latest.json" | |
| def _sample_portrait(): | |
| e = R.get("sample_portrait") | |
| if R.verify(e): | |
| R.install(e) | |
| import cv2 | |
| return cv2.imread(str(R.resolve("sample_portrait"))) # BGR uint8 | |
| def smoke_openvino(device: str) -> dict: | |
| import openvino as ov | |
| core = ov.Core() | |
| out = {"openvino": ov.__version__, "devices": core.available_devices, "models": {}} | |
| for eid in sorted(R.entries()): | |
| if not eid.startswith("omz_"): | |
| continue | |
| d = R.resolve(eid) | |
| xml = next(d.glob("*.xml")) | |
| t0 = time.time() | |
| cm = core.compile_model(core.read_model(str(xml)), "CPU") | |
| ins = [{"name": i.get_any_name(), "shape": list(i.shape), "dtype": str(i.get_element_type())} for i in cm.inputs] | |
| outs = [{"name": o.get_any_name(), "shape": list(o.shape), "dtype": str(o.get_element_type())} for o in cm.outputs] | |
| feed = {i.get_any_name(): np.random.RandomState(0).rand(*list(i.shape)).astype(np.float32) for i in cm.inputs} | |
| res = cm(feed) | |
| out["models"][eid] = {"xml": xml.name, "compile_ms": round((time.time() - t0) * 1e3, 1), "inputs": ins, "outputs": outs, | |
| "synthetic_output_shapes": {cm.outputs[k].get_any_name(): list(v.shape) for k, v in enumerate(res.values())}} | |
| print(f" [openvino] {xml.name}: in {[(i['name'], i['shape']) for i in ins]} -> out {[(o['name'], o['shape']) for o in outs]}") | |
| return out | |
| def smoke_mediapipe(device: str) -> dict: | |
| import cv2 | |
| import mediapipe as mp | |
| from mediapipe.tasks import python as mpp | |
| from mediapipe.tasks.python import vision | |
| opts = vision.FaceLandmarkerOptions(base_options=mpp.BaseOptions(model_asset_path=str(R.resolve("mediapipe_face_landmarker"))), | |
| running_mode=vision.RunningMode.IMAGE, num_faces=1, output_face_blendshapes=True, | |
| output_facial_transformation_matrixes=True) | |
| lmk = vision.FaceLandmarker.create_from_options(opts) | |
| syn = (np.random.RandomState(0).rand(256, 256, 3) * 255).astype(np.uint8) | |
| r0 = lmk.detect(mp.Image(image_format=mp.ImageFormat.SRGB, data=syn)) | |
| img = cv2.cvtColor(_sample_portrait(), cv2.COLOR_BGR2RGB) | |
| t0 = time.time() | |
| r1 = lmk.detect(mp.Image(image_format=mp.ImageFormat.SRGB, data=img)) | |
| dt = (time.time() - t0) * 1e3 | |
| names = [c.category_name for c in r1.face_blendshapes[0]] if r1.face_blendshapes else [] | |
| out = {"mediapipe": mp.__version__, | |
| "synthetic_256": {"faces": len(r0.face_landmarks), "blendshape_lists": len(r0.face_blendshapes), "matrices": len(r0.facial_transformation_matrixes)}, | |
| "portrait": {"faces": len(r1.face_landmarks), "landmarks_per_face": len(r1.face_landmarks[0]) if r1.face_landmarks else 0, | |
| "landmark_fields": ["x", "y", "z"], "n_blendshapes": len(names), | |
| "matrix_shape": list(r1.facial_transformation_matrixes[0].shape) if r1.facial_transformation_matrixes else None, "ms": round(dt, 1)}, | |
| "blendshape_names": names} | |
| lmk.close() | |
| print(f" [mediapipe] synthetic faces={out['synthetic_256']['faces']}; portrait landmarks={out['portrait']['landmarks_per_face']} blendshapes={len(names)} matrix={out['portrait']['matrix_shape']}") | |
| return out | |
| def smoke_l2cs(device: str) -> dict: | |
| import torch | |
| from l2cs import Pipeline | |
| dev = torch.device(device if ":" in device or device == "cpu" else device + ":0") | |
| p = Pipeline(weights=R.resolve("l2cs"), arch="ResNet50", device=dev) | |
| syn = (np.random.RandomState(0).rand(224, 224, 3) * 255).astype(np.uint8) | |
| pitch, yaw = p.predict_gaze(syn) | |
| img = _sample_portrait() | |
| t0 = time.time() | |
| r = p.step(img) | |
| dt = (time.time() - t0) * 1e3 | |
| out = {"arch": "ResNet50", "device": str(dev), "params_M": round(sum(t.numel() for t in p.model.parameters()) / 1e6, 3), | |
| "synthetic_224_predict_gaze": {"pitch_shape": list(pitch.shape), "yaw_shape": list(yaw.shape), "unit": "radians"}, | |
| "portrait_step": {"fields": list(r.__dataclass_fields__), "pitch": list(r.pitch.shape), "yaw": list(r.yaw.shape), "bboxes": list(r.bboxes.shape), | |
| "landmarks": list(r.landmarks.shape), "scores": list(r.scores.shape), "ms": round(dt, 1)}} | |
| print(f" [l2cs] loaded {out['params_M']} M params on {dev}; synthetic pitch/yaw {pitch.shape}; portrait fields {out['portrait_step']['fields']} bboxes {r.bboxes.shape}") | |
| return out | |
| def smoke_videomaev2(device: str) -> dict: | |
| import torch | |
| sys.path.insert(0, str(R.HERE / "vendor")) | |
| import videomaev2_model as vm | |
| m = vm.vit_small_patch16_224(num_classes=710, all_frames=16, tubelet_size=2) | |
| ck = torch.load(R.resolve("videomaev2"), map_location="cpu", weights_only=False) | |
| res = m.load_state_dict(ck["module"], strict=True) | |
| m = m.to(device).half().eval() | |
| x = torch.randn(1, 3, 16, 224, 224, device=device, dtype=torch.float16) | |
| with torch.no_grad(): | |
| t0 = time.time() | |
| f = m.forward_features(x) | |
| torch.cuda.synchronize() if device.startswith("cuda") else None | |
| dt = (time.time() - t0) * 1e3 | |
| logits = m.head(f) | |
| out = {"load_state_dict": str(res), "input_shape": list(x.shape), "dtype": str(x.dtype), "feature_shape": list(f.shape), "feature_dim": int(f.shape[-1]), | |
| "logits_shape": list(logits.shape), "finite": bool(torch.isfinite(f).all()), "params_M": round(sum(p.numel() for p in m.parameters()) / 1e6, 3), "ms_first_call": round(dt, 1)} | |
| print(f" [videomaev2] {out['load_state_dict']}; {tuple(x.shape)} fp16 -> feature {tuple(f.shape)} (dim {out['feature_dim']}), logits {tuple(logits.shape)}") | |
| return out | |
| def smoke_openface3(device: str) -> dict: | |
| import cv2 | |
| import torch | |
| from baselines_dms import of3_compat | |
| fd, ld, mt = of3_compat.load(device=device) | |
| ld68 = of3_compat.landmark_detector(device=device, landmarks=68) | |
| syn_path = str(R.WEIGHTS / "_samples" / "synthetic_256.png") | |
| cv2.imwrite(syn_path, (np.random.RandomState(0).rand(256, 256, 3) * 255).astype(np.uint8)) | |
| face0, dets0 = fd.get_face(syn_path) | |
| img = _sample_portrait() | |
| t0 = time.time() | |
| face, dets = fd.get_face(str(R.resolve("sample_portrait"))) | |
| dt = (time.time() - t0) * 1e3 | |
| lm98 = ld.detect_landmarks(img, dets) | |
| lm68 = ld68.detect_landmarks(img, dets) | |
| emo, gaze, au = mt.predict(face) | |
| out = {"device": device, "output_fields": list(of3_compat.OUTPUT_FIELDS), "emotion_classes": list(of3_compat.EMOTIONS), | |
| "synthetic_256": {"face": None if face0 is None else list(face0.shape), "dets": None if dets0 is None else list(dets0.shape)}, | |
| "portrait": {"face_crop": list(face.shape), "dets": list(dets.shape), "dets_row_fields": "x1,y1,x2,y2,score + 5 landmark (x,y) = 15", "n_dets_above_landmark_threshold_0.5": len(lm98), "ms_detect_first_call": round(dt, 1), | |
| "landmarks_98_WFLW": [len(lm98), list(lm98[0].shape)], "landmarks_68_300W": [len(lm68), list(lm68[0].shape)], | |
| "emotion_logits": list(emo.shape), "gaze_yaw_pitch": list(gaze.shape), "action_units": list(au.shape)}} | |
| print(f" [openface3] synthetic face={out['synthetic_256']['face']}; portrait dets {dets.shape} lm98 {lm98[0].shape} lm68 {lm68[0].shape} emotion {tuple(emo.shape)} gaze {tuple(gaze.shape)} AU {tuple(au.shape)}") | |
| return out | |
| SECTIONS = {"openvino": smoke_openvino, "mediapipe": smoke_mediapipe, "l2cs": smoke_l2cs, "videomaev2": smoke_videomaev2, "openface3": smoke_openface3} | |
| def main(argv=None) -> int: | |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) | |
| ap.add_argument("--only", default=",".join(SECTIONS)) | |
| ap.add_argument("--device", default="cuda:0") | |
| a = ap.parse_args(argv) | |
| (R.WEIGHTS / "_samples").mkdir(parents=True, exist_ok=True) | |
| report = {"utc": R.utc_now(), "device": a.device, "sections": {}} | |
| rc = 0 | |
| for name in a.only.split(","): | |
| print(f"[smoke] {name}") | |
| try: | |
| report["sections"][name] = {"status": "OK", **SECTIONS[name](a.device)} | |
| except Exception as e: # record, do not abort | |
| report["sections"][name] = {"status": "FAIL", "error": f"{type(e).__name__}: {e}", "traceback": traceback.format_exc()[-2000:]} | |
| print(f" [FAIL] {name}: {type(e).__name__}: {e}") | |
| rc = 1 | |
| OUT.write_text(json.dumps(report, indent=1, ensure_ascii=False) + "\n") | |
| print(f"[smoke] wrote {OUT}; {sum(s['status'] == 'OK' for s in report['sections'].values())}/{len(report['sections'])} sections OK") | |
| return rc | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 9.51 kB
- Xet hash:
- 9d50bf804f5e93035977a13b59d246c811f4ed3bdc0617efa75c7e8578990707
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.