From 810ce59af4985898bb84ebdc1fa1099b0ef5c8d1 Mon Sep 17 00:00:00 2001 From: Tool Backend Date: Tue, 3 Nov 2026 02:25:41 +0000 Subject: [PATCH] change: Implement 5: Save selected images to the caller directory and downscale only behind the resize flag 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 --- .agent-work/ledger.json | 5 + tests/test_cli.py | 8 +- tests/test_download_resize.py | 326 ++++++++++++++++++++++++ tools/example.py | 454 ++++++++++------------------------ tools/images.py | 368 +++++++++++++++++++++++++++ 5 files changed, 841 insertions(+), 320 deletions(-) create mode 100644 tests/test_download_resize.py diff --git a/.agent-work/ledger.json b/.agent-work/ledger.json index 152f680..4ee58a1 100644 --- a/.agent-work/ledger.json +++ b/.agent-work/ledger.json @@ -21,6 +21,11 @@ "Implemented byte-based format and minimum-size filtering for the image search path. New tools/images.py: detect_format (png/ico plus jpeg/gif/bmp/webp so \"other format\" is distinguishable from \"unreadable\"), read_png_dimensions (IHDR), read_ico_dimensions (ICONDIR entries, 0 => 256, largest entry), inspect_image/ImageInfo, ImageFormatError (code image_unreadable), and png_bytes/ico_bytes builders. tools/example.py: candidate_cap(count) bounded window (count*4, max 50) requested from the source, _select() fetches and parses every candidate, KeptImage.to_item() reports source/url/format/actual width+height/size_bytes, _download() writes the already verified bytes, and _execute() distinguishes ok/no_format/no_size/access_error/partial_download with the existing exit codes; metadata-based filtering removed. tests/test_images.py added (31 tests) and tests/test_cli.py fixtures aligned to real image bytes. All facade commands pass: compileall, pytest (114 passed), smoke, help, package import.", "Addressed changes-requested review on PR #14: the CLI no longer turns common live-source results into partial_download/access_error. Any payload whose real format is not the requested png/ico (jpeg, gif, bmp, webp, svg, tiff, pdf, or an unrecognised body such as an HTML page) is now a format mismatch feeding no_format, with observed_formats/other_format_candidates detail; only unfetchable candidates are failures, and access_error is emitted only when nothing at all could be fetched. images.detect_format gained svg/tiff/pdf recognition and ImageFormatError is scoped to genuinely unusable png/ico payloads. Statuses/exits unchanged (no_format 3, no_size 4, access_error 5). Tests extended in tests/test_images.py (svg/tiff/pdf detection, mixed-format window stays ok, ico-only-foreign yields no_format, no_size preserved alongside other formats, non-image body is no_format, unfetchable candidate does not demote a search, corrupt png does not hide a match, empty window) and tests/test_cli.py aligned. Facade validate/test/smoke/help all green (129 passed) on the integrated branch; no protected facade paths touched." ] + }, + "projects/product-54354138-1#7": { + "events": [ + "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." + ] } } } diff --git a/tests/test_cli.py b/tests/test_cli.py index 9adbc3f..c01c902 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -477,11 +477,13 @@ def test_download_resize_downscales_stored_image( assert download["ok"] is True assert download["resize"]["applied"] is True assert download["resize"]["from"] == {"width": 200, "height": 100} - assert download["resize"]["to"] == {"width": 100, "height": 50} + # --resize stores exactly the requested width and height, whatever the + # source aspect ratio was. + assert download["resize"]["to"] == {"width": 100, "height": 100} stored = Path(download["path"]).read_bytes() - assert png_size(stored) == (100, 50) + assert png_size(stored) == (100, 100) assert download["width"] == 100 - assert download["height"] == 50 + assert download["height"] == 100 def test_download_without_resize_keeps_original_payload( diff --git a/tests/test_download_resize.py b/tests/test_download_resize.py new file mode 100644 index 0000000..9dcd614 --- /dev/null +++ b/tests/test_download_resize.py @@ -0,0 +1,326 @@ +"""Saving selected images into the caller directory and downscaling on demand. + +The contract under test: + +* ``--download`` writes every selected image into the directory named by the + caller and reports paths that really exist and really open as the requested + format; +* ``--download`` without ``--output-dir`` is an ``invalid_request`` -- there is + no hidden default directory; +* one item whose save fails is reported in that item while the run ends in the + distinct ``partial_download`` status (and exit code); +* ``--resize`` stores exactly the requested width and height, re-encoded in the + requested format, and never enlarges or otherwise rewrites an image that + already fits; without the flag the original bytes and dimensions are kept. + +Every fixture is a real payload built with :mod:`tools.images`, so the tests +assert on decoded files rather than on helper functions agreeing with +themselves. Nothing here touches the network. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Sequence + +import pytest + +from tools import envelope, example, images +from tools.errors import RequestFieldError + + +class FakeCandidate: + """Minimal stand-in for :class:`tools.sources.SourceCandidate`.""" + + def __init__(self, title: str, payload: bytes, *, fail: bool = False) -> None: + self.source_id = "fake" + self.title = title + self.file_url = f"https://example.test/{title.replace(' ', '-')}" + self.page_url = self.file_url + ".html" + self.mime_type = "image/png" + self.license = "CC0" + self.creator = "tester" + self.width: int | None = None + self.height: int | None = None + self._payload = payload + self._fail = fail + + def to_dict(self) -> dict[str, Any]: + return {"source_id": self.source_id, "title": self.title, "file_url": self.file_url} + + def fetch_bytes(self) -> bytes: + if self._fail: + raise OSError("download refused") + return self._payload + + +class FakeRegistry: + def __init__(self, candidates: Sequence[FakeCandidate] = ()) -> None: + self._candidates = list(candidates) + + def available_sources(self) -> Sequence[str]: + return ["fake"] + + def search(self, source_id: str, topic: str, limit: int = 10) -> list[FakeCandidate]: + return self._candidates[:limit] + + +def run_main( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + registry: FakeRegistry, + *args: str, +) -> tuple[int, dict[str, Any]]: + monkeypatch.setattr(example, "default_registry", lambda: registry) + code = example.main(list(args)) + captured = capsys.readouterr() + assert captured.err == "" + lines = [line for line in captured.out.splitlines() if line.strip()] + assert len(lines) == 1, captured.out + return code, json.loads(lines[0]) + + +def png(width: int, height: int) -> bytes: + return images.png_bytes(width, height) + + +def ico(width: int, height: int) -> bytes: + return images.ico_bytes(width, height) + + +def saved_image(path: Path) -> images.DecodedImage: + """Decode a reported path, failing the test if it cannot be opened.""" + + assert path.is_file(), f"reported path does not exist: {path}" + return images.decode_image(path.read_bytes()) + + +PNG_ARGS = ("--topic", "fox", "--format", "png", "--source", "fake", "--width", "100", "--height", "100") +ICO_ARGS = ("--topic", "fox", "--format", "ico", "--source", "fake", "--width", "50", "--height", "50") + + +# --------------------------------------------------------------------------- +# tools.images: the PNG/ICO helpers the saver is built on +# --------------------------------------------------------------------------- + + +def test_encode_and_decode_png_round_trip_the_requested_size() -> None: + rows = (bytes((10, 20, 30, 255)) * 4,) * 3 + + payload = images.encode_png(4, 3, rows) + + assert images.detect_format(payload) == images.PNG + decoded = images.decode_image(payload) + assert (decoded.format, decoded.width, decoded.height) == ("png", 4, 3) + assert decoded.rows == rows + + +def test_ico_wrap_and_unwrap_round_trip_the_embedded_png() -> None: + inner = png(6, 5) + + payload = images.ico_wrap(inner, 6, 5) + + assert images.detect_format(payload) == images.ICO + assert images.ico_unwrap(payload) == (6, 5, inner) + decoded = images.decode_image(payload) + assert (decoded.format, decoded.width, decoded.height) == ("ico", 6, 5) + + +def test_resample_rows_hits_the_target_size_exactly() -> None: + rows = tuple(bytes((x, x, x, 255)) * 4 for x in range(4)) + + resized = images.resample_rows(rows, 4, 4, 2, 3) + + assert len(resized) == 3 + assert {len(row) for row in resized} == {2 * 4} + + +def test_encode_image_re_encodes_both_accepted_formats() -> None: + rows = (bytes((1, 2, 3, 255)) * 2,) * 2 + + png_payload = images.encode_image("png", 2, 2, rows) + ico_payload = images.encode_image("ico", 2, 2, rows) + + assert images.decode_image(png_payload).format == images.PNG + assert images.decode_image(ico_payload).format == images.ICO + + +# --------------------------------------------------------------------------- +# --download: caller directory, reported paths, missing directory, save failure +# --------------------------------------------------------------------------- + + +def test_download_writes_into_the_caller_directory( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + registry = FakeRegistry([FakeCandidate("fox one", png(100, 100))]) + + code, document = run_main(monkeypatch, capsys, registry, *PNG_ARGS, "--download", "--output-dir", str(tmp_path)) + + assert code == envelope.EXIT_OK + assert document["status"] == envelope.STATUS_OK + download = document["items"][0]["download"] + assert download["ok"] is True + target = Path(download["path"]) + assert target.parent == tmp_path + assert target.suffix == ".png" + stored = saved_image(target) + assert (stored.format, stored.width, stored.height) == ("png", 100, 100) + assert download["format"] == "png" + assert (download["width"], download["height"]) == (100, 100) + assert sorted(path.name for path in tmp_path.iterdir()) == [target.name] + + +def test_download_without_output_dir_is_invalid_request( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + registry = FakeRegistry([FakeCandidate("fox one", png(100, 100))]) + + code, document = run_main(monkeypatch, capsys, registry, *PNG_ARGS, "--download") + + assert code == envelope.EXIT_INVALID_REQUEST + assert document["status"] == envelope.STATUS_INVALID_REQUEST + assert document["error"]["code"] == "request_field" + assert document["error"]["details"]["field"] == "output_dir" + # Nothing was written anywhere, least of all into a hidden default. + assert list(tmp_path.iterdir()) == [] + + +def test_request_document_without_output_dir_is_invalid_request() -> None: + with pytest.raises(RequestFieldError) as excinfo: + example.normalize_request({"topic": "fox", "format": "png", "download": True}) + + assert excinfo.value.field == "output_dir" + # Without --download the directory is not needed, so it stays optional. + request = example.normalize_request({"topic": "fox", "format": "png"}) + assert request["download"] is False + assert request["output_dir"] is None + + +def test_failing_save_yields_partial_download( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + registry = FakeRegistry( + [FakeCandidate("fox one", png(100, 100)), FakeCandidate("fox two", png(100, 100))] + ) + # The second destination is a directory, so writing the file must fail while + # the first item still saves successfully. + blocked = tmp_path / "002-fox-two.png" + blocked.mkdir() + + code, document = run_main( + monkeypatch, capsys, registry, *PNG_ARGS, "--count", "2", "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_PARTIAL_DOWNLOAD == 6 + assert document["status"] == envelope.STATUS_PARTIAL_DOWNLOAD + first, second = document["items"] + assert first["download"]["ok"] is True + assert saved_image(Path(first["download"]["path"])).width == 100 + assert second["download"]["ok"] is False + assert second["download"]["error"]["code"] + assert "path" not in second["download"] + assert document["error"]["code"] == "download_incomplete" + assert document["error"]["details"]["failed"] == 1 + # The path list only names files that exist and open as the requested format. + saved_paths = document["error"]["details"]["saved_paths"] + assert saved_paths == [first["download"]["path"]] + for path in saved_paths: + assert saved_image(Path(path)).format == "png" + + +# --------------------------------------------------------------------------- +# --resize: exact requested size, requested format, never enlarge +# --------------------------------------------------------------------------- + + +def test_resize_stores_exactly_the_requested_size( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + # 200x100 is larger than the 100x100 request in one dimension only: --resize + # still stores exactly 100x100 rather than preserving the aspect ratio. + registry = FakeRegistry([FakeCandidate("fox one", png(200, 100))]) + + code, document = run_main( + monkeypatch, capsys, registry, *PNG_ARGS, "--resize", "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert download["resize"]["applied"] is True + assert download["resize"]["from"] == {"width": 200, "height": 100} + assert download["resize"]["to"] == {"width": 100, "height": 100} + stored = saved_image(Path(download["path"])) + assert (stored.format, stored.width, stored.height) == ("png", 100, 100) + + +def test_resize_re_encodes_ico_in_the_requested_format( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + registry = FakeRegistry([FakeCandidate("fox one", ico(100, 100))]) + + code, document = run_main( + monkeypatch, capsys, registry, *ICO_ARGS, "--resize", "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + target = Path(download["path"]) + assert target.suffix == ".ico" + stored = saved_image(target) + assert (stored.format, stored.width, stored.height) == ("ico", 50, 50) + assert images.read_ico_dimensions(target.read_bytes()) == (50, 50) + + +def test_resize_never_enlarges_or_rewrites_a_fitting_image( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + payload = png(100, 100) + registry = FakeRegistry([FakeCandidate("fox one", payload)]) + + code, document = run_main( + monkeypatch, capsys, registry, *PNG_ARGS, "--resize", "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert download["resize"]["applied"] is False + assert download["resize"]["reason"] == "already_fits" + assert Path(download["path"]).read_bytes() == payload + assert (download["width"], download["height"]) == (100, 100) + + +def test_without_resize_the_original_bytes_and_dimensions_are_kept( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + payload = png(200, 100) + registry = FakeRegistry([FakeCandidate("fox one", payload)]) + + code, document = run_main( + monkeypatch, capsys, registry, *PNG_ARGS, "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert "resize" not in download + assert Path(download["path"]).read_bytes() == payload + stored = saved_image(Path(download["path"])) + assert (stored.format, stored.width, stored.height) == ("png", 200, 100) + + +def test_without_resize_an_ico_keeps_its_container_and_size( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + payload = ico(100, 100) + registry = FakeRegistry([FakeCandidate("fox one", payload)]) + + code, document = run_main( + monkeypatch, capsys, registry, *ICO_ARGS, "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert Path(download["path"]).read_bytes() == payload + stored = saved_image(Path(download["path"])) + assert (stored.format, stored.width, stored.height) == ("ico", 100, 100) diff --git a/tools/example.py b/tools/example.py index ce667c4..36fdb03 100644 --- a/tools/example.py +++ b/tools/example.py @@ -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 ", + 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(" len(payload): - raise ImageResizeError("truncated ICO directory") - width, height, _colours, _reserved, _planes, _bits, size, offset = struct.unpack( - " 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(" 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, }, ), diff --git a/tools/images.py b/tools/images.py index 5c1f31a..fd16617 100644 --- a/tools/images.py +++ b/tools/images.py @@ -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(" 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}, + )