跳转到内容

检查中间图像:使用 .v 与 vipsdisp 调试流水线

复制本页

最终图像变暗时,问题可能发生在解码、Alpha 处理、合成或编码阶段。把选定阶段保存为原生 VIPS 格式(.v),再检查它的像素与属性,可以逐步缩小排查范围。本例故意在合成前多做一次预乘(premultiplication),用四个检查点找到颜色首次变化的位置。

输入是项目自有的 720 × 480 透明 SVG。背景固定为 #f0e6f2,合成空间显式设为 srgb,位置不变,也不缩放。

重复预乘使半透明紫色圆环与彩色圆点变暗的合成结果
错误:额外预乘后仍按直通 Alpha 合成720 × 480,43,713 字节,含 Alpha,4 通道

由 pyvips 3.2.0 / libvips 8.17.3 实际生成。

移除额外预乘后,紫色圆环和彩色圆点保持预期亮度的合成结果
修复:移除额外预乘720 × 480,49,524 字节,含 Alpha,4 通道

由 pyvips 3.2.0 / libvips 8.17.3 实际生成。

这里的 PNG 都由对应 .v 重新读取后,显式执行 rint().cast("uchar") 再保存:浮点值先舍入,再转换成 8 位无符号整数。它们是显示副本;查看精确浮点像素时使用下面的 .v。结果有四个波段,但合成后的 Alpha 全为 255,有 Alpha 通道不等于图像仍有透明像素。

普通图片导出可能转换像素类型、量化或舍弃不支持的字段。比如本例 premultiply() 返回的 float 像素包含小数;先导出 8 位 PNG,再转成 .v,无法恢复这些小数。

本例逐像素验证 .v 保存前后的一致性,同时检查尺寸、波段(bands)、像素类型(band format)、色彩解释(interpretation)、编码类型、分辨率、偏移,以及人为添加的 checkpoint-stage 字符串。浮点容差为 1e-5。这项验证不代表所有种类的元数据都已经覆盖。

准备 Python、pyvips 和 libvips;CLI 对照还需要 vips 命令。下载上面的 SVG,保存为 input.svg,把下面的完整源码保存为 main.py。pyvips Tab 是本例的 Reference;CLI Tab 使用同一个 Python 驱动调用 vips 操作,再执行同一组断言。它的每个中间步骤都会落盘,执行方式与 Python 惰性流水线不同。

运行命令

python3 main.py input.svg output/

完整源码

"""Save and verify native VIPS checkpoints; optionally build them with the CLI."""

import argparse
import json
import math
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

import pyvips


STAGES = ("source", "premultiplied", "wrong", "correct")
POINTS = ((238, 126), (502, 356), (360, 240), (0, 0))
BACKGROUND = (240, 230, 242)
TOLERANCE = 1e-5


def build_stage(input_path, stage):
    # Each checkpoint gets a fresh input, never a second consumer of a stream.
    source = pyvips.Image.new_from_file(str(input_path), access="random")
    if (source.width, source.height, source.bands) != (720, 480, 4):
        raise ValueError("expected the supplied 720x480 RGBA SVG")
    if stage == "source":
        return source
    if stage == "premultiplied":
        return source.premultiply()
    background = pyvips.Image.black(source.width, source.height).new_from_image(
        list(BACKGROUND)
    ).cast("float").copy(interpretation="srgb")
    # A float cast on the correct path keeps quantisation out of this comparison.
    overlay = source.premultiply() if stage == "wrong" else source.cast("float")
    return background.composite2(
        overlay, "over", compositing_space="srgb", premultiplied=False
    )


def build_cli_stage(input_path, stage, destination):
    """CLI operations materialise each intermediate; Python only drives them."""
    def vips(*args):
        subprocess.run(["vips", *map(str, args)], check=True, capture_output=True, text=True)

    with tempfile.TemporaryDirectory(prefix="vips-cli-checkpoint-") as directory:
        work = Path(directory)
        if stage == "source":
            vips("copy", input_path, destination)
        elif stage == "premultiplied":
            vips("premultiply", input_path, destination)
        else:
            vips("black", work / "black.v", 720, 480, "--bands", 3)
            vips("linear", work / "black.v", work / "background.v", "0 0 0", "240 230 242")
            vips("copy", work / "background.v", work / "srgb.v", "--interpretation", "srgb")
            if stage == "wrong":
                vips("premultiply", input_path, work / "overlay.v")
            else:
                vips("cast", input_path, work / "overlay.v", "float")
            # premultiplied defaults to false in the CLI; do not pass its true flag.
            vips("composite2", work / "srgb.v", work / "overlay.v", destination,
                 "over", "--compositing-space", "srgb")


def properties(image):
    return {name: image.get(name) for name in (
        "width", "height", "bands", "format", "coding", "interpretation",
        "xres", "yres", "xoffset", "yoffset",
    )}


def samples(image):
    return [{"x": x, "y": y, "values": [round(v, 6) for v in image.getpoint(x, y)]}
            for x, y in POINTS]


def require(condition, message):
    if not condition:
        raise ValueError(message)


def verify_roundtrip(original, restored, stage):
    before, after = properties(original), properties(restored)
    for name, value in before.items():
        matches = abs(value - after[name]) <= TOLERANCE if name in ("xres", "yres") else value == after[name]
        require(matches, f"{stage}: changed {name}: {value} -> {after[name]}")
    difference = (original.cast("double") - restored.cast("double")).abs().max()
    require(math.isfinite(difference) and difference <= TOLERANCE,
            f"{stage}: pixel roundtrip error {difference}")
    require(restored.get("checkpoint-stage") == stage, f"{stage}: missing stage metadata")


def verify_case(images):
    source, premultiplied = images["source"], images["premultiplied"]
    for x, y in POINTS:
        before, after = source.getpoint(x, y), premultiplied.getpoint(x, y)
        require(before[3] == after[3], "premultiply changed alpha")
        for channel in range(3):
            require(abs(after[channel] - before[channel] * before[3] / 255) <= TOLERANCE,
                    "premultiply did not scale RGB by alpha")
        wrong, correct = images["wrong"].getpoint(x, y), images["correct"].getpoint(x, y)
        require(wrong[3] == correct[3] == 255, "expected an opaque composite")
        if (x, y) == (0, 0):
            require(wrong[:3] == correct[:3] == list(BACKGROUND), "transparent control changed")
        else:
            alpha = before[3] / 255
            for channel in range(3):
                expected = before[channel] * alpha + BACKGROUND[channel] * (1 - alpha)
                require(abs(correct[channel] - expected) < 1e-4, "correct over pixel is wrong")
                require(correct[channel] - wrong[channel] > 10, "expected a visibly darker wrong pixel")


def view_checkpoint(image, viewer="vipsdisp"):
    executable = shutil.which(viewer)
    if executable is None:
        raise ValueError(f"viewer not installed: {viewer}")
    # A private directory avoids fixed /tmp names and filename races.
    with tempfile.TemporaryDirectory(prefix="vips-view-") as directory:
        filename = Path(directory) / "checkpoint.v"
        image.write_to_file(str(filename))
        subprocess.run([executable, str(filename)], check=True)
        # An existing viewer instance may accept the file and return immediately.
        input("Close this image in the viewer, then press Enter to remove the temporary file: ")


def check_saved(input_path, output_dir):
    """Validate persisted evidence too, including native files without byte baselines."""
    report = json.loads((output_dir / "checkpoints.json").read_text(encoding="utf-8"))
    images = {}
    require([entry["stage"] for entry in report["checkpoints"]] == list(STAGES),
            "unexpected checkpoint report stages")
    for stage, entry in zip(STAGES, report["checkpoints"]):
        original = build_stage(input_path, stage)
        restored = pyvips.Image.new_from_file(str(output_dir / f"{stage}.v"), access="random")
        verify_roundtrip(original, restored, stage)
        require(entry["path"] == f"{stage}.v" and entry["properties"] == properties(restored)
                and entry["samples"] == samples(restored) and entry["roundtrip"] == "passed"
                and entry["metadata"] == {"checkpoint-stage": stage}, f"{stage}: stale report")
        images[stage] = restored
    verify_case(images)
    for stage in ("wrong", "correct"):
        preview = pyvips.Image.new_from_file(str(output_dir / f"{stage}.png"), access="random")
        require((preview.cast("double") - images[stage].rint().cast("uchar")).abs().max() == 0,
                f"{stage}: stale display copy")
    return images


def generate(input_path, output_dir, use_cli=False):
    require(input_path.is_file(), "input file does not exist")
    output_dir.mkdir(parents=True, exist_ok=True)
    images, checkpoints = {}, []
    for stage in STAGES:
        original = build_stage(input_path, stage).copy()
        original.set_type(pyvips.GValue.gstr_type, "checkpoint-stage", stage)
        path = output_dir / f"{stage}.v"
        if use_cli:
            # The CLI snapshot is independently compared against the intended stage.
            # Add the same diagnostic string through a separate, lossless VIPS write.
            with tempfile.TemporaryDirectory(prefix="vips-stage-") as directory:
                candidate = Path(directory) / "stage.v"
                build_cli_stage(input_path, stage, candidate)
                cli_image = pyvips.Image.new_from_file(str(candidate), access="random").copy()
                cli_image.set_type(pyvips.GValue.gstr_type, "checkpoint-stage", stage)
                cli_image.write_to_file(str(path))
        else:
            original.write_to_file(str(path))
        restored = pyvips.Image.new_from_file(str(path), access="random")
        verify_roundtrip(original, restored, stage)
        images[stage] = restored
        checkpoints.append({"stage": stage, "path": path.name,
                            "properties": properties(restored), "samples": samples(restored),
                            "roundtrip": "passed", "metadata": {"checkpoint-stage": stage}})

    verify_case(images)
    outputs = []
    for stage in ("wrong", "correct"):
        # The PNG is a rounded 8-bit display copy, not the float checkpoint.
        preview = images[stage].rint().cast("uchar")
        path = output_dir / f"{stage}.png"
        preview.write_to_file(str(path), compression=6)
        decoded = pyvips.Image.new_from_file(str(path), access="random")
        require((decoded.cast("double") - preview.cast("double")).abs().max() == 0,
                f"{stage}: PNG differs from explicit display conversion")
        outputs.append({"id": stage, "path": path.name, "format": "png",
                        "width": decoded.width, "height": decoded.height, "bands": decoded.bands,
                        "hasAlpha": decoded.hasalpha(), "size": path.stat().st_size,
                        "parameters": {"compositing_space": "srgb", "premultiplied": False,
                                       "extraPremultiply": stage == "wrong", "preview": "rint + uchar"}})

    generator = {"pyvips": pyvips.__version__,
                 "libvips": ".".join(str(pyvips.version(i)) for i in range(3))}
    if use_cli:
        generator = {"vipsCli": generator["libvips"], "libvips": generator["libvips"],
                     "verificationPyvips": pyvips.__version__}
    for name, data in (
        ("result.json", {"recipeId": "pipeline-checkpoints", "generator": generator,
                         "input": input_path.name, "outputs": outputs}),
        ("checkpoints.json", {"generator": generator, "tolerance": TOLERANCE,
                              "checkpoints": checkpoints, "diagnosis": "extra-premultiply",
                              "scope": "image state only; not pipeline execution or performance"}),
    ):
        (output_dir / name).write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
    return images


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument("--cli", action="store_true", help="build snapshots with vips CLI operations")
    parser.add_argument("--check", action="store_true", help="verify saved evidence without writing files")
    parser.add_argument("--view", choices=STAGES, help="interactively inspect a private temporary copy")
    args = parser.parse_args()
    try:
        if args.check:
            images = check_saved(args.input.resolve(), args.output.resolve())
        else:
            images = generate(args.input.resolve(), args.output.resolve(), args.cli)
        if args.view:
            view_checkpoint(images[args.view])
    except (pyvips.Error, OSError, ValueError, subprocess.SubprocessError, EOFError) as error:
        print(f"pipeline-checkpoints: {error}", file=sys.stderr)
        if isinstance(error, subprocess.CalledProcessError) and error.stderr:
            print(error.stderr, file=sys.stderr)
        return 1
    print("pipeline-checkpoints: four native snapshots, two previews, all assertions passed")
    return 0


if __name__ == "__main__":
    sys.exit(main())

程序为每个检查点重新加载输入,而不是从同一条顺序流水线追加多个消费者。错误与正确合成都在浮点类型下计算,排除提前量化这个额外变量;程序的断言失败或文件读写失败时返回非零退出码。

从下表下载 .v,或打开本地运行生成的文件。安装方式见 vipsdisp 项目说明,然后运行:

终端窗口
vipsdisp output/premultiplied.v

从菜单 View → Properties(Alt+Return)核对波段、像素类型和元数据,再开启 View → Info bar。把光标移到表中指定的 (x, y) 像素,在底部信息栏读取坐标与 RGBA。坐标从左上角 (0, 0) 开始。显示亮度或透明棋盘有助于观察,诊断依据仍是数值与属性。

从实际保存的 .v 文件重新读取的属性
阶段尺寸波段/类型/色彩解释原始文件
原始透明图720 × 4804 / uchar / srgbsource.v
手动预乘后720 × 4804 / float / srgbpremultiplied.v
错误合成720 × 4804 / float / srgbwrong.v
正确合成720 × 4804 / float / srgbcorrect.v
像素 (238, 126) · RGBA
阶段实际值(最多三位小数)
原始透明图237, 118, 168, 174
手动预乘后161.718, 80.518, 114.635, 174
错误合成186.584, 128, 155.092, 255
正确合成237.953, 153.576, 191.506, 255
像素 (502, 356) · RGBA
阶段实际值(最多三位小数)
原始透明图133, 213, 113, 130
手动预乘后67.804, 108.588, 57.608, 130
错误合成152.214, 168.104, 147.996, 255
正确合成185.451, 221.333, 176.235, 255
像素 (360, 240) · RGBA
阶段实际值(最多三位小数)
原始透明图214, 229, 243, 219
手动预乘后183.788, 196.671, 208.694, 219
错误合成191.724, 201.376, 213.396, 255
正确合成217.671, 229.141, 242.859, 255
像素 (0, 0) · RGBA
阶段实际值(最多三位小数)
原始透明图0, 0, 0, 0
手动预乘后0, 0, 0, 0
错误合成240, 230, 242, 255
正确合成240, 230, 242, 255

下载完整属性、六位小数采样与验证结果(JSON)

没有 GUI 时也可以检查同一文件:

终端窗口
vipsheader --all output/premultiplied.v
vips getpoint output/premultiplied.v 238 126
vips getpoint output/wrong.v 238 126
vips getpoint output/correct.v 238 126

按 source → premultiplied → wrong 检查:手动预乘后 Alpha 不变,RGB 已经乘上 Alpha / 255;随后 composite2(..., premultiplied=False) 仍把叠加层当作直通 Alpha(straight alpha)输入,又在合成内部预乘一次。

对本例的 8 位来源,把归一化 Alpha 记作 a = Alpha / 255。在显式 sRGB 空间、完全不透明背景的条件下,正确路径每个颜色通道为 C × a + B × (1 − a);错误路径为 C × a² + B × (1 − a)。这不是通用的线性光合成建议。

修复就是移除合成前那次多余的 premultiply()。正确路径显式转成 float 后直接合成,Alpha 约定保持一致。(0, 0) 是完全透明的对照点,两条路径都应得到背景色;其余三个半透明点的错误 RGB 都明显更暗。程序同时检查这些关系及 PNG 与明确舍入结果的一致性。

下一次排查其他异常时,可以先保存疑似操作前后两个检查点,比较关键字段与相同坐标像素,再逐段缩小范围。不要只凭“看起来修好了”结束验证。

主命令把结果保留在你指定的输出目录,便于反复检查。需要临时副本时运行:

终端窗口
python3 main.py input.svg output/ --view premultiplied

查看模式通过 TemporaryDirectory 创建私有目录,以参数数组调用查看器,不使用固定 /tmp 文件名或 Shell 拼接。查看器可能把请求交给已运行的实例后立即退出,所以程序还会等待你确认:先在查看器中关闭该图像,再按 Enter 清理临时文件。启动失败、读取确认时遇到 EOF 或异常退出都会清理该私有目录。

保存、重新加载、继续执行的图像断点

Section titled “保存、重新加载、继续执行的图像断点”

维护者建议把检查点扩展成图像断点(image breakpoint):保存当前图像,打开查看器,再返回从 .v 重新加载的图像。调用方必须重新赋值,例如 image = image_breakpoint(image, "before vertical flip")。这样,后续操作读取的是实体化后的文件,不会沿着旧图像再次消费已读完的顺序输入。

这个流程也适合连续设置多个断点:JPEG 顺序加载 → 断点 1(保存并重载)→ 垂直翻转 → 断点 2(保存并重载)→ PNG 输出。这里刻意使用垂直翻转,因为它需要自底向上读取来源;.v 支持直接访问,可以满足这个读取顺序。

下载项目自有的 1938 × 1292 JPEG,保存为 input.jpg。把下方源码保存为 breakpoints.py,运行:

终端窗口
python3 breakpoints.py input.jpg flipped.png

每个断点都使用带序号和阶段名称的文件,便于识别当前停在哪一步。检查完后,先在查看器中关闭当前图像,再在终端按 Enter 继续。即使阶段名称相同,两次调用也不会覆盖同一个文件。需要无 GUI 验证时运行:

终端窗口
python3 breakpoints.py input.jpg flipped.png --no-viewer

--no-viewer 只跳过查看器与人工确认,仍执行两次 .v 保存和重新加载,并在最后逐像素比较输出与独立随机访问对照。

完整可运行源码:命名断点与顺序 JPEG 案例
breakpoints.py
"""Inspect and reload named image breakpoints before continuing a pipeline."""
import argparse
from contextlib import contextmanager
import json
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import pyvips
@contextmanager
def image_breakpoints(viewer="vipsdisp"):
"""Keep native files alive until all downstream evaluation has finished."""
executable = shutil.which(viewer) if viewer is not None else None
if viewer is not None and executable is None:
raise ValueError(f"viewer not installed: {viewer}")
with tempfile.TemporaryDirectory(prefix="vips-breakpoints-") as directory:
active = True
sequence = 0
def image_breakpoint(image, name="breakpoint"):
nonlocal sequence
if not active:
raise ValueError("breakpoint session has already ended")
sequence += 1
label = re.sub(r"[^A-Za-z0-9_-]+", "-", name).strip("-")[:64]
filename = Path(directory) / f"{sequence:02d}-{label or 'breakpoint'}.v"
# The directory is private; each call has a new filename, even for
# repeated labels. Never overwrite a file backing an earlier image.
image.write_to_file(str(filename))
if executable is not None:
subprocess.run([executable, str(filename)], check=True)
# vipsdisp may forward this file to an existing instance.
input(f"Inspect {name!r}, close this image in the viewer, "
"then press Enter to continue: ")
# Reassign the caller's image to this new source. Continuing from
# the old image would consume the original sequential stream again.
return pyvips.Image.new_from_file(str(filename), access="random")
try:
yield image_breakpoint
finally:
active = False
def run(input_path, output_path, viewer="vipsdisp"):
if input_path.resolve() == output_path.resolve():
raise ValueError("output must not overwrite the input")
with image_breakpoints(viewer) as image_breakpoint:
image = pyvips.Image.jpegload(str(input_path), access="sequential")
image = image_breakpoint(image, "before vertical flip")
image = image.flip("vertical")
image = image_breakpoint(image, "after vertical flip")
# Complete the sink while both lazy file-backed sources still exist.
image.pngsave(str(output_path))
# Independent decoding, rather than rereading the consumed sequential input.
reference = pyvips.Image.jpegload(str(input_path), access="random").flip("vertical")
actual = pyvips.Image.new_from_file(str(output_path), access="random")
fields = ("width", "height", "bands", "format", "interpretation")
if any(actual.get(field) != reference.get(field) for field in fields):
raise ValueError("output properties differ from the random-access control")
difference = (actual.cast("double") - reference.cast("double")).abs().max()
if difference != 0:
raise ValueError(f"output pixels differ from the control: {difference}")
return {
"pyvips": pyvips.__version__,
"libvips": ".".join(str(pyvips.version(i)) for i in range(3)),
"breakpoints": ["before vertical flip", "after vertical flip"],
"properties": {field: actual.get(field) for field in fields},
"maxPixelDifference": difference,
}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("input", type=Path, help="JPEG input")
parser.add_argument("output", type=Path, help="PNG output")
parser.add_argument("--no-viewer", action="store_true",
help="run the same materialisation and reloads without a GUI")
args = parser.parse_args()
try:
report = run(args.input, args.output, None if args.no_viewer else "vipsdisp")
except (pyvips.Error, OSError, ValueError, subprocess.SubprocessError, EOFError) as error:
print(f"Breakpoint failed: {error}", file=sys.stderr)
return 1
print(json.dumps(report, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())

2026-10-26 在 pyvips 3.2.0 / libvips 8.17.3 下,用上述 JPEG 实际验证:

路径 验证结果
顺序加载 → 仅保存副本 → 继续使用旧图像翻转 复现 out of order read
顺序加载 → 两个保存并重载的断点 → 输出 成功;1938 × 1292、3 波段、uchar、srgb
断点输出与独立随机访问翻转对照 全图最大像素差为 0

这是无 GUI 模式的执行证据;查看器的实际界面复核见下节。断点增加了未压缩文件的磁盘占用与读写,还替换了执行图的上游依赖。它能帮助观察多个阶段,但不能据此评价原流水线的流式行为、峰值内存或性能,也不能修复发生在第一个断点之前的读取顺序错误。

本例在 2026-10-24 使用 pyvips 3.2.0、vips CLI 8.17.3 与 libvips 8.17.3 实际生成。自动验证覆盖四个检查点的全图像素往返、上述字段、阶段字符串、四个采样点及显示副本转换;CLI 检查点也分别与预期阶段比较。其他 Recipe 的跨 binding 像素比较状态不因此改变。

查看器步骤使用 Flathub 的 vipsdisp 4.2.0 实际复核:四个文件均可打开,Properties 展示预期属性,Info bar 的 (238, 126) 浮点读数与此处证据一致。查看器自身的运行环境与生成环境分开记录,不用查看器的版本替代生成结果的 libvips 版本。

govips v2.20.0 与 sharp 0.35.5 都能进行 Alpha 合成,但本页所需的原生 .v 状态导出不在它们当前的公开保存 API 中。PNG 或 raw buffer 可以用于其他诊断,但不能替代本例验证的完整字段集合。本例不引入未发布的 govips API。

继续阅读顺序访问与实体化、Alpha 基础、多图层合成与 out of order read 排查。