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

change: Implement 4: Filter candidates by real format and minimum size from downloaded bytes
Repository Validate / repository-checks (pull_request) Successful in 3s
Details

Browse Source 
Added tools/images.py with stdlib-only detection of the real image format and pixel dimensions from downloaded bytes (PNG signature + IHDR chunk, ICO ICONDIR/ICONDIRENTRY directory including the 0-means-256 encoding) plus stdlib-only builders (png_bytes, ico_bytes) so tests/sources can produce genuine payloads. Wired it into the CLI search path in tools/example.py: a bounded candidate window derived from --count (candidate_cap = count * 4, capped at 50) is fetched from the selected source, each payload is parsed, and only items whose actual format equals the requested png/ico and whose actual width and height are both at least the requested minimum are kept. Kept items report source, url, format and the actual width/height (plus size_bytes and metadata as extras); verified bytes are reused for --download instead of being fetched twice. Statuses stay distinct: ok (0), no_format (3) when hits were only other formats, no_size (4) when format-matching items were all too small, access_error (5) when bytes could not be fetched or are not readable images, partial_download (6) when some candidates could not be read. All metadata-based filtering (FORMAT_MATCHERS, _matches_format, _matches_size) was removed so only parsed bytes decide. Added tests/test_images.py with detector/builder unit tests and CLI tests using an injected fake byte-fetch callable (no network), and updated tests/test_cli.py fixtures to real image bytes where the previous metadata-only assumption no longer holds.

Tests:
Facade commands run with .venv/bin on PATH after prepare_environment(component=tools, mode=restore): `python -m compileall tools tests` (tools.validate) succeeded; `pytest` (tools.test) => 114 passed, including the new tests/test_images.py (31 tests); `python -m tools.example examples/input.json` (tools.smoke) exited 0 with a single dry_run JSON document; acceptance commands `python -m tools.example --help` exited 0 and `import tools` resolved. Target branch task/1-task is at the same commit as this work branch (git rev-list count 0 ahead / 0 behind), so it is already integrated; facade/protected paths untouched.

Progress:
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.

Published by developer-agent through Tool Backend.

Closes #6
This commit is contained in:
Tool Backend
2026-11-03 02:11:36 +00:00
parent 6a7538c07f
commit b544968cb8
5 changed files with 1093 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/example.py
+219 -60
View File
@@ -17,9 +17,11 @@ import re
import struct
import sys
import zlib
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence
from . import images
from .envelope import (
STATUS_ACCESS_ERROR,
STATUS_NO_FORMAT,
@@ -41,6 +43,7 @@ from .errors import (
UnknownSourceError,
UnsupportedFormatError,
)
from .images import ImageFormatError, ImageInfo
from .sources import DEFAULT_LIMIT, SourceCandidate, default_registry
PROG = "python -m tools.example"
@@ -66,11 +69,21 @@ DEFAULT_OUTPUT_DIR = "downloads"
MIN_COUNT = 1
MAX_COUNT = 50
#: MIME types (and filename suffixes) that count as a hit for each format.
FORMAT_MATCHERS: dict[str, tuple[str, ...]] = {
"png": ("image/png", ".png"),
"ico": ("image/x-icon", "image/vnd.microsoft.icon", "image/ico", ".ico"),
}
#: How many candidates are fetched per wanted image. Nothing a source reports
#: about a hit is trusted, so a candidate is only known to be usable once its
#: bytes have been downloaded and parsed; the source is therefore asked for a
#: window of ``count * SAMPLE_FACTOR`` hits and the window is cut short as soon
#: as ``count`` usable images have been found.
SAMPLE_FACTOR = 4
#: Hard upper bound on the number of candidates inspected in one invocation.
MAX_CANDIDATE_CAP = MAX_COUNT
def candidate_cap(count: int) -> int:
"""Return the bounded number of candidates to fetch for ``count`` images."""
wanted = max(MIN_COUNT, int(count))
return min(MAX_CANDIDATE_CAP, wanted * SAMPLE_FACTOR)
class _Parser(argparse.ArgumentParser):
@@ -312,49 +325,145 @@ def normalize_request(raw: Mapping[str, Any]) -> dict[str, Any]:
}
def _matches_format(candidate: SourceCandidate, image_format: str) -> bool:
matchers = FORMAT_MATCHERS[image_format]
mime = str(getattr(candidate, "mime_type", None) or "").split(";")[0].strip().lower()
if mime and mime in matchers:
return True
url = str(getattr(candidate, "file_url", "") or "").lower()
return any(url.endswith(suffix) for suffix in matchers if suffix.startswith("."))
def _meets_minimum_size(info: ImageInfo, request: Mapping[str, Any]) -> bool:
"""True when the *real* pixel size of an image is usable for the request.
def _matches_size(candidate: SourceCandidate, request: Mapping[str, Any]) -> bool:
"""True when ``candidate`` is usable for the requested target size.
The requested width and height are a target size, not an exact match: any
image that is at least that large is usable because it can always be
downscaled, while a smaller image would have to be scaled up and is
therefore rejected. ``--resize`` only decides whether the downloaded bytes
are downscaled; it never widens or narrows this filter.
The requested width and height are a minimum, not an exact match: any image
that is at least that large is usable because it can always be downscaled,
while a smaller image would have to be scaled up and is therefore rejected.
``--resize`` only decides whether the downloaded bytes are downscaled; it
never widens or narrows this filter.
"""
wanted_width = request.get("width")
wanted_height = request.get("height")
if wanted_width is None and wanted_height is None:
return True
actual_width = getattr(candidate, "width", None)
actual_height = getattr(candidate, "height", None)
if isinstance(actual_width, bool) or isinstance(actual_height, bool):
if wanted_width is not None and info.width < wanted_width:
return False
if not isinstance(actual_width, int) or not isinstance(actual_height, int):
return False
if wanted_width is not None and actual_width < wanted_width:
return False
if wanted_height is not None and actual_height < wanted_height:
if wanted_height is not None and info.height < wanted_height:
return False
return True
def _select(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[SourceCandidate], list[SourceCandidate]]:
"""Return ``(matching, rejected_by_size)`` split for the request."""
def _failure(candidate: SourceCandidate, fragment: Mapping[str, Any]) -> dict[str, Any]:
"""Describe one candidate whose bytes could not be turned into an image."""
by_format = [c for c in candidates if _matches_format(c, request["format"])]
matching = [c for c in by_format if _matches_size(c, request)]
rejected = [c for c in by_format if not _matches_size(c, request)]
return matching, rejected
return {
"source": str(getattr(candidate, "source_id", "") or ""),
"url": str(getattr(candidate, "file_url", "") or ""),
"error": dict(fragment),
}
@dataclass(frozen=True)
class KeptImage:
"""One candidate whose downloaded bytes really match the request."""
candidate: SourceCandidate
info: ImageInfo
payload: bytes
def to_item(self) -> dict[str, Any]:
"""Report ``source``, ``url``, ``format`` and the *actual* dimensions."""
source = str(getattr(self.candidate, "source_id", "") or "")
url = str(getattr(self.candidate, "file_url", "") or "")
item: dict[str, Any] = {
"source": source,
"url": url,
# Kept as an alias so existing consumers keep working unchanged.
"file_url": url,
"format": self.info.format,
"width": self.info.width,
"height": self.info.height,
"size_bytes": len(self.payload),
}
for key in ("title", "page_url", "mime_type"):
value = getattr(self.candidate, key, None)
if value is not None:
item[key] = value
return item
@dataclass(frozen=True)
class Selection:
"""Outcome of inspecting a bounded window of downloaded candidates."""
kept: tuple[KeptImage, ...] = ()
format_rejects: int = 0
size_rejects: int = 0
failures: tuple[dict[str, Any], ...] = ()
@property
def inspected(self) -> int:
"""Number of candidates whose bytes were read successfully."""
return len(self.kept) + self.format_rejects + self.size_rejects
def _select(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> Selection:
"""Fetch and inspect candidates, keeping only what the real bytes justify.
Every kept item has been downloaded and parsed: its format comes from the
container signature and its width/height come from the PNG ``IHDR`` or the
ICO directory, never from what the source claimed. A candidate that serves
other-format bytes counts as a format rejection, a candidate of the wanted
format that is too small counts as a size rejection, and a candidate whose
bytes cannot be fetched or parsed is recorded as a failure instead of being
silently dropped.
"""
wanted_format = str(request["format"])
limit = int(request["count"])
kept: list[KeptImage] = []
format_rejects = 0
size_rejects = 0
failures: list[dict[str, Any]] = []
for candidate in candidates:
if len(kept) >= limit:
break
try:
payload = candidate.fetch_bytes()
except ToolkitError as error:
failures.append(_failure(candidate, error.to_dict()))
continue
except OSError as error:
failures.append(
_failure(candidate, error_fragment("fetch_failed", str(error)))
)
continue
detected = images.detect_format(payload)
if detected is None:
failures.append(
_failure(
candidate,
error_fragment(
"image_unreadable",
"downloaded payload is not a recognised image format",
details={"bytes": len(payload)},
),
)
)
continue
if detected != wanted_format:
format_rejects += 1
continue
try:
info = images.inspect_image(payload)
except ImageFormatError as error:
failures.append(_failure(candidate, error.to_dict()))
continue
if not _meets_minimum_size(info, request):
size_rejects += 1
continue
kept.append(KeptImage(candidate=candidate, info=info, payload=payload))
return Selection(
kept=tuple(kept),
format_rejects=format_rejects,
size_rejects=size_rejects,
failures=tuple(failures),
)
def _slug(candidate: SourceCandidate) -> str:
@@ -665,12 +774,16 @@ def resize_image(payload: bytes, request: Mapping[str, Any]) -> tuple[bytes, dic
return output, fragment
def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[dict[str, Any]], int]:
"""Download ``candidates``; return ``(items, failure_count)``.
def _download(
selected: Sequence[KeptImage], request: Mapping[str, Any]
) -> tuple[list[dict[str, Any]], int]:
"""Write the already-fetched payloads; return ``(items, failure_count)``.
With ``resize`` enabled every stored file is downscaled to the requested
target size first. A payload that cannot be decoded or downscaled is
reported as a failed item rather than silently stored at the wrong size.
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.
"""
directory = Path(request["output_dir"]).expanduser()
@@ -685,11 +798,11 @@ def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any])
suffix = "." + str(request["format"])
items: list[dict[str, Any]] = []
failures = 0
for index, candidate in enumerate(candidates, start=1):
record = dict(candidate.to_dict())
destination = directory / f"{index:03d}-{_slug(candidate)}{suffix}"
for index, entry in enumerate(selected, start=1):
record = entry.to_item()
destination = directory / f"{index:03d}-{_slug(entry.candidate)}{suffix}"
payload = entry.payload
try:
payload = candidate.fetch_bytes()
resize_fragment = resize_image(payload, request) if request["resize"] else None
if resize_fragment is not None:
payload, resize_fragment = resize_fragment
@@ -756,7 +869,21 @@ def _request_from(argv: Sequence[str]) -> tuple[dict[str, Any], str, Any]:
def _execute(argv: Sequence[str]) -> Response:
"""Run one request and return the response that must be printed."""
"""Run one request and return the response that must be printed.
The search path downloads a bounded window of candidates and filters that
window on the parsed bytes, so the answer distinguishes three different
situations through both the status and the exit code:
* ``ok`` -- at least one image really is the requested format and at least
the requested width and height;
* ``no_format`` -- the source had hits, but none of them really are the
requested format;
* ``no_size`` -- images of the requested format existed, but all of them
are smaller than requested;
* ``access_error`` -- the source could not be searched or served bytes that
are not readable images, which is never reported as a "no match".
"""
request, source, registry = _request_from(argv)
builder = ResponseBuilder(source=source, request=request)
@@ -764,24 +891,36 @@ def _execute(argv: Sequence[str]) -> Response:
if request["dry_run"]:
return builder.dry_run()
candidates = registry.search(source, request["topic"], request["count"])
matching, rejected = _select(candidates, request)
cap = candidate_cap(request["count"])
candidates = registry.search(source, request["topic"], cap)
selection = _select(candidates, request)
unreadable = list(selection.failures)
if not matching:
if not selection.kept:
details = {
"requested_format": request["format"],
"requested_size": {"width": request["width"], "height": request["height"]},
"minimum_size": {"width": request["width"], "height": request["height"]},
"candidate_count": len(candidates),
"format_matches": len(matching) + len(rejected),
"candidates_inspected": selection.inspected,
"format_matches": selection.size_rejects,
"size_matches": 0,
"unreadable_candidates": len(unreadable),
}
if rejected:
if unreadable:
return builder.access_error(
error=error_fragment(
"candidate_unreadable",
"the source served images that could not be read",
details={**details, "failures": unreadable},
)
)
if selection.size_rejects:
return builder.build(
STATUS_NO_SIZE,
error=error_fragment(
"no_size_match",
"no image is at least the requested width and height",
"the source returned no images at least the requested width and height",
details=details,
),
)
@@ -795,27 +934,47 @@ def _execute(argv: Sequence[str]) -> Response:
)
if not request["download"]:
return builder.build(STATUS_OK, items=[c.to_dict() for c in matching])
if unreadable:
return builder.build(
STATUS_PARTIAL_DOWNLOAD,
items=[entry.to_item() for entry in selection.kept],
error=error_fragment(
"candidate_incomplete",
f"{len(unreadable)} of {selection.inspected + len(unreadable)} "
"candidates could not be read",
details={
"failed": len(unreadable),
"total": selection.inspected + len(unreadable),
"failures": unreadable,
},
),
)
return builder.build(STATUS_OK, items=[entry.to_item() for entry in selection.kept])
items, failures = _download(matching, request)
if failures and failures == len(items):
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": failures, "total": len(items)},
details={"failed": download_failures, "total": len(items)},
),
)
if failures:
if failed:
return builder.build(
STATUS_PARTIAL_DOWNLOAD,
items=items,
error=error_fragment(
"download_incomplete",
f"{failures} of {len(items)} downloads failed",
details={"failed": failures, "total": len(items)},
f"{failed} of {len(items) + len(unreadable)} images could not be handled",
details={
"failed": failed,
"total": len(items) + len(unreadable),
"failures": unreadable,
},
),
)
return builder.build(STATUS_OK, items=items)
tools/images.py
+367
View File
@@ -0,0 +1,367 @@
"""Stdlib-only inspection and construction of the image formats we accept.
The CLI decides what to keep from the **downloaded bytes** and never from the
metadata a source reports about a file: a search hit that claims to be a
``image/png`` of 800x600 pixels is kept only when the bytes really are a PNG of
at least the requested size. This module is the single place that knows how to
read the real format and the real pixel dimensions out of those bytes.
It deliberately depends on nothing but the standard library:
* :func:`detect_format` recognises the container signatures of the two accepted
formats (``png``, ``ico``) plus a handful of common formats (``jpeg``,
``gif``, ``bmp``, ``webp``). Recognising the *other* formats is what makes
the ``no_format`` outcome possible: "the source has images, but none of them
are the format you asked for" is a different answer from "the source has no
usable images at all".
* :func:`read_png_dimensions` reads the ``IHDR`` header of a PNG.
* :func:`read_ico_dimensions` reads the ICONDIR/ICONDIRENTRY directory of an
ICO file (the directory stores the real sizes, including the ``0``-means-256
encoding).
* :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.
Every failure raises :class:`ImageFormatError`, a :class:`~tools.errors.ToolkitError`
subclass whose ``code`` is stable so the CLI can report an unreadable payload as
an access problem instead of silently treating it as "no match".
"""
from __future__ import annotations
import struct
import zlib
from dataclasses import dataclass
from typing import Any, Iterable, Sequence
from .errors import ToolkitError
__all__ = [
"BMP",
"GIF",
"ICO",
"ICO_ENTRY_SIZE",
"ICO_HEADER_SIZE",
"ICO_MAX_DIMENSION",
"ImageFormatError",
"ImageInfo",
"JPEG",
"KNOWN_FORMATS",
"PNG",
"PNG_SIGNATURE",
"WEBP",
"detect_format",
"ico_bytes",
"ico_entries",
"inspect_image",
"png_bytes",
"read_ico_dimensions",
"read_png_dimensions",
]
#: The two formats the CLI accepts as search requests.
PNG = "png"
ICO = "ico"
#: Formats that are recognised (so they can be reported as "another format")
#: but never accepted by the CLI.
JPEG = "jpeg"
GIF = "gif"
BMP = "bmp"
WEBP = "webp"
#: Every format :func:`detect_format` can name.
KNOWN_FORMATS: tuple[str, ...] = (PNG, ICO, JPEG, GIF, BMP, WEBP)
#: The 8-byte signature every PNG file starts with.
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
#: ICONDIR: reserved (0), type (1 = icon) and the entry count.
ICO_HEADER_SIZE = 6
#: One ICONDIRENTRY: 16 bytes of width/height/colours/planes/bpp/size/offset.
ICO_ENTRY_SIZE = 16
#: A single ICO dimension byte of 0 encodes 256 pixels.
ICO_MAX_DIMENSION = 256
#: Magic prefixes of the recognised but unsupported formats.
_JPEG_SIGNATURE = b"\xff\xd8\xff"
_GIF_SIGNATURES = (b"GIF87a", b"GIF89a")
_BMP_SIGNATURE = b"BM"
_RIFF_SIGNATURE = b"RIFF"
_WEBP_SIGNATURE = b"WEBP"
#: Number of payload bytes worth quoting in an error detail.
_SIGNATURE_PROBE = 16
class ImageFormatError(ToolkitError):
"""Raised when downloaded bytes are not a readable image of a known format.
The bytes could not be attributed to any format the toolkit understands, or
they claim to be an accepted format but their header is truncated/corrupt.
It derives from :class:`~tools.errors.ToolkitError`, so the CLI reports it
as an access problem (the source handed us something unusable) rather than
as a "no such format" / "no such size" verdict about the topic.
"""
code = "image_unreadable"
def __init__(self, message: str, *, details: dict[str, Any] | None = None) -> None:
super().__init__(message, details=details)
@dataclass(frozen=True)
class ImageInfo:
"""The real format and real pixel size of one payload."""
format: str
width: int
height: int
def to_dict(self) -> dict[str, Any]:
"""JSON-ready projection used in CLI items."""
return {"format": self.format, "width": self.width, "height": self.height}
@property
def area(self) -> int:
return self.width * self.height
def _probe(payload: bytes) -> bytes:
return bytes(payload[:_SIGNATURE_PROBE])
def detect_format(payload: bytes) -> str | None:
"""Return the real container format of *payload*, or ``None``.
``None`` means "these bytes are not an image format the toolkit knows", which
is an access/payload problem. A recognised format other than ``png``/``ico``
is a perfectly good answer that the CLI turns into ``no_format``.
"""
if not isinstance(payload, (bytes, bytearray, memoryview)):
return None
data = bytes(payload)
if data.startswith(PNG_SIGNATURE):
return PNG
if data[:4] == b"\x00\x00\x01\x00":
return ICO
if data.startswith(_JPEG_SIGNATURE):
return JPEG
if data.startswith(_GIF_SIGNATURES):
return GIF
if data.startswith(_BMP_SIGNATURE) and len(data) >= 14:
return BMP
if data.startswith(_RIFF_SIGNATURE) and data[8:12] == _WEBP_SIGNATURE:
return WEBP
return None
def read_png_dimensions(payload: bytes) -> tuple[int, int]:
"""Return ``(width, height)`` from the ``IHDR`` chunk of a PNG payload."""
data = bytes(payload)
if not data.startswith(PNG_SIGNATURE):
raise ImageFormatError(
"payload does not start with the PNG signature",
details={"signature": _probe(data).hex()},
)
offset = len(PNG_SIGNATURE)
if len(data) < offset + 8:
raise ImageFormatError("PNG header is truncated")
(length,) = struct.unpack(">I", data[offset : offset + 4])
chunk_type = data[offset + 4 : offset + 8]
if chunk_type != b"IHDR":
raise ImageFormatError(
"PNG payload does not start with an IHDR chunk",
details={"first_chunk": chunk_type.decode("latin-1")},
)
if length != 13:
raise ImageFormatError(
"PNG IHDR chunk has an unexpected length",
details={"length": length},
)
header = data[offset + 8 : offset + 8 + length]
if len(header) != length:
raise ImageFormatError("PNG IHDR chunk is truncated")
width, height = struct.unpack(">II", header[:8])
if width < 1 or height < 1:
raise ImageFormatError(
"PNG image declares no pixels",
details={"width": width, "height": height},
)
return int(width), int(height)
def ico_entries(payload: bytes) -> tuple[tuple[int, int, int, int], ...]:
"""Return ``(width, height, size, offset)`` for every ICO directory entry."""
data = bytes(payload)
if len(data) < ICO_HEADER_SIZE:
raise ImageFormatError("ICO header is truncated")
reserved, kind, count = struct.unpack("<HHH", data[:ICO_HEADER_SIZE])
if reserved != 0 or kind != 1 or count < 1:
raise ImageFormatError(
"payload is not an ICO image",
details={"reserved": reserved, "type": kind, "count": count},
)
directory_end = ICO_HEADER_SIZE + count * ICO_ENTRY_SIZE
if len(data) < directory_end:
raise ImageFormatError(
"ICO directory is truncated",
details={"count": count, "bytes": len(data)},
)
entries: list[tuple[int, int, int, int]] = []
for index in range(count):
start = ICO_HEADER_SIZE + index * ICO_ENTRY_SIZE
raw_width, raw_height, _colours, _reserved, _planes, _bits, size, offset = struct.unpack(
"<BBBBHHII", data[start : start + ICO_ENTRY_SIZE]
)
width = raw_width or ICO_MAX_DIMENSION
height = raw_height or ICO_MAX_DIMENSION
if size < 1 or offset < directory_end or offset + size > len(data):
raise ImageFormatError(
"ICO directory entry points outside the payload",
details={"index": index, "size": size, "offset": offset, "bytes": len(data)},
)
entries.append((width, height, size, offset))
return tuple(entries)
def read_ico_dimensions(payload: bytes) -> tuple[int, int]:
"""Return the dimensions of the largest image stored in an ICO payload.
The ICONDIRENTRY directory is authoritative: it carries one width/height pair
per stored image, so no decoding of the embedded bitmap or PNG is needed.
"""
entries = ico_entries(payload)
width, height, _size, _offset = max(entries, key=lambda entry: entry[0] * entry[1])
return int(width), int(height)
def inspect_image(payload: bytes) -> ImageInfo:
"""Return the real format and pixel size of *payload*.
Raises :class:`ImageFormatError` when the bytes are not an image at all, or
when they are not one of the two formats whose dimensions can be read.
"""
image_format = detect_format(payload)
if image_format is None:
raise ImageFormatError(
"downloaded payload is not a recognised image format",
details={"signature": _probe(bytes(payload)).hex()},
)
if image_format == PNG:
width, height = read_png_dimensions(payload)
return ImageInfo(format=PNG, width=width, height=height)
if image_format == ICO:
width, height = read_ico_dimensions(payload)
return ImageInfo(format=ICO, width=width, height=height)
raise ImageFormatError(
f"pixel dimensions are only readable for png and ico, not {image_format}",
details={"format": image_format},
)
# ---------------------------------------------------------------------------
# Builders: the same formats, produced with the standard library only
# ---------------------------------------------------------------------------
def _png_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)
def png_bytes(
width: int,
height: int,
*,
rgba: tuple[int, int, int, int] = (255, 255, 255, 255),
rows: Sequence[bytes] | None = None,
level: int = 6,
) -> bytes:
"""Build a valid 8-bit RGBA PNG of exactly ``width`` x ``height`` pixels.
``rows`` may supply raw (filter-0) RGBA scanlines; otherwise every pixel is
filled with ``rgba``. The result is a real PNG file: a correct signature,
a 13-byte ``IHDR``, a zlib-compressed ``IDAT`` and an empty ``IEND``.
"""
if isinstance(width, bool) or isinstance(height, bool):
raise ValueError("width and height must be integers")
if width < 1 or height < 1:
raise ValueError("width and height must be >= 1")
if rows is None:
pixel = bytes(rgba)
line = pixel * width
raw = bytearray()
for _ in range(height):
raw.append(0)
raw += line
raw_bytes = bytes(raw)
else:
if len(rows) != height:
raise ValueError("rows must contain exactly height scanlines")
raw = bytearray()
for row in rows:
if len(row) != width * 4:
raise ValueError("each scanline must hold width * 4 bytes")
raw.append(0)
raw += row
raw_bytes = bytes(raw)
header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
return (
PNG_SIGNATURE
+ _png_chunk(b"IHDR", header)
+ _png_chunk(b"IDAT", zlib.compress(raw_bytes, level))
+ _png_chunk(b"IEND", b"")
)
def ico_bytes(
width: int,
height: int,
*,
rgba: tuple[int, int, int, int] = (255, 255, 255, 255),
frames: Iterable[tuple[int, int]] | None = None,
) -> bytes:
"""Build an ICO payload whose directory lists the requested sizes.
By default the ICO stores one embedded PNG of ``width`` x ``height`` pixels
and the directory entry repeats those dimensions (``0`` encodes 256). Pass
``frames`` to describe several directory entries instead, for example
``ico_bytes(48, 48, frames=[(16, 16), (48, 48)])``: the dimensions of the
*largest* entry then become the reported size, which is exactly what
:func:`read_ico_dimensions` returns.
"""
sizes = list(frames) if frames is not None else [(width, height)]
if not sizes:
raise ValueError("an ICO payload needs at least one frame")
payloads: list[bytes] = []
for frame_width, frame_height in sizes:
if not 1 <= frame_width <= ICO_MAX_DIMENSION:
raise ValueError("ICO widths must be between 1 and 256 pixels")
if not 1 <= frame_height <= ICO_MAX_DIMENSION:
raise ValueError("ICO heights must be between 1 and 256 pixels")
payloads.append(png_bytes(frame_width, frame_height, rgba=rgba))
header = struct.pack("<HHH", 0, 1, len(payloads))
offset = ICO_HEADER_SIZE + ICO_ENTRY_SIZE * len(payloads)
entries = bytearray()
for (frame_width, frame_height), payload in zip(sizes, payloads):
entries += struct.pack(
"<BBBBHHII",
frame_width % ICO_MAX_DIMENSION,
frame_height % ICO_MAX_DIMENSION,
0,
0,
1,
32,
len(payload),
offset,
)
offset += len(payload)
return header + bytes(entries) + b"".join(payloads)