orisys-tech/orisys-sensor-design
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f81418c542ba04ade98d5941f9c691ddfe84ece6
orisys-sensor-design/blender/scripts/probe_render.py
T
ykcheungab 59f48d6d06 initial version of random angle strokes
2026-10-03 20:41:30 +08:00

1055 lines
42 KiB
Python
Raw Blame History

"""Blender headless probe rendering.
blender -b sensors/finger-xynova/r01/source/geometry.blend \\
-P blender/scripts/probe_render.py -- \\
--job <abs/build/probe/finger-xynova/r01/_render_job.json>
Reads the job written by ``orisys_sensor_design.render``, loads the
``probe_frames.npz`` produced by the deform stage, installs ONE emission
material bound to the marker texture over the configured UV map, and renders
every frame of the plan to ``frames/<name>.png`` plus a second reference render
at a different seed (``frames/reference_null.png``) that the tracker uses to
measure its zero-input floor.
THE .BLEND IS NEVER WRITTEN. No ``bpy.ops.wm.save_mainfile`` /
``save_as_mainfile`` call exists in this script; every scene change is in memory
only.
WHY THE VIEW TRANSFORM IS FORCED AND CHECKED
--------------------------------------------
The shipping .blend has ``view_transform = 'AgX'`` and ``cycles.samples = 1024``
with denoising on. AgX is a filmic tonemapper: it silently rewrites every marker
pixel, which corrupts every flow number downstream. This script therefore forces
``view_transform='Standard'``, ``look='None'``, ``exposure=0.0``,
``use_denoising=False``, ``use_adaptive_sampling=False``, ``samples`` from the
job, a fixed ``cycles.seed``, ``interpolation='Closest'`` and dithering off, then
reads the resolved view transform back and REFUSES with exit code 2 if it is not
``Standard``.
DETERMINISM
-----------
Cycles CPU floating-point accumulation order is not fixed, so a boundary pixel
can tip by one 8-bit step even with an identical scene and seed. That is thread
scheduling, not irreproducibility, and because it is content-dependent a strict
byte gate silently disqualifies some marker patterns and biases a pattern
search.
So: after every render this script still strips ONLY the known-volatile ``tEXt``
chunks (``Date``, ``Time``, ``RenderTime``, ``cycles.ViewLayer.*_time``) and
keeps every image-defining chunk byte-for-byte. ``reference_deterministic`` is
then decided from the DECODED PIXELS of two frame-0 renders in the same session:
identical shape, ``max |delta| <= 1`` per 8-bit channel, and a differing-pixel
fraction ``<= 0.05%``. The measured ``max_abs_delta`` and ``differing_fraction``
are recorded in the manifest either way, and a tolerance breach still fails with
exit 1. Raw file bytes are never the comparison: PNG compression and metadata
are not the measurement.
MEASUREMENTS
------------
Two pixel scales are recorded, separately, because confusing them would corrupt
the scorer's units:
* ``mesh_px_per_mm`` is mesh-wide: p05/p50/p95/min/max of (projected mesh edge
length in px) / (3D edge length in mm) over every mesh edge of the reference
frame, projected with ``bpy_extras.object_utils.world_to_camera_view``. For the
shipping revision that reproduces the contract's 13.430 / 10.499 / 38.594.
* ``probe_px_per_mm`` is probe-local and is the gain the scorer must use:
``|centroid_px(step) - centroid_px(0)| / step_mm`` per axis (each axis using
its largest step), summarised by the median over axes. It is anisotropic: on
this sensor ``z`` (surface-normal motion) is much smaller than x/y because
normal motion is largely out of the image plane. That is recorded, not fixed.
``probe_roi_px`` is the reference probe centroid projected to pixels (u=column,
v=row, origin top-left) plus ``radius_mm * probe_px_per_mm.p50``. Each frame
entry carries ``centroid_px`` for THAT frame's deformed probe centroid.
Exit codes: 0 success, 1 scene/geometry error, 2 config/job error,
3 missing/empty required artifact.
"""
# pyright: reportMissingImports=false
from __future__ import annotations
import argparse
import hashlib
import json
import re
import struct
import sys
import time
import zlib
from datetime import datetime, timezone
from pathlib import Path
import bpy
import numpy as np
from bpy_extras.object_utils import world_to_camera_view
from mathutils import Vector
JOB_SCHEMA = "orisys-sensor-design/probe-render-job-v1"
MANIFEST_SCHEMA = "orisys-sensor-design/probe-render-manifest-v1"
MANIFEST_FILENAME = "render_manifest.json"
FRAMES_DIRNAME = "frames"
NULL_FRAME_NAME = "reference_null"
REQUIRED_VIEW_TRANSFORM = "Standard"
DEFAULT_INTERPOLATE = "Closest"
DEFAULT_ENGINE = "CYCLES"
DEFAULT_SAMPLES = 16
DEFAULT_SEED = 0
_PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
#: Volatile PNG text chunks Blender adds per render. They carry wall-clock or
#: render timings and are the ONLY reason two identical renders differ on disk.
#: The set is explicit on purpose: a new volatile chunk must fail the
#: determinism check loudly, not be silently averaged away.
_VOLATILE_TEXT_KEYS = frozenset({"Date", "Time", "RenderTime"})
_VOLATILE_TEXT_PATTERN = re.compile(r"^cycles\.ViewLayer\..*_time$")
#: Determinism gate. Cycles CPU thread-order rounding can move a boundary pixel
#: by one 8-bit step; anything beyond that, or on more than 0.05 % of pixels, is
#: real nondeterminism and must stop the run.
DETERMINISM_MAX_ABS_DELTA = 1
DETERMINISM_MAX_DIFFERING_FRACTION = 0.0005
MESH_PX_PER_MM_BASIS = "per-mesh-edge projected px / 3D mm, all {count} edges"
PROBE_PX_PER_MM_BASIS = (
"probe-local: |centroid_px(step) - centroid_px(0)| / step_mm, "
"per axis at its largest step"
)
_REQUIRED_NPZ_KEYS = (
"base_vertices",
"vertices",
"frame_names",
"frame_axes",
"frame_steps_mm",
"weights",
"origin",
"basis",
"probe_indices",
)
class ConfigError(ValueError):
"""Invalid job/flags (exit code 2)."""
class SceneError(RuntimeError):
"""Missing, inconsistent or unusable scene data (exit code 1)."""
class ArtifactError(RuntimeError):
"""A required artifact is missing or empty (exit code 3)."""
# ----------------------------------------------------------------------
# CLI
# ----------------------------------------------------------------------
def _argv_after_double_dash():
if "--" not in sys.argv:
return []
return sys.argv[sys.argv.index("--") + 1 :]
def parse_args(argv):
parser = argparse.ArgumentParser(prog="probe_render.py")
parser.add_argument(
"--job",
type=Path,
required=True,
help="JSON render job written by orisys-design render",
)
parser.add_argument("--width", type=int, default=None, help="Override render width")
parser.add_argument("--height", type=int, default=None, help="Override render height")
parser.add_argument("--samples", type=int, default=None, help="Override Cycles samples")
parser.add_argument("--seed", type=int, default=None, help="Override Cycles seed")
parser.add_argument("--uv-map", dest="uv_map", default=None, help="Override mesh UV map")
parser.add_argument(
"--texture", type=Path, default=None, help="Override marker texture PNG"
)
return parser.parse_args(argv)
def _load_job(path):
if not path.is_file():
raise ConfigError(f"Render job not found: {path}")
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ConfigError(f"Render job is not valid JSON: {path}: {exc}") from exc
if not isinstance(data, dict):
raise ConfigError(f"Render job must be a JSON object: {path}")
return data
def _as_int(value, label):
if isinstance(value, bool) or not isinstance(value, int):
raise ConfigError(f"{label}: must be an integer (got {value!r})")
return int(value)
def _as_positive_int(value, label):
number = _as_int(value, label)
if number <= 0:
raise ConfigError(f"{label}: must be a positive integer (got {value!r})")
return number
def _as_name(value, label):
if not isinstance(value, str) or not value.strip():
raise ConfigError(f"{label}: must be a non-empty name (got {value!r})")
return value.strip()
def _utc_now():
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def _sha256(path):
digest = hashlib.sha256()
with Path(path).open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _require_non_empty(path, label):
candidate = Path(path)
if not candidate.is_file():
raise ArtifactError(f"{label} not found: {candidate}")
if candidate.stat().st_size == 0:
raise ArtifactError(f"{label} is empty: {candidate}")
return candidate
# ----------------------------------------------------------------------
# PNG metadata
# ----------------------------------------------------------------------
def strip_volatile_text_chunks(path):
"""Drop only the volatile ``tEXt`` chunks; keep every image chunk verbatim.
Returns the number of dropped chunks. The file stays a valid PNG and its
IHDR/IDAT/IEND bytes are untouched, so the pixel data is exactly what Cycles
produced.
"""
data = Path(path).read_bytes()
if data[:8] != _PNG_SIGNATURE:
raise SceneError(f"renderer wrote a file that is not a PNG: {path}")
out = bytearray(_PNG_SIGNATURE)
pos = 8
dropped = 0
while pos + 8 <= len(data):
(length,) = struct.unpack(">I", data[pos : pos + 4])
ctype = data[pos + 4 : pos + 8]
end = pos + 12 + length
if end > len(data):
raise SceneError(f"truncated PNG: {path}")
if ctype == b"tEXt":
key = data[pos + 8 : pos + 8 + length].split(b"\0", 1)[0].decode("latin-1")
if key in _VOLATILE_TEXT_KEYS or _VOLATILE_TEXT_PATTERN.match(key):
dropped += 1
pos = end
continue
out += data[pos:end]
pos = end
if dropped:
Path(path).write_bytes(bytes(out))
return dropped
# ----------------------------------------------------------------------
# PNG pixel comparison (the determinism gate)
# ----------------------------------------------------------------------
def png_dimensions(path):
"""Return ``(width, height)`` from a PNG's IHDR, without decoding pixels."""
data = Path(path).read_bytes()
if data[:8] != _PNG_SIGNATURE:
raise SceneError(f"not a PNG: {path}")
if data[12:16] != b"IHDR":
raise SceneError(f"PNG does not start with IHDR: {path}")
width, height = struct.unpack(">II", data[16:24])
return int(width), int(height)
def decode_png_rgba8(path):
"""Decode an 8-bit RGBA, non-interlaced PNG to a ``(H, W, 4)`` uint8 array.
Only the format this pipeline writes is accepted; anything else fails loudly
rather than being compared under a false assumption. Decoding here (rather
than through Blender's image API) keeps the comparison exact: no colour
management, no float round-trip.
"""
data = Path(path).read_bytes()
if data[:8] != _PNG_SIGNATURE:
raise SceneError(f"not a PNG: {path}")
header = None
idat = bytearray()
pos = 8
while pos + 8 <= len(data):
(length,) = struct.unpack(">I", data[pos : pos + 4])
ctype = data[pos + 4 : pos + 8]
end = pos + 12 + length
if end > len(data):
raise SceneError(f"truncated PNG: {path}")
if ctype == b"IHDR":
header = struct.unpack(">IIBBBBB", data[pos + 8 : pos + 8 + length])
elif ctype == b"IDAT":
idat += data[pos + 8 : pos + 8 + length]
elif ctype == b"IEND":
break
pos = end
if header is None:
raise SceneError(f"PNG has no IHDR: {path}")
width, height, depth, color_type, compression, filter_method, interlace = header
if (depth, color_type, interlace) != (8, 6, 0):
raise SceneError(
"pixel comparison expects an 8-bit RGBA non-interlaced PNG, got "
f"depth={depth} color_type={color_type} interlace={interlace}: {path}"
)
if compression != 0 or filter_method != 0:
raise SceneError(f"unsupported PNG compression/filter method: {path}")
try:
raw = zlib.decompress(bytes(idat))
except zlib.error as exc:
raise SceneError(f"cannot decompress PNG pixel data: {path}: {exc}") from exc
stride = width * 4
if len(raw) != height * (stride + 1):
raise SceneError(
f"PNG pixel data is {len(raw)} bytes, expected {height * (stride + 1)}: {path}"
)
pixels = np.empty((height, width, 4), dtype=np.uint8)
previous = bytearray(stride)
cursor = 0
for y in range(height):
filter_type = raw[cursor]
cursor += 1
line = bytearray(raw[cursor : cursor + stride])
cursor += stride
if filter_type == 0:
pass
elif filter_type == 1:
for i in range(4, stride):
line[i] = (line[i] + line[i - 4]) & 0xFF
elif filter_type == 2:
for i in range(stride):
line[i] = (line[i] + previous[i]) & 0xFF
elif filter_type == 3:
for i in range(stride):
left = line[i - 4] if i >= 4 else 0
line[i] = (line[i] + ((left + previous[i]) >> 1)) & 0xFF
elif filter_type == 4:
for i in range(stride):
a = line[i - 4] if i >= 4 else 0
b = previous[i]
c = previous[i - 4] if i >= 4 else 0
p = a + b - c
pa = abs(p - a)
pb = abs(p - b)
pc = abs(p - c)
if pa <= pb and pa <= pc:
pred = a
elif pb <= pc:
pred = b
else:
pred = c
line[i] = (line[i] + pred) & 0xFF
else:
raise SceneError(f"unknown PNG filter type {filter_type}: {path}")
pixels[y] = np.frombuffer(bytes(line), dtype=np.uint8).reshape(width, 4)
previous = line
return pixels
def compare_reference_renders(path_a, path_b):
"""Measure how far two frame-0 renders differ, in decoded pixels.
Returns a result dict with ``passed``, ``max_abs_delta``,
``differing_pixels``, ``total_pixels``, ``differing_fraction`` and the
tolerances applied. A byte-identical pair short-circuits to zero deltas; any
other pair is decoded and compared channel by channel. Shape mismatch, or any
unreadable/unsupported PNG, raises :class:`SceneError`.
"""
bytes_a = Path(path_a).read_bytes()
bytes_b = Path(path_b).read_bytes()
shape_a = png_dimensions(path_a)
shape_b = png_dimensions(path_b)
if shape_a != shape_b:
raise SceneError(
f"frame-0 renders have different shapes: {shape_a} vs {shape_b} "
f"({path_a}, {path_b})"
)
height, width = shape_a[1], shape_a[0]
total_pixels = height * width
if bytes_a == bytes_b:
max_abs_delta = 0
differing_pixels = 0
else:
pixels_a = decode_png_rgba8(path_a)
pixels_b = decode_png_rgba8(path_b)
if pixels_a.shape != pixels_b.shape:
raise SceneError(
f"decoded frame-0 renders have different shapes: "
f"{pixels_a.shape} vs {pixels_b.shape}"
)
delta = np.abs(pixels_a.astype(np.int16) - pixels_b.astype(np.int16))
max_abs_delta = int(delta.max()) if delta.size else 0
differing_pixels = int((delta.max(axis=2) > 0).sum())
differing_fraction = differing_pixels / total_pixels if total_pixels else 0.0
passed = (
max_abs_delta <= DETERMINISM_MAX_ABS_DELTA
and differing_fraction <= DETERMINISM_MAX_DIFFERING_FRACTION
)
return {
"passed": bool(passed),
"max_abs_delta": max_abs_delta,
"differing_pixels": differing_pixels,
"total_pixels": int(total_pixels),
"differing_fraction": float(differing_fraction),
"max_abs_delta_tolerance": DETERMINISM_MAX_ABS_DELTA,
"differing_fraction_tolerance": DETERMINISM_MAX_DIFFERING_FRACTION,
"shape": [int(width), int(height)],
}
# ----------------------------------------------------------------------
# Scene lookup
# ----------------------------------------------------------------------
def _require_object(name, object_type, label):
obj = bpy.data.objects.get(name)
if obj is None:
raise SceneError(f"Scene has no {label} named {name!r}")
if obj.type != object_type:
raise SceneError(
f"Object {name!r} is {obj.type}, expected {object_type} for the {label}"
)
return obj
def _require_uv_layer(obj, name):
for layer in obj.data.uv_layers:
if layer.name == name:
if len(layer.data) == 0:
raise SceneError(f"UV map {name!r} on {obj.name!r} has no loop data")
return layer
available = ", ".join(sorted(layer.name for layer in obj.data.uv_layers)) or "none"
raise SceneError(
f"Mesh {obj.name!r} has no UV map named {name!r} (available: {available})"
)
class Projector:
"""World coordinates -> render pixels (u=column, v=row, origin top-left)."""
def __init__(self, scene, camera, width, height):
self.scene = scene
self.camera = camera
self.width = float(width)
self.height = float(height)
def project(self, world_co):
u, v, depth = world_to_camera_view(
self.scene, self.camera, Vector((float(world_co[0]), float(world_co[1]), float(world_co[2])))
)
return (u * self.width, (1.0 - v) * self.height, depth)
def project_many(self, world_coords):
return np.array([self.project(p)[:2] for p in world_coords], dtype=np.float64)
def _mesh_px_per_mm_stats(reference_coords, edges, projector, matrix_world):
"""Mesh-wide sampling scale: projected-edge px / 3D-edge mm over every edge.
This describes how the whole surface is sampled by the camera. It is NOT the
gain that converts the probe's measured flow into millimetres; see
:func:`_probe_px_per_mm` for that. Measured on the reference frame from the
actual projection; never assumed.
"""
if len(edges) == 0:
raise SceneError("Mesh has no edges; cannot measure the projection scale")
world = np.array(
[(matrix_world @ Vector(tuple(c))).to_tuple() for c in reference_coords],
dtype=np.float64,
)
px = projector.project_many(world)
d3 = np.linalg.norm(reference_coords[edges[:, 0]] - reference_coords[edges[:, 1]], axis=1)
d2 = np.linalg.norm(px[edges[:, 0]] - px[edges[:, 1]], axis=1)
usable = d3 > 0.0
if not np.any(usable):
raise SceneError("Every mesh edge has zero length; cannot measure the projection scale")
ratios = d2[usable] / d3[usable]
stats = {
"min": float(ratios.min()),
"p05": float(np.percentile(ratios, 5)),
"p50": float(np.percentile(ratios, 50)),
"p95": float(np.percentile(ratios, 95)),
"max": float(ratios.max()),
}
stats["basis"] = MESH_PX_PER_MM_BASIS.format(count=int(len(edges)))
if not all(np.isfinite(value) for key, value in stats.items() if key != "basis"):
raise SceneError(f"Non-finite projection scale measured: {stats}")
return stats
def _probe_px_per_mm(frames, centroids):
"""Probe-local scale per axis: |centroid_px(step) - centroid_px(0)| / step_mm.
This is the gain the scorer must use to convert the probe's measured flow to
millimetres. It is anisotropic; on this sensor ``z`` (surface-normal motion)
is much smaller than x/y because normal motion is largely out of the image
plane. That is a real property, recorded rather than corrected.
Each axis uses its largest step, which is the most numerically robust finite
difference of the same locally-linear map (all 6 probe vertices move by the
same vector, so the centroid tracks it exactly).
"""
reference = np.asarray(centroids[0], dtype=np.float64)
per_axis: dict[str, float] = {}
for axis in ("x", "y", "z"):
positions = [i for i, frame in enumerate(frames) if frame["axis"] == axis]
if not positions:
continue
best = max(positions, key=lambda i: frames[i]["step_mm"])
step_mm = float(frames[best]["step_mm"])
if step_mm <= 0.0:
raise SceneError(f"axis {axis!r} has no positive step; cannot measure its scale")
delta = np.asarray(centroids[best], dtype=np.float64) - reference
per_axis[axis] = float(np.linalg.norm(delta) / step_mm)
if not per_axis:
raise SceneError("Frame plan has no axis frames; cannot measure the probe-local scale")
stats: dict[str, object] = {"p50": float(np.median(list(per_axis.values())))}
for axis in ("x", "y", "z"):
if axis in per_axis:
stats[axis] = per_axis[axis]
stats["basis"] = PROBE_PX_PER_MM_BASIS
if not all(np.isfinite(value) for value in per_axis.values()):
raise SceneError(f"Non-finite probe-local scale measured: {per_axis}")
return stats
# ----------------------------------------------------------------------
# Config resolution
# ----------------------------------------------------------------------
def _resolve_config(job, args):
schema = job.get("schema")
if schema != JOB_SCHEMA:
raise ConfigError(f"Job schema is {schema!r}, expected {JOB_SCHEMA!r}")
settings = job.get("settings")
if not isinstance(settings, dict):
raise ConfigError("Job 'settings' must be a mapping")
objects = job.get("objects")
if not isinstance(objects, dict):
raise ConfigError("Job 'objects' must be a mapping")
artifacts = job.get("artifacts")
if not isinstance(artifacts, dict):
raise ConfigError("Job 'artifacts' must be a mapping")
output_dir = job.get("output_dir")
if not isinstance(output_dir, str) or not output_dir:
raise ConfigError("Job 'output_dir' must be a non-empty path")
frames_npz = artifacts.get("frames_npz")
if not isinstance(frames_npz, str) or not frames_npz:
raise ConfigError("Job 'artifacts.frames_npz' must be a non-empty path")
raw_frames = job.get("frames")
if not isinstance(raw_frames, list) or not raw_frames:
raise ConfigError("Job 'frames' must be a non-empty list")
frames = []
for position, entry in enumerate(raw_frames):
if not isinstance(entry, dict):
raise ConfigError(f"Job frames[{position}] must be a mapping")
frame = {
"index": _as_int(entry.get("index"), f"frames[{position}].index"),
"name": _as_name(entry.get("name"), f"frames[{position}].name"),
"axis": entry.get("axis"),
"step_mm": entry.get("step_mm"),
}
if frame["axis"] is not None:
frame["axis"] = _as_name(frame["axis"], f"frames[{position}].axis")
if not isinstance(frame["step_mm"], (int, float)) or isinstance(frame["step_mm"], bool):
raise ConfigError(
f"frames[{position}].step_mm must be a number (got {frame['step_mm']!r})"
)
frame["step_mm"] = float(frame["step_mm"])
frames.append(frame)
names = [frame["name"] for frame in frames]
if len(set(names)) != len(names):
raise ConfigError("Job 'frames' contains duplicate names")
if frames[0]["index"] != 0:
raise ConfigError(f"Job frames[0].index must be 0 (got {frames[0]['index']})")
null_raw = job.get("null_frame") or {}
if not isinstance(null_raw, dict):
raise ConfigError("Job 'null_frame' must be a mapping")
null_name = _as_name(null_raw.get("name", NULL_FRAME_NAME), "null_frame.name")
# --- the hard, non-negotiable render settings ---
requested_view_transform = _as_name(
settings.get("view_transform", REQUIRED_VIEW_TRANSFORM), "settings.view_transform"
)
if requested_view_transform != REQUIRED_VIEW_TRANSFORM:
raise ConfigError(
"settings.view_transform must be "
f"{REQUIRED_VIEW_TRANSFORM!r} (got {requested_view_transform!r}); AgX/Filmic "
"silently rewrites marker pixel values and invalidates the measurement"
)
config = {
"output_dir": Path(output_dir).resolve(),
"frames_npz": Path(frames_npz).resolve(),
"source_blend": Path(str(job.get("source_blend", ""))).resolve(),
"mesh": _as_name(objects.get("mesh"), "objects.mesh"),
"camera": _as_name(objects.get("camera"), "objects.camera"),
"frames": frames,
"null_name": null_name,
"null_seed": _as_int(
null_raw.get("seed", _as_int(settings.get("seed", DEFAULT_SEED), "settings.seed") + 1),
"null_frame.seed",
),
"uv_map": _as_name(args.uv_map or settings.get("uv_map"), "settings.uv_map"),
"texture": Path(args.texture or settings.get("texture") or "").expanduser().resolve(),
"width": _as_positive_int(
args.width if args.width is not None else settings.get("width"), "settings.width"
),
"height": _as_positive_int(
args.height if args.height is not None else settings.get("height"), "settings.height"
),
"engine": _as_name(settings.get("engine", DEFAULT_ENGINE), "settings.engine"),
"samples": _as_positive_int(
args.samples if args.samples is not None else settings.get("samples", DEFAULT_SAMPLES),
"settings.samples",
),
"denoise": settings.get("denoise", False),
"view_transform": requested_view_transform,
"interpolate": _as_name(
settings.get("interpolate", DEFAULT_INTERPOLATE), "settings.interpolate"
),
"seed": _as_int(
args.seed if args.seed is not None else settings.get("seed", DEFAULT_SEED),
"settings.seed",
),
"radius_mm": settings.get("radius_mm"),
}
if not isinstance(config["denoise"], bool):
raise ConfigError(f"settings.denoise must be true or false (got {config['denoise']!r})")
if not isinstance(config["radius_mm"], (int, float)) or isinstance(config["radius_mm"], bool):
raise ConfigError(
"settings.radius_mm must be a positive number (got "
f"{config['radius_mm']!r}); probe_roi_px.radius = radius_mm * px_per_mm"
)
config["radius_mm"] = float(config["radius_mm"])
if config["radius_mm"] <= 0.0:
raise ConfigError(f"settings.radius_mm must be > 0 (got {config['radius_mm']!r})")
if args.texture is None and (not str(config["texture"]) or str(config["texture"]) == "."):
raise ConfigError("settings.texture must be a non-empty path")
return config
# ----------------------------------------------------------------------
# Scene preparation
# ----------------------------------------------------------------------
def _apply_scene_settings(scene, config):
"""Force the contract's render settings and refuse a non-Standard transform."""
try:
scene.view_settings.view_transform = REQUIRED_VIEW_TRANSFORM
except TypeError as exc:
raise ConfigError(
f"This Blender build cannot resolve view transform {REQUIRED_VIEW_TRANSFORM!r}: {exc}"
) from exc
try:
scene.view_settings.look = "None"
except TypeError:
# Blender 4.x renames looks; the transform is what matters and 'None' is
# valid there. A build without 'None' keeps its default look, recorded
# as applied in the manifest.
pass
scene.view_settings.exposure = 0.0
resolved_view_transform = scene.view_settings.view_transform
if resolved_view_transform != REQUIRED_VIEW_TRANSFORM:
raise ConfigError(
"resolved view transform is "
f"{resolved_view_transform!r}, expected {REQUIRED_VIEW_TRANSFORM!r}; refusing to "
"render marker values through a filmic transform"
)
scene.render.engine = config["engine"]
scene.render.resolution_x = config["width"]
scene.render.resolution_y = config["height"]
scene.render.resolution_percentage = 100
scene.render.film_transparent = False
scene.render.use_file_extension = True
# 8-bit output must not be dithered: dithering is random and makes the
# saved bytes irreproducible.
scene.render.dither_intensity = 0.0
scene.render.image_settings.file_format = "PNG"
scene.render.image_settings.color_mode = "RGBA"
# Stamp metadata is pure noise for a measurement; volatile cycles timing
# chunks are stripped after each render regardless.
for attribute in dir(scene.render):
if attribute.startswith("use_stamp_"):
try:
setattr(scene.render, attribute, False)
except (AttributeError, TypeError):
pass
cycles = getattr(scene, "cycles", None)
if cycles is not None:
cycles.samples = config["samples"]
# Denoising and adaptive sampling both change pixel values and would
# make the sweep depend on a stochastic estimator.
cycles.use_denoising = False
cycles.use_adaptive_sampling = False
cycles.seed = config["seed"]
if hasattr(cycles, "use_animated_seed"):
cycles.use_animated_seed = False
elif config["engine"] == "CYCLES":
raise SceneError("Scene has no Cycles settings while the engine is CYCLES")
return {
"engine": config["engine"],
"samples": config["samples"],
"denoise": False,
"view_transform": resolved_view_transform,
"look": scene.view_settings.look,
"exposure": float(scene.view_settings.exposure),
"interpolate": config["interpolate"],
"seed": config["seed"],
}
def _install_marker_material(mesh_obj, image, uv_map, interpolate):
"""Install ONE emission material driven by the marker image over ``uv_map``."""
material = bpy.data.materials.new("probe_marker_emission")
material.use_nodes = True
node_tree = material.node_tree
for node in list(node_tree.nodes):
node_tree.nodes.remove(node)
output = node_tree.nodes.new("ShaderNodeOutputMaterial")
emission = node_tree.nodes.new("ShaderNodeEmission")
texture = node_tree.nodes.new("ShaderNodeTexImage")
uv_node = node_tree.nodes.new("ShaderNodeUVMap")
uv_node.uv_map = uv_map
texture.image = image
try:
texture.interpolation = interpolate
except TypeError as exc:
raise ConfigError(f"settings.interpolate {interpolate!r} is not a valid mode") from exc
emission.inputs["Strength"].default_value = 1.0
node_tree.links.new(uv_node.outputs["UV"], texture.inputs["Vector"])
node_tree.links.new(texture.outputs["Color"], emission.inputs["Color"])
node_tree.links.new(emission.outputs["Emission"], output.inputs["Surface"])
mesh = mesh_obj.data
mesh.materials.clear()
mesh.materials.append(material)
for polygon in mesh.polygons:
polygon.material_index = 0
return material
def _write_vertices(mesh, coords):
flat = np.ascontiguousarray(coords, dtype=np.float32).ravel()
mesh.vertices.foreach_set("co", flat)
mesh.update()
def _render_to(scene, path):
Path(path).parent.mkdir(parents=True, exist_ok=True)
scene.render.filepath = str(path)
bpy.ops.render.render(write_still=True)
strip_volatile_text_chunks(path)
return Path(path)
# ----------------------------------------------------------------------
# Main
# ----------------------------------------------------------------------
def main():
args = parse_args(_argv_after_double_dash())
try:
job = _load_job(args.job)
config = _resolve_config(job, args)
except (ConfigError, json.JSONDecodeError, OSError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
scene = bpy.context.scene
try:
settings_applied = _apply_scene_settings(scene, config)
except (ConfigError, SceneError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2 if isinstance(exc, ConfigError) else 1
try:
npz_path = _require_non_empty(config["frames_npz"], "probe_frames.npz")
texture_path = _require_non_empty(config["texture"], "marker texture")
mesh_obj = _require_object(config["mesh"], "MESH", "mesh")
camera_obj = _require_object(config["camera"], "CAMERA", "camera")
_require_uv_layer(mesh_obj, config["uv_map"])
except (ArtifactError, SceneError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 3 if isinstance(exc, ArtifactError) else 1
# The scene may be saved with a production-view camera active. Projection,
# rendered frames, and the reported ROI must all use the job's sensor camera.
scene.camera = camera_obj
started = time.time()
mesh = mesh_obj.data
try:
with np.load(str(npz_path), allow_pickle=False) as data:
missing = [key for key in _REQUIRED_NPZ_KEYS if key not in data.files]
if missing:
raise SceneError(
f"{npz_path} is missing key(s) {', '.join(missing)} "
f"(present: {', '.join(sorted(data.files))})"
)
base_vertices = np.asarray(data["base_vertices"], dtype=np.float64)
vertices = np.asarray(data["vertices"], dtype=np.float64)
npz_names = [str(name) for name in np.asarray(data["frame_names"]).tolist()]
probe_indices = np.asarray(data["probe_indices"], dtype=np.int64).ravel()
except SceneError:
raise
except Exception as exc: # noqa: BLE001 - surface any npz read failure
print(f"ERROR: cannot read {npz_path}: {exc}", file=sys.stderr)
return 3
plan_names = [frame["name"] for frame in config["frames"]]
try:
if vertices.ndim != 3 or vertices.shape[2] != 3:
raise SceneError(f"probe_frames.npz 'vertices' must be (F, V, 3), got {vertices.shape}")
if base_vertices.ndim != 2 or base_vertices.shape[1] != 3:
raise SceneError(
f"probe_frames.npz 'base_vertices' must be (V, 3), got {base_vertices.shape}"
)
if len(vertices) != len(npz_names):
raise SceneError(
f"probe_frames.npz has {len(vertices)} frames but {len(npz_names)} frame names"
)
if len(npz_names) != len(plan_names):
raise SceneError(
f"probe_frames.npz has {len(npz_names)} frames but the job plan has "
f"{len(plan_names)}"
)
mismatched = [
(index, plan, npz)
for index, (plan, npz) in enumerate(zip(plan_names, npz_names))
if plan != npz
]
if mismatched:
index, plan, npz = mismatched[0]
raise SceneError(
f"frame order disagrees at index {index}: job plan {plan!r} vs "
f"probe_frames.npz {npz!r}"
)
if vertices.shape[1] != len(mesh.vertices):
raise SceneError(
f"probe_frames.npz has {vertices.shape[1]} vertices but mesh {mesh_obj.name!r} "
f"has {len(mesh.vertices)}"
)
if base_vertices.shape[0] != len(mesh.vertices):
raise SceneError(
f"probe_frames.npz 'base_vertices' has {base_vertices.shape[0]} rows but mesh "
f"{mesh_obj.name!r} has {len(mesh.vertices)} vertices"
)
if probe_indices.size == 0:
raise SceneError("probe_frames.npz 'probe_indices' is empty")
if probe_indices.min() < 0 or probe_indices.max() >= len(mesh.vertices):
raise SceneError(
f"probe_frames.npz 'probe_indices' out of range 0..{len(mesh.vertices) - 1}"
)
except SceneError as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
try:
image = bpy.data.images.load(str(texture_path), check_existing=False)
except RuntimeError as exc:
print(f"ERROR: cannot load marker texture {texture_path}: {exc}", file=sys.stderr)
return 3
if image.size[0] <= 0 or image.size[1] <= 0:
print(f"ERROR: marker texture has no pixels: {texture_path}", file=sys.stderr)
return 3
image.colorspace_settings.name = "sRGB"
output_dir = config["output_dir"]
frames_dir = output_dir / FRAMES_DIRNAME
frames_dir.mkdir(parents=True, exist_ok=True)
_install_marker_material(mesh_obj, image, config["uv_map"], config["interpolate"])
projector = Projector(scene, camera_obj, config["width"], config["height"])
matrix_world = mesh_obj.matrix_world
edges = np.array(
[tuple(edge.vertices) for edge in mesh.edges], dtype=np.int64
).reshape(-1, 2)
def project_local(local_co):
return projector.project(matrix_world @ Vector(tuple(float(x) for x in local_co)))
try:
mesh_px_per_mm = _mesh_px_per_mm_stats(vertices[0], edges, projector, matrix_world)
# Every probe vertex moves by the same vector, so the centroid of each
# frame's probe members is exactly origin + applied displacement.
centroids = [
(float(project_local(vertices[i][probe_indices].mean(axis=0))[0]),
float(project_local(vertices[i][probe_indices].mean(axis=0))[1]))
for i in range(len(config["frames"]))
]
probe_px_per_mm = _probe_px_per_mm(config["frames"], centroids)
except SceneError as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
roi_u, roi_v = centroids[0]
roi_radius = config["radius_mm"] * float(probe_px_per_mm["p50"])
# --- frames, in plan order ---
frame_entries = []
determinism = None
reference_png_sha = None
for position, frame in enumerate(config["frames"]):
centroid = centroids[position]
_write_vertices(mesh, vertices[position])
png = _render_to(scene, frames_dir / f"{frame['name']}.png")
digest = _sha256(png)
if position == 0:
reference_png_sha = digest
# Same session, same seed, same scene: the DECODED PIXELS must agree
# within one 8-bit step on at most 0.05% of pixels. Raw file bytes are
# not the comparison; Cycles thread ordering can flip a boundary pixel.
repeat = _render_to(scene, output_dir / "checks" / f"{frame['name']}_repeat.png")
determinism = compare_reference_renders(png, repeat)
if not determinism["passed"]:
print(
"ERROR: frame 0 rendered twice with the same seed is not reproducible: "
f"max_abs_delta={determinism['max_abs_delta']} "
f"(tolerance {determinism['max_abs_delta_tolerance']}), "
f"differing_fraction={determinism['differing_fraction']:.8f} "
f"(tolerance {determinism['differing_fraction_tolerance']}), "
f"differing_pixels={determinism['differing_pixels']}/"
f"{determinism['total_pixels']}; every later number would be invalid",
file=sys.stderr,
)
return 1
frame_entries.append(
{
"index": frame["index"],
"name": frame["name"],
"axis": frame["axis"],
"step_mm": frame["step_mm"],
"png": f"{FRAMES_DIRNAME}/{frame['name']}.png",
"sha256": digest,
"width": config["width"],
"height": config["height"],
"centroid_px": [centroid[0], centroid[1]],
}
)
print(
f"[{position + 1}/{len(config['frames'])}] {frame['name']} -> {png.name} "
f"({png.stat().st_size} bytes, centroid {centroid[0]:.4f},{centroid[1]:.4f})",
flush=True,
)
if reference_png_sha is None:
print("ERROR: the job plan has no reference frame", file=sys.stderr)
return 1
if abs(centroids[0][0] - roi_u) > 1e-9 or abs(centroids[0][1] - roi_v) > 1e-9:
print(
f"ERROR: reference centroid_px {centroids[0]} does not match probe_roi_px "
f"centre ({roi_u}, {roi_v})",
file=sys.stderr,
)
return 1
# --- null frame: the reference again at a different seed ---
_write_vertices(mesh, vertices[0])
cycles = getattr(scene, "cycles", None)
if cycles is not None:
cycles.seed = config["null_seed"]
null_png = _render_to(scene, frames_dir / f"{config['null_name']}.png")
null_sha = _sha256(null_png)
print(
f"null frame {config['null_name']} seed={config['null_seed']} -> {null_png.name} "
f"({null_png.stat().st_size} bytes)",
flush=True,
)
if config["source_blend"].is_file():
source_sha = _sha256(config["source_blend"])
else:
source_sha = ""
manifest = {
"schema": MANIFEST_SCHEMA,
"generated_at": _utc_now(),
"source_blend": {
"path": str(config["source_blend"]),
"sha256": source_sha,
},
"texture": {
"path": str(texture_path),
"sha256": _sha256(texture_path),
},
"uv_map": config["uv_map"],
"camera": camera_obj.name,
"width": config["width"],
"height": config["height"],
"settings_applied": settings_applied,
"mesh_px_per_mm": mesh_px_per_mm,
"probe_px_per_mm": probe_px_per_mm,
"probe_roi_px": {
"u": float(roi_u),
"v": float(roi_v),
"radius": float(roi_radius),
},
"reference_index": 0,
"reference_deterministic": bool(determinism["passed"]),
"reference_determinism": determinism,
"frames": frame_entries,
"null": {
"name": config["null_name"],
"seed": config["null_seed"],
"png": f"{FRAMES_DIRNAME}/{config['null_name']}.png",
"sha256": null_sha,
},
}
manifest_path = output_dir / MANIFEST_FILENAME
manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
print(
f"Render complete: {len(frame_entries)} frames + 1 null in "
f"{time.time() - started:.1f}s"
)
print(
" mesh_px_per_mm min/p05/p50/p95/max "
f"{mesh_px_per_mm['min']:.4f} / {mesh_px_per_mm['p05']:.4f} / "
f"{mesh_px_per_mm['p50']:.4f} / {mesh_px_per_mm['p95']:.4f} / "
f"{mesh_px_per_mm['max']:.4f}"
)
print(
" probe_px_per_mm p50/x/y/z "
f"{probe_px_per_mm['p50']:.4f} / {float(probe_px_per_mm.get('x', float('nan'))):.4f} / "
f"{float(probe_px_per_mm.get('y', float('nan'))):.4f} / "
f"{float(probe_px_per_mm.get('z', float('nan'))):.4f}"
)
print(f" probe_roi_px u/v/radius {roi_u:.4f} / {roi_v:.4f} / {roi_radius:.4f}")
print(
f" reference_deterministic {determinism['passed']} "
f"(max_abs_delta {determinism['max_abs_delta']}/"
f"{determinism['max_abs_delta_tolerance']}, differing_fraction "
f"{determinism['differing_fraction']:.8f}/"
f"{determinism['differing_fraction_tolerance']})"
)
print(f" reference sha256 {reference_png_sha}")
print(f" manifest {manifest_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink