def benchmark_configuration(reference, target, patch_size, max_disparity, direction="left_to_right", warmups=2, measured=10):
result = None
for _ in range(warmups):
result = run_match(reference, target, max_disparity=max_disparity, patch_size=patch_size, direction=direction)
runtimes = []
for _ in range(measured):
start = time.perf_counter()
result = run_match(reference, target, max_disparity=max_disparity, patch_size=patch_size, direction=direction)
runtimes.append(time.perf_counter() - start)
assert result is not None
median_runtime = float(np.median(runtimes))
return result, median_runtime, {
"median_runtime_seconds": median_runtime,
"min_runtime_seconds": float(np.min(runtimes)),
"max_runtime_seconds": float(np.max(runtimes)),
"warmup_runs": warmups,
"measured_runs": measured,
}
def as_float_tensor(array):
if isinstance(array, torch.Tensor):
return array.to(device=DEVICE, dtype=torch.float32)
return torch.as_tensor(array, dtype=torch.float32, device=DEVICE)
def tensor_to_numpy(array):
if isinstance(array, torch.Tensor):
return array.detach().cpu().numpy()
return np.asarray(array)
def masked_invalid_numpy(array):
return np.ma.masked_invalid(tensor_to_numpy(array))
def finite_quantile(array, q):
tensor = as_float_tensor(array)
values = tensor[torch.isfinite(tensor)]
return float(torch.quantile(values, q).item())
def finite_min(array):
tensor = as_float_tensor(array)
values = tensor[torch.isfinite(tensor)]
return float(values.min().item())
def finite_max(array):
tensor = as_float_tensor(array)
values = tensor[torch.isfinite(tensor)]
return float(values.max().item())
def box_filter(image, radius):
image_t = as_float_tensor(image)
if radius <= 0:
return image_t.clone()
kernel = 2 * radius + 1
padded = F.pad(image_t[None, None], (radius, radius, radius, radius), mode="replicate")
filtered = F.avg_pool2d(padded, kernel_size=kernel, stride=1)
return filtered[0, 0]
def disparity_to_depth(disparity, focal_length=FOCAL_LENGTH_PX, baseline=BASELINE_UNITS):
disparity_t = as_float_tensor(disparity)
depth = torch.full_like(disparity_t, float("nan"))
valid = torch.isfinite(disparity_t)
depth[valid] = focal_length * baseline / torch.clamp(disparity_t[valid], min=1e-3)
return depth if isinstance(disparity, torch.Tensor) else tensor_to_numpy(depth)
def depth_to_disparity(depth, focal_length=FOCAL_LENGTH_PX, baseline=BASELINE_UNITS):
depth_t = as_float_tensor(depth)
disparity = torch.full_like(depth_t, float("nan"))
valid = depth_t > 1e-6
disparity[valid] = focal_length * baseline / depth_t[valid]
return disparity if isinstance(depth, torch.Tensor) else tensor_to_numpy(disparity)
def create_synthetic_scene(height=HEIGHT, width=WIDTH, focal_length=FOCAL_LENGTH_PX, baseline=BASELINE_UNITS):
y, x = torch.meshgrid(
torch.arange(height, dtype=torch.float32, device=DEVICE),
torch.arange(width, dtype=torch.float32, device=DEVICE),
indexing="ij",
)
left = 0.28 + 0.10 * torch.sin(x / 6.0) + 0.08 * torch.cos(y / 8.0) + 0.04 * torch.sin((x + 1.3 * y) / 11.0)
disparity = torch.full((height, width), 2.2, dtype=torch.float32, device=DEVICE)
region = np.full((height, width), "background", dtype=object)
checker = (x > 16) & (x < 60) & (y > 18) & (y < 66)
left[checker] = 0.38 + 0.24 * torch.remainder(torch.floor(x[checker] / 5.0) + torch.floor(y[checker] / 5.0), 2.0)
disparity[checker] = 8.7
region[tensor_to_numpy(checker)] = "checker"
textureless = (x > 18) & (x < 64) & (y > 72) & (y < 91)
left[textureless] = 0.63
disparity[textureless] = 3.1
region[tensor_to_numpy(textureless)] = "textureless"
repeated = (x > 96) & (x < 138) & (y > 16) & (y < 52)
left[repeated] = 0.34 + 0.22 * torch.remainder(torch.floor((x[repeated] - 96.0) / 4.0), 2.0)
disparity[repeated] = 8.0
region[tensor_to_numpy(repeated)] = "repeated"
occluder = (x > 66) & (x < 76) & (y > 14) & (y < 82)
left[occluder] = 0.86 - 0.10 * torch.remainder(torch.floor((y[occluder] - 14.0) / 6.0), 2.0)
disparity[occluder] = 9.6
region[tensor_to_numpy(occluder)] = "occluder"
circle = (x - 88.0) ** 2 + (y - 38.0) ** 2 < 17.0 ** 2
left[circle] = 0.22 + 0.55 * torch.exp(-((x[circle] - 88.0) ** 2 + (y[circle] - 38.0) ** 2) / 140.0)
disparity[circle] = 4.6
region[tensor_to_numpy(circle)] = "circle"
ramp = (x > 78) & (x < 132) & (y > 60) & (y < 89)
left[ramp] = 0.25 + 0.30 * ((x[ramp] - 78.0) / (132.0 - 78.0))
disparity[ramp] = 3.5 + 2.8 * ((x[ramp] - 78.0) / (132.0 - 78.0))
region[tensor_to_numpy(ramp)] = "ramp"
left = torch.clamp(left, 0.0, 1.0)
depth = disparity_to_depth(disparity, focal_length, baseline)
right = torch.full_like(left, float("nan"))
right_disp = torch.full_like(disparity, float("-inf"))
source_visible = torch.zeros((height, width), dtype=torch.bool, device=DEVICE)
invalid_projection = torch.zeros((height, width), dtype=torch.bool, device=DEVICE)
source_xr = torch.full((height, width), -1, dtype=torch.long, device=DEVICE)
for yy in range(height):
for xx in range(width):
d = float(disparity[yy, xx].item())
xr = int(round(xx - d))
source_xr[yy, xx] = xr
if xr < 0 or xr >= width:
invalid_projection[yy, xx] = True
continue
if d > float(right_disp[yy, xr].item()):
right_disp[yy, xr] = d
right[yy, xr] = left[yy, xx]
row_x = torch.arange(width, dtype=torch.float32, device=DEVICE)
for yy in range(height):
valid_cols = torch.isfinite(right[yy])
if int(valid_cols.sum().item()) >= 2:
xp = row_x[valid_cols]
fp = right[yy, valid_cols]
right_idx = torch.searchsorted(xp, row_x)
left_idx = torch.clamp(right_idx - 1, 0, xp.numel() - 1)
right_idx = torch.clamp(right_idx, 0, xp.numel() - 1)
x0 = xp[left_idx]
x1 = xp[right_idx]
y0 = fp[left_idx]
y1 = fp[right_idx]
denom = torch.where(torch.abs(x1 - x0) < 1e-6, torch.ones_like(x1), x1 - x0)
alpha = torch.where(
torch.abs(x1 - x0) < 1e-6,
torch.zeros_like(row_x),
(row_x - x0) / denom,
)
fill = y0 + alpha * (y1 - y0)
right[yy] = fill
elif int(valid_cols.sum().item()) == 1:
right[yy] = right[yy, valid_cols][0]
else:
right[yy] = 0.0
for yy in range(height):
for xx in range(width):
xr = int(source_xr[yy, xx].item())
if xr < 0 or xr >= width:
continue
source_visible[yy, xx] = abs(float(right_disp[yy, xr].item()) - float(disparity[yy, xx].item())) < 1e-6
occlusion_mask = ~source_visible
return {
"left": left,
"right": right,
"gt_disparity": disparity,
"gt_depth": depth,
"visible_mask": source_visible,
"occlusion_mask": occlusion_mask,
"invalid_projection": invalid_projection,
"region": region,
}
def shift_for_disparity(image, disparity, direction):
image_t = as_float_tensor(image)
shifted = torch.full_like(image_t, float("nan"))
valid = torch.zeros_like(image_t, dtype=torch.bool)
if disparity == 0:
shifted[:] = image_t
valid[:] = True
return shifted, valid
if direction == "left_to_right":
shifted[:, disparity:] = image_t[:, :-disparity]
valid[:, disparity:] = True
elif direction == "right_to_left":
shifted[:, :-disparity] = image_t[:, disparity:]
valid[:, :-disparity] = True
else:
raise ValueError(f"Unsupported direction: {direction}")
return shifted, valid
def build_support_mask(height, width, radius, max_disparity, direction):
mask = torch.zeros((height, width), dtype=torch.bool, device=DEVICE)
y0, y1 = radius, height - radius
if direction == "left_to_right":
x0, x1 = radius + max_disparity, width - radius
elif direction == "right_to_left":
x0, x1 = radius, width - radius - max_disparity
else:
raise ValueError(f"Unsupported direction: {direction}")
if y1 > y0 and x1 > x0:
mask[y0:y1, x0:x1] = True
return mask
def build_cost_volume(reference, target, patch_size, max_disparity, direction):
reference_t = as_float_tensor(reference)
target_t = as_float_tensor(target)
radius = patch_size // 2
h, w = reference_t.shape
cost_volume = torch.full((max_disparity + 1, h, w), float("nan"), dtype=torch.float32, device=DEVICE)
support_mask = build_support_mask(h, w, radius, max_disparity, direction)
for d in range(max_disparity + 1):
aligned, aligned_valid = shift_for_disparity(target_t, d, direction)
diff = (reference_t - aligned) ** 2
diff = torch.where(aligned_valid, diff, torch.full_like(diff, 1e3))
cost = box_filter(diff, radius)
valid = support_mask & aligned_valid
layer = torch.full((h, w), float("nan"), dtype=torch.float32, device=DEVICE)
layer[valid] = cost[valid]
cost_volume[d] = layer
return cost_volume, support_mask
def disparity_from_cost_volume(cost_volume):
cost_volume_t = as_float_tensor(cost_volume)
filled = torch.where(torch.isfinite(cost_volume_t), cost_volume_t, torch.full_like(cost_volume_t, float("inf")))
best = torch.argmin(filled, dim=0).to(torch.float32)
valid = torch.isfinite(cost_volume_t).any(dim=0)
best = torch.where(valid, best, torch.full_like(best, float("nan")))
best_cost = torch.min(filled, dim=0).values
best_cost = torch.where(valid, best_cost, torch.full_like(best_cost, float("nan")))
return best, valid, best_cost
def compute_error_maps(pred_disparity, gt_disparity, gt_depth):
pred_disparity_t = as_float_tensor(pred_disparity)
gt_disparity_t = as_float_tensor(gt_disparity)
gt_depth_t = as_float_tensor(gt_depth)
disparity_error = torch.abs(pred_disparity_t - gt_disparity_t)
pred_depth = as_float_tensor(disparity_to_depth(pred_disparity_t))
depth_error = torch.abs(pred_depth - gt_depth_t)
return disparity_error, pred_depth, depth_error
def summarize_metrics(pred_disparity, gt_disparity, gt_depth, eval_mask, runtime_seconds, consistency_mask=None):
eval_mask_t = torch.as_tensor(eval_mask, dtype=torch.bool, device=DEVICE)
disparity_error, pred_depth, depth_error = compute_error_maps(pred_disparity, gt_disparity, gt_depth)
disp_values = disparity_error[eval_mask_t]
useful_depth = eval_mask_t & torch.isfinite(depth_error) & torch.isfinite(as_float_tensor(pred_disparity)) & (as_float_tensor(pred_disparity) > 0.25)
depth_values = depth_error[useful_depth]
metrics = {
"disparity_mae_px": float(disp_values.mean().item()),
"bad_pixel_rate_gt_1px": float((disp_values > BAD_PIXEL_THRESHOLD).to(torch.float32).mean().item()),
"valid_pixel_ratio": float(eval_mask_t.to(torch.float32).mean().item()),
"depth_rmse_scene_units": float(torch.sqrt((depth_values ** 2).mean()).item()),
"runtime_seconds": float(runtime_seconds),
}
if consistency_mask is not None:
consistency_mask_t = torch.as_tensor(consistency_mask, dtype=torch.bool, device=DEVICE)
metrics["left_right_consistency_rate"] = float(consistency_mask_t[eval_mask_t].to(torch.float32).mean().item())
return metrics, disparity_error, pred_depth, depth_error
def run_match(reference, target, max_disparity, patch_size, direction="left_to_right"):
start = time.perf_counter()
cost_volume, support_mask = build_cost_volume(reference, target, patch_size, max_disparity, direction)
pred_disparity, valid_mask, min_cost = disparity_from_cost_volume(cost_volume)
runtime = time.perf_counter() - start
return {
"cost_volume": cost_volume,
"support_mask": support_mask,
"pred_disparity": pred_disparity,
"valid_mask": valid_mask,
"min_cost": min_cost,
"runtime_seconds": runtime,
"direction": direction,
}
def sample_cost_curve(cost_volume, yy, xx):
return as_float_tensor(cost_volume)[:, yy, xx]
def bilateral_brightness_variant(left, right):
left_t = as_float_tensor(left)
right_t = as_float_tensor(right)
right_bright = torch.clamp(0.12 + 0.82 * right_t, 0.0, 1.0)
return left_t.clone(), right_bright
def compute_right_to_left_consistency(left_disp, right_disp, left_valid_mask, right_valid_mask, eval_mask, tol=1.0):
left_disp_t = as_float_tensor(left_disp)
right_disp_t = as_float_tensor(right_disp)
left_valid_t = torch.as_tensor(left_valid_mask, dtype=torch.bool, device=DEVICE)
right_valid_t = torch.as_tensor(right_valid_mask, dtype=torch.bool, device=DEVICE)
eval_mask_t = torch.as_tensor(eval_mask, dtype=torch.bool, device=DEVICE)
h, w = left_disp_t.shape
x_coords = torch.arange(w, device=DEVICE).view(1, w).expand(h, w)
disp_indices = torch.round(torch.where(torch.isfinite(left_disp_t), left_disp_t, torch.zeros_like(left_disp_t))).to(torch.long)
xr = x_coords - disp_indices
in_bounds = (xr >= 0) & (xr < w)
xr_safe = xr.clamp(0, w - 1)
sampled_right_disp = torch.gather(right_disp_t, 1, xr_safe)
sampled_right_valid = torch.gather(right_valid_t.to(torch.int64), 1, xr_safe).to(torch.bool)
consistency = (
eval_mask_t
& left_valid_t
& torch.isfinite(left_disp_t)
& in_bounds
& sampled_right_valid
& torch.isfinite(sampled_right_disp)
& (torch.abs(left_disp_t - sampled_right_disp) <= tol)
)
return consistency
def fit_subpixel_quadratic(cost_curve, best_disp):
curve_t = as_float_tensor(cost_curve)
if best_disp <= 0 or best_disp >= len(curve_t) - 1:
return float(best_disp), None
c1 = float(curve_t[best_disp - 1].item())
c2 = float(curve_t[best_disp].item())
c3 = float(curve_t[best_disp + 1].item())
if not np.all(np.isfinite([c1, c2, c3])):
return float(best_disp), None
denom = c1 - 2.0 * c2 + c3
if abs(denom) < 1e-9:
return float(best_disp), None
offset = 0.5 * (c1 - c3) / denom
refined = float(best_disp + offset)
xs = np.linspace(best_disp - 1, best_disp + 1, 200)
ys = c2 + 0.5 * denom * (xs - best_disp) ** 2 + 0.5 * (c3 - c1) * (xs - best_disp)
return refined, (xs, ys)
def create_constant_disparity_pair(height=72, width=112, disparity_px=4):
y, x = torch.meshgrid(
torch.arange(height, dtype=torch.float32, device=DEVICE),
torch.arange(width, dtype=torch.float32, device=DEVICE),
indexing="ij",
)
left = 0.28 + 0.22 * torch.sin(x / 3.7) + 0.18 * torch.cos(y / 5.1) + 0.11 * torch.sin((1.9 * x + 0.8 * y) / 6.3)
left = torch.clamp(left, 0.0, 1.0)
right = torch.zeros_like(left)
right[:, :-disparity_px] = left[:, disparity_px:]
left_visible_mask = torch.zeros((height, width), dtype=torch.bool, device=DEVICE)
left_visible_mask[:, disparity_px:] = True
right_visible_mask = torch.zeros((height, width), dtype=torch.bool, device=DEVICE)
right_visible_mask[:, : width - disparity_px] = True
return {
"left": left,
"right": right,
"left_visible_mask": left_visible_mask,
"right_visible_mask": right_visible_mask,
"disparity_px": float(disparity_px),
}
def run_constant_disparity_sanity_test():
sanity = create_constant_disparity_pair()
left_match = run_match(
sanity["left"],
sanity["right"],
max_disparity=8,
patch_size=7,
direction="left_to_right",
)
right_match = run_match(
sanity["right"],
sanity["left"],
max_disparity=8,
patch_size=7,
direction="right_to_left",
)
left_eval_mask = sanity["left_visible_mask"] & left_match["valid_mask"]
right_eval_mask = sanity["right_visible_mask"] & right_match["valid_mask"]
consistency = compute_right_to_left_consistency(
left_match["pred_disparity"],
right_match["pred_disparity"],
left_match["valid_mask"],
right_match["valid_mask"],
left_eval_mask,
tol=0.5,
)
median_left = float(torch.median(left_match["pred_disparity"][left_eval_mask]).item())
median_right = float(torch.median(right_match["pred_disparity"][right_eval_mask]).item())
consistency_rate = float(consistency[left_eval_mask].to(torch.float32).mean().item())
print(
f"Controlled 4 px sanity test: median left-to-right disparity = {median_left:.3f} px, "
f"median right-to-left disparity = {median_right:.3f} px, "
f"left-right consistency rate = {consistency_rate:.3f}"
)
assert abs(median_left - 4.0) <= 0.25
assert abs(median_right - 4.0) <= 0.25
assert consistency_rate >= 0.95
return {
"median_left_disparity": median_left,
"median_right_disparity": median_right,
"left_right_consistency_rate": consistency_rate,
}
def warp_left_to_right_fractional(left_image, disparity_px):
left_t = as_float_tensor(left_image)
h, w = left_t.shape
ys, xs = torch.meshgrid(
torch.arange(h, dtype=torch.float32, device=DEVICE),
torch.arange(w, dtype=torch.float32, device=DEVICE),
indexing="ij",
)
source_x = xs + disparity_px
source_y = ys
valid = (source_x >= 0.0) & (source_x <= (w - 1))
grid_x = 2.0 * source_x / max(w - 1, 1) - 1.0
grid_y = 2.0 * source_y / max(h - 1, 1) - 1.0
grid = torch.stack((grid_x, grid_y), dim=-1)[None]
warped = F.grid_sample(
left_t[None, None],
grid,
mode="bilinear",
padding_mode="zeros",
align_corners=True,
)[0, 0]
return warped, valid
def run_fractional_disparity_experiment(ground_truth_disparity=8.7):
height, width = 48, 96
y, x = torch.meshgrid(
torch.arange(height, dtype=torch.float32, device=DEVICE),
torch.arange(width, dtype=torch.float32, device=DEVICE),
indexing="ij",
)
left = 0.31 + 0.24 * torch.sin(x / 2.9) + 0.19 * torch.cos(y / 4.1) + 0.12 * torch.sin((1.4 * x + 0.7 * y) / 5.3)
left = torch.clamp(left, 0.0, 1.0)
right, valid = warp_left_to_right_fractional(left, ground_truth_disparity)
match = run_match(left, right, max_disparity=12, patch_size=9, direction="left_to_right")
yy, xx = 22, 48
assert bool(valid[yy, xx].item()), "Fractional-disparity sample point fell outside the valid warp support."
curve = sample_cost_curve(match["cost_volume"], yy, xx)
best_disp = int(torch.argmin(torch.where(torch.isfinite(curve), curve, torch.full_like(curve, float("inf")))).item())
refined_disp, fitted = fit_subpixel_quadratic(curve, best_disp)
integer_error = abs(best_disp - ground_truth_disparity)
refined_error = abs(refined_disp - ground_truth_disparity)
print(
f"Fractional disparity experiment: gt = {ground_truth_disparity:.3f} px, "
f"integer estimate = {best_disp:.3f} px, refined estimate = {refined_disp:.3f} px, "
f"integer error = {integer_error:.3f} px, refined error = {refined_error:.3f} px"
)
assert refined_error < integer_error
assert refined_error < 0.15
return {
"left": left,
"right": right,
"valid_mask": valid,
"curve": curve,
"yy": yy,
"xx": xx,
"ground_truth_disparity": float(ground_truth_disparity),
"integer_disparity": float(best_disp),
"refined_disparity": float(refined_disp),
"integer_error": float(integer_error),
"refined_error": float(refined_error),
"fitted_curve": fitted,
}
def rot_x(theta):
c, s = np.cos(theta), np.sin(theta)
return np.array([[1.0, 0.0, 0.0], [0.0, c, -s], [0.0, s, c]], dtype=np.float64)
def rot_y(theta):
c, s = np.cos(theta), np.sin(theta)
return np.array([[c, 0.0, s], [0.0, 1.0, 0.0], [-s, 0.0, c]], dtype=np.float64)
def skew(vec):
tx, ty, tz = vec
return np.array([[0.0, -tz, ty], [tz, 0.0, -tx], [-ty, tx, 0.0]], dtype=np.float64)
def project_points(K, R, t, points):
pixels = []
for point in points:
cam = R @ point + t
pix = K @ cam
pixels.append(np.array([pix[0] / pix[2], pix[1] / pix[2], 1.0], dtype=np.float64))
return np.stack(pixels)
def line_segment_in_frame(line, width, height):
a, b, c = line
points = []
for x in [0.0, width - 1.0]:
if abs(b) > 1e-9:
y = -(a * x + c) / b
if -10.0 <= y <= height + 10.0:
points.append((x, y))
for y in [0.0, height - 1.0]:
if abs(a) > 1e-9:
x = -(b * y + c) / a
if -10.0 <= x <= width + 10.0:
points.append((x, y))
unique = []
for point in points:
if all((point[0] - other[0]) ** 2 + (point[1] - other[1]) ** 2 > 1e-6 for other in unique):
unique.append(point)
return unique[:2] if len(unique) >= 2 else None
scene = create_synthetic_scene()
eval_mask_template = scene["visible_mask"]
global_artifacts = {"sanity_test": run_constant_disparity_sanity_test()}