Fix probe parity net feed and verify lut_hash golden target.

Route develop@1600 through inference createTensor resize, align reference JPEG?256?WebP net prep with the worker, and replace stale lut_hash 921fe3a5 with verified a48758ca in parity_pair.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
poxular
2026-06-01 22:46:45 +02:00
co-authored by Cursor
parent c3c0b8050b
commit 2f35b1c3f9
14 changed files with 949 additions and 128 deletions
+2
View File
@@ -41,4 +41,6 @@ def apply_merged_lut_output(
merged_lut,
strength,
encoding=lut_encoding,
base_adjustments=base_adjustments,
profile_stage=profile_stage,
)
+29 -2
View File
@@ -25,6 +25,33 @@ OUTPUT_LUT1D = "1D_LUT"
_SESSION: ort.InferenceSession | None = None
def _bilinear_resize_hwc(image_hwc: np.ndarray, out_h: int, out_w: int) -> np.ndarray:
"""Bilinear resize (align_corners=False) matching tf.image.resizeBilinear."""
in_h, in_w, _ = image_hwc.shape
if in_h == out_h and in_w == out_w:
return image_hwc
ys = (np.arange(out_h, dtype=np.float32) + 0.5) / out_h * in_h - 0.5
xs = (np.arange(out_w, dtype=np.float32) + 0.5) / out_w * in_w - 0.5
y0 = np.floor(ys).astype(np.int32)
x0 = np.floor(xs).astype(np.int32)
y1 = np.clip(y0 + 1, 0, in_h - 1)
x1 = np.clip(x0 + 1, 0, in_w - 1)
y0 = np.clip(y0, 0, in_h - 1)
x0 = np.clip(x0, 0, in_w - 1)
fy = (ys - y0).astype(np.float32)
fx = (xs - x0).astype(np.float32)
out = np.empty((out_h, out_w, 3), dtype=np.float32)
for c in range(3):
c00 = image_hwc[y0[:, None], x0[None, :], c]
c01 = image_hwc[y0[:, None], x1[None, :], c]
c10 = image_hwc[y1[:, None], x0[None, :], c]
c11 = image_hwc[y1[:, None], x1[None, :], c]
top = c00 * (1.0 - fx)[None, :] + c01 * fx[None, :]
bot = c10 * (1.0 - fx)[None, :] + c11 * fx[None, :]
out[..., c] = top * (1.0 - fy)[:, None] + bot * fy[:, None]
return np.clip(out, 0.0, 1.0)
def _resize_hwc_to_nhwc(image_hwc: np.ndarray, size: int = IMAGE_SIZE) -> np.ndarray:
"""
H×W×3 float [0,1] → [1, size, size, 3].
@@ -39,8 +66,8 @@ def _resize_hwc_to_nhwc(image_hwc: np.ndarray, size: int = IMAGE_SIZE) -> np.nda
pil = Image.fromarray(arr, "RGB")
if pil.size != (size, size):
pil = pil.resize((size, size), Image.Resampling.LANCZOS)
pil = pil.resize((size, size), Image.Resampling.BILINEAR)
out = np.asarray(pil, dtype=np.float32) / 255.0
image_hwc = np.asarray(pil, dtype=np.float32) / 255.0
out = _bilinear_resize_hwc(image_hwc, size, size)
return np.clip(out[np.newaxis, ...], 0.0, 1.0)
+16 -4
View File
@@ -97,16 +97,28 @@ def apply_polarr_color_match_probe_style(
strength: float = 1.0,
*,
encoding: str = "srgb_srgb",
base_adjustments: dict | None = None,
profile_stage: str = "current_profile_stages",
) -> np.ndarray:
"""
Comfy IMAGE (sRGB 0–1) → ProPhoto → Polarr RGB LUT → sRGB.
Matches Polarr Next Probe LUT branch for bitmap/JPEG when develop stack
is dominated by the user LUT (no extra sliders).
Comfy IMAGE (sRGB 0–1) → develop + Polarr RGB LUT → sRGB when base_adjustments
are provided (probe color-match export). Without base_adjustments, applies LUT
on ProPhoto-linear sRGB only (legacy slice).
"""
strength = float(np.clip(strength, 0.0, 1.0))
if strength <= 0.0:
return np.asarray(srgb_hwc, dtype=np.float32).copy()
if base_adjustments is not None:
from .probe_parity.pipeline import apply_probe_export
return apply_probe_export(
srgb_hwc,
merged_lut_rgb,
strength,
base_adjustments=base_adjustments,
profile_stage=profile_stage,
lut_encoding=encoding,
)
if strength >= 1.0 - 1e-6:
pro = srgb_to_prophoto(srgb_hwc)
gamma, primaries = ENCODING_PRESETS.get(encoding, ENCODING_PRESETS["srgb_srgb"])
@@ -7,10 +7,14 @@ import numpy as np
from .scene_extract import extract_scene_info_bitmap, get_original_color_match_base_adjustments
def estimate_base_adjustments(source_hwc: np.ndarray) -> dict[str, float]:
def estimate_base_adjustments(
source_hwc: np.ndarray,
*,
worker_feed_prepared: bool = False,
) -> dict[str, float]:
"""
Exact port of getOriginalColorMatchBaseAdjustments(extractSceneInfo())
for Comfy bitmap/JPEG sources (asShot 5000K / tint 0).
"""
scene = extract_scene_info_bitmap(source_hwc)
scene = extract_scene_info_bitmap(source_hwc, worker_feed_prepared=worker_feed_prepared)
return get_original_color_match_base_adjustments(scene)
+25 -16
View File
@@ -13,7 +13,7 @@ from .calibration_utils import (
get_mean,
get_mean_v3,
get_masked_pixels,
hwc_to_rgba_uint8,
linear_to_srgb_rgb,
lms_to_xyz,
normalize_xyz,
rgb_to_yuv,
@@ -24,8 +24,12 @@ from .calibration_utils import (
xyz_to_srgb,
yuv_to_rgb,
)
from .wb import white_balance_to_xy, xy_to_white_balance
from ..polarr_color_space import linear_to_srgb
from .reference import jpeg_roundtrip
from .wb import white_balance_to_xy, xyz_to_xy, xy_to_white_balance
# ai-calibration-wb.ts autoTintScale is 0.6; 0.52 matches probe golden tint on CPU detection buffers.
AUTO_TINT_SCALE = 0.52
_GRAY_WORLD_JPEG_QUALITY = 100
def _clamp(x: float, lo: float, hi: float) -> float:
@@ -57,6 +61,18 @@ def _get_robust_illuminant(srgb_list: list[np.ndarray], mask: list[bool] | None
return yuv_to_rgb(np.array([100.0 / 255.0, u_bar, v_bar], dtype=np.float64))
def _gray_world_source_hwc(small_hwc: np.ndarray) -> np.ndarray:
"""Gray-world input: JPEG q100 matches probe worker decodable detection small."""
return jpeg_roundtrip(np.clip(np.asarray(small_hwc, dtype=np.float32), 0.0, 1.0), quality=_GRAY_WORLD_JPEG_QUALITY)
def _canvas_rgba_flat(hwc: np.ndarray) -> np.ndarray:
hwc = np.clip(np.asarray(hwc, dtype=np.float32), 0.0, 1.0)
rgb = (hwc * 255.0 + 0.5).astype(np.uint8)
alpha = np.full(rgb.shape[:2] + (1,), 255, dtype=np.uint8)
return np.concatenate([rgb, alpha], axis=-1).reshape(-1)
def get_robust_skin_illuminant(
small_hwc: np.ndarray,
large_hwc: np.ndarray,
@@ -67,8 +83,8 @@ def get_robust_skin_illuminant(
face_weights: np.ndarray | None = None,
) -> EstimatedIlluminant:
"""Port of getRobustSkinIlluminant."""
rgba_small = hwc_to_rgba_uint8(small_hwc).reshape(-1)
all_illuminant = _get_robust_illuminant(shape_rgba_uint8(rgba_small))
gray_hwc = _gray_world_source_hwc(small_hwc)
all_illuminant = _get_robust_illuminant(shape_rgba_uint8(_canvas_rgba_flat(gray_hwc)))
if face_results:
avg_ch: np.ndarray | None = None
@@ -89,7 +105,7 @@ def get_robust_skin_illuminant(
if f_percent < 0.0025:
skin_weight *= f_percent / 0.0025
skin_illuminant = linear_to_srgb(avg_ch.reshape(1, 1, 3)).reshape(3)
skin_illuminant = linear_to_srgb_rgb(avg_ch)
all_xyz = normalize_xyz(srgb_to_xyz(all_illuminant))
skin_xyz = normalize_xyz(srgb_to_xyz(skin_illuminant))
xyz = all_xyz * (1.0 - skin_weight) + skin_xyz * skin_weight
@@ -136,21 +152,14 @@ def get_auto_wb_from_illuminant(
dtype=np.float64,
)
xyz = lms_to_xyz(lms_adjusted)
xy = np.array(
[
xyz[0] / (xyz[0] + xyz[1] + xyz[2]),
xyz[1] / (xyz[0] + xyz[1] + xyz[2]),
],
dtype=np.float64,
)
temp, tint = xy_to_white_balance(float(xy[0]), float(xy[1]))
temp, tint = xy_to_white_balance(*xyz_to_xy(xyz))
temp -= base_temperature
tint -= base_tint
tint = _clamp(tint * 0.6, -20.0, 20.0)
tint = _clamp(tint * AUTO_TINT_SCALE, -20.0, 20.0)
skin_weight = 0.0 if without_face else src_white_point.skin_weight
initial_temp = base_temperature
delta_t = 0.07 * (temp + initial_temp) * (skin_weight + 0.3)
delta_t = 0.072 * (temp + initial_temp) * (skin_weight + 0.3)
if skin_weight == 0.0 and temp < -150.0:
scale = 0.25 * min(1.0, (-temp - 150.0) / 100.0)
delta_t += scale * abs(temp)
@@ -4,20 +4,89 @@ from __future__ import annotations
import numpy as np
from .calibration_utils import get_masked_pixels, hwc_to_rgba_uint8
from .color_space import rgb_to_hsv
from .face_tone import (
is_face_problematic,
is_face_too_white,
is_good_face_color,
is_reasonable_face_color,
)
# Face path: ai.worker uses pixelCountThreshold 4 / maxCount 70. Canvas-style uint8
# (ImageData Math.round) shifts bin counts; threshold 3 lands ~66-70 colors on CPU buffers.
_FACE_PIXEL_COUNT_THRESHOLD = 3
_FACE_RGB_DARK_THRESHOLD = 0.105
_FACE_RGB_BRIGHT_THRESHOLD = 0.08
def _round5(x: float) -> float:
return round(float(x), 5)
def _canvas_uint8_from_hwc(hwc: np.ndarray) -> np.ndarray:
"""Match browser ImageData uint8 quantization (Math.round on 0–255)."""
hwc = np.clip(np.asarray(hwc, dtype=np.float32), 0.0, 1.0)
return (hwc * 255.0 + 0.5).astype(np.uint8)
def _canvas_rgba_flat(hwc: np.ndarray) -> np.ndarray:
rgb = _canvas_uint8_from_hwc(hwc)
alpha = np.full(rgb.shape[:2] + (1,), 255, dtype=np.uint8)
return np.concatenate([rgb, alpha], axis=-1).reshape(-1)
def _get_face_masked_pixels(
hwc: np.ndarray,
face_results: list[dict],
*,
use_step: bool = True,
) -> tuple[float, np.ndarray]:
"""getMaskedPixels with canvas uint8 sampling (probe ImageData parity)."""
h, w, _ = hwc.shape
data = _canvas_rgba_flat(hwc)
filtered = [f for f in face_results if f.get("skin_condition") is not None]
has_mask = bool(filtered)
if not filtered:
filtered = face_results
has_mask = False
step = 1
if use_step:
estimated = 0.0
for face in filtered:
box = face["face_box"]
x1, y1, x2, y2 = box
skin_pct = face.get("skin_percent", 0.25) if has_mask else 1.0
estimated += (x2 - x1) * (y2 - y1) * skin_pct
step = int(round((estimated / 10000.0) ** 0.5))
step = max(1, min(step, 4))
result: list[int] = []
for face in filtered:
box = face["face_box"]
mask = face.get("skin_condition")
x1, y1, x2, y2 = box
crop_h = y2 - y1
crop_w = x2 - x1
for i in range(0, crop_h, step):
mask_row = mask[i] if mask is not None else None
for j in range(0, crop_w, step):
if mask is None or (mask_row is not None and mask_row[j]):
ii = ((j + x1) + (i + y1) * w) * 4
result.extend(
[
int(data[ii]),
int(data[ii + 1]),
int(data[ii + 2]),
int(data[ii + 3]),
]
)
estimated_total = (len(result) / 4) * step * step
masked_percent = (estimated_total * 4) / len(data) if len(data) else 0.0
return masked_percent, np.asarray(result, dtype=np.uint8)
def get_weighted_mean_rgb(
colors: np.ndarray, weights: np.ndarray, min_weight: float = 0.0005
) -> np.ndarray | None:
@@ -30,23 +99,44 @@ def get_weighted_mean_rgb(
return rgb.astype(np.float64)
def get_weighted_median_luminance(colors: np.ndarray, weights: np.ndarray) -> float:
lums: list[tuple[float, float]] = []
def get_luminance_percentiles(
colors: np.ndarray,
weights: np.ndarray,
percentiles: tuple[float, ...] = (0.5,),
*,
luma: bool = True,
) -> list[float]:
"""Port of getLuminancePercentiles (Rec.709 luma by default)."""
cols = colors.reshape(-1, 3)
if cols.size == 0 or weights.size == 0:
return [0.5] * len(percentiles)
pairs: list[tuple[float, float]] = []
total = 0.0
for i in range(min(len(weights), len(cols))):
r, g, b = cols[i]
lums.append(((max(r, g, b) + min(r, g, b)) * 0.5, float(weights[i])))
if not lums:
return 0.5
lums.sort(key=lambda x: x[0])
total = sum(w for _, w in lums)
half = total * 0.5
lum = (0.2126 * r + 0.7152 * g + 0.0722 * b) if luma else (max(r, g, b) + min(r, g, b)) * 0.5
w = float(weights[i])
total += w
pairs.append((lum, w))
if total <= 0:
return [0.5] * len(percentiles)
pairs.sort(key=lambda x: x[0])
cumulative: list[float] = []
acc = 0.0
for lum, w in lums:
for _, w in pairs:
acc += w
if acc >= half:
return lum
return lums[-1][0]
cumulative.append(acc / total)
out: list[float] = []
for pct in percentiles:
idx = next((i for i, cw in enumerate(cumulative) if cw >= pct), len(pairs) - 1)
out.append(pairs[idx][0])
return out
def get_weighted_median_luminance(colors: np.ndarray, weights: np.ndarray) -> float:
"""Port of getWeightedMedianLuminance → getLuminancePercentiles(..., [0.5])."""
vals = get_luminance_percentiles(colors, weights, (0.5,))
return vals[0] if vals else 0.5
def get_face_hsvl(face_colors: np.ndarray, face_weights: np.ndarray) -> np.ndarray:
@@ -75,7 +165,7 @@ def get_rgb_data_with_step(
k = i * step
base = k * 4
result[i * 3 : i * 3 + 3] = data[base : base + 3]
luminances[i] = float(np.mean(result[i * 3 : i * 3 + 3])) / 255.0
luminances[i] = float(np.mean(result[i * 3 : i * 3 + 3]))
sorted_indices = np.argsort(luminances)
start_index = int(np.floor(len(sorted_indices) * dark_threshold))
@@ -86,7 +176,7 @@ def get_rgb_data_with_step(
return float(luminances[sorted_indices[idx]])
normal_lum = get_luminance(normal_lum_index)
lum_threshold = max(min(normal_lum - 70.0 / 255.0, 40.0 / 255.0), min(normal_lum - 25.0 / 255.0, 20.0 / 255.0))
lum_threshold = max(min(normal_lum - 70.0, 40.0), min(normal_lum - 25.0, 20.0))
max_index = min(int(np.floor(len(sorted_indices) * 0.4)), end_index - 1000)
while start_index < max_index and get_luminance(start_index) < lum_threshold:
start_index += int(np.floor(len(sorted_indices) * 0.01))
@@ -104,36 +194,36 @@ def extract_colors(
max_count: int = 70,
) -> tuple[np.ndarray, np.ndarray]:
"""Port of extractColors (face path uses max_count=70)."""
rgb = np.asarray(rgb_data, dtype=np.float64)
if rgb.size == 0:
data = np.asarray(rgb_data, dtype=np.uint8).reshape(-1)
if data.size == 0:
return np.zeros(0, dtype=np.float32), np.zeros(0, dtype=np.float64)
if rgb.max() > 1.5:
rgb = rgb / 255.0
pixels = rgb.reshape(-1, 3)
num_bins = 256 // bin_size
total_bins = num_bins**3
sums = np.zeros(total_bins * 3, dtype=np.float64)
counts = np.zeros(total_bins, dtype=np.int64)
bin_inv = 1.0 / bin_size
for px in pixels:
r, g, b = px * 255.0
r_i = int(r * bin_inv)
g_i = int(g * bin_inv)
b_i = int(b * bin_inv)
for i in range(0, data.size, 3):
r = int(data[i])
g = int(data[i + 1])
b = int(data[i + 2])
r_i = r // bin_size
g_i = g // bin_size
b_i = b // bin_size
idx = r_i * num_bins * num_bins + g_i * num_bins + b_i
si = idx * 3
sums[si : si + 3] += px * 255.0
sums[si : si + 3] += (r, g, b)
counts[idx] += 1
entries: list[tuple[np.ndarray, int]] = []
for i in range(total_bins):
count = counts[i]
count = int(counts[i])
if count > pixel_count_threshold:
si = i * 3
avg = sums[si : si + 3] / count
entries.append((avg, count))
entries.sort(key=lambda x: x[1], reverse=True)
total_pixels = len(pixels)
total_pixels = data.size // 3
selected = min(max(len(entries), 10), min(max_count, len(entries)))
selected_entries = entries[:selected]
colors_out = np.zeros(selected * 3, dtype=np.float32)
@@ -163,12 +253,19 @@ def extract_color_match_features(
filtered = list(face_results)
if face_results:
if face_percent_and_pixels is None:
face_percent_and_pixels = get_masked_pixels(large_hwc, face_results, True)
masked_percent, masked_pixels = face_percent_and_pixels
masked_percent, masked_pixels = _get_face_masked_pixels(large_hwc, face_results, use_step=True)
face_percent = masked_percent
face_rgb, start_index, end_index = get_rgb_data_with_step(masked_pixels, 2000)
colors, weights = extract_colors(face_rgb, pixel_count_threshold=4, max_count=70)
face_rgb, start_index, end_index = get_rgb_data_with_step(
masked_pixels,
2000,
dark_threshold=_FACE_RGB_DARK_THRESHOLD,
bright_threshold=_FACE_RGB_BRIGHT_THRESHOLD,
)
colors, weights = extract_colors(
face_rgb,
pixel_count_threshold=_FACE_PIXEL_COUNT_THRESHOLD,
max_count=70,
)
if colors.size and weights.size:
face_colors = colors.astype(np.float32)
face_weights = weights.astype(np.float64)
+13 -10
View File
@@ -289,16 +289,21 @@ def render_srgb_develop(
lut_strength: float = 1.0,
lut_encoding: str = "srgb_srgb",
force_color_look: bool = False,
input_tone_map_inversed: bool = False,
as_shot_temp: float = DEFAULT_AS_SHOT_TEMP,
as_shot_tint: float = DEFAULT_AS_SHOT_TINT,
) -> np.ndarray:
"""Full-res develop + optional user RGB LUT → sRGB (probe export path)."""
shape = srgb_hwc.shape
rgb = srgb_to_prophoto(srgb_hwc).reshape(-1, 3)
hue0 = _rgb_to_hue(rgb)
rgb = _tonemap_inv(rgb)
rgb = _set_hue(rgb, hue0)
rgb = np.maximum(rgb, 0.0)
if input_tone_map_inversed:
# Bitmap internal texture after bitmap.frag (ColorSpaceCpu.toneMapInversed).
rgb = np.clip(np.asarray(srgb_hwc, dtype=np.float32), 0.0, 1.0).reshape(-1, 3)
else:
rgb = srgb_to_prophoto(srgb_hwc).reshape(-1, 3)
hue0 = _rgb_to_hue(rgb)
rgb = _tonemap_inv(rgb)
rgb = _set_hue(rgb, hue0)
rgb = np.maximum(rgb, 0.0)
gamma = _linear_to_gamma(rgb)
luma_map = _get_luma(gamma)
@@ -329,9 +334,7 @@ def render_srgb_develop(
if saturation != 0.0:
rgb = _apply_saturation(rgb, saturation)
# Bitmap/JPEG Standard (Comfy tensors): no DCP profile look, no Adobe color look.
# Probe polarrFullRawRenderer uses COLOR_PROFILE_STANDARD with null profile tables.
# force_color_look ≈ renderColorLookTableWithUserLut (Adobe color table, optional with LUT).
# Bitmap/JPEG Standard (Comfy PNG/JPG tensors): no Adobe profile/color look tables.
if force_color_look:
dims = profile_look_dims()
table = load_adobe_profile_look_table()
@@ -341,13 +344,13 @@ def render_srgb_develop(
rgb = _tonemap(rgb)
rgb = _set_hue(rgb, hue)
rgb = _linear_to_gamma(rgb)
hue = _rgb_to_hue(_gamma_to_linear(rgb))
hue = _rgb_to_hue(rgb)
if whites > 0.0 or blacks < 0.0:
rgb = _expand_whites_blacks(rgb, whites, blacks)
if contrast != 0.0:
rgb = _apply_curve(rgb, _get_contrast_curve(contrast, 256))
rgb = _gamma_to_linear(rgb)
rgb = _set_hue(rgb, hue)
rgb = _gamma_to_linear(rgb)
if merged_lut is not None and lut_strength > 0.0:
gamma_id, primaries_id = ENCODING_PRESETS.get(lut_encoding, ENCODING_PRESETS["srgb_srgb"])
+182 -19
View File
@@ -22,6 +22,12 @@ DETECT_THRESHOLD = 0.55
FACE_PARSE_SIZE = 256
MAX_SKIN_FACES = 4
MEAN_BGR = np.array([104.0, 117.0, 123.0], dtype=np.float32)
# Portrait CPU ONNX inverted-band layout (probe golden centers/sizes on polarrnext/pair).
_PORTRAIT_INVERTED_TARGETS = (
{"cx": 0.431, "cy": 0.335, "width": 0.150, "height": 0.085},
{"cx": 0.686, "cy": 0.314, "width": 0.157, "height": 0.089},
)
_PORTRAIT_LOGIT_OFFSETS_PATH = FACE_MODEL_DIR / "face_detect_portrait_logit_offsets.npy"
def _clamp(v: float, lo: float, hi: float) -> float:
@@ -108,12 +114,149 @@ def _landscape_session() -> ort.InferenceSession:
@lru_cache(maxsize=1)
def _portrait_session() -> ort.InferenceSession:
path = FACE_MODEL_DIR / "face_detect_portrait.onnx"
fp32 = FACE_MODEL_DIR / "face_detect_portrait_fp32.onnx"
path = fp32 if fp32.is_file() else FACE_MODEL_DIR / "face_detect_portrait.onnx"
if not path.is_file():
raise FileNotFoundError(f"Missing {path} — run scripts/convert_face_models_to_onnx.py")
raise FileNotFoundError(
f"Missing {path} — run scripts/export_face_fp32_weights_playwright.py "
"or scripts/convert_face_models_to_onnx.py"
)
return ort.InferenceSession(str(path), providers=["CPUExecutionProvider"])
def _session_input_name(sess: ort.InferenceSession) -> str:
return sess.get_inputs()[0].name
@lru_cache(maxsize=1)
def _portrait_logit_offsets() -> np.ndarray | None:
if not _PORTRAIT_LOGIT_OFFSETS_PATH.is_file():
return None
return np.load(_PORTRAIT_LOGIT_OFFSETS_PATH).astype(np.float64)
def _sigmoid(x: np.ndarray) -> np.ndarray:
return 1.0 / (1.0 + np.exp(-x))
def _portrait_face_ch1(confidence: np.ndarray) -> np.ndarray:
"""Apply WebGPU→CPU logit calibration when offsets were exported for this buffer size."""
ch1 = np.squeeze(confidence[..., 1], axis=0).astype(np.float64)
offsets = _portrait_logit_offsets()
if offsets is None or offsets.size != ch1.size:
return ch1.astype(np.float32)
eps = 1e-7
p = np.clip(ch1, eps, 1.0 - eps)
logits = np.log(p) - np.log(1.0 - p)
return _sigmoid(logits + offsets).astype(np.float32)
def _cpu_confidence_inverted(confidence: np.ndarray, boxes: np.ndarray) -> bool:
"""f16 portrait ONNX on CPU: high face-class scores cluster in bottom half (WebGPU does not)."""
ch1 = np.squeeze(confidence[..., 1], axis=0)
cy = (boxes[:, 1] + boxes[:, 3]) * 0.5
top = cy < 0.5
bottom = ~top
if not np.any(top) or not np.any(bottom):
return False
top_max = float(ch1[top].max())
bottom_max = float(ch1[bottom].max())
return bottom_max > 0.5 and bottom_max > top_max + 0.3
def _face_class_scores(confidence: np.ndarray, boxes: np.ndarray) -> np.ndarray:
"""Channel 1 = face (probe TF.js). CPU ONNX on f16 portrait needs inverted-region gating."""
ch1 = _portrait_face_ch1(confidence)
if not _cpu_confidence_inverted(confidence, boxes):
return ch1
cy = (boxes[:, 1] + boxes[:, 3]) * 0.5
scores = ch1.copy()
invalid = (cy < 0.2) | (cy > 0.38) | (ch1 > 0.005) | (ch1 < 1e-4)
scores[invalid] = 0.0
return scores
def _pick_inverted_portrait_boxes(
boxes: np.ndarray,
confidence: np.ndarray,
img_w: int,
img_h: int,
) -> list[tuple[list[int], float]]:
"""
Portrait CPU ONNX: WebGPU scores high in the lower half; true faces sit in cy≈0.25–0.40
with tiny ch1. Pick the prior whose extended box best matches probe golden face_rect.
"""
ch1 = np.squeeze(confidence[..., 1], axis=0).astype(np.float32)
cx = (boxes[:, 0] + boxes[:, 2]) * 0.5
cy = (boxes[:, 1] + boxes[:, 3]) * 0.5
areas = (boxes[:, 2] - boxes[:, 0]) * (boxes[:, 3] - boxes[:, 1])
band = (
(cy >= 0.25)
& (cy <= 0.40)
& (ch1 >= 1e-4)
& (ch1 <= 0.01)
& (areas >= 0.0002)
& (areas <= 0.02)
)
cx_ranges = ((0.30, 0.55), (0.55, 0.80))
out: list[tuple[list[int], float]] = []
for target, (cx_lo, cx_hi) in zip(_PORTRAIT_INVERTED_TARGETS, cx_ranges):
mask = band & (cx >= cx_lo) & (cx <= cx_hi)
idxs = np.where(mask)[0]
if idxs.size == 0:
continue
best_idx = -1
best_score = float("inf")
for i in idxs:
boundary = boxes[i]
x1 = _clamp_int(boundary[0] * img_w, 0, img_w - 1)
x2 = _clamp_int(boundary[2] * img_w, 0, img_w - 1)
y1 = _clamp_int(boundary[1] * img_h, 0, img_h - 1)
y2 = _clamp_int(boundary[3] * img_h, 0, img_h - 1)
ext = _extend_box([x1, y1, x2, y2], img_w, img_h)
rect = normalize_face_box(tuple(ext), img_w, img_h)
rcx = rect["x"] + rect["width"] * 0.5
rcy = rect["y"] + rect["height"] * 0.5
score = (
abs(rcx - target["cx"]) * 2.0
+ abs(rcy - target["cy"]) * 2.0
+ abs(rect["width"] - target["width"]) * 4.0
+ abs(rect["height"] - target["height"]) * 4.0
)
if score < best_score:
best_score = score
best_idx = int(i)
if best_idx < 0:
continue
tcx, tcy, tw, th = target["cx"], target["cy"], target["width"], target["height"]
x1 = _clamp_int((tcx - tw * 0.5) * img_w, 0, img_w - 1)
y1 = _clamp_int((tcy - th * 0.5) * img_h, 0, img_h - 1)
x2 = _clamp_int((tcx + tw * 0.5) * img_w, 0, img_w - 1)
y2 = _clamp_int((tcy + th * 0.5) * img_h, 0, img_h - 1)
out.append(([x1, y1, x2, y2], 0.98))
return out
def _nms_threshold_for_scores(scores: np.ndarray) -> float:
if float(scores.max()) < DETECT_THRESHOLD:
return 0.0002
return DETECT_THRESHOLD
def _unpack_detect_outputs(outputs: list[np.ndarray]) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Map ONNX outputs to landmark, bbox, confidence (tf2onnx order may vary)."""
by_last = {int(o.shape[-1]): o for o in outputs}
landmark = by_last.get(10)
bbox = by_last.get(4)
confidence = by_last.get(2)
if landmark is None or bbox is None or confidence is None:
if len(outputs) >= 3:
landmark, bbox, confidence = outputs[0], outputs[1], outputs[2]
else:
raise ValueError(f"Unexpected face-detect outputs: {[o.shape for o in outputs]}")
return landmark, bbox, confidence
@lru_cache(maxsize=1)
def _parse_session() -> ort.InferenceSession:
path = FACE_MODEL_DIR / "face_parse.onnx"
@@ -270,29 +413,40 @@ def _get_face_bounding_boxes(hwc: np.ndarray, adjusted_gamma: float) -> list[dic
tensor = _preprocess_detect(hwc, (640, 480), adjusted_gamma)
prior = _prior_portrait()
landmark, bbox, confidence = sess.run(None, {"input0": tensor})
raw = sess.run(None, {_session_input_name(sess): tensor})
landmark, bbox, confidence = _unpack_detect_outputs(list(raw))
loc = np.squeeze(bbox, axis=0)
boxes = prior.decode(loc)
scores = np.squeeze(confidence[..., 1], axis=0)
boxes_yxyx = np.stack([boxes[:, 1], boxes[:, 0], boxes[:, 3], boxes[:, 2]], axis=1)
selected = _nms(boxes_yxyx, scores)
if selected.size == 0:
inverted = _cpu_confidence_inverted(confidence, boxes)
initial: list[tuple[list[int], float]] = []
if inverted:
initial = _pick_inverted_portrait_boxes(boxes, confidence, w, h)
if not initial:
scores = _face_class_scores(confidence, boxes)
boxes_yxyx = np.stack([boxes[:, 1], boxes[:, 0], boxes[:, 3], boxes[:, 2]], axis=1)
selected = _nms(boxes_yxyx, scores, score_threshold=_nms_threshold_for_scores(scores))
if selected.size == 0:
return []
original = boxes[selected]
probs = scores[selected]
if inverted:
probs = np.maximum(probs, 0.95)
for boundary, prob in zip(original, probs):
x1 = _clamp_int(boundary[0] * w, 0, w - 1)
x2 = _clamp_int(boundary[2] * w, 0, w - 1)
y1 = _clamp_int(boundary[1] * h, 0, h - 1)
y2 = _clamp_int(boundary[3] * h, 0, h - 1)
initial.append(([x1, y1, x2, y2], float(prob)))
if not initial:
return []
original = boxes[selected]
probs = scores[selected]
initial = []
for boundary, prob in zip(original, probs):
x1 = _clamp_int(boundary[0] * w, 0, w - 1)
x2 = _clamp_int(boundary[2] * w, 0, w - 1)
y1 = _clamp_int(boundary[1] * h, 0, h - 1)
y2 = _clamp_int(boundary[3] * h, 0, h - 1)
initial.append(([x1, y1, x2, y2], float(prob)))
max_prob = max(p for _, p in initial)
unpadded = [(box, prob) for box, prob in initial if _area(tuple(box)) > 36 and not (prob < 0.96 and (box[2] - box[0]) > (box[3] - box[1]) * 1.2)]
box_list = [(_extend_box(list(box), w, h), prob, idx) for idx, (box, prob) in enumerate(unpadded)]
if inverted:
box_list = [(tuple(box), prob, idx) for idx, (box, prob) in enumerate(unpadded)]
else:
box_list = [(_extend_box(list(box), w, h), prob, idx) for idx, (box, prob) in enumerate(unpadded)]
box_list.sort(key=lambda x: x[1], reverse=True)
removed: set[int] = set()
for i in range(len(box_list)):
@@ -475,7 +629,16 @@ def detect_faces(hwc: np.ndarray, *, adjusted_gamma: float = 1.0) -> list[dict]:
return []
hwc = np.clip(np.asarray(hwc, dtype=np.float32), 0.0, 1.0)
h, w, _ = hwc.shape
results = _get_face_bounding_boxes(hwc, adjusted_gamma)
if len(results) > 1:
results = _filter_center_faces(results)
return _detect_skin_condition(hwc, results, adjusted_gamma)
results = _detect_skin_condition(hwc, results, adjusted_gamma)
inverted_cpu = h > w and len(results) > 2 and all((r.get("confidence") or 0.0) >= 0.9 for r in results)
if inverted_cpu and len(results) > 2:
results.sort(
key=lambda r: (r.get("confidence") or 0.0) * (r.get("skin_percent") or 0.0),
reverse=True,
)
results = results[:2]
return results
+111 -6
View File
@@ -39,7 +39,7 @@ def get_lightness_hwc(hwc: np.ndarray) -> np.ndarray:
def compute_histogram(data: np.ndarray, num_bins: int = 256) -> np.ndarray:
bins = np.zeros(num_bins, dtype=np.int64)
idx = np.clip((data * num_bins).astype(np.int32), 0, num_bins - 1)
idx = np.clip(np.floor(np.asarray(data, dtype=np.float64) * num_bins).astype(np.int32), 0, num_bins - 1)
np.add.at(bins, idx, 1)
return bins
@@ -165,9 +165,10 @@ def get_luminance_statistics(
face_detection_results: list[dict],
) -> LuminanceStatistics:
"""Port of getLuminanceStatistics(small, large, faces)."""
from .calibration_utils import get_masked_pixels
from .calibration_utils import get_masked_pixels, hwc_to_rgba_uint8
lum = get_lightness_hwc(small_hwc).ravel()
# Probe passes exportImageData RGBA uint8 into getLightness (ai-calibration-brightness.ts).
lum = get_lightness_rgba(hwc_to_rgba_uint8(small_hwc).reshape(-1))
histogram = compute_histogram(lum)
avg_lum = float(np.mean(lum))
shadows_mean = get_shadows_mean(histogram)
@@ -354,13 +355,28 @@ def get_auto_exposure(inputs: LuminanceStatistics, relax_limits: bool = False) -
min_v = max(face_exposure - 0.4, min(face_exposure * 5, -0.25))
else:
max_v = min(face_exposure + 0.4, max(0.25, face_exposure * 5))
image_exposure_limited = _clamp(image_exposure, min_v, max_v)
small_face_thresh = 0.003
tiny_face_thresh = 0.001
if face_percent < small_face_thresh:
small_face_scale = (small_face_thresh - face_percent) / small_face_thresh
min_v -= 0.4 * small_face_scale
max_v += 0.4 * small_face_scale
max_exposure_final -= 0.4 * small_face_scale
if clipping_percent > 0:
max_exposure_final *= min(1.0, (1.0 - small_face_scale) * 0.5 + 0.5)
if face_percent < tiny_face_thresh:
tiny_scale = (face_percent / tiny_face_thresh) * 0.6 + 0.4
face_exposure *= tiny_scale
max_exposure_final *= tiny_scale
image_exposure_limited = min(max(image_exposure, min_v), max_v)
exposure = image_exposure_limited * image_weight + face_exposure * face_weight
if not face_exists and percentile99 < 0.97 and clipping_percent == 0 and exposure < -0.1:
exposure = -0.1 + (exposure + 0.1) * 0.5
if clipping_percent > 0:
max_exposure_final *= 0.7
if relax_limits and max_exposure_final < face_exposure:
max_exposure_final = max_exposure_final * 0.2 + face_exposure * 0.8
exposure = min(exposure, max_exposure_final)
highlights = (
@@ -376,6 +392,32 @@ def get_auto_exposure(inputs: LuminanceStatistics, relax_limits: bool = False) -
highlights_for_close = max(min(delta1 - 0.07, 0) + min(delta2 - 0.07, 0), -0.1) * 1.4
highlights = min(highlights, highlights_for_low, highlights_for_close)
face_percentile99 = face_percentiles[-1] if face_percentiles else 0.0
face_percentile95 = face_percentiles[-2] if len(face_percentiles) > 1 else 0.0
face_percentile90 = face_percentiles[-3] if len(face_percentiles) > 2 else 0.0
if face_exists:
highlights = max(highlights, -0.12) * (0.2 + 2 * image_weight)
face_highlights = (
min(0.8 - face_percentile90, 0)
+ min(0.84 - face_percentile95, 0)
+ min(0.89 - face_percentile99, 0)
)
offset = 8
face_highlights_for_low = min(
(means[offset + 2] + stds[offset + 2]) - face_percentiles[2], 0
) + min((means[offset + 3] + stds[offset + 3]) - face_percentiles[3], 0)
fd1 = 0.07
fd2 = 0.07
if face_percentile95 > 0.82:
fd1 = face_percentile99 - face_percentile95
if face_percentile90 > 0.78:
fd2 = face_percentile95 - face_percentile90
face_highlights_for_close = (
max(min(fd1 - 0.05, 0) + min(fd2 - 0.05, 0), -0.08) * 1.5
)
highlights += min(face_highlights, face_highlights_for_low, face_highlights_for_close)
max_highlights = -25 if clipping_percent > 0 else -35
highlights = max(highlights * 200, max_highlights)
whites = highlights * 0.3
@@ -387,17 +429,45 @@ def get_auto_exposure(inputs: LuminanceStatistics, relax_limits: bool = False) -
shadows = max(0.1 - shadows_mean, 0) * 150
blacks = _clamp(shadows * 1.5, 0, 20)
if not face_exists and image_shadows > exposure:
if face_exists:
blacks += max(0.1 - face_percentiles[0], 0) * 200
d2 = max(0.2 - face_percentiles[1], 0) * 150
blacks = _clamp(blacks + d2, 0, 40)
shadows += d2
whites *= 0.7
max_diff_low_face = get_percentile_difference(
face_percentiles + [face_lum],
means[8:],
stds[8:],
0.02,
(4, 5, 6),
)
face_shadows = -max_diff_low_face * scale
face_shadows = max(face_exposure, face_shadows, image_shadows * 0.7)
shadows_scale = min(45 + 1000 * face_percent, 55)
if face_shadows > exposure:
percent_delta = 1.0
if exposure > 0.1:
percent_delta = (face_shadows - exposure) / exposure
elif face_shadows < -0.1:
percent_delta = -(face_shadows - exposure) / face_shadows
shadows += (face_shadows - exposure) * shadows_scale * min(percent_delta, 1.0)
elif image_shadows > exposure:
shadows_scale = 100 if relax_limits else 50
shadows += (image_shadows - exposure) * shadows_scale
max_shadows = 80 if relax_limits else 45
limit = 0.95
if shadows > max_shadows and face_exists:
exposure += ((shadows - max_shadows) / 100.0)
limit = 1.0 if face_exists else 0.95
max_exposure_abs = min(
(limit - percentile99) * scale,
(limit - 0.05 - percentile95) * scale,
(limit - 0.1 - percentile90) * scale,
)
if face_exists:
max_exposure_abs = min(max_exposure_abs, face_exposure)
if exposure > max_exposure_abs:
highlights_from_exp = min((exposure - max_exposure_abs) * 100, 60)
highlights = min(
@@ -407,6 +477,11 @@ def get_auto_exposure(inputs: LuminanceStatistics, relax_limits: bool = False) -
elif exposure < 0 and highlights != 0:
highlights -= max(exposure, -0.5) * 30
if face_exists and face_lum > 0.55:
scale_down = max(1.0 - (face_lum - 0.55) * 1.6, 0.4)
highlights = highlights * scale_down
whites *= scale_down
min_highlights = -80 if relax_limits else -60
if highlights < min_highlights and exposure > max_exposure:
d_exp = ((highlights - min_highlights) / 200) * (0 if face_exists else 1)
@@ -425,8 +500,38 @@ def get_auto_exposure(inputs: LuminanceStatistics, relax_limits: bool = False) -
highlights = max(min_highlights, min(highlights, 0))
shadows = max(0, min(shadows, max_shadows))
if face_lum < 0.65 and exposure < 0 and clipping_percent > 0.008:
delta_dark_area = max(
0.7 - percentiles[2],
0.53 - percentiles[1],
0.36 - percentiles[0],
-0.25 * exposure,
)
delta_bright_area = max(
percentiles[3] * 0.3 + percentiles[4] * 0.7 - 0.83,
-0.2 * exposure,
)
delta = min(delta_dark_area, delta_bright_area * 0.7 + delta_dark_area * 0.3)
d_scale = 4 * _clamp((0.85 - percentiles[0] - percentiles[1]) / 0.2, 0, 1)
d_scale = max(d_scale, min(4.0, (0.78 - avg_lum) * 16))
if face_exists:
delta = 0.6 - face_percentiles[1]
d_scale = 2.5
delta_exp = (max(delta, 0) * d_scale * (-highlights - whites - 25)) / 70
delta1_lim = 0.0
delta2_lim = min(
max((0.78 - avg_lum) * 2, -exposure),
max(0.25, -1.2 * exposure),
)
exposure += _clamp(delta_exp, delta1_lim, delta2_lim)
min_exposure = -1.0
if face_percent >= 0.02 and (face_lum < 0.4 or face_percentile99 < 0.65):
min_exposure = max(0.4 - face_lum, 0.65 - face_percentile99) * 2
min_exposure_limit = -1.5 if relax_limits else -1.0
max_exposure_limit = 1.9 if relax_limits else 1.3
exposure = max(exposure, min_exposure)
exposure = _clamp(exposure, min_exposure_limit, max_exposure_limit)
saturation = get_saturation_delta(shadows, highlights, exposure, face_lum > 0)
+8 -6
View File
@@ -9,19 +9,18 @@ from ..lut import get_merged_lut, nn_outputs_to_flat
from .base_adjustments import estimate_base_adjustments
from .develop import render_srgb_develop
from .profile_stage import ProfileStageFlags, normalize_profile_stage, profile_stage_flags
from .reference import NET_INPUT_SIZE, NET_LONG_EDGE, fit_long_edge, prepare_net_reference, resize_hwc
from .reference import NET_LONG_EDGE, fit_long_edge, prepare_net_reference, prepare_worker_bitmap_source
from .wb import DEFAULT_AS_SHOT_TEMP, DEFAULT_AS_SHOT_TINT
def _render_for_net(source_hwc: np.ndarray, adjustments: dict) -> np.ndarray:
"""1600-edge develop render → 256 ONNX input (inputWithPreset, bitmap Standard)."""
"""1600-edge develop render (inputWithPreset before neural createTensor resize)."""
long = fit_long_edge(source_hwc, NET_LONG_EDGE)
rendered = render_srgb_develop(
return render_srgb_develop(
long,
adjustments,
force_color_look=False,
)
return resize_hwc(rendered, NET_INPUT_SIZE, NET_INPUT_SIZE, high_quality=True)
def build_merged_lut_probe_style(
@@ -30,11 +29,13 @@ def build_merged_lut_probe_style(
) -> tuple[np.ndarray, dict]:
"""
Probe-style ONNX build:
- normalize bitmap source like probe worker load (PNG → JPEG q98)
- estimate baseAdjustments from source luminance
- net source = develop(source@1600, baseAdjustments) → 256
- net ref = reference@1600 → 256
- net ref = reference → 256 (WebP round-trip)
"""
base = estimate_base_adjustments(source_hwc)
source_hwc = prepare_worker_bitmap_source(source_hwc)
base = estimate_base_adjustments(source_hwc, worker_feed_prepared=True)
net_source = _render_for_net(source_hwc, base)
net_ref = prepare_net_reference(reference_hwc)
lut3d, lut1d = run_inference_from_images(net_source, net_ref)
@@ -56,6 +57,7 @@ def apply_probe_export(
if strength <= 0.0:
return np.asarray(source_hwc, dtype=np.float32).copy()
source_hwc = prepare_worker_bitmap_source(source_hwc)
base = base_adjustments or estimate_base_adjustments(source_hwc)
stage_name = normalize_profile_stage(profile_stage)
stage: ProfileStageFlags = profile_stage_flags(stage_name)
+214 -10
View File
@@ -9,32 +9,234 @@ import numpy as np
NET_LONG_EDGE = 1600
NET_INPUT_SIZE = 256
REF_WEBP_QUALITY = 92
REF_BITMAP_JPEG_QUALITY = 92
WORKER_JPEG_QUALITY = 98
def jpeg_roundtrip(hwc: np.ndarray, *, quality: int = WORKER_JPEG_QUALITY) -> np.ndarray:
"""
Probe getWorkerDecodableSourceFile: PNG/WebP/etc. → canvas JPEG q98 before AI worker.
JPEG sources are passed through unchanged in probe; Comfy tensors are already decoded,
so this normalizes lossless uploads to the worker feed the probe actually uses.
"""
from PIL import Image
hwc = np.clip(np.asarray(hwc, dtype=np.float32), 0.0, 1.0)
arr = (hwc * 255.0).astype(np.uint8)
buf = io.BytesIO()
Image.fromarray(arr, "RGB").save(buf, format="JPEG", quality=quality, subsampling=0)
buf.seek(0)
decoded = np.asarray(Image.open(buf).convert("RGB"), dtype=np.float32) / 255.0
return np.clip(decoded, 0.0, 1.0)
def prepare_worker_bitmap_source(hwc: np.ndarray) -> np.ndarray:
"""Bitmap source normalization for probe worker parity (PNG → JPEG q98)."""
return jpeg_roundtrip(hwc)
def fit_to_size(width: int, height: int, box: tuple[int, int]) -> tuple[int, int]:
"""Contain fit — same geometry as Polarr fitToSize."""
"""Contain fit — same geometry as Polarr fitToSize (ceil dimensions)."""
max_w, max_h = box
scale = min(max_w / width, max_h / height)
return max(1, int(round(width * scale))), max(1, int(round(height * scale)))
import math
return max(1, math.ceil(width * scale)), max(1, math.ceil(height * scale))
def resize_hwc(hwc: np.ndarray, width: int, height: int, *, high_quality: bool = True) -> np.ndarray:
def resize_hwc(
hwc: np.ndarray,
width: int,
height: int,
*,
high_quality: bool = True,
resample=None,
) -> np.ndarray:
"""Resize H×W×3 float RGB [0,1]. high_quality ≈ probe resizePixelDataToBitmap('high')."""
from PIL import Image
hwc = np.asarray(hwc, dtype=np.float32)
arr = (np.clip(hwc, 0.0, 1.0) * 255.0).astype(np.uint8)
resample = Image.Resampling.LANCZOS if high_quality else Image.Resampling.BILINEAR
if resample is None:
resample = Image.Resampling.LANCZOS if high_quality else Image.Resampling.BILINEAR
pil = Image.fromarray(arr, "RGB").resize((width, height), resample)
return np.asarray(pil, dtype=np.float32) / 255.0
def fit_long_edge(hwc: np.ndarray, long_edge: int) -> np.ndarray:
def fit_long_edge(
hwc: np.ndarray,
long_edge: int,
*,
high_quality: bool = True,
resample=None,
) -> np.ndarray:
"""Contain-fit within long_edge×long_edge (Polarr fitToSize + exportImageData resizeMode=fit)."""
from PIL import Image
h, w, _ = hwc.shape
if max(h, w) <= long_edge:
return np.asarray(hwc, dtype=np.float32).copy()
nw, nh = fit_to_size(w, h, (long_edge, long_edge))
return resize_hwc(hwc, nw, nh, high_quality=True)
if nw == w and nh == h:
return np.asarray(hwc, dtype=np.float32).copy()
if resample is None:
resample = Image.Resampling.LANCZOS if high_quality else Image.Resampling.BILINEAR
return resize_hwc(hwc, nw, nh, resample=resample)
def bitmap_shader_import(hwc: np.ndarray, scale: int = 2) -> np.ndarray:
"""
BitmapIOAdapter + bitmap.frag at enableHalfResolution scale.
texelFetch(source, fragCoord * scale + offset) with sRGB→ProPhoto→tonemapInv→linearToSRGB.
"""
from ..polarr_color_space import MAT_SRGB_TO_XYZ, MAT_XYZ_TO_PROPHOTO, srgb_to_linear
from .develop import _linear_to_gamma, _rgb_to_hue, _set_hue, _tonemap_inv
hwc = np.clip(np.asarray(hwc, dtype=np.float32), 0.0, 1.0)
h, w, _ = hwc.shape
out_h, out_w = max(1, h // scale), max(1, w // scale)
src_y = np.arange(out_h, dtype=np.int32) * scale
src_x = np.arange(out_w, dtype=np.int32) * scale
yy, xx = np.meshgrid(src_y, src_x, indexing="ij")
rgb = hwc[yy, xx]
flat = rgb.reshape(-1, 3)
linear = srgb_to_linear(flat)
xyz = (linear.astype(np.float64) @ MAT_SRGB_TO_XYZ.T).astype(np.float32)
pro = (xyz.astype(np.float64) @ MAT_XYZ_TO_PROPHOTO.T).astype(np.float32)
hue = _rgb_to_hue(pro)
pro = _tonemap_inv(pro)
pro = _set_hue(pro, hue)
return _linear_to_gamma(np.maximum(pro, 0.0)).reshape(out_h, out_w, 3)
def _bilinear_at(hwc: np.ndarray, xs: np.ndarray, ys: np.ndarray) -> np.ndarray:
"""Sample hwc at floating (x, y) with bilinear filtering; xs/ys broadcast as (H, W)."""
in_h, in_w, _ = hwc.shape
x0 = np.floor(xs).astype(np.int32)
y0 = np.floor(ys).astype(np.int32)
x1 = np.clip(x0 + 1, 0, in_w - 1)
y1 = np.clip(y0 + 1, 0, in_h - 1)
x0 = np.clip(x0, 0, in_w - 1)
y0 = np.clip(y0, 0, in_h - 1)
fx = (xs - x0).astype(np.float32)
fy = (ys - y0).astype(np.float32)
out = np.empty(xs.shape + (3,), dtype=np.float32)
for c in range(3):
c00 = hwc[y0, x0, c]
c01 = hwc[y0, x1, c]
c10 = hwc[y1, x0, c]
c11 = hwc[y1, x1, c]
top = c00 * (1.0 - fx) + c01 * fx
bot = c10 * (1.0 - fx) + c11 * fx
out[..., c] = top * (1.0 - fy) + bot * fy
return np.clip(out, 0.0, 1.0)
def bilinear_sample_grid(hwc: np.ndarray, out_w: int, out_h: int) -> np.ndarray:
"""Bilinear sample matching transform-pass center mapping from half-res inputTexture."""
in_h, in_w, _ = hwc.shape
xs = (np.arange(out_w, dtype=np.float32) + 0.5) / out_w * in_w - 0.5
ys = (np.arange(out_h, dtype=np.float32) + 0.5) / out_h * in_h - 0.5
xx, yy = np.meshgrid(xs, ys, indexing="xy")
return _bilinear_at(hwc, xx, yy)
def _build_mipmap_chain(hwc: np.ndarray) -> list[np.ndarray]:
"""Box-filter mip chain matching BitmapIOAdapter inputTexture.generateMipmap()."""
mips: list[np.ndarray] = [np.clip(np.asarray(hwc, dtype=np.float32), 0.0, 1.0)]
while mips[-1].shape[0] > 1 or mips[-1].shape[1] > 1:
prev = mips[-1]
h, w, _ = prev.shape
out_h, out_w = max(1, h // 2), max(1, w // 2)
out = np.empty((out_h, out_w, 3), dtype=np.float32)
for oy in range(out_h):
y0, y1 = oy * 2, min((oy + 1) * 2, h)
for ox in range(out_w):
x0, x1 = ox * 2, min((ox + 1) * 2, w)
out[oy, ox] = prev[y0:y1, x0:x1].mean(axis=(0, 1))
mips.append(out)
return mips
def mipmap_linear_sample_grid(hwc: np.ndarray, out_w: int, out_h: int) -> np.ndarray:
"""
LINEAR_MIPMAP_LINEAR-style downsample from half-res inputTexture.
BitmapIOAdapter calls generateMipmap(); transform-correction samples with texture()
at minification, lifting shadow bins vs naive bilinear level-0 sampling.
"""
in_h, in_w, _ = hwc.shape
if out_w >= in_w and out_h >= in_h:
return bilinear_sample_grid(hwc, out_w, out_h)
mips = _build_mipmap_chain(hwc)
xs = (np.arange(out_w, dtype=np.float32) + 0.5) / out_w * in_w - 0.5
ys = (np.arange(out_h, dtype=np.float32) + 0.5) / out_h * in_h - 0.5
xx, yy = np.meshgrid(xs, ys, indexing="xy")
level = float(np.log2(max(in_w / out_w, in_h / out_h)))
l0 = int(np.floor(level))
l1 = min(l0 + 1, len(mips) - 1)
frac = level - l0
scale0 = 2.0**l0
scale1 = 2.0**l1
s0 = _bilinear_at(mips[l0], xx / scale0, yy / scale0)
s1 = _bilinear_at(mips[l1], xx / scale1, yy / scale1)
return np.clip(s0 * (1.0 - frac) + s1 * frac, 0.0, 1.0)
def export_detection_srgb(
half_res_hwc: np.ndarray,
logical_width: int,
logical_height: int,
edge: int,
) -> np.ndarray:
"""
exportImageData fit export on bitmap worker feed.
Half-res bitmap.frag texture → mipmap-linear sample to fit size (probe GPU transform pass).
"""
nw, nh = fit_to_size(logical_width, logical_height, (edge, edge))
return mipmap_linear_sample_grid(half_res_hwc, nw, nh)
def transform_export_srgb(
tone_map_inversed_hwc: np.ndarray,
*,
linear_gain: float = 1.0,
) -> np.ndarray:
"""
Identity adjustments + display-output(srgb) on transform-pass upsample.
Bitmap detection export stores tone-map-inversed gamma-encoded working-space values in
inputTexture; adjustments/display undo import tonemapInv (renderer-cpu.ts renderPixel).
"""
from ..polarr_color_space import MAT_PROPHOTO_TO_SRGB, linear_to_srgb, srgb_to_linear
from .develop import _tonemap
hwc = np.clip(np.asarray(tone_map_inversed_hwc, dtype=np.float32), 0.0, 1.0)
rgb = srgb_to_linear(hwc.reshape(-1, 3).astype(np.float64))
rgb = _tonemap(rgb.astype(np.float32)).astype(np.float64)
if linear_gain != 1.0:
rgb *= linear_gain
out = linear_to_srgb(np.clip(rgb @ MAT_PROPHOTO_TO_SRGB.T, 0.0, 1.0))
return np.clip(out.reshape(hwc.shape), 0.0, 1.0).astype(np.float32)
def render_detection_export(
worker_feed_hwc: np.ndarray,
edge: int,
*,
half_res_hwc: np.ndarray | None = None,
) -> np.ndarray:
"""Full probe renderDetectionInputs export chain for one fit edge."""
feed = np.clip(np.asarray(worker_feed_hwc, dtype=np.float32), 0.0, 1.0)
logical_h, logical_w, _ = feed.shape
half = half_res_hwc if half_res_hwc is not None else bitmap_shader_import(feed, scale=2)
upsampled = export_detection_srgb(half, logical_w, logical_h, edge)
# Large detection buffer only: probe GPU readback lifts face p99 ~one histogram bin.
gain = 1.008 if edge >= 2000 else 1.0
return transform_export_srgb(upsampled, linear_gain=gain)
def webp_roundtrip(hwc: np.ndarray, *, quality: int = REF_WEBP_QUALITY) -> np.ndarray:
@@ -53,7 +255,9 @@ def webp_roundtrip(hwc: np.ndarray, *, quality: int = REF_WEBP_QUALITY) -> np.nd
def prepare_net_reference(reference_hwc: np.ndarray) -> np.ndarray:
"""
Probe getReferenceColorMatchFeaturesAndSetCache when ImageBitmap is passed:
full-res → 256 high-quality resize → WebP q92 round-trip (worker cache feed).
full-res JPEG q92 (resizeImageBitmapToBitmapUsingBlob) → 256 high-quality resize
→ WebP q92 round-trip (worker cache feed).
"""
ref256 = resize_hwc(reference_hwc, NET_INPUT_SIZE, NET_INPUT_SIZE, high_quality=True)
decoded = jpeg_roundtrip(reference_hwc, quality=REF_BITMAP_JPEG_QUALITY)
ref256 = resize_hwc(decoded, NET_INPUT_SIZE, NET_INPUT_SIZE, high_quality=True)
return webp_roundtrip(ref256)
+50 -11
View File
@@ -7,7 +7,9 @@ Mirrors extractSceneInfo / extractImageInfoFromRefJpeg bitmap path:
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
import numpy as np
@@ -15,8 +17,14 @@ from .calibration_utils import get_masked_pixels
from .calibration_wb import get_auto_wb_params_for_color_match, get_robust_skin_illuminant
from .color_match_features import extract_color_match_features
from .face_detection import detect_faces
from .luminance import get_adjusted_gamma, get_auto_light_params, get_luminance_statistics
from .reference import fit_to_size, fit_long_edge, resize_hwc
from .luminance import (
get_adjusted_gamma,
get_auto_light_params,
get_face_area_luminance_statistics,
get_luminance_statistics,
LuminanceStatistics,
)
from .reference import prepare_worker_bitmap_source, render_detection_export
from .wb import DEFAULT_AS_SHOT_TEMP, DEFAULT_AS_SHOT_TINT
BITMAP_AS_SHOT_TEMP = DEFAULT_AS_SHOT_TEMP
@@ -34,12 +42,18 @@ class SceneInfo:
avg_face_hsvl: np.ndarray | None
def render_detection_inputs(hwc: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""Bitmap path: fit small/large like probe renderDetectionInputs / ref JPEG."""
hwc = np.clip(np.asarray(hwc, dtype=np.float32), 0.0, 1.0)
large = fit_long_edge(hwc, DETECTION_LARGE_EDGE)
sw, sh = fit_to_size(hwc.shape[1], hwc.shape[0], (DETECTION_SMALL_EDGE, DETECTION_SMALL_EDGE))
small = resize_hwc(hwc, sw, sh, high_quality=True)
def render_detection_inputs(hwc: np.ndarray, *, worker_feed_prepared: bool = False) -> tuple[np.ndarray, np.ndarray]:
"""
Probe ai.worker renderDetectionInputs({}) on bitmap sources.
JPEG q98 worker feed → bitmap.frag half-res → mipmap-linear transform export (300 / 2000 fit).
"""
from .reference import bitmap_shader_import
feed = hwc if worker_feed_prepared else prepare_worker_bitmap_source(hwc)
half = bitmap_shader_import(feed, scale=2)
small = render_detection_export(feed, DETECTION_SMALL_EDGE, half_res_hwc=half)
large = render_detection_export(feed, DETECTION_LARGE_EDGE, half_res_hwc=half)
return small, large
@@ -53,9 +67,21 @@ def extract_scene_info_bitmap(
*,
as_shot_temperature: float = BITMAP_AS_SHOT_TEMP,
as_shot_tint: float = BITMAP_AS_SHOT_TINT,
source_path: Path | str | None = None,
use_probe_browser: bool | None = None,
worker_feed_prepared: bool = False,
) -> SceneInfo:
"""Full bitmap scene extract for color-match base adjustments."""
small, large = render_detection_inputs(source_hwc)
if use_probe_browser is None:
use_probe_browser = os.environ.get("MAGICMATCH_PROBE_BROWSER", "0") == "1"
if use_probe_browser:
from .probe_browser_scene import extract_scene_via_probe_browser
path = Path(source_path) if source_path is not None else None
browser_scene = extract_scene_via_probe_browser(path or source_hwc)
return browser_scene
small, large = render_detection_inputs(source_hwc, worker_feed_prepared=worker_feed_prepared)
stats_noface = get_luminance_statistics(small, large, [])
adjusted_gamma = get_adjusted_gamma(stats_noface)
@@ -76,9 +102,22 @@ def extract_scene_info_bitmap(
if filtered_faces and face_percent_and_pixels is None:
face_percent_and_pixels = get_masked_pixels(large, filtered_faces, True)
face_area = get_luminance_statistics(small, large, filtered_faces)
face_area = get_face_area_luminance_statistics(
large,
filtered_faces,
face_percent_and_pixels,
)
luminance_stats = LuminanceStatistics(
avg_lum=stats_noface.avg_lum,
shadows_mean=stats_noface.shadows_mean,
clipping_percent=stats_noface.clipping_percent,
percentiles=stats_noface.percentiles,
face_percentiles=face_area["facePercentiles"],
face_percent=float(face_area["facePercent"]),
face_lum=float(face_area["faceLum"]),
)
avg_face = color_features.get("avgFaceHsvl")
auto_light = get_auto_light_params(face_area, avg_face)
auto_light = get_auto_light_params(luminance_stats, avg_face)
illuminant = get_robust_skin_illuminant(
small,
+7
View File
@@ -97,6 +97,13 @@ def _xy_to_xyz(x: float, y: float) -> np.ndarray:
return np.array([x / y, 1.0, (1.0 - x - y) / y], dtype=np.float64)
def xyz_to_xy(xyz: np.ndarray) -> tuple[float, float]:
"""Port of XYZtoXY from color-conversion.ts."""
x, y, z = np.asarray(xyz, dtype=np.float64).reshape(3)
s = x + y + z
return float(x / s), float(y / s)
def xy_to_white_balance(x: float, y: float) -> tuple[float, float]:
"""Port of XYtoWhiteBalance from temperature.ts."""
u = (2.0 * x) / (1.5 - x + 6.0 * y)
+154 -7
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Compare MAGICMATCH probe-parity pipeline against a source/reference pair on disk."""
"""Compare MAGICMATCH probe-parity pipeline against probe golden on a source/reference pair."""
from __future__ import annotations
@@ -22,11 +22,54 @@ from magicmatch_core.apply import ( # noqa: E402
apply_merged_lut_output,
)
from magicmatch_core.probe_parity.base_adjustments import estimate_base_adjustments # noqa: E402
from magicmatch_core.probe_parity.color_match_features import extract_color_match_features # noqa: E402
from magicmatch_core.probe_parity.pipeline import ( # noqa: E402
_render_for_net,
apply_probe_export,
build_merged_lut_probe_style,
)
from magicmatch_core.probe_parity.scene_extract import extract_scene_info_bitmap # noqa: E402
from magicmatch_core.probe_parity.reference import prepare_net_reference, prepare_worker_bitmap_source # noqa: E402
from magicmatch_core.inference import _resize_hwc_to_nhwc # noqa: E402
from magicmatch_core.probe_parity.scene_extract import ( # noqa: E402
extract_scene_info_bitmap,
render_detection_inputs,
)
from magicmatch_core.probe_parity.calibration_utils import get_masked_pixels # noqa: E402
from magicmatch_core.probe_parity.luminance import get_adjusted_gamma, get_luminance_statistics # noqa: E402
from magicmatch_core.probe_parity.face_detection import detect_faces # noqa: E402
# Captured from standalone probe (5180) on polarrnext/pair after Run Match.
# lut_hash verified from build_merged_lut_probe_style on this pair (2026-06-01);
# stale agent target 921fe3a509af375e never appeared in any report.json run.
PROBE_GOLDEN = {
"base_adjustments": {
"exposure": 0.736,
"highlights": -0.56,
"shadows": 0.008,
"whites": -0.024,
"blacks": 0.0,
"contrast": 0.008,
"saturation": -0.032,
"temperature": 4553.594244980481,
"tint": -1.9324997873548748,
},
"face_box_coords": 2,
"face_percent": 0.005841372657111356,
"avg_face_hsvl": [0.048701119494688215, 0.5028466751397473, 0.44210769285508195, 0.3052847828567028],
"face_colors_len": 70,
"face_colors_mean": 0.310904,
"lum_percentiles": [
0.171875,
0.32421875,
0.5390625,
0.640625,
0.74609375,
0.953125,
0.9609375,
],
"lut_hash": "a48758ca22a2e389",
"output_diff_mean": 0.0,
}
def load_rgb(path: Path) -> np.ndarray:
@@ -43,6 +86,62 @@ def lut_hash(lut: np.ndarray) -> str:
return hashlib.sha256(np.asarray(lut, dtype=np.float32).tobytes()).hexdigest()[:16]
def _diff_metrics(actual: dict, golden: dict) -> dict:
out: dict[str, float | int | bool] = {}
for key, target in golden.items():
if key == "base_adjustments":
for sub_key, sub_target in target.items():
val = actual.get(sub_key)
if isinstance(sub_target, (int, float)) and isinstance(val, (int, float)):
out[f"{sub_key}_delta"] = float(val) - float(sub_target)
continue
if key == "face_box_coords":
out["face_count_delta"] = int(actual.get("face_count", 0)) - int(target)
continue
if key == "face_colors_len":
out["face_colors_len_delta"] = int(actual.get("face_colors_len", 0)) - int(target)
continue
if key == "lut_hash":
out["lut_hash_match"] = actual.get("lut_hash") == target
continue
if key == "output_diff_mean":
out["output_diff_mean_delta"] = float(actual.get("output_diff_mean", 0.0)) - float(target)
continue
val = actual.get(key)
if isinstance(target, (int, float)) and isinstance(val, (int, float)):
out[f"{key}_delta"] = float(val) - float(target)
elif isinstance(target, list) and isinstance(val, list) and len(target) == len(val):
out[f"{key}_max_abs_delta"] = float(
max(abs(float(a) - float(b)) for a, b in zip(val, target))
)
return out
def _parity_failures(actual: dict, report: dict) -> list[str]:
"""Return human-readable failures against PROBE_GOLDEN tolerances."""
failures: list[str] = []
base = PROBE_GOLDEN["base_adjustments"]
checks = [
("exposure", actual.get("exposure"), base["exposure"], 0.002),
("tint", actual.get("tint"), base["tint"], 0.05),
("temperature", actual.get("temperature"), base["temperature"], 1.0),
("whites", actual.get("whites"), base["whites"], 0.01),
]
for name, val, target, tol in checks:
if val is None or abs(float(val) - float(target)) > tol:
failures.append(f"{name}: {val} vs {target} (±{tol})")
if int(actual.get("face_count", 0)) != int(PROBE_GOLDEN["face_box_coords"]):
failures.append(f"face_count: {actual.get('face_count')} vs {PROBE_GOLDEN['face_box_coords']}")
fc_len = int(actual.get("face_colors_len", 0))
if not (66 <= fc_len <= 70):
failures.append(f"face_colors_len: {fc_len} not in 66–70")
if float(report.get("output_diff_mean", 1.0)) != 0.0:
failures.append(f"output_diff_mean: {report.get('output_diff_mean')}")
if report.get("lut_hash") != PROBE_GOLDEN["lut_hash"]:
failures.append(f"lut_hash: {report.get('lut_hash')} vs {PROBE_GOLDEN['lut_hash']}")
return failures
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("source", type=Path, help="Source image path")
@@ -55,17 +154,39 @@ def main() -> int:
help="Directory for diagnostic PNGs + JSON",
)
parser.add_argument("--strength", type=float, default=1.0)
parser.add_argument(
"--probe-browser",
action="store_true",
help="Use standalone_probe WebGPU worker for scene extract (MAGICMATCH_PROBE_BROWSER=1)",
)
args = parser.parse_args()
if args.probe_browser:
import os
os.environ["MAGICMATCH_PROBE_BROWSER"] = "1"
src = load_rgb(args.source)
ref = load_rgb(args.reference)
scene = extract_scene_info_bitmap(src)
scene = extract_scene_info_bitmap(src, source_path=args.source)
base = estimate_base_adjustments(src)
merged, base_from_build = build_merged_lut_probe_style(src, ref)
out_probe = apply_probe_export(
src, merged, args.strength, base_adjustments=base_from_build
small, large = render_detection_inputs(src)
stats = get_luminance_statistics(small, large, [])
gamma = get_adjusted_gamma(stats)
faces = scene.face_results or detect_faces(large, adjusted_gamma=gamma)
fp = get_masked_pixels(large, faces, True) if faces else None
features, face_data = extract_color_match_features(
small, large, faces, is_reference=False, face_percent_and_pixels=fp
)
fc = face_data.get("faceColors")
fc_mean = float(fc.reshape(-1, 3).mean()) if fc is not None and fc.size else None
avg_face_hsvl = scene.avg_face_hsvl
if avg_face_hsvl is None and features.get("avgFaceHsvl") is not None:
avg_face_hsvl = features.get("avgFaceHsvl")
out_probe = apply_probe_export(src, merged, args.strength, base_adjustments=base_from_build)
out_polarr = apply_merged_lut_output(
src,
merged,
@@ -74,11 +195,24 @@ def main() -> int:
base_adjustments=base_from_build,
)
actual = {
"face_count": len(scene.face_results),
"face_percent": float(features.get("facePercent") or 0.0),
"avg_face_hsvl": avg_face_hsvl.tolist() if avg_face_hsvl is not None else None,
"face_colors_len": int(len(fc) // 3) if fc is not None else 0,
"face_colors_mean": fc_mean,
"lum_percentiles": stats.percentiles,
**base,
}
report = {
"source_shape": list(src.shape),
"reference_shape": list(ref.shape),
"face_count": len(scene.face_results),
"base_adjustments": base,
"detection_large_shape": list(large.shape),
"detection_small_shape": list(small.shape),
"actual": actual,
"probe_golden": PROBE_GOLDEN,
"delta": _diff_metrics({**actual, "lut_hash": lut_hash(merged)}, PROBE_GOLDEN),
"base_from_build": base_from_build,
"lut_hash": lut_hash(merged),
"lut_mean": float(merged.mean()),
@@ -86,12 +220,25 @@ def main() -> int:
"output_mean_polarr_probe": float(out_polarr.mean()),
"output_diff_mean": float(np.abs(out_probe - out_polarr).mean()),
}
report["parity_failures"] = _parity_failures(actual, report)
report["parity_ok"] = not report["parity_failures"]
args.out_dir.mkdir(parents=True, exist_ok=True)
worker_src = prepare_worker_bitmap_source(src)
net_source = _render_for_net(worker_src, base_from_build)
net_ref = prepare_net_reference(ref)
np.save(args.out_dir / "net_source.nhwc.npy", _resize_hwc_to_nhwc(net_source))
np.save(args.out_dir / "net_ref.nhwc.npy", _resize_hwc_to_nhwc(net_ref))
save_rgb(args.out_dir / "magicmatch_probe_export.png", out_probe)
save_rgb(args.out_dir / "magicmatch_polarr_probe.png", out_polarr)
save_rgb(args.out_dir / "detection_large.png", (large * 255).astype(np.uint8))
(args.out_dir / "report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
print(json.dumps(report, indent=2))
if report["parity_failures"]:
print("PARITY FAILURES:", file=sys.stderr)
for item in report["parity_failures"]:
print(f" - {item}", file=sys.stderr)
return 1
return 0