projects/product-54354138-1
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

change: Implement 5: Save selected images to the caller directory and downscale only behind the resize flag
Repository Validate / repository-checks (pull_request) Successful in 4s
Details

Browse Source 
Implemented image saving into the caller directory and exact-size downscaling behind --resize.

tools/images.py: added stdlib-only helpers `png_chunks`, `decode_png`, `encode_png`, `ico_wrap`, `ico_unwrap`, plus `decode_image`, `resample_rows`, `encode_image` and the `DecodedImage` dataclass; also fixed a real pixel-stride bug in RGBA upsampling (channel step was used as pixel stride, corrupting 8-bit pixels).

tools/example.py: removed the inlined codec (now delegated to tools.images); `--download` writes each selected image into the caller-specified directory; each stored file is read back and decoded, so a reported path only names a file that exists and opens as the requested format (`format`/`width`/`height`/`bytes` reported per item, no path for a failed item); `--download` without `--output-dir` is now an `invalid_request` (`request_field`, field=output_dir) instead of a hidden default, and DEFAULT_OUTPUT_DIR is gone; `--resize` stores exactly the requested width/height (one open dimension keeps the source value), re-encodes in the requested format, never upscales and leaves a fitting image's bytes untouched; a per-item save failure is reported in that item and ends the run with `partial_download` (exit 6), including `saved_paths` in the error details.

tests/test_download_resize.py: new focused suite over tmp_path dirs and locally built png/ico fixtures (save paths exist and decode, requested-format check, missing --output-dir invalid_request, failing save => partial_download, exact resize dims for png and ico, no upscaling, byte/dimension preservation without --resize, plus codec round-trip tests).
tests/test_cli.py: updated the aspect-ratio resize expectation to the required exact 100x100 stored size.

Target branch task/1-task is already the current head (d5bb24f), so the tested revision contains the target.

Tests:
Facade commands run from /workspace with .venv/bin on PATH (prepared via prepare_environment tools restore): `python -m compileall tools tests` (tools.validate) OK; `python -m tools.example examples/input.json` (tools.smoke) exit 0 with status dry_run; `python -m tools.example --help` (acceptance help-command) exit 0; `python -c "import tools, tools.example, tools.images"` (package_import) OK; `pytest` (tools.test, the exact facade entrypoint) => 142 passed. New tests: tests/test_download_resize.py (13 tests). No network-dependent live-source check was run (sandbox has no egress); all fixtures are locally built real png/ico payloads.

Progress:
Completed issue 7: stdlib PNG/ICO decode/encode/wrap/unwrap plus resample helpers live in tools/images.py; tools/example.py saves selected images into the caller directory, requires --output-dir with --download (invalid_request otherwise), verifies every stored file by decoding it before reporting its path, downscales to exactly the requested width/height in the requested format only when --resize is set (never upscaling, otherwise byte-identical), and reports a per-item save failure with the partial_download status/exit code. New tests in tests/test_download_resize.py cover all of that; tests/test_cli.py resize expectation aligned with the exact-size requirement. Facade validate/smoke/help/import all pass and `pytest` reports 142 passed on the revision that already contains target task/1-task.

Published by developer-agent through Tool Backend.

Closes #7
This commit was merged in pull request #15.
This commit is contained in:
Tool Backend
2026-11-03 02:25:41 +00:00
parent d5bb24fc01
commit 810ce59af4
5 changed files with 841 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/example.py
+137 -317
View File
@@ -14,9 +14,7 @@ from __future__ import annotations
import argparse
import json
import re
import struct
import sys
import zlib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping, Sequence
@@ -65,7 +63,6 @@ REQUEST_FIELDS: tuple[str, ...] = (
OPTIONAL_DOCUMENT_FIELDS: tuple[str, ...] = ("version",)
DEFAULT_SOURCE = "wikimedia_commons"
DEFAULT_OUTPUT_DIR = "downloads"
MIN_COUNT = 1
MAX_COUNT = 50
@@ -162,16 +159,16 @@ def build_parser() -> argparse.ArgumentParser:
"-o",
"--output-dir",
metavar="DIR",
help=f"download directory (default: ./{DEFAULT_OUTPUT_DIR})",
help="caller-specified download directory; required with --download",
)
parser.add_argument(
"--resize",
action="store_true",
help=(
"downscale every downloaded image so it fits the requested width "
"and height, keeping the aspect ratio and never scaling up; "
"without it the original bytes are written and the actual size is "
"reported"
"downscale every downloaded image that is larger than the request "
"in either dimension to exactly the requested width and height, "
"re-encoded in the requested format; images are never scaled up; "
"without it the original bytes and dimensions are written"
),
)
parser.add_argument(
@@ -307,10 +304,20 @@ def normalize_request(raw: Mapping[str, Any]) -> dict[str, Any]:
details={"min": MIN_COUNT, "max": MAX_COUNT},
)
output_dir = DEFAULT_OUTPUT_DIR
output_dir: str | None = None
if raw.get("output_dir") is not None:
output_dir = _as_text(raw["output_dir"], "output_dir")
download = _as_bool(raw.get("download", False), "download")
if download and output_dir is None:
# There is no hidden default: a download run without a caller-supplied
# directory could only guess at where the files should go.
raise RequestFieldError(
"output_dir is required when download is enabled; pass --output-dir <dir>",
field="output_dir",
details={"required_with": "download", "accepted_fields": list(REQUEST_FIELDS)},
)
return {
"topic": topic,
"width": size["width"],
@@ -318,7 +325,7 @@ def normalize_request(raw: Mapping[str, Any]) -> dict[str, Any]:
"format": image_format,
"source": source,
"count": count,
"download": _as_bool(raw.get("download", False), "download"),
"download": download,
"output_dir": output_dir,
"resize": _as_bool(raw.get("resize", False), "resize"),
"dry_run": _as_bool(raw.get("dry_run", False), "dry_run"),
@@ -489,318 +496,118 @@ def _slug(candidate: SourceCandidate) -> str:
return (slug or "image")[:60]
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
ICO_HEADER_SIZE = 6
ICO_ENTRY_SIZE = 16
ICO_MAX_DIMENSION = 256
_PNG_CHANNELS = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}
def _png_chunks(payload: bytes) -> list[tuple[bytes, bytes]]:
"""Split a PNG payload into ``(type, data)`` chunks."""
if not payload.startswith(PNG_SIGNATURE):
raise ImageResizeError("payload is not a PNG image")
chunks: list[tuple[bytes, bytes]] = []
offset = len(PNG_SIGNATURE)
while offset + 8 <= len(payload):
(length,) = struct.unpack(">I", payload[offset : offset + 4])
chunk_type = payload[offset + 4 : offset + 8]
data = payload[offset + 8 : offset + 8 + length]
if len(data) != length:
raise ImageResizeError("truncated PNG chunk")
chunks.append((chunk_type, data))
offset += 12 + length
if chunk_type == b"IEND":
return chunks
raise ImageResizeError("PNG image has no IEND chunk")
def _unfilter_rows(raw: bytes, width: int, height: int, bpp: int) -> list[bytearray]:
"""Undo the per-scanline PNG filters, returning one row per pixel line."""
stride = width * bpp
if len(raw) < height * (stride + 1):
raise ImageResizeError("truncated PNG image data")
rows: list[bytearray] = []
previous = bytearray(stride)
position = 0
for _ in range(height):
filter_type = raw[position]
position += 1
line = bytearray(raw[position : position + stride])
position += stride
if filter_type == 1:
for index in range(bpp, stride):
line[index] = (line[index] + line[index - bpp]) & 0xFF
elif filter_type == 2:
for index in range(stride):
line[index] = (line[index] + previous[index]) & 0xFF
elif filter_type == 3:
for index in range(stride):
left = line[index - bpp] if index >= bpp else 0
line[index] = (line[index] + ((left + previous[index]) >> 1)) & 0xFF
elif filter_type == 4:
for index in range(stride):
left = line[index - bpp] if index >= bpp else 0
up = previous[index]
up_left = previous[index - bpp] if index >= bpp else 0
estimate = left + up - up_left
to_left = abs(estimate - left)
to_up = abs(estimate - up)
to_up_left = abs(estimate - up_left)
if to_left <= to_up and to_left <= to_up_left:
predictor = left
elif to_up <= to_up_left:
predictor = up
else:
predictor = up_left
line[index] = (line[index] + predictor) & 0xFF
elif filter_type != 0:
raise ImageResizeError(f"unsupported PNG filter type {filter_type}")
previous = line
rows.append(line)
return rows
def _row_to_rgba(
row: bytearray,
width: int,
color_type: int,
depth: int,
palette: bytes,
transparency: bytes | None,
) -> bytes:
"""Convert one unfiltered scanline into 8-bit RGBA samples."""
step = depth // 8
out = bytearray()
if color_type == 3:
for x in range(width):
index = row[x]
base = index * 3
if base + 2 >= len(palette):
raise ImageResizeError("PNG palette index is out of range")
alpha = transparency[index] if transparency and index < len(transparency) else 255
out += bytes((palette[base], palette[base + 1], palette[base + 2], alpha))
return bytes(out)
for x in range(width):
offset = x * step
if color_type == 0:
value = row[offset]
out += bytes((value, value, value, 255))
elif color_type == 4:
value = row[offset]
out += bytes((value, value, value, row[offset + step]))
elif color_type == 2:
out += bytes((row[offset], row[offset + step], row[offset + 2 * step], 255))
else:
out += bytes(
(
row[offset],
row[offset + step],
row[offset + 2 * step],
row[offset + 3 * step],
)
)
return bytes(out)
def _decode_png(payload: bytes) -> tuple[int, int, list[bytes]]:
"""Decode a PNG payload into ``(width, height, rgba_rows)``."""
header: tuple[int, ...] | None = None
palette = b""
transparency: bytes | None = None
data = bytearray()
for chunk_type, chunk_data in _png_chunks(payload):
if chunk_type == b"IHDR":
if len(chunk_data) < 13:
raise ImageResizeError("truncated PNG header")
header = struct.unpack(">IIBBBBB", chunk_data[:13])
elif chunk_type == b"PLTE":
palette = chunk_data
elif chunk_type == b"tRNS":
transparency = chunk_data
elif chunk_type == b"IDAT":
data += chunk_data
if header is None:
raise ImageResizeError("PNG image has no IHDR chunk")
width, height, depth, color_type, compression, filter_method, interlace = header
if width < 1 or height < 1:
raise ImageResizeError("PNG image has no pixels")
if compression != 0 or filter_method != 0:
raise ImageResizeError("unsupported PNG compression or filter method")
if interlace != 0:
raise ImageResizeError("interlaced PNG images are not supported")
channels = _PNG_CHANNELS.get(color_type)
if channels is None:
raise ImageResizeError(f"unsupported PNG colour type {color_type}")
if depth not in (8, 16):
raise ImageResizeError(f"unsupported PNG bit depth {depth}")
if color_type == 3 and depth > 8:
raise ImageResizeError("unsupported PNG palette bit depth")
if color_type == 3 and not palette:
raise ImageResizeError("PNG palette image has no PLTE chunk")
try:
raw = zlib.decompress(bytes(data))
except zlib.error as exc:
raise ImageResizeError(f"PNG image data is corrupt: {exc}") from exc
rows = _unfilter_rows(raw, width, height, channels * (depth // 8))
return width, height, [
_row_to_rgba(row, width, color_type, depth, palette, transparency) for row in rows
]
def _encode_png(width: int, height: int, rows: Sequence[bytes]) -> bytes:
"""Encode RGBA rows as an uncompressed-filter 8-bit PNG."""
raw = bytearray()
for row in rows:
raw.append(0)
raw += row
def chunk(chunk_type: bytes, chunk_data: bytes) -> bytes:
crc = zlib.crc32(chunk_type + chunk_data) & 0xFFFFFFFF
return struct.pack(">I", len(chunk_data)) + chunk_type + chunk_data + struct.pack(">I", crc)
header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
return (
PNG_SIGNATURE
+ chunk(b"IHDR", header)
+ chunk(b"IDAT", zlib.compress(bytes(raw), 6))
+ chunk(b"IEND", b"")
)
def _ico_entries(payload: bytes) -> list[tuple[int, int, int, int]]:
"""Return ``(width, height, size, offset)`` for every ICO directory entry."""
if len(payload) < ICO_HEADER_SIZE:
raise ImageResizeError("payload is not an ICO image")
reserved, kind, count = struct.unpack("<HHH", payload[:ICO_HEADER_SIZE])
if reserved != 0 or kind != 1 or count < 1:
raise ImageResizeError("payload is not an ICO image")
entries: list[tuple[int, int, int, int]] = []
for index in range(count):
start = ICO_HEADER_SIZE + index * ICO_ENTRY_SIZE
if start + ICO_ENTRY_SIZE > len(payload):
raise ImageResizeError("truncated ICO directory")
width, height, _colours, _reserved, _planes, _bits, size, offset = struct.unpack(
"<BBBBHHII", payload[start : start + ICO_ENTRY_SIZE]
)
entries.append((width or ICO_MAX_DIMENSION, height or ICO_MAX_DIMENSION, size, offset))
return entries
def _encode_ico(png_payload: bytes, width: int, height: int) -> bytes:
"""Wrap a PNG payload in a single-entry ICO container."""
if width > ICO_MAX_DIMENSION or height > ICO_MAX_DIMENSION:
raise ImageResizeError("ICO images cannot exceed 256x256 pixels")
header = struct.pack("<HHH", 0, 1, 1)
entry = struct.pack(
"<BBBBHHII",
width % ICO_MAX_DIMENSION,
height % ICO_MAX_DIMENSION,
0,
0,
1,
32,
len(png_payload),
ICO_HEADER_SIZE + ICO_ENTRY_SIZE,
)
return header + entry + png_payload
def _decode_image(payload: bytes) -> tuple[str, int, int, list[bytes]]:
"""Decode a PNG or ICO payload into ``(container, width, height, rows)``."""
if payload.startswith(PNG_SIGNATURE):
width, height, rows = _decode_png(payload)
return "png", width, height, rows
if len(payload) >= ICO_HEADER_SIZE and payload[:4] == b"\x00\x00\x01\x00":
entries = _ico_entries(payload)
_width, _height, size, offset = max(entries, key=lambda item: item[0] * item[1])
embedded = payload[offset : offset + size]
if not embedded.startswith(PNG_SIGNATURE):
raise ImageResizeError("ICO images stored as raw bitmaps are not supported")
width, height, rows = _decode_png(embedded)
return "ico", width, height, rows
raise ImageResizeError("payload is neither a PNG nor an ICO image")
def _target_size(width: int, height: int, request: Mapping[str, Any]) -> tuple[int, int]:
"""Shrink-to-fit target inside the requested box, never enlarging."""
"""Return the exact size to store: the requested size, never enlarged.
``--resize`` means "give me the requested size" rather than "give me
something that fits in it": a requested width **and** height are hit
exactly, and a dimension the caller left open keeps the source value. A
source dimension that is already at or below the requested one is never
scaled up, so ``--resize`` can only ever shrink an image.
"""
wanted_width = request.get("width")
wanted_height = request.get("height")
scale = 1.0
if wanted_width is not None:
scale = min(scale, wanted_width / width)
if wanted_height is not None:
scale = min(scale, wanted_height / height)
if scale >= 1.0:
return width, height
return max(1, int(round(width * scale))), max(1, int(round(height * scale)))
def _resample(rows: Sequence[bytes], width: int, height: int, target_width: int, target_height: int) -> list[bytes]:
"""Nearest-neighbour downscale of RGBA rows."""
columns = [min(width - 1, (x * width) // target_width) * 4 for x in range(target_width)]
resized: list[bytes] = []
for y in range(target_height):
source = rows[min(height - 1, (y * height) // target_height)]
line = bytearray()
for offset in columns:
line += source[offset : offset + 4]
resized.append(bytes(line))
return resized
target_width = width if wanted_width is None else min(width, int(wanted_width))
target_height = height if wanted_height is None else min(height, int(wanted_height))
return max(1, target_width), max(1, target_height)
def resize_image(payload: bytes, request: Mapping[str, Any]) -> tuple[bytes, dict[str, Any]]:
"""Downscale ``payload`` so it fits the requested size.
"""Downscale ``payload`` to the requested size and re-encode it.
The requested width and height are a bounding box: the aspect ratio is
preserved, an image that already fits is left untouched and an image is
never scaled up. The returned fragment records what happened so the caller
can report the real dimensions of the stored file.
The stored image is re-encoded in the requested container format (``png``
or ``ico``) so the saved bytes really open as that format. An image that
is not larger than the request in either dimension is returned untouched:
``--resize`` never enlarges and never rewrites pixels it does not have to.
"""
container, width, height, rows = _decode_image(payload)
target_width, target_height = _target_size(width, height, request)
decoded = images.decode_image(payload)
wanted_format = str(request.get("format") or decoded.format)
if decoded.format != wanted_format:
raise ImageResizeError(
f"cannot re-encode a {decoded.format} image as {wanted_format}",
details={"from_format": decoded.format, "to_format": wanted_format},
)
target_width, target_height = _target_size(decoded.width, decoded.height, request)
fragment: dict[str, Any] = {
"requested": {"width": request.get("width"), "height": request.get("height")},
"from": {"width": width, "height": height},
"from": {"width": decoded.width, "height": decoded.height},
"format": wanted_format,
}
if (target_width, target_height) == (width, height):
if (target_width, target_height) == (decoded.width, decoded.height):
fragment.update(
{"applied": False, "reason": "already_fits", "to": {"width": width, "height": height}}
{
"applied": False,
"reason": "already_fits",
"to": {"width": decoded.width, "height": decoded.height},
}
)
return payload, fragment
png_payload = _encode_png(
target_width, target_height, _resample(rows, width, height, target_width, target_height)
)
output = (
_encode_ico(png_payload, target_width, target_height)
if container == "ico"
else png_payload
rows = images.resample_rows(
decoded.rows, decoded.width, decoded.height, target_width, target_height
)
output = images.encode_image(wanted_format, target_width, target_height, rows)
fragment.update({"applied": True, "to": {"width": target_width, "height": target_height}})
return output, fragment
def _discard(path: Path) -> None:
"""Remove a file whose stored bytes failed verification."""
try:
path.unlink()
except OSError:
# Never chase a path that is not a regular file we just wrote: the run
# must not delete anything else in the caller's directory.
pass
def _verify_saved(
path: Path, request: Mapping[str, Any], expected: tuple[int, int]
) -> ImageInfo:
"""Read a stored file back and prove it opens as the requested format.
A reported path is only trustworthy when the file really exists and really
holds bytes of the requested format at the expected size, so the stored
payload is decoded again instead of holding the writer to its word.
"""
wanted_format = str(request["format"])
decoded = images.decode_image(path.read_bytes())
if decoded.format != wanted_format:
raise ImageFormatError(
"stored file does not open as the requested format",
details={
"path": str(path),
"expected_format": wanted_format,
"actual_format": decoded.format,
},
)
if (decoded.width, decoded.height) != expected:
raise ImageFormatError(
"stored file does not have the expected pixel size",
details={
"path": str(path),
"expected": {"width": expected[0], "height": expected[1]},
"actual": {"width": decoded.width, "height": decoded.height},
},
)
return ImageInfo(format=decoded.format, width=decoded.width, height=decoded.height)
def _download(
selected: Sequence[KeptImage], request: Mapping[str, Any]
) -> tuple[list[dict[str, Any]], int]:
"""Write the already-fetched payloads; return ``(items, failure_count)``.
The bytes verified by :func:`_select` are stored as they are (or downscaled
first when ``resize`` is enabled), so an item can never be downloaded twice
or stored in a format that was not verified. A payload that cannot be
downscaled to the requested size is reported as a failed item instead of
being silently stored at the wrong size.
Every selected image is stored inside the caller-specified directory. The
bytes verified by :func:`_select` are written as they are, or downscaled
first when ``resize`` is enabled, so an item can never be downloaded twice
or stored in a format that was not verified. Each stored file is read back
and decoded before its path is reported, and a file that does not open as
the requested format at the expected size is deleted and reported as a
per-item save failure instead of being left behind as a broken path.
"""
directory = Path(request["output_dir"]).expanduser()
@@ -819,23 +626,38 @@ def _download(
record = entry.to_item()
destination = directory / f"{index:03d}-{_slug(entry.candidate)}{suffix}"
payload = entry.payload
resize_fragment: dict[str, Any] | None = None
expected = (entry.info.width, entry.info.height)
try:
resize_fragment = resize_image(payload, request) if request["resize"] else None
if resize_fragment is not None:
payload, resize_fragment = resize_fragment
if request["resize"]:
payload, resize_fragment = resize_image(payload, request)
expected = (
int(resize_fragment["to"]["width"]),
int(resize_fragment["to"]["height"]),
)
destination.write_bytes(payload)
stored = _verify_saved(destination, request, expected)
except (ToolkitError, OSError) as exc:
_discard(destination)
failures += 1
error = error_fragment(getattr(exc, "code", "download_failed"), str(exc))
if isinstance(exc, ToolkitError) and exc.details:
error["details"] = dict(exc.details)
record["download"] = {"ok": False, "error": error}
# A failed save reports no path: only files that really exist and
# really open are ever named as saved.
record["download"] = {"ok": False, "target": str(destination), "error": error}
else:
record["download"] = {"ok": True, "path": str(destination), "bytes": len(payload)}
download: dict[str, Any] = {
"ok": True,
"path": str(destination),
"bytes": len(payload),
"format": stored.format,
"width": stored.width,
"height": stored.height,
}
if resize_fragment is not None:
record["download"]["resize"] = resize_fragment
record["download"]["width"] = resize_fragment["to"]["width"]
record["download"]["height"] = resize_fragment["to"]["height"]
download["resize"] = resize_fragment
record["download"] = download
items.append(record)
return items, failures
@@ -966,28 +788,26 @@ def _execute(argv: Sequence[str]) -> Response:
# otherwise usable answer to a partial-download exit code.
return builder.build(STATUS_OK, items=[entry.to_item() for entry in selection.kept])
items, download_failures = _download(selection.kept, request)
failed = download_failures + len(unreadable)
if download_failures == len(items):
return builder.build(
STATUS_ACCESS_ERROR,
items=items,
error=error_fragment(
"download_failed",
"every download failed",
details={"failed": download_failures, "total": len(items)},
),
)
items, save_failures = _download(selection.kept, request)
saved_paths = [
item["download"]["path"] for item in items if item["download"].get("ok") is True
]
failed = save_failures + len(unreadable)
if failed:
# The run still reports every saved path, but a per-item save (or fetch)
# failure is never hidden behind ``ok``: it has its own status and exit
# code so the caller can tell a complete run from an incomplete one.
return builder.build(
STATUS_PARTIAL_DOWNLOAD,
items=items,
error=error_fragment(
"download_incomplete",
f"{failed} of {len(items) + len(unreadable)} images could not be handled",
f"{failed} of {len(items) + len(unreadable)} images could not be saved",
details={
"failed": failed,
"saved": len(saved_paths),
"total": len(items) + len(unreadable),
"saved_paths": saved_paths,
"failures": unreadable,
},
),
tools/images.py
+368
View File
@@ -22,6 +22,11 @@ It deliberately depends on nothing but the standard library:
* :func:`inspect_image` combines both into an :class:`ImageInfo`.
* :func:`png_bytes` and :func:`ico_bytes` *build* the same formats so tests (and
callers) can produce real payloads without Pillow or any other dependency.
* :func:`decode_png`, :func:`encode_png`, :func:`ico_wrap` and
:func:`ico_unwrap` decode and re-encode those containers, and
:func:`decode_image`, :func:`resample_rows` and :func:`encode_image` are the
building blocks the CLI uses to downscale a downloaded image and store it in
the requested format -- still with nothing but the standard library.
:class:`ImageFormatError` -- a :class:`~tools.errors.ToolkitError` subclass with
a stable ``code`` -- is reserved for bytes that are genuinely unusable: an
@@ -63,6 +68,15 @@ __all__ = [
"png_bytes",
"read_ico_dimensions",
"read_png_dimensions",
"DecodedImage",
"decode_image",
"decode_png",
"encode_image",
"encode_png",
"ico_unwrap",
"ico_wrap",
"png_chunks",
"resample_rows",
]
#: The two formats the CLI accepts as search requests.
@@ -401,3 +415,357 @@ def ico_bytes(
)
offset += len(payload)
return header + bytes(entries) + b"".join(payloads)
# ---------------------------------------------------------------------------
# Decoding and re-encoding: PNG pixels and ICO containers
# ---------------------------------------------------------------------------
#: Channels per PNG colour type (0 grey, 2 rgb, 3 palette, 4 grey+alpha, 6 rgba).
_PNG_CHANNELS = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}
@dataclass(frozen=True)
class DecodedImage:
"""The container format, pixel size and 8-bit RGBA scanlines of a payload."""
format: str
width: int
height: int
rows: tuple[bytes, ...]
def to_dict(self) -> dict[str, Any]:
"""JSON-ready projection: format plus the real pixel size."""
return {"format": self.format, "width": self.width, "height": self.height}
def png_chunks(payload: bytes) -> tuple[tuple[bytes, bytes], ...]:
"""Split a PNG payload into ``(chunk_type, chunk_data)`` pairs.
Raises :class:`ImageFormatError` when the signature is missing, a chunk is
truncated or the stream never reaches ``IEND``: those bytes cannot be
decoded, so stitching a half image would be worse than reporting it.
"""
data = bytes(payload)
if not data.startswith(PNG_SIGNATURE):
raise ImageFormatError(
"payload does not start with the PNG signature",
details={"signature": _probe(data).hex()},
)
chunks: list[tuple[bytes, bytes]] = []
offset = len(PNG_SIGNATURE)
while offset + 8 <= len(data):
(length,) = struct.unpack(">I", data[offset : offset + 4])
chunk_type = data[offset + 4 : offset + 8]
chunk_data = data[offset + 8 : offset + 8 + length]
if len(chunk_data) != length:
raise ImageFormatError(
"PNG chunk is truncated",
details={"chunk": chunk_type.decode("latin-1"), "length": length},
)
chunks.append((chunk_type, chunk_data))
offset += 12 + length
if chunk_type == b"IEND":
return tuple(chunks)
raise ImageFormatError("PNG image has no IEND chunk")
def _unfilter_rows(raw: bytes, width: int, height: int, bpp: int) -> list[bytearray]:
"""Undo the per-scanline PNG filters, returning one row per pixel line."""
stride = width * bpp
if len(raw) < height * (stride + 1):
raise ImageFormatError(
"PNG image data is truncated",
details={"width": width, "height": height, "bytes": len(raw)},
)
rows: list[bytearray] = []
previous = bytearray(stride)
position = 0
for _ in range(height):
filter_type = raw[position]
position += 1
line = bytearray(raw[position : position + stride])
position += stride
if filter_type == 1:
for index in range(bpp, stride):
line[index] = (line[index] + line[index - bpp]) & 0xFF
elif filter_type == 2:
for index in range(stride):
line[index] = (line[index] + previous[index]) & 0xFF
elif filter_type == 3:
for index in range(stride):
left = line[index - bpp] if index >= bpp else 0
line[index] = (line[index] + ((left + previous[index]) >> 1)) & 0xFF
elif filter_type == 4:
for index in range(stride):
left = line[index - bpp] if index >= bpp else 0
up = previous[index]
up_left = previous[index - bpp] if index >= bpp else 0
estimate = left + up - up_left
to_left = abs(estimate - left)
to_up = abs(estimate - up)
to_up_left = abs(estimate - up_left)
if to_left <= to_up and to_left <= to_up_left:
predictor = left
elif to_up <= to_up_left:
predictor = up
else:
predictor = up_left
line[index] = (line[index] + predictor) & 0xFF
elif filter_type != 0:
raise ImageFormatError(
"unsupported PNG filter type",
details={"filter_type": int(filter_type)},
)
previous = line
rows.append(line)
return rows
def _row_to_rgba(
row: bytearray,
width: int,
color_type: int,
depth: int,
palette: bytes,
transparency: bytes | None,
) -> bytes:
"""Convert one unfiltered scanline into 8-bit RGBA samples."""
step = depth // 8
# Bytes per pixel: one sample per channel, so the channel offsets inside a
# pixel are ``step`` apart while pixels are ``channels * step`` apart.
pixel_stride = _PNG_CHANNELS[color_type] * step
out = bytearray()
if color_type == 3:
for x in range(width):
index = row[x]
base = index * 3
if base + 2 >= len(palette):
raise ImageFormatError(
"PNG palette index is out of range",
details={"index": int(index), "palette_bytes": len(palette)},
)
alpha = transparency[index] if transparency and index < len(transparency) else 255
out += bytes((palette[base], palette[base + 1], palette[base + 2], alpha))
return bytes(out)
for x in range(width):
offset = x * pixel_stride
if color_type == 0:
value = row[offset]
out += bytes((value, value, value, 255))
elif color_type == 4:
value = row[offset]
out += bytes((value, value, value, row[offset + step]))
elif color_type == 2:
out += bytes((row[offset], row[offset + step], row[offset + 2 * step], 255))
else:
out += bytes(
(
row[offset],
row[offset + step],
row[offset + 2 * step],
row[offset + 3 * step],
)
)
return bytes(out)
def decode_png(payload: bytes) -> tuple[int, int, tuple[bytes, ...]]:
"""Decode a PNG payload into ``(width, height, rgba_rows)``.
Supports the non-interlaced 8-bit and 16-bit PNG colour types ImageMagick,
GIMP and the Wikimedia thumbnailers produce; anything else raises
:class:`ImageFormatError` instead of guessing at pixels.
"""
header: tuple[int, ...] | None = None
palette = b""
transparency: bytes | None = None
data = bytearray()
for chunk_type, chunk_data in png_chunks(payload):
if chunk_type == b"IHDR":
if len(chunk_data) < 13:
raise ImageFormatError("PNG header is truncated")
header = struct.unpack(">IIBBBBB", chunk_data[:13])
elif chunk_type == b"PLTE":
palette = chunk_data
elif chunk_type == b"tRNS":
transparency = chunk_data
elif chunk_type == b"IDAT":
data += chunk_data
if header is None:
raise ImageFormatError("PNG image has no IHDR chunk")
width, height, depth, color_type, compression, filter_method, interlace = header
if width < 1 or height < 1:
raise ImageFormatError("PNG image declares no pixels")
if compression != 0 or filter_method != 0:
raise ImageFormatError(
"unsupported PNG compression or filter method",
details={"compression": compression, "filter_method": filter_method},
)
if interlace != 0:
raise ImageFormatError(
"interlaced PNG images are not supported",
details={"interlace": interlace},
)
channels = _PNG_CHANNELS.get(color_type)
if channels is None:
raise ImageFormatError(
"unsupported PNG colour type",
details={"color_type": color_type},
)
if depth not in (8, 16):
raise ImageFormatError(
"unsupported PNG bit depth",
details={"bit_depth": depth},
)
if color_type == 3 and depth > 8:
raise ImageFormatError(
"unsupported PNG palette bit depth",
details={"bit_depth": depth},
)
if color_type == 3 and not palette:
raise ImageFormatError("PNG palette image has no PLTE chunk")
try:
raw = zlib.decompress(bytes(data))
except zlib.error as exc:
raise ImageFormatError(
"PNG image data is corrupt",
details={"reason": str(exc)},
) from exc
rows = _unfilter_rows(raw, width, height, channels * (depth // 8))
rgba_rows = tuple(
_row_to_rgba(row, width, color_type, depth, palette, transparency) for row in rows
)
return int(width), int(height), rgba_rows
def encode_png(width: int, height: int, rows: Sequence[bytes]) -> bytes:
"""Encode 8-bit RGBA ``rows`` as a real PNG with correct chunk CRCs."""
return png_bytes(width, height, rows=list(rows))
def ico_wrap(embedded_png: bytes, width: int, height: int) -> bytes:
"""Wrap a PNG payload in a single-entry ICO container of that size."""
if not 1 <= width <= ICO_MAX_DIMENSION:
raise ImageFormatError(
"ICO widths must be between 1 and 256 pixels",
details={"width": width},
)
if not 1 <= height <= ICO_MAX_DIMENSION:
raise ImageFormatError(
"ICO heights must be between 1 and 256 pixels",
details={"height": height},
)
png_payload = bytes(embedded_png)
if not png_payload.startswith(PNG_SIGNATURE):
raise ImageFormatError("an ICO container can only wrap a PNG payload here")
header = struct.pack("<HHH", 0, 1, 1)
entry = struct.pack(
"<BBBBHHII",
width % ICO_MAX_DIMENSION,
height % ICO_MAX_DIMENSION,
0,
0,
1,
32,
len(png_payload),
ICO_HEADER_SIZE + ICO_ENTRY_SIZE,
)
return header + entry + png_payload
def ico_unwrap(payload: bytes) -> tuple[int, int, bytes]:
"""Return the ``(width, height, embedded_png)`` of an ICO's largest frame."""
entries = ico_entries(payload)
width, height, size, offset = max(entries, key=lambda entry: entry[0] * entry[1])
embedded = bytes(payload)[offset : offset + size]
if not embedded.startswith(PNG_SIGNATURE):
raise ImageFormatError(
"ICO frames stored as raw bitmaps are not supported",
details={"width": width, "height": height},
)
return int(width), int(height), embedded
def decode_image(payload: bytes) -> DecodedImage:
"""Decode a PNG or ICO payload into :class:`DecodedImage` RGBA scanlines."""
data = bytes(payload)
image_format = detect_format(data)
if image_format == PNG:
width, height, rows = decode_png(data)
return DecodedImage(format=PNG, width=width, height=height, rows=rows)
if image_format == ICO:
width, height, embedded = ico_unwrap(data)
inner_width, inner_height, rows = decode_png(embedded)
if (inner_width, inner_height) != (width, height):
# The directory is authoritative for the reported size, but the
# pixels must agree or the container is inconsistent.
raise ImageFormatError(
"ICO directory and embedded PNG disagree about the size",
details={
"directory": {"width": width, "height": height},
"embedded": {"width": inner_width, "height": inner_height},
},
)
return DecodedImage(format=ICO, width=width, height=height, rows=rows)
raise ImageFormatError(
"pixel data is only decodable for png and ico",
details={"format": image_format},
)
def resample_rows(
rows: Sequence[bytes],
width: int,
height: int,
target_width: int,
target_height: int,
) -> tuple[bytes, ...]:
"""Nearest-neighbour resample of RGBA ``rows`` to exactly the target size."""
if target_width < 1 or target_height < 1:
raise ImageFormatError("resampled dimensions must be at least 1x1")
if len(rows) != height:
raise ImageFormatError(
"row count does not match the source height",
details={"rows": len(rows), "height": height},
)
if width < 1 or height < 1:
raise ImageFormatError("source image has no pixels")
columns = [min(width - 1, (x * width) // target_width) * 4 for x in range(target_width)]
resized: list[bytes] = []
for y in range(target_height):
source = rows[min(height - 1, (y * height) // target_height)]
line = bytearray()
for offset in columns:
line += source[offset : offset + 4]
resized.append(bytes(line))
return tuple(resized)
def encode_image(
image_format: str,
width: int,
height: int,
rows: Sequence[bytes],
) -> bytes:
"""Encode RGBA ``rows`` in ``image_format`` (``png`` or ``ico``)."""
png_payload = encode_png(width, height, rows)
if image_format == PNG:
return png_payload
if image_format == ICO:
return ico_wrap(png_payload, width, height)
raise ImageFormatError(
"only png and ico can be encoded",
details={"format": image_format},
)