#!/usr/bin/env python3 """Protected, rig-name-independent RobotExpressive evaluator. This host-side driver launches fresh Blender processes for evidence extraction and sparse replay rendering, then scores the submitted 50/20/15/15 contract. It compares evaluated mesh landmarks, never unrelated rigs' raw transforms. """ import argparse import json import math import re import statistics import subprocess from pathlib import Path ROOT = Path(__file__).resolve().parents[2] DEFAULT_BLENDER = Path("/Applications/Blender.app/Contents/MacOS/Blender") def run(cmd, log): log.parent.mkdir(parents=True, exist_ok=True) proc = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) log.write_text(proc.stdout) if proc.returncode: raise RuntimeError(f"Command failed ({proc.returncode}); see {log}") return proc.stdout def load(path): return json.loads(Path(path).read_text()) def lerp_error(value, full, zero): if value <= full: return 1.0 if value >= zero: return 0.0 return (zero - value) / (zero - full) def lerp_similarity(value, zero, full): if value <= zero: return 0.0 if value >= full: return 1.0 return (value - zero) / (full - zero) def descriptor_cost(a, b): # Counts dominate correspondence. Shape terms rank equal-topology matches. count = ( abs(a["vertex_count"] - b["vertex_count"]) / max(a["vertex_count"], b["vertex_count"], 1) + abs(a["edge_count"] - b["edge_count"]) / max(a["edge_count"], b["edge_count"], 1) + abs(a["polygon_count"] - b["polygon_count"]) / max(a["polygon_count"], b["polygon_count"], 1) ) shape = sum(abs(x - y) for x, y in zip( a["bbox_ratios_sorted"], b["bbox_ratios_sorted"] )) / 3 edges = sum(abs(x - y) for x, y in zip( a["edge_length_quantiles"], b["edge_length_quantiles"] )) / 5 volume = min(1.0, abs( a.get("signed_volume_normalized", 0.0) - b.get("signed_volume_normalized", 0.0) ) * 20) return count * 3 + shape + edges + volume def match_components(ref, cand): refs = sorted((c for c in ref["components"] if c.get("canonical_part")), key=lambda c: c["descriptor"]["vertex_count"], reverse=True) candidates = {c["id"]: c for c in cand["components"]} # Establish rough yaw/scale/translation only from descriptor-unambiguous # canonical anchors. Helpers and names cannot influence this normalization. anchors = [] for rc in refs: ranked = sorted( (descriptor_cost(rc["descriptor"], cc["descriptor"]), tuple(cand["frames"]["1"][cid]["centroid"]), cid) for cid, cc in candidates.items()) if ranked and ranked[0][0] <= 0.20 and ( len(ranked) == 1 or ranked[1][0] - ranked[0][0] > 1e-5): anchors.append((rc["id"], ranked[0][2])) if len(anchors) < 3: # Preserve a deterministic, name-independent fallback. Ambiguous # correspondence is reported through reduced coverage downstream. anchors = [] used = set() for rc in refs: ranked = sorted( (descriptor_cost(rc["descriptor"], cc["descriptor"]), tuple(cand["frames"]["1"][cid]["centroid"]), cid) for cid, cc in candidates.items() if cid not in used) if ranked and ranked[0][0] <= 0.20: anchors.append((rc["id"], ranked[0][2])) used.add(ranked[0][2]) if len(anchors) >= 8: break rp = [ref["frames"]["1"][rid]["centroid"] for rid, _cid in anchors] cp = [cand["frames"]["1"][cid]["centroid"] for _rid, cid in anchors] if rp and cp: ref_center = [statistics.mean(v[i] for v in rp) for i in range(3)] cand_center = [statistics.mean(v[i] for v in cp) for i in range(3)] rr = [vec_sub(v, ref_center) for v in rp] cr = [vec_sub(v, cand_center) for v in cp] a = sum(c[0] * r[0] + c[1] * r[1] for c, r in zip(cr, rr)) b = sum(c[0] * r[1] - c[1] * r[0] for c, r in zip(cr, rr)) rough_yaw = math.atan2(b, a) denom = sum(sum(x * x for x in v) for v in cr) numer = sum(sum(x * x for x in v) for v in rr) rough_scale = math.sqrt(numer / denom) if denom > 1e-12 else 1.0 else: ref_center = cand_center = [0.0, 0.0, 0.0] rough_scale, rough_yaw = 1.0, 0.0 ref_radius = max( math.sqrt(statistics.mean(vec_len(vec_sub(v, ref_center)) ** 2 for v in rp)) if rp else 1.0, 1e-6) def spatial_cost(rid, cid): rp = ref["frames"]["1"][rid]["centroid"] cp0 = cand["frames"]["1"][cid]["centroid"] local = vec_sub(cp0, cand_center) co, si = math.cos(rough_yaw), math.sin(rough_yaw) cp = [ ref_center[0] + rough_scale * (co * local[0] - si * local[1]), ref_center[1] + rough_scale * (si * local[0] + co * local[1]), ref_center[2] + rough_scale * local[2], ] return vec_len(vec_sub(rp, cp)) / max(ref_radius, 1e-9) matches = {} costs = {} for rc in refs: ranked = sorted( ((descriptor_cost(rc["descriptor"], cc["descriptor"]) + 0.20 * spatial_cost(rc["id"], cid), tuple(cand["frames"]["1"][cid]["centroid"]), cid) for cid, cc in candidates.items())) if ranked and ranked[0][0] <= 0.55: _assignment_cost, _stable_position, cid = ranked[0] matches[rc["id"]] = cid costs[rc["id"]] = descriptor_cost( rc["descriptor"], candidates[cid]["descriptor"]) del candidates[cid] total_vertices = sum(c["descriptor"]["vertex_count"] for c in refs) matched_vertices = sum( c["descriptor"]["vertex_count"] for c in refs if c["id"] in matches ) return matches, costs, matched_vertices / max(total_vertices, 1) def vec_add(a, b): return [a[i] + b[i] for i in range(3)] def vec_sub(a, b): return [a[i] - b[i] for i in range(3)] def vec_len(a, axis_weights=(1, 1, 1)): return math.sqrt(sum(axis_weights[i] * a[i] * a[i] for i in range(3))) def aggregate_parts(evidence, ref, matches=None): """Aggregate connected components into protected canonical part IDs.""" component_info = {c["id"]: c for c in evidence["components"]} ref_info = {c["id"]: c for c in ref["components"]} by_part = {} if matches is None: entries = [(c["id"], c["id"], c["canonical_part"]) for c in ref["components"] if c["canonical_part"]] else: entries = [(rid, cid, ref_info[rid]["canonical_part"]) for rid, cid in matches.items() if ref_info[rid]["canonical_part"]] for rid, cid, part in entries: by_part.setdefault(part, []).append(( cid, ref_info[rid]["descriptor"]["vertex_count"] )) result = {} for frame, components in evidence["frames"].items(): result[frame] = {} for part, members in by_part.items(): valid = [(components[cid], weight) for cid, weight in members if cid in components] if not valid: continue total = sum(w for _v, w in valid) center = [ sum(v["centroid"][i] * w for v, w in valid) / total for i in range(3) ] lo = [min(v["bbox_min"][i] for v, _w in valid) for i in range(3)] hi = [max(v["bbox_max"][i] for v, _w in valid) for i in range(3)] landmark_names = set.intersection(*[ set(v.get("landmarks", {})) for v, _w in valid ]) if valid else set() landmarks = { name: [ sum(v["landmarks"][name][i] * w for v, w in valid) / total for i in range(3) ] for name in landmark_names } result[frame][part] = { "centroid": center, "bbox_min": lo, "bbox_max": hi, "landmarks": landmarks, } return result def solve_alignment(ref_parts, cand_parts): common = sorted(set(ref_parts["1"]) & set(cand_parts["1"])) if len(common) < 3: return {"scale": 1.0, "yaw": 0.0, "translation": [0, 0, 0]} r = [ref_parts["1"][p]["centroid"] for p in common] c = [cand_parts["1"][p]["centroid"] for p in common] rc = [statistics.mean(v[i] for v in r) for i in range(3)] cc = [statistics.mean(v[i] for v in c) for i in range(3)] rr = [vec_sub(v, rc) for v in r] cr = [vec_sub(v, cc) for v in c] a = sum(cv[0] * rv[0] + cv[1] * rv[1] for cv, rv in zip(cr, rr)) b = sum(cv[0] * rv[1] - cv[1] * rv[0] for cv, rv in zip(cr, rr)) yaw = math.atan2(b, a) denom = sum(sum(x * x for x in v) for v in cr) numer = sum(sum(x * x for x in v) for v in rr) scale = math.sqrt(numer / denom) if denom > 1e-12 else 1.0 co, si = math.cos(yaw), math.sin(yaw) rotated_cc = [ scale * (co * cc[0] - si * cc[1]), scale * (si * cc[0] + co * cc[1]), scale * cc[2], ] translation = vec_sub(rc, rotated_cc) return {"scale": scale, "yaw": yaw, "translation": translation} def transform(point, alignment): s, a = alignment["scale"], alignment["yaw"] co, si = math.cos(a), math.sin(a) return [ s * (co * point[0] - si * point[1]) + alignment["translation"][0], s * (si * point[0] + co * point[1]) + alignment["translation"][1], s * point[2] + alignment["translation"][2], ] def transform_parts(parts, alignment): out = {} for frame, pdata in parts.items(): out[frame] = {} for part, values in pdata.items(): out[frame][part] = { "centroid": transform(values["centroid"], alignment), "bbox_min": transform(values["bbox_min"], alignment), "bbox_max": transform(values["bbox_max"], alignment), "landmarks": { k: transform(v, alignment) for k, v in values.get("landmarks", {}).items() }, } # Rotation can swap x/y bbox extrema; repair min/max conservatively. out[frame][part]["bbox_min"] = [ min(out[frame][part]["bbox_min"][i], out[frame][part]["bbox_max"][i]) for i in range(3) ] out[frame][part]["bbox_max"] = [ max(transform(values["bbox_min"], alignment)[i], transform(values["bbox_max"], alignment)[i]) for i in range(3) ] return out def mean_motion_error(ref_parts, cand_parts, parts, frames, axis_weights): errors = [] for f in frames: sf = str(f) for p in parts: if p in ref_parts.get(sf, {}) and p in cand_parts.get(sf, {}): errors.append(vec_len(vec_sub( ref_parts[sf][p]["centroid"], cand_parts[sf][p]["centroid"] ), axis_weights)) return statistics.mean(errors) if errors else float("inf") def part_path_length(parts, part, start, end, axis_weights): total = 0.0 previous = None for frame in range(start, end + 1): current = parts.get(str(frame), {}).get(part) if current: point = current["centroid"] if previous is not None: total += vec_len(vec_sub(point, previous), axis_weights) previous = point return total def max_drift(parts, part, spans): worst = 0.0 for start, end in spans: pts = [ parts[str(f)][part].get("landmarks", {}).get( "dynamic_sole", parts[str(f)][part].get("landmarks", {}).get( "z_low", parts[str(f)][part]["centroid"])) for f in range(start, end + 1) if part in parts.get(str(f), {})] if pts: base = pts[0] worst = max(worst, max(vec_len(vec_sub(p, base), (1, 1, 0.25)) for p in pts)) return worst def max_acceleration(parts): worst = 0.0 for part in set().union(*(v.keys() for v in parts.values())): for f in range(2, 192): a = parts.get(str(f - 1), {}).get(part) b = parts.get(str(f), {}).get(part) c = parts.get(str(f + 1), {}).get(part) if a and b and c: second = vec_add(vec_sub(a["centroid"], [ 2 * x for x in b["centroid"] ]), c["centroid"]) worst = max(worst, vec_len(second)) return worst def best_timing_lag(ref_parts, cand_parts, part="part_006"): def speeds(data): out = {} for f in range(2, 193): if part in data.get(str(f), {}) and part in data.get(str(f - 1), {}): out[f] = vec_len(vec_sub(data[str(f)][part]["centroid"], data[str(f - 1)][part]["centroid"])) return out rs, cs = speeds(ref_parts), speeds(cand_parts) best = (float("inf"), 0) for lag in range(-18, 19): pairs = [(rs[f], cs[f + lag]) for f in range(62, 157) if f in rs and f + lag in cs] if pairs: err = statistics.mean(abs(a - b) for a, b in pairs) if err < best[0]: best = (err, lag) return best[1], best[0] def material_palette(evidence): colors = set() for comp in evidence["components"]: for mat in comp.get("materials", []): colors.add(tuple(mat["rgb"])) return sorted(colors) def palette_error(ref, cand): if not ref or not cand: return 1.0 return statistics.mean(min(vec_len(vec_sub(list(a), list(b))) for b in cand) for a in ref) def ffprobe(preview): cmd = [ "ffprobe", "-v", "error", "-count_frames", "-select_streams", "v:0", "-show_entries", "stream=r_frame_rate,nb_read_frames,width,height", "-show_entries", "format=duration", "-of", "json", str(preview) ] try: return json.loads(subprocess.check_output(cmd, text=True)) except Exception: return None def compare_preview(preview, replay_dir, config, work_dir): if not preview or not Path(preview).is_file(): return {"present": False, "timing_fraction": 0.0, "ssim_mean": 0.0, "ssim_fraction": 0.0, "motion_correlation": 0.0, "motion_fraction": 0.0} info = ffprobe(preview) timing = 0.0 if info and info.get("streams"): stream = info["streams"][0] rate = stream.get("r_frame_rate", "0/1").split("/") fps = float(rate[0]) / max(float(rate[1]), 1) frames = int(stream.get("nb_read_frames", 0)) duration = float(info["format"].get("duration", 0)) timing = sum((abs(fps - 24) < 0.01, frames == 192, abs(duration - 8.0) < 0.08)) / 3 extracted = [] ssims = [] work_dir.mkdir(parents=True, exist_ok=True) for frame in config["replay_comparison"]["sample_frames"]: dst = work_dir / f"preview_{frame:04d}.png" cmd = [ "ffmpeg", "-y", "-v", "error", "-i", str(preview), "-vf", f"select=eq(n\\,{frame - 1}),scale=320:320", "-frames:v", "1", str(dst) ] subprocess.run(cmd, check=False) replay = replay_dir / f"replay_{frame:04d}.png" if dst.exists() and replay.exists(): extracted.append((frame, dst, replay)) proc = subprocess.run([ "ffmpeg", "-v", "info", "-i", str(dst), "-i", str(replay), "-lavfi", "ssim", "-f", "null", "-" ], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) match = re.findall(r"All:([0-9.]+)", proc.stdout) if match: ssims.append(float(match[-1])) mean_ssim = statistics.mean(ssims) if ssims else 0.0 # Temporal motion agreement from consecutive-frame SSIM energy. This uses # FFmpeg only, remains compression tolerant, and requires no Python imaging # dependency. motion_corr = 0.0 def image_ssim(a, b): proc = subprocess.run([ "ffmpeg", "-v", "info", "-i", str(a), "-i", str(b), "-lavfi", "ssim", "-f", "null", "-" ], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) matches = re.findall(r"All:([0-9.]+)", proc.stdout) return float(matches[-1]) if matches else 1.0 pm, rm = [], [] for current, previous in zip(extracted[1:], extracted[:-1]): pm.append(1.0 - image_ssim(current[1], previous[1])) rm.append(1.0 - image_ssim(current[2], previous[2])) if len(pm) >= 3 and statistics.pstdev(pm) and statistics.pstdev(rm): ap, ar = statistics.mean(pm), statistics.mean(rm) motion_corr = sum((x - ap) * (y - ar) for x, y in zip(pm, rm)) motion_corr /= math.sqrt( sum((x - ap) ** 2 for x in pm) * sum((y - ar) ** 2 for y in rm) ) motion_corr = max(0.0, motion_corr) tol = config["tolerances"] return { "present": True, "ffprobe": info, "timing_fraction": timing, "ssim_mean": mean_ssim, "ssim_fraction": lerp_similarity( mean_ssim, tol["preview_ssim_zero_credit"], tol["preview_ssim_full_credit"]), "motion_correlation": motion_corr, "motion_fraction": lerp_similarity( motion_corr, tol["preview_motion_correlation_zero_credit"], tol["preview_motion_correlation_full_credit"]), } def main(): parser = argparse.ArgumentParser() parser.add_argument("--scene", required=True) parser.add_argument("--preview") parser.add_argument("--reference-scene", required=True) parser.add_argument("--config", required=True) parser.add_argument("--output", required=True) parser.add_argument("--blender", default=str(DEFAULT_BLENDER)) parser.add_argument("--reference-evidence") args = parser.parse_args() config = load(args.config) out = Path(args.output) out.parent.mkdir(parents=True, exist_ok=True) run_dir = out.parent / (out.stem + "_evidence") run_dir.mkdir(parents=True, exist_ok=True) extract_script = Path(__file__).with_name("evaluator_extract.py") replay_script = Path(__file__).with_name("evaluator_replay.py") ref_evidence_path = (Path(args.reference_evidence) if args.reference_evidence else run_dir / "reference_evidence.json") if not ref_evidence_path.exists(): run([ args.blender, "-b", "--factory-startup", "--disable-autoexec", "-P", str(extract_script), "--", args.reference_scene, str(ref_evidence_path), args.config, "reference" ], run_dir / "reference_extract.log") cand_evidence_path = run_dir / "candidate_evidence.json" run([ args.blender, "-b", "--factory-startup", "--disable-autoexec", "-P", str(extract_script), "--", args.scene, str(cand_evidence_path), args.config, "candidate" ], run_dir / "candidate_extract.log") replay_dir = run_dir / "replay" run([ args.blender, "-b", "--factory-startup", "--disable-autoexec", "-P", str(replay_script), "--", args.scene, str(replay_dir), args.config ], run_dir / "replay.log") ref, cand = load(ref_evidence_path), load(cand_evidence_path) matches, match_costs, coverage = match_components(ref, cand) cand_component_info = {c["id"]: c for c in cand["components"]} candidate_vertices = sum( c["descriptor"]["vertex_count"] for c in cand["components"]) matched_candidate_vertices = sum( cand_component_info[cid]["descriptor"]["vertex_count"] for cid in matches.values()) unmatched_candidate_share = ( 1.0 - matched_candidate_vertices / max(candidate_vertices, 1) ) ref_parts = aggregate_parts(ref, ref) cand_parts_raw = aggregate_parts(cand, ref, matches) alignment = solve_alignment(ref_parts, cand_parts_raw) cand_parts = transform_parts(cand_parts_raw, alignment) h = config["coordinate_conventions"]["reference_character_height"] awd = config["coordinate_conventions"]["observable_axis_weights"] axis_weights = (awd["x"], awd["y_depth"], awd["z"]) tol = config["tolerances"] all_parts = sorted(config["canonical_parts"]) motion_thresholds = ( tol["motion_error_full_credit_height_fraction"] * h, tol["motion_error_zero_credit_height_fraction"] * h, ) motion_defs = [ ("whole_body_trajectory_points", 15.0, all_parts, range(1, 193)), ("step_and_foot_trajectories", 15.0, ["part_001", "part_012", "part_008", "part_010", "part_009", "part_011", "part_002"], range(25, 61)), ("right_arm_and_two_waves", 12.0, ["part_007", "part_013", "part_006"], range(61, 157)), ("head_torso_response", 4.0, ["part_003", "part_002"], range(1, 193)), ("return_to_start", 4.0, all_parts, [192]), ] motion_details = {} motion_score = 0.0 for name, weight, parts, frames in motion_defs: error = mean_motion_error(ref_parts, cand_parts, parts, frames, axis_weights) frac = lerp_error(error, *motion_thresholds) extra = {} if name == "right_arm_and_two_waves": # The broad trajectory error can miss a short absent wave when # averaged over 96 frames. Reserve explicit credit for each wave's # hand-path excursion so two readable waves are operationalized. wave_fractions = [] for wave_name, (start, end) in ( ("wave_1", config["motion_events"]["wave_1"]), ("wave_2", config["motion_events"]["wave_2"])): # The evaluated Shoulder.R shell is the most stable geometric # end-effector landmark for this source: unlike the skinned # hand centroid, it clearly resolves both wave excursions. ref_length = part_path_length( ref_parts, "part_006", start, end, axis_weights) cand_length = part_path_length( cand_parts, "part_006", start, end, axis_weights) ratio = cand_length / max(ref_length, 1e-9) wave_fraction = lerp_error( abs(ratio - 1.0), tol["wave_path_length_ratio_full_credit_error"], tol["wave_path_length_ratio_zero_credit_error"]) wave_fractions.append(wave_fraction) extra[wave_name] = { "reference_path_length": ref_length, "candidate_path_length": cand_length, "ratio": ratio, "fraction": wave_fraction, } points = 8.0 * frac + 2.0 * wave_fractions[0] + 2.0 * wave_fractions[1] else: points = weight * frac motion_score += points motion_details[name] = { "mean_error_units": error, "mean_error_height_fraction": error / h, "fraction": frac, "points": points, "maximum": weight, **extra, } # Centroids cannot establish segment orientation. Reserve 10/50 motion # points for explicit principal-axis pose review under the configured # angular tolerances, so a rotated segment cannot receive full motion # credit from positional agreement alone. centroid_motion_score = motion_score motion_score *= 0.8 for detail in motion_details.values(): detail["points"] *= 0.8 detail["maximum"] *= 0.8 motion_details["pose_orientation_review"] = { "points": 0.0, "maximum": 10.0, "manual_pending_points": 10.0, "status": "MANUAL_RUBRIC_REQUIRED", "angular_tolerances": config["angular_tolerances"], "centroid_only_score_before_orientation_reserve": centroid_motion_score, } match_cost = statistics.mean(match_costs.values()) if match_costs else 1.0 palette_err = palette_error(material_palette(ref), material_palette(cand)) ref_component_info = {c["id"]: c for c in ref["components"]} framing_samples = [] for frame in cand["frames"].values(): weighted, total_weight = 0.0, 0 for rid, cid in matches.items(): if cid in frame: weight = ref_component_info[rid]["descriptor"]["vertex_count"] weighted += frame[cid].get("camera_visible_fraction", 0.0) * weight total_weight += weight if total_weight: framing_samples.append(weighted / total_weight) framing = statistics.mean(framing_samples) if framing_samples else 0.0 visual_details = { "canonical_component_coverage": { "value": coverage, "points": 8 * min(1.0, coverage), "maximum": 8}, "geometry_and_proportions": { "descriptor_error": match_cost, "points": 6 * lerp_error( match_cost, tol["geometry_descriptor_full_credit"], tol["geometry_descriptor_zero_credit"]), "maximum": 6}, "material_palette": { "rgb_error": palette_err, "points": 3 * lerp_error(palette_err, 0.01, 0.25), "maximum": 3}, "render_readability_and_framing": { "mean_visible_vertex_fraction": framing, "points": 3 * lerp_similarity(framing, 0.65, 0.98), "maximum": 3}, } visual_score = sum(v["points"] for v in visual_details.values()) lag, lag_error = best_timing_lag(ref_parts, cand_parts) event_frac = lerp_error( abs(lag), tol["event_timing_full_credit_frames"], tol["event_timing_zero_credit_frames"]) left_drift = max_drift(cand_parts, "part_001", config["contact_windows"]["left_foot_planted"]) right_drift = max_drift(cand_parts, "part_012", config["contact_windows"]["right_foot_planted"]) drift = max(left_drift, right_drift) drift_frac = lerp_error( drift, tol["planted_foot_drift_full_credit_height_fraction"] * h, tol["planted_foot_drift_zero_credit_height_fraction"] * h) accel = max_acceleration(cand_parts) accel_frac = lerp_error( accel, tol["centroid_acceleration_full_credit_height_fraction"] * h, tol["centroid_acceleration_zero_credit_height_fraction"] * h) floor_z = cand.get("detected_floor_z") floor_errors = [] if floor_z is not None: for part, key in (("part_001", "left_foot_planted"), ("part_012", "right_foot_planted")): for start, end in config["contact_windows"][key]: for frame in range(start, end + 1): values = cand_parts_raw.get(str(frame), {}).get(part) if values: floor_errors.append(abs(values["bbox_min"][2] - floor_z)) floor_error = max(floor_errors, default=None) floor_frac = ( lerp_error( floor_error, tol["foot_floor_tolerance_height_fraction"] * h, tol["foot_floor_tolerance_height_fraction"] * h * 5) if floor_error is not None else None ) contact_points = 3.0 * drift_frac contact_manual_pending = 0.0 if floor_frac is None: contact_manual_pending = 2.0 else: contact_points += 2.0 * floor_frac temporal_details = { "event_timing": {"best_lag_frames": lag, "shape_error": lag_error, "points": 5 * event_frac, "maximum": 5}, "foot_contact_and_slide": { "left_drift": left_drift, "right_drift": right_drift, "detected_floor_z": floor_z, "max_abs_planted_foot_floor_gap": floor_error, "points": contact_points, "maximum": 5, "manual_pending_points": contact_manual_pending, "limitation": "Geometry-derived contact-patch drift can conflate " "a controlled foot roll with sliding."}, "smoothness": {"max_centroid_second_difference": accel, "points": 3 * accel_frac, "maximum": 3}, "intersections_and_detachment": { "points": 0.0, "maximum": 2, "manual_pending_points": 2.0, "status": "MANUAL_RUBRIC_REQUIRED", "limitation": "No unrelated proxy is substituted. Reliable arbitrary-" "topology hand/shin intersection and detached-geometry " "judgment requires the documented BVH/visual rubric; " "improving reference contact is never penalized."}, } temporal_score = sum(v["points"] for v in temporal_details.values()) preview = compare_preview( args.preview, replay_dir, config, run_dir / "preview_samples") preview_details = { "duration_frame_rate_frame_count": { "fraction": preview["timing_fraction"], "points": 3 * preview["timing_fraction"], "maximum": 3}, "sampled_perceptual_replay_match": { "ssim": preview["ssim_mean"], "fraction": preview["ssim_fraction"], "points": 8 * preview["ssim_fraction"], "maximum": 8}, "sampled_temporal_motion_agreement": { "correlation": preview["motion_correlation"], "fraction": preview["motion_fraction"], "points": 4 * preview["motion_fraction"], "maximum": 4}, } preview_score = sum(v["points"] for v in preview_details.values()) max_displacement = 0.0 for p in all_parts: if p in cand_parts.get("1", {}): base = cand_parts["1"][p]["centroid"] for f in cand_parts.values(): if p in f: max_displacement = max( max_displacement, vec_len(vec_sub(f[p]["centroid"], base))) driven_component_ids = set(cand.get("armature_driven_component_ids", [])) total_canonical_vertices = sum( component["descriptor"]["vertex_count"] for component in ref["components"] if component.get("canonical_part") ) armature_driven_matched_vertices = sum( ref_component_info[rid]["descriptor"]["vertex_count"] for rid, cid in matches.items() if cid in driven_component_ids ) armature_driven_coverage = ( armature_driven_matched_vertices / max(total_canonical_vertices, 1) ) minimum_coverage = tol["minimum_component_match_coverage"] preview_readable = bool( preview["present"] and preview.get("ffprobe") and preview["ffprobe"].get("streams") and preview["timing_fraction"] == 1.0 ) gates = { "fresh_process_open": cand["fresh_process_open"], "visible_character_geometry": cand["component_count"] > 0, "canonical_component_match_coverage": { "value": coverage, "minimum": minimum_coverage, "pass": coverage >= minimum_coverage, }, "functional_animated_armature_drives_matched_geometry": { "armature_count": cand.get("armature_count", 0), "animated_armature_count": cand.get("animated_armature_count", 0), "matched_geometry_coverage": armature_driven_coverage, "minimum": minimum_coverage, "pass": ( cand.get("armature_count", 0) > 0 and cand.get("animated_armature_count", 0) > 0 and armature_driven_coverage >= minimum_coverage ), }, "evaluated_mesh_motion": { "max_component_displacement": max_displacement, "pass": max_displacement > 0.01 * h, }, "required_preview": { "present": preview["present"], "timing_fraction": preview["timing_fraction"], "pass": preview_readable, }, "timing": { "frame_range": [cand["frame_start"], cand["frame_end"]], "fps": cand["fps"], "pass": cand["frame_start"] <= 1 and cand["frame_end"] >= 192 and abs(cand["fps"] - 24) < 0.01, }, } numeric_valid = all([ gates["fresh_process_open"], gates["visible_character_geometry"], gates["canonical_component_match_coverage"]["pass"], gates["functional_animated_armature_drives_matched_geometry"]["pass"], gates["evaluated_mesh_motion"]["pass"], gates["timing"]["pass"], gates["required_preview"]["pass"], ]) total = motion_score + visual_score + temporal_score + preview_score manual_pending_points = 12.0 + contact_manual_pending total_range = [total, total + manual_pending_points] awarded = (lambda value: value) if numeric_valid else (lambda _value: 0.0) report = { "schema_version": 1, "evaluator_config_version": config["config_version"], "status": config["status"], "scene": str(Path(args.scene).resolve()), "preview": str(Path(args.preview).resolve()) if args.preview else None, "reference_scene": str(Path(args.reference_scene).resolve()), "representation": { "raw_bones_compared": False, "candidate_bone_names_used": False, "connected_component_matches": len(matches), "component_match_coverage": coverage, "unmatched_candidate_geometry_share": unmatched_candidate_share, "helper_dominance_reported": True, "fixed_global_alignment": alignment, "per_frame_alignment": False, }, "validity_gates": gates, "numerical_score_valid": numeric_valid, "scores": { "motion_reproduction": {"score": awarded(motion_score), "maximum": 50.0, "diagnostic_score_before_gate": motion_score, "details": motion_details}, "visual_structural_fidelity": {"score": awarded(visual_score), "maximum": 20.0, "diagnostic_score_before_gate": visual_score, "details": visual_details}, "temporal_contact_quality": {"score": awarded(temporal_score), "maximum": 15.0, "diagnostic_score_before_gate": temporal_score, "details": temporal_details}, "preview_replay_consistency": {"score": awarded(preview_score), "maximum": 15.0, "diagnostic_score_before_gate": preview_score, "details": preview_details}, "total": { "score": awarded(total), "diagnostic_score_when_invalid": None if numeric_valid else total, "manual_pending_range": total_range if numeric_valid else None, "manual_pending_points": manual_pending_points if numeric_valid else 0.0, "maximum": 100.0, }, }, "preview_evidence": preview, "manual_rubric_flags": [ "Professional artistic suitability is never automated.", "Segment pose orientation requires principal-axis review under the configured angular tolerances.", "Hand/shin contact requires visual/BVH review for arbitrary topology.", ], "artifacts": { "candidate_evidence": str(cand_evidence_path), "reference_evidence": str(ref_evidence_path), "replay_directory": str(replay_dir), "logs": str(run_dir), }, } out.write_text(json.dumps(report, indent=2)) if numeric_valid: print(f"[evaluate] {total:.2f}/100 (eligible, " f"+0..{manual_pending_points:g} manual) -> {out}") else: print(f"[evaluate] 0.00/100 (validity gate failed; " f"raw diagnostic {total:.2f}) -> {out}") if __name__ == "__main__": main()