diff --git a/.dockerignore b/.dockerignore index 9cc1876..c37dcdd 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,3 +1,6 @@ +# Local reports, captures and scratch files +.user/ + ### Python template # Byte-compiled / optimized / DLL files __pycache__/ diff --git a/.gitignore b/.gitignore index 1b2aedb..04497c6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +/.user/ __pycache__ .pytest_cache .DS_Store diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 2175768..d6a4660 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -23,23 +23,24 @@ repos: # flake8 python linter with all the fixins - repo: https://github.com/PyCQA/flake8 - rev: 6.0.0 + rev: 7.3.0 hooks: # additional configuration of flake8 and extensions in pyproject.toml - id: flake8 additional_dependencies: [ - flake8-annotations, - flake8-bandit, - flake8-bugbear, - flake8-black, - flake8-docstrings, - flake8-import-order, - darglint, - mypy, - pycodestyle, - pydocstyle, - Flake8-pyproject + flake8-annotations==3.2.0, + flake8-bandit==4.1.1, + bandit==1.9.4, + flake8-bugbear==24.12.12, + flake8-black==0.4.0, + black==23.1.0, + flake8-docstrings==1.7.0, + flake8-import-order==0.19.2, + darglint==1.8.1, + pycodestyle==2.14.0, + pydocstyle==6.3.0, + Flake8-pyproject==1.2.4 ] # removed unused imports and variables diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..5b999a6 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,104 @@ +# Repository guide + +Python 3.11 / Flask service that converts raster images to SVG using NumPy, +OpenCV, FAISS, scikit-image and Potrace. Requests can return raw SVG or upload it +to S3. Start with `README.md` for setup and usage. + +## Code map + +- `vectorizing/__init__.py`: app factory, request validation, cropping and API responses. +- `vectorizing/solvers/binary/`: binary masks and tracing. +- `vectorizing/solvers/color/`: quantization, cluster cleanup, ordered masks and tracing. +- `vectorizing/geometry/` and `vectorizing/svg/`: path conversion, bounds and SVG serialization. +- `vectorizing/util/`: image loading, color-mode normalization and size limits. +- `vectorizing/server/`: environment settings, S3, logging and timing. +- `vectorizing/tests/`: API contracts, image gallery and algorithm regressions. +- `scripts/`: local infrastructure, dependency generation and offline comparison/benchmark tools. + +## Development and checks + +Prefer Docker Compose; it supplies Moto S3 and needs no AWS account or `.env` file. +Run from the repository root: + +```sh +docker compose up --build +pre-commit run --all-files +docker compose --profile testing run --build --rm test +``` + +For a focused test run: + +```sh +docker compose --profile testing run --build --rm test \ + python -m pytest vectorizing/tests/test_color_bitmap_rims.py +``` + +- The API defaults to `http://localhost:8000` in Compose. +- Application source is copied into the image, not bind-mounted. Rebuild after edits. +- Use `docker compose down` to stop services you started; Moto data is discarded. + Isolate project names and image tags when testing alongside other sessions/worktrees. +- Native execution needs the Python dependencies and system libraries from the + Dockerfile/dev-container setup. Package imports require `PORT` and `S3_BUCKET`; + integration tests also require a distinct `S3_TEST_BUCKET` and a local S3 endpoint. +- Preserve normal AWS behavior unless `AWS_ENDPOINT_URL` is explicitly configured. + Use Moto for validation, not live buckets. +- CI runs the pre-commit hooks with Python 3.11 and the Compose test command. + +## Output and behavior contracts + +- Keep smooth anti-aliasing and transparent exteriors. Do not hide seams with SVG + strokes or `shape-rendering="crispEdges"`, or reduce normal tracing quality. +- Color masks use bounded overlaps, not full underpainting. The current rim is + two processed-image pixels; zoom enlarges it and very thin features may remain + covered. Do not promise universally exact holes and seam-free rendering. +- Preserve palette/centroid ordering, dtypes and overflow behavior, connectivity, + and independent mask storage when optimizing arrays. Rim conversion runs forward + because the next mask must still be cumulative. +- Preserve public defaults, response/error shapes, return containers, SVG paint + order and numeric formatting during refactors or performance work. +- Image readers currently wrap all loading failures, including interrupts, and URL + downloads have no timeout. These are tested legacy contracts, not recommendations; + change them deliberately with updated tests, never incidentally during lint cleanup. +- Comments and docstrings should explain behavior and invariants, not edit history. + +## Regression evidence + +- Add focused tests for changed behavior. Keep reference implementations independent + of the optimized code they check. +- The gallery can auto-create missing PNG baselines. A newly seeded baseline does + **not** prove equivalence; do not delete or replace baselines just to make tests pass. +- For output-preserving changes, capture before editing (or use a trusted baseline + checkout), then compare SVG bytes, RGBA pixels, palettes, dimensions and bounds + with `scripts/vectorization_outputs.py`. See the README for commands. +- For layer/geometry changes, check layer deletion, zero alpha and half opacity, + transparent backgrounds and thin features, including `aftermath.png`. An assembled + render alone cannot establish editability. +- Use `scripts/benchmark_vectorization.py` for measured performance claims. Retain + raw samples, check equivalent outputs, and report runtime/hardware context. Its + timing excludes input decoding, HTTP/S3 and rasterization; process peak RSS includes + startup and warmups. Local results are not production or concurrent-load guarantees. + +## Style, dependencies and artifacts + +- Follow `.pre-commit-config.yaml` and `pyproject.toml`: Black, 88 columns, isort, + annotations and NumPy-style docstrings. Avoid broad lint suppressions and keep + correctness/security checks enabled. Passing lint is not a static-type-check claim. +- Prefer existing dependencies. Edit `requirements/prod.in` or `requirements/dev.in`, + then use `scripts/compile_requirements.sh` and `scripts/compile_envs.sh` when dependency + updates are needed. The latter requires the configured Conda `dev` environment. + Keep system dependency setup consistent with `scripts/install_system_dependencies.sh` + and the Dockerfile. +- Keep generated captures, benchmarks, investigations and scratch source under `.user/` + or temporary directories, not tracked code. `.user/` is excluded from Git and Docker. + Historical notes there may be stale; verify them against current source. +- Keep visual evidence local unless upload is explicitly requested. Never include + credentials in reports or commits. + +## Pull requests + +Keep descriptions short and plain-language. Fetch `.github/pull_request_template.md` +from the `Kittl/.github` repository rather than inventing a replacement. +Leave owner attestations unchecked without explicit approval. Do not add agent/tool +badges. After addressing human review feedback, reply but leave the thread open +unless the user explicitly asks to resolve it; uncertain reviewer identity counts +as human. diff --git a/README.md b/README.md index 1946d65..2d534bb 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,39 @@ pre-commit run --all-files The first execution might take a bit longer, as it will set up the virtual environment where the linter and the formatter will run. +## Output comparisons and benchmarks + +These developer commands use local fixtures and need no AWS credentials or Moto. +Run them in the dev container or an environment with `requirements/dev.txt` and +its system libraries installed. They do not change application code or dependencies. + +Capture each version into a different directory, then compare: + +```bash +python scripts/vectorization_outputs.py capture .user/before --source-root /path/to/baseline +python scripts/vectorization_outputs.py capture .user/after +python scripts/vectorization_outputs.py compare .user/before .user/after +python scripts/benchmark_vectorization.py --output .user/benchmark.json --case aftermath-16 +``` + +- Repeat `--case` to select fixtures; `--help` lists them. The default is all 15 cases. +- Comparison checks SVG bytes, decoded RGBA pixels, colors, dimensions, bounds and input identity. Exit codes: `0` equal, `1` changed, `2` invalid/incomplete artifacts. Nonempty output directories and existing benchmark reports are never overwritten. +- Benchmark timing covers the solver, SVG serialization and bounds, not file decoding, HTTP/S3 or rasterization. Each sample uses a fresh process; peak RSS includes imports, input preparation, warmups and output checks. Linux and macOS memory units are normalized to MiB. +- Add `--baseline-root /path/to/baseline` to benchmark two trusted checkouts in alternating order, with exact-output checks. `--repeats` defaults to 3 and `--warmups` to 1 per process. Output mismatches exit `1` but still save the report; timings have no pass/fail threshold. +- Reports include runtime/source fingerprints and raw samples. `--label` can record commit IDs. Checkouts share the installed dependencies; this is not a dependency-isolated or production/concurrency benchmark. Use only trusted source roots. + +For Docker, build the current scripts and mount only the output directory: + +```bash +mkdir -p .user +docker build -t vectorizing:tools . +docker run --rm --user "$(id -u):$(id -g)" -v "$PWD/.user:/results" vectorizing:tools \ + python scripts/vectorization_outputs.py capture /results/candidate +``` + +A separate baseline checkout must also be mounted when using `--source-root` or +`--baseline-root`. Generated artifacts under `.user/` are excluded from Git and Docker builds. + ## Server The server has a single endpoint that receives `POST` requests. @@ -111,6 +144,9 @@ To run tests against the local Moto service, run: docker compose --profile testing run --build --rm test ``` +Tests are split into `test_*.py` files in `vectorizing/tests`. Compose and CI +run them all. `color_reference.py` keeps the old code for before/after checks. + When running pytest outside Compose, configure the AWS variables for an S3-compatible service first. ### Required setup diff --git a/docker-compose.yml b/docker-compose.yml index 9389ea5..c0a2cee 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -57,4 +57,4 @@ services: depends_on: moto-init: condition: service_completed_successfully - command: python -m pytest vectorizing/tests + command: python -m pytest diff --git a/pyproject.toml b/pyproject.toml index 158e06f..1bfb6f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,12 @@ extend-ignore = [ 'I100', 'I101', 'I202', + # Black controls whitespace in complex slices. + 'E203', + # The containing class already determines the type of self. + 'ANN101', ] +per-file-ignores = ['vectorizing/tests/*:S101'] import-order-style = "google" docstring-convention = "numpy" strictness = "short" diff --git a/pytest.ini b/pytest.ini index 7a166d7..ab009e8 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,3 +1,3 @@ [pytest] pythonpath = . -python_files = test.py test_*.py *_test.py \ No newline at end of file +testpaths = vectorizing/tests diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..fa849dd --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Local development and maintenance commands.""" diff --git a/scripts/_vectorization.py b/scripts/_vectorization.py new file mode 100644 index 0000000..cd8941b --- /dev/null +++ b/scripts/_vectorization.py @@ -0,0 +1,143 @@ +"""Shared fixtures and offline solver execution for developer commands.""" + +import hashlib +import math +import os +import platform +import sys +from collections.abc import Callable +from importlib.metadata import version +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +# Explicit cases keep developer reports independent of pytest's module layout. +CASES = { + "1px_binary": ("1px.jpg", "binary", None), + "1px_color": ("1px.jpg", "color", 9), + "black_rectangle_binary": ("black_rectangle.png", "binary", None), + "black_rectangle_color": ("black_rectangle.png", "color", 5), + "bubbles": ("bubbles.png", "color", 5), + "geo_logo": ("geo_logo.png", "binary", None), + "shapes": ("shapes.png", "color", 4), + "shapes_2": ("shapes_2.png", "color", 7), + "white_text": ("white_text.png", "binary", None), + "aftermath-2": ("aftermath.png", "color", 2), + "aftermath-6": ("aftermath.png", "color", 6), + "aftermath-16": ("aftermath.png", "color", 16), + "aftermath-64": ("aftermath.png", "color", 64), + "empty-binary": ("empty.png", "binary", None), + "empty-color": ("empty.png", "color", 6), +} + + +def file_hash(path: Path) -> str: + """Hash file bytes without retaining the entire file in memory.""" + with path.open("rb") as source: + return hashlib.file_digest(source, "sha256").hexdigest() + + +def fingerprint(svg: str, details: dict[str, object]) -> dict[str, object]: + """Record exact SVG bytes alongside palette, dimensions, bounds and input.""" + return {"svg_sha256": hashlib.sha256(svg.encode("utf-8")).hexdigest(), **details} + + +def source_hash(source_root: Path) -> str: + """Fingerprint runtime Python sources without requiring Git in the container.""" + digest = hashlib.sha256() + for path in sorted((source_root / "vectorizing").rglob("*.py")): + relative = path.relative_to(source_root) + if "tests" not in relative.parts: + digest.update(str(relative).replace(os.sep, "/").encode("utf-8") + b"\0") + digest.update(path.read_bytes() + b"\0") + return digest.hexdigest() + + +def environment() -> dict[str, object]: + """Report the numerical runtime and native thread counts without credentials.""" + import cv2 + import faiss + + packages = ( + "numpy", + "Pillow", + "scipy", + "scikit-image", + "faiss-cpu", + "opencv-python-headless", + "pypotrace", + "skia-pathops", + "CairoSVG", + ) + return { + "python": platform.python_version(), + "platform": platform.platform(), + "machine": platform.machine(), + "cpus": os.cpu_count(), + "faiss_threads": faiss.omp_get_max_threads(), + "opencv_threads": cv2.getNumThreads(), + "versions": {name: version(name) for name in packages}, + } + + +def prepare_case( + name: str, + source_root: Path, +) -> Callable[[], tuple[str, dict[str, object]]]: + """Load a fixture once and return a fresh solver/SVG/bounds run for its checkout.""" + source_root = source_root.resolve() + if not (source_root / "vectorizing" / "__init__.py").is_file(): + raise ValueError(f"Not a vectorizing checkout: {source_root}") + sys.path.insert(0, str(source_root)) + # Package imports require these settings; no server or S3 client is created. + os.environ.setdefault("PORT", "8000") + os.environ.setdefault("S3_BUCKET", "unused-vectorization-tools") + + import numpy as np + from PIL import Image + + import vectorizing + from vectorizing.geometry.bounds import compound_paths_bounds + from vectorizing.server.timer import Timer + from vectorizing.solvers.binary.BinarySolver import BinarySolver + from vectorizing.solvers.color.ColorSolver import ColorSolver + from vectorizing.svg.markup import generate_SVG_markup + from vectorizing.util.read import convert_RGB_A + + if Path(vectorizing.__file__).resolve().parent != source_root / "vectorizing": + raise ValueError( + "A different checkout is already imported; use a fresh process", + ) + filename, solver, count = CASES[name] + path = source_root / "vectorizing" / "tests" / "images" / filename + with Image.open(path) as source: + image = convert_RGB_A(source).copy() + source_info = { + "image": filename, + "sha256": file_hash(path), + "solver": solver, + "color_count": count, + } + + def run() -> tuple[str, dict[str, object]]: + result = ( + BinarySolver(image).solve() + if solver == "binary" + else ColorSolver(image, count, Timer()).solve() + ) + paths, colors, width, height = result + bounds = compound_paths_bounds(paths) + # Empty artwork has infinite bounds; strings keep the report valid JSON. + bounds = { + key: float(value) if math.isfinite(value) else str(float(value)) + for key, value in bounds.items() + } + palette = np.asarray(colors) + return generate_SVG_markup(*result), { + "input": source_info, + "colors": palette.tolist(), + "colors_dtype": str(palette.dtype), + "dimensions": [width, height], + "bounds": bounds, + } + + return run diff --git a/scripts/benchmark_vectorization.py b/scripts/benchmark_vectorization.py new file mode 100644 index 0000000..00fdd06 --- /dev/null +++ b/scripts/benchmark_vectorization.py @@ -0,0 +1,214 @@ +"""Measure offline solver/SVG/bounds time and fresh-process peak memory.""" + +import argparse +import json +import os +import statistics +import subprocess # noqa: S404 - Only the fixed Python worker is launched. +import sys +import time +from pathlib import Path + + +def peak_rss_mib() -> float: + """Convert the process high-water mark to MiB on Linux and macOS.""" + if sys.platform != "darwin" and not sys.platform.startswith("linux"): + raise ValueError("Peak RSS measurement supports Linux and macOS only") + import resource + + value = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + return value / (1024 * 1024 if sys.platform == "darwin" else 1024) + + +def sample(name: str, source_root: Path, warmups: int) -> dict[str, object]: + """Warm one case, time one run and verify stable output in this fresh process.""" + from scripts._vectorization import ( + environment, + fingerprint, + prepare_case, + source_hash, + ) + + run = prepare_case(name, source_root) + revision = source_hash(source_root) + expected = None + consistent = True + for _ in range(warmups + 1): + start = time.perf_counter() + svg, details = run() + elapsed = (time.perf_counter() - start) * 1000 + output = fingerprint(svg, details) + if expected is not None and expected != output: + consistent = False + expected = output + runtime = environment() + return { + "elapsed_ms": elapsed, + "peak_rss_mib": peak_rss_mib(), + "output": output, + "consistent": consistent, + "environment": runtime, + "pid": os.getpid(), + "source_sha256": revision, + } + + +def run_sample(name: str, source_root: Path, warmups: int) -> dict[str, object]: + """Run this script in a new interpreter, propagating worker failures.""" + command = [ + sys.executable, + str(Path(__file__).resolve()), + "--worker", + "--case", + name, + "--source-root", + str(source_root), + "--warmups", + str(warmups), + ] + # The executable and script are fixed; arguments are passed without a shell. + result = subprocess.run(command, capture_output=True, text=True) # noqa: S603 + if result.returncode: + raise RuntimeError( + f"Benchmark worker failed for {name}: {result.stderr.strip()}", + ) + return json.loads(result.stdout) + + +def benchmark( + names: list[str], + roots: dict[str, Path], + repeats: int, + warmups: int, +) -> tuple[dict[str, object], bool]: + """Alternate variants, retaining raw samples and exact-output equality checks.""" + cases = {} + all_equal = True + for name in names: + samples = {variant: [] for variant in roots} + for iteration in range(repeats): + order = list(roots) if iteration % 2 == 0 else list(reversed(roots)) + for variant in order: + samples[variant].append(run_sample(name, roots[variant], warmups)) + reference = next(iter(samples.values()))[0]["output"] + equal = all( + row["consistent"] and row["output"] == reference + for rows in samples.values() + for row in rows + ) + all_equal &= equal + medians = { + variant: { + "elapsed_ms": statistics.median(row["elapsed_ms"] for row in rows), + "peak_rss_mib": statistics.median(row["peak_rss_mib"] for row in rows), + } + for variant, rows in samples.items() + } + cases[name] = {"samples": samples, "medians": medians, "outputs_equal": equal} + for variant, values in medians.items(): + print( + f"{name} {variant}: {values['elapsed_ms']:.2f} ms, " + f"{values['peak_rss_mib']:.2f} MiB", + ) + if not equal: + print( + f"{name}: outputs differ; timings are not an equivalent comparison.", + file=sys.stderr, + ) + return cases, all_equal + + +def positive(value: str) -> int: + """Require at least one measured sample.""" + count = int(value) + if count < 1: + raise argparse.ArgumentTypeError("must be at least 1") + return count + + +def nonnegative(value: str) -> int: + """Allow zero warmups, but reject negative counts.""" + count = int(value) + if count < 0: + raise argparse.ArgumentTypeError("must be at least 0") + return count + + +def main(argv: list[str] | None = None) -> int: + """Write a benchmark report, returning one when any outputs differ.""" + from scripts._vectorization import CASES, ROOT + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, help="New JSON report path (required)") + parser.add_argument( + "--source-root", + type=Path, + default=ROOT, + help="Trusted candidate checkout", + ) + parser.add_argument( + "--baseline-root", + type=Path, + help="Optional trusted baseline checkout", + ) + parser.add_argument( + "--case", + action="append", + choices=CASES, + help="Repeat to select cases; default: all", + ) + parser.add_argument("--repeats", type=positive, default=3) + parser.add_argument( + "--warmups", + type=nonnegative, + default=1, + help="Warmups per fresh process", + ) + parser.add_argument( + "--label", + help="Optional revisions or description for this report", + ) + parser.add_argument("--worker", action="store_true", help=argparse.SUPPRESS) + args = parser.parse_args(argv) + names = list(dict.fromkeys(args.case or CASES)) + try: + if args.worker: + if len(names) != 1: + raise ValueError("A worker measures exactly one case") + print( + json.dumps( + sample(names[0], args.source_root, args.warmups), + allow_nan=False, + ), + ) + return 0 + if args.output is None: + raise ValueError("--output is required") + if args.output.exists(): + raise ValueError(f"Refusing to overwrite: {args.output}") + roots = {"baseline": args.baseline_root.resolve()} if args.baseline_root else {} + roots["candidate"] = args.source_root.resolve() + cases, equal = benchmark(names, roots, args.repeats, args.warmups) + report = { + "schema_version": 1, + "label": args.label, + "source_roots": {name: str(path) for name, path in roots.items()}, + "repeats": args.repeats, + "warmups_per_process": args.warmups, + "timing_scope": "solver, SVG serialization, colors, dimensions and bounds", + "rss_scope": "fresh process: imports, input, warmups, output checks", + "cases": cases, + "outputs_equal": equal, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("x", encoding="utf-8") as destination: + json.dump(report, destination, indent=2, sort_keys=True, allow_nan=False) + destination.write("\n") + return int(not equal) + except (OSError, ValueError, RuntimeError, ImportError) as error: + parser.error(str(error)) + + +if __name__ == "__main__": + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + raise SystemExit(main()) diff --git a/scripts/init_moto.py b/scripts/init_moto.py index 4bad201..b6426b4 100644 --- a/scripts/init_moto.py +++ b/scripts/init_moto.py @@ -1,8 +1,9 @@ +"""Create separate local application and test buckets if they do not exist.""" + import os import boto3 - s3 = boto3.client("s3", endpoint_url=os.environ["AWS_ENDPOINT_URL"]) existing = {bucket["Name"] for bucket in s3.list_buckets()["Buckets"]} diff --git a/scripts/vectorization_outputs.py b/scripts/vectorization_outputs.py new file mode 100644 index 0000000..1b86e0d --- /dev/null +++ b/scripts/vectorization_outputs.py @@ -0,0 +1,181 @@ +"""Capture local fixtures and compare exact SVG, pixel and solver outputs.""" + +import argparse +import hashlib +import json +import sys +from pathlib import Path + + +def capture( + output: Path, + names: list[str], + source_root: Path, + label: str | None, +) -> None: + """Write a complete manifest last, refusing to replace existing artifacts.""" + from cairosvg import svg2png + + from scripts._vectorization import ( + environment, + fingerprint, + prepare_case, + source_hash, + ) + + if output.exists() and (not output.is_dir() or any(output.iterdir())): + raise ValueError(f"Output must be a new or empty directory: {output}") + output.mkdir(parents=True, exist_ok=True) + cases = {} + for name in names: + svg, details = prepare_case(name, source_root)() + svg_bytes = svg.encode("utf-8") + for extension, data in { + "svg": svg_bytes, + "png": svg2png(bytestring=svg_bytes), + }.items(): + with (output / f"{name}.{extension}").open("xb") as artifact: + artifact.write(data) + pixel_hash, shape = pixels(output / f"{name}.png") + cases[name] = { + **fingerprint(svg, details), + "rgba_sha256": pixel_hash, + "rgba_shape": shape, + } + print(f"Captured {name}") + report = { + "schema_version": 1, + "label": label, + "source_root": str(source_root.resolve()), + "source_sha256": source_hash(source_root), + "environment": environment(), + "cases": cases, + } + with (output / "outputs.json").open("x", encoding="utf-8") as destination: + json.dump(report, destination, indent=2, sort_keys=True, allow_nan=False) + destination.write("\n") + + +def pixels(path: Path) -> tuple[str, list[int]]: + """Hash decoded RGBA pixels, independent of PNG compression.""" + from PIL import Image + + with Image.open(path) as image: + rgba = image.convert("RGBA") + return hashlib.sha256(rgba.tobytes()).hexdigest(), [rgba.height, rgba.width, 4] + + +def read_capture(folder: Path) -> dict[str, object]: + """Validate a manifest and its SVG/PNG artifacts before trusting its hashes.""" + from scripts._vectorization import CASES, file_hash + + report = json.loads((folder / "outputs.json").read_text(encoding="utf-8")) + # Reject non-standard NaN/Infinity literals in edited or external reports. + json.dumps(report, allow_nan=False) + if not isinstance(report, dict) or report.get("schema_version") != 1: + raise ValueError(f"Unsupported capture format: {folder}") + cases = report.get("cases") + if not isinstance(cases, dict) or not cases: + raise ValueError(f"Capture has no cases: {folder}") + required = { + "input", + "colors", + "colors_dtype", + "dimensions", + "bounds", + "svg_sha256", + "rgba_sha256", + "rgba_shape", + } + for name, record in cases.items(): + if ( + name not in CASES + or not isinstance(record, dict) + or not required <= record.keys() + ): + raise ValueError(f"Invalid capture case: {name}") + if file_hash(folder / f"{name}.svg") != record["svg_sha256"]: + raise ValueError(f"SVG does not match its manifest: {folder}/{name}") + digest, shape = pixels(folder / f"{name}.png") + if digest != record["rgba_sha256"] or shape != record["rgba_shape"]: + raise ValueError(f"PNG does not match its manifest: {folder}/{name}") + return report + + +def compare(baseline: Path, candidate: Path) -> int: + """Return one for changed outputs or case sets, and zero for exact equality.""" + before, after = read_capture(baseline), read_capture(candidate) + old, new = before["cases"], after["cases"] + changed = False + for name in sorted(old.keys() | new.keys()): + if name not in old or name not in new: + print(f"{name}: {'added' if name in new else 'missing'} case") + changed = True + elif old[name] != new[name]: + fields = sorted( + key + for key in old[name].keys() | new[name].keys() + if key not in old[name] + or key not in new[name] + or old[name][key] != new[name][key] + ) + print(f"{name}: changed {', '.join(fields)}") + changed = True + if before.get("environment") != after.get("environment"): + print( + "Note: capture environments differ; see outputs.json metadata.", + file=sys.stderr, + ) + if not changed: + print(f"All {len(old)} cases match exactly.") + return int(changed) + + +def main(argv: list[str] | None = None) -> int: + """Capture outputs or compare two saved directories without AWS access.""" + from scripts._vectorization import CASES, ROOT + + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + save = commands.add_parser("capture", help="Save SVGs, PNGs and a JSON manifest") + save.add_argument("output", type=Path) + save.add_argument( + "--source-root", + type=Path, + default=ROOT, + help="Trusted checkout to run", + ) + save.add_argument( + "--case", + action="append", + choices=CASES, + help="Repeat to select cases; default: all", + ) + save.add_argument( + "--label", + help="Optional revision or description for the manifest", + ) + diff = commands.add_parser( + "compare", + help="Compare complete captures; exit 1 on changes", + ) + diff.add_argument("baseline", type=Path) + diff.add_argument("candidate", type=Path) + args = parser.parse_args(argv) + try: + if args.command == "compare": + return compare(args.baseline, args.candidate) + capture( + args.output, + list(dict.fromkeys(args.case or CASES)), + args.source_root, + args.label, + ) + return 0 + except (OSError, ValueError, ImportError) as error: + parser.error(str(error)) + + +if __name__ == "__main__": + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + raise SystemExit(main()) diff --git a/vectorizing/__init__.py b/vectorizing/__init__.py index 879184e..26d04e5 100644 --- a/vectorizing/__init__.py +++ b/vectorizing/__init__.py @@ -1,17 +1,24 @@ +"""Expose image vectorization and health checks through Flask.""" + import os -import sentry_sdk from types import SimpleNamespace -from flask import Flask, request, jsonify +from typing import Literal, NoReturn -from vectorizing.server.timer import Timer +import numpy as np +import sentry_sdk +from flask import Flask, Response, jsonify, request +from pathops import Path +from PIL import Image + +from vectorizing.geometry.bounds import compound_paths_bounds +from vectorizing.server.env import get_optional, get_required from vectorizing.server.logs import setup_logs from vectorizing.server.s3 import upload_markup +from vectorizing.server.timer import Timer +from vectorizing.solvers.binary.BinarySolver import BinarySolver +from vectorizing.solvers.color.ColorSolver import ColorSolver from vectorizing.svg.markup import generate_SVG_markup from vectorizing.util.read import try_read_image_from_url -from vectorizing.server.env import get_required, get_optional -from vectorizing.solvers.color.ColorSolver import ColorSolver -from vectorizing.solvers.binary.BinarySolver import BinarySolver -from vectorizing.geometry.bounds import compound_paths_bounds # 0 -> BinarySolver # 1 -> ColorSolver @@ -24,30 +31,37 @@ S3_BUCKET, ) = get_required() -( - SENTRY_DSN, - S3_TEST_BUCKET -) = get_optional() +(SENTRY_DSN, S3_TEST_BUCKET) = get_optional() setup_logs() -def process_binary(img): + +def process_binary(img: Image.Image) -> tuple[list[Path], list[list[int]], int, int]: + """Trace an image into black paths and return its processed dimensions.""" solver = BinarySolver(img) return solver.solve() -def process_color(img, color_count, timer): + +def process_color( + img: Image.Image, + color_count: int | None, + timer: Timer, +) -> list[list[Path] | list[np.ndarray] | int]: + """Trace color layers with bounded overlaps and return processed dimensions.""" solver = ColorSolver(img, color_count, timer) return solver.solve() -def validate_args(args): - if not 'url' in args: + +def validate_args(args: dict[str, object]) -> SimpleNamespace | Literal[False]: + """Validate the URL key, solver choice and optional four-integer crop box.""" + if "url" not in args: return False - solver = args.get('solver', DEFAULT_SOLVER) - if not solver in SOLVERS: + solver = args.get("solver", DEFAULT_SOLVER) + if solver not in SOLVERS: return False - box = args.get('crop_box') + box = args.get("crop_box") if box: if len(box) != 4: return False @@ -61,21 +75,22 @@ def validate_args(args): solver=solver, url=args.get("url"), raw=args.get("raw"), - color_count=args.get("color_count") + color_count=args.get("color_count"), ) -def invalid_args(): - return jsonify({ - "success": False, - "error": "INVALID_PARAMETERS" - }), 400 -def create_app(test_config=None): +def invalid_args() -> tuple[Response, int]: + """Return the API's invalid-parameters response.""" + return jsonify({"success": False, "error": "INVALID_PARAMETERS"}), 400 + + +def create_app(test_config: dict[str, object] | None = None) -> Flask: + """Create the HTTP app; the optional test_config argument is unused.""" app = Flask(__name__, instance_relative_config=True) - app.debug = PYTHON_ENV == 'development' + app.debug = PYTHON_ENV == "development" - @app.route('/', methods = ['POST']) - def index(): + @app.route("/", methods=["POST"]) + def index() -> str | Response | tuple[Response, int]: args = request.json args = validate_args(args) @@ -91,7 +106,7 @@ def index(): try: timer = Timer() - timer.start_timer('Image Reading') + timer.start_timer("Image Reading") img = try_read_image_from_url(url) timer.end_timer() @@ -99,66 +114,72 @@ def index(): img = img.crop(tuple(crop_box)) if solver == 0: - timer.start_timer('Binary Solver - Total') + timer.start_timer("Binary Solver - Total") solved = process_binary(img) timer.end_timer() else: - timer.start_timer('Color Solver - Total') + timer.start_timer("Color Solver - Total") solved = process_color(img, color_count, timer) timer.end_timer() compound_paths, colors, width, height = solved - timer.start_timer('Markup Creation') + timer.start_timer("Markup Creation") markup = generate_SVG_markup(compound_paths, colors, width, height) timer.end_timer() if raw: - return markup + return Response(markup, mimetype="image/svg+xml") - timer.start_timer('Markup Upload') + timer.start_timer("Markup Upload") cuid_str = upload_markup(markup, S3_BUCKET) timer.end_timer() - timer.start_timer('Bounds Creation') + timer.start_timer("Bounds Creation") bounds = compound_paths_bounds(compound_paths) timer.end_timer() app.logger.info(timer.timelog()) - return jsonify({ - 'success': True, - 'objectId': cuid_str, - 'info': { - 'bounds': bounds, - 'image_width': width, - 'image_height': height - } - }) - - except (Exception) as e: + return jsonify( + { + "success": True, + "objectId": cuid_str, + "info": { + "bounds": bounds, + "image_width": width, + "image_height": height, + }, + }, + ) + + except Exception as e: app.logger.error(e) - return jsonify({ - "success": False, - "error": "INTERNAL_SERVER_ERROR" - }), 500 - - @app.route('/health', methods = ['GET']) - def healthcheck(): - return jsonify({ - "success": True, - }), 200 - - @app.route('/test-error', methods = ['GET']) - def test_error(): - raise Exception('Test Error') - - app.logger.info(f'Vectorizing server running on port: {PORT}, environment: {PYTHON_ENV}') + return jsonify({"success": False, "error": "INTERNAL_SERVER_ERROR"}), 500 + + @app.route("/health", methods=["GET"]) + def healthcheck() -> tuple[Response, int]: + return ( + jsonify( + { + "success": True, + }, + ), + 200, + ) + + @app.route("/test-error", methods=["GET"]) + def test_error() -> NoReturn: + raise Exception("Test Error") + + app.logger.info( + f"Vectorizing server running on port: {PORT}, environment: {PYTHON_ENV}", + ) if SENTRY_DSN: sentry_sdk.init( - dsn = SENTRY_DSN, - traces_sample_rate = 0.1, - environment = PYTHON_ENV, + dsn=SENTRY_DSN, + traces_sample_rate=0.1, + environment=PYTHON_ENV, ) return app diff --git a/vectorizing/geometry/bounds.py b/vectorizing/geometry/bounds.py index 0931cc9..cc563d4 100644 --- a/vectorizing/geometry/bounds.py +++ b/vectorizing/geometry/bounds.py @@ -1,17 +1,13 @@ -import numpy as np +"""Compute document bounds from traced vector paths.""" +from collections.abc import Iterable -# Calculates bounds of a list of compound paths -def compound_paths_bounds(compound_paths): - """ - Calculates the combined bounds of a list of compound paths. +import numpy as np +from pathops import Path - Parameters: - compound_paths: The list of compound paths (SKPath). - Returns: - An object containing combined bounds info. - """ +def compound_paths_bounds(compound_paths: Iterable[Path]) -> dict[str, float]: + """Return combined bounds, retaining infinite extents for an empty iterable.""" min_x = np.inf min_y = np.inf max_x = -np.inf diff --git a/vectorizing/geometry/potrace.py b/vectorizing/geometry/potrace.py index 521cdb7..64b8232 100644 --- a/vectorizing/geometry/potrace.py +++ b/vectorizing/geometry/potrace.py @@ -1,19 +1,11 @@ -import numpy as np -from pathops import Path, FillType +"""Convert Potrace curves into paths suitable for SVG serialization.""" +import potrace +from pathops import FillType, Path -def potrace_path_to_compound_path(potrace_path): - """ - Converts a potrace path into an SKPath. - This conversion is needed to later perform boolean operations - on compound paths. - Parameters: - potrace_path: The potrace path - - Returns: - The SKPath - """ +def potrace_path_to_compound_path(potrace_path: potrace.Path) -> Path: + """Convert Potrace curves to an even-odd path of lines and cubic Beziers.""" compound_path = Path(fillType=FillType.EVEN_ODD) for potrace_curve in potrace_path: diff --git a/vectorizing/server/env.py b/vectorizing/server/env.py index b7f37c9..4adb9ae 100644 --- a/vectorizing/server/env.py +++ b/vectorizing/server/env.py @@ -1,33 +1,32 @@ +"""Read and cast the application's ordered environment settings.""" + import os -REQUIRED_ENVIRONMENT_VARIABLES = { - 'PORT': int, - 'S3_BUCKET': str -} +REQUIRED_ENVIRONMENT_VARIABLES = {"PORT": int, "S3_BUCKET": str} + +OPTIONAL_ENVIRONMENT_VARIABLES = {"SENTRY_DSN": str, "S3_TEST_BUCKET": str} -OPTIONAL_ENVIRONMENT_VARIABLES = { - 'SENTRY_DSN': str, - 'S3_TEST_BUCKET': str -} class VariableNotDefinedException(Exception): - pass + """Signal that at least one required environment setting is missing.""" -def get_required(): - missing_required = any([ - not key in os.environ - for key in REQUIRED_ENVIRONMENT_VARIABLES - ]) + +def get_required() -> list[int | str]: + """Return port and bucket, raising if either required setting is missing.""" + missing_required = any( + [key not in os.environ for key in REQUIRED_ENVIRONMENT_VARIABLES], + ) if missing_required: raise VariableNotDefinedException() return [ - cast(os.environ[key]) - for key, cast in REQUIRED_ENVIRONMENT_VARIABLES.items() + cast(os.environ[key]) for key, cast in REQUIRED_ENVIRONMENT_VARIABLES.items() ] -def get_optional(): + +def get_optional() -> list[str | None]: + """Return Sentry DSN and test bucket in order, using None for unset values.""" return [ cast(os.environ[key]) if key in os.environ else None for key, cast in OPTIONAL_ENVIRONMENT_VARIABLES.items() diff --git a/vectorizing/server/logs.py b/vectorizing/server/logs.py index d63b170..93b24b8 100644 --- a/vectorizing/server/logs.py +++ b/vectorizing/server/logs.py @@ -1,33 +1,24 @@ +"""Configure application logging to standard output.""" + from logging.config import dictConfig -def setup_logs(): + +def setup_logs() -> None: + """Configure the root logger with the service's INFO-level stdout format.""" # Set up Flask logging to STDOUT - dictConfig({ - 'version': 1, - 'formatters': - { - 'default': - { - 'format': '[%(levelname)s] %(message)s' - } + dictConfig( + { + "version": 1, + "formatters": {"default": {"format": "[%(levelname)s] %(message)s"}}, + "handlers": { + "stdout": { + "class": "logging.StreamHandler", + "formatter": "default", + "stream": "ext://sys.stdout", + }, + }, + "loggers": { + "": {"handlers": ["stdout"], "level": "INFO", "propagate": True}, }, - 'handlers': - { - 'stdout': - { - 'class': 'logging.StreamHandler', - 'formatter': 'default', - 'stream': 'ext://sys.stdout' - } - }, - 'loggers': - { - '': - { - 'handlers': ['stdout'], - 'level': 'INFO', - 'propagate': True - } - } - } - ) \ No newline at end of file + }, + ) diff --git a/vectorizing/server/s3.py b/vectorizing/server/s3.py index 4316e4f..d5c07e2 100644 --- a/vectorizing/server/s3.py +++ b/vectorizing/server/s3.py @@ -1,13 +1,17 @@ +"""Store SVGs and test images in AWS S3 or an explicitly configured endpoint.""" + import os from functools import lru_cache import boto3 import cuid +from botocore.client import BaseClient from botocore.exceptions import ClientError @lru_cache(maxsize=1) -def get_s3_client(): +def get_s3_client() -> BaseClient: + """Cache the S3 client, honoring AWS_ENDPOINT_URL only when explicitly set.""" endpoint_url = os.getenv("AWS_ENDPOINT_URL") or None return boto3.client( "s3", @@ -16,7 +20,8 @@ def get_s3_client(): ) -def upload_markup(markup, s3_bucket_name): +def upload_markup(markup: str, s3_bucket_name: str) -> str: + """Upload UTF-8 SVG markup under a generated key and return that key.""" cuid_str = cuid.cuid() get_s3_client().put_object( @@ -29,7 +34,8 @@ def upload_markup(markup, s3_bucket_name): return cuid_str -def get_object_url(s3_file_key, s3_bucket_name): +def get_object_url(s3_file_key: str, s3_bucket_name: str) -> str | None: + """Return an existing object's URL, keeping signatures for custom endpoints.""" s3 = get_s3_client() try: s3.head_object( @@ -48,7 +54,12 @@ def get_object_url(s3_file_key, s3_bucket_name): return object_url if os.getenv("AWS_ENDPOINT_URL") else object_url.split("?")[0] -def upload_file(local_file_path, s3_bucket_name, s3_file_key): +def upload_file( + local_file_path: str | os.PathLike[str], + s3_bucket_name: str, + s3_file_key: str, +) -> str | None: + """Upload a local file and return the stored object's URL.""" get_s3_client().upload_file( local_file_path, s3_bucket_name, diff --git a/vectorizing/server/timer.py b/vectorizing/server/timer.py index 4e6cb0c..c91c60d 100644 --- a/vectorizing/server/timer.py +++ b/vectorizing/server/timer.py @@ -1,23 +1,31 @@ +"""Collect named timings, including nested processing stages.""" + import time + class Timer: - def __init__(self): + """Measure nested stages and retain the latest duration for each name.""" + + def __init__(self) -> None: self.entries = {} self.actions_stack = [] self.time_stack = [] - def start_timer(self, action_name): + def start_timer(self, action_name: str) -> None: + """Start a named stage without ending any enclosing stage.""" self.time_stack.append(time.time()) self.actions_stack.append(action_name) - - def end_timer(self): + + def end_timer(self) -> None: + """Record elapsed time for the most recently started stage.""" self.entries[self.actions_stack.pop()] = time.time() - self.time_stack.pop() - def timelog(self): - log = '' + def timelog(self) -> str: + """Format recorded stages in insertion order with a trailing separator.""" + log = "" for action_name in self.entries: time = self.entries[action_name] - log += f'{action_name}: {time} / ' - - return log \ No newline at end of file + log += f"{action_name}: {time} / " + + return log diff --git a/vectorizing/solvers/binary/BinarySolver.py b/vectorizing/solvers/binary/BinarySolver.py index bd8b5a2..5eb1367 100644 --- a/vectorizing/solvers/binary/BinarySolver.py +++ b/vectorizing/solvers/binary/BinarySolver.py @@ -1,5 +1,9 @@ +"""Trace an image into a single black vector layer.""" + import numpy as np import potrace +from pathops import Path +from PIL import Image from vectorizing.geometry.potrace import potrace_path_to_compound_path from vectorizing.solvers.binary.bitmap import compute_bitmap @@ -7,13 +11,16 @@ class BinarySolver: - def __init__(self, img): + """Resize, threshold and trace an RGB or RGBA image.""" + + def __init__(self, img: Image.Image) -> None: self.img = limit_size(img) # Init image array self.img_arr = np.asarray(self.img).astype(np.uint8) - def solve(self): + def solve(self) -> tuple[list[Path], list[list[int]], int, int]: + """Return the traced path, black RGBA color and processed image size.""" width, height = self.img.size bitmap = compute_bitmap(self.img_arr) traced_bitmap = potrace.Bitmap(bitmap).trace() diff --git a/vectorizing/solvers/binary/bitmap.py b/vectorizing/solvers/binary/bitmap.py index ffb2334..845d13c 100644 --- a/vectorizing/solvers/binary/bitmap.py +++ b/vectorizing/solvers/binary/bitmap.py @@ -1,9 +1,11 @@ +"""Prepare foreground masks for monochrome tracing.""" + import cv2 import numpy as np -# Inverts an image -def invert(img_arr): +def invert(img_arr: np.ndarray) -> np.ndarray: + """Invert RGB channels while leaving any alpha channel unchanged.""" channel_count = img_arr.shape[-1] if channel_count == 4: @@ -16,10 +18,8 @@ def invert(img_arr): return 255 - img_arr -# Blends a transparent image with a white background by -# interpolating linearly. In the end, an RGBA image -# is turned into RGB -def alpha_blend(img_arr): +def alpha_blend(img_arr: np.ndarray) -> np.ndarray: + """Composite uint8 RGBA pixels over white and return uint8 RGB pixels.""" r, g, b, a = cv2.split(img_arr) n_alpha = a / 255 @@ -30,8 +30,8 @@ def alpha_blend(img_arr): return cv2.merge((r, g, b)).astype(np.uint8) -# Compute threshold -def threshold(img_arr): +def threshold(img_arr: np.ndarray) -> np.ndarray: + """Apply Otsu thresholding, marking dark pixels as foreground.""" channel_count = img_arr.shape[-1] if channel_count == 4: @@ -46,11 +46,11 @@ def threshold(img_arr): return np.where(thresholded >= 128, 0, 1) -# Creates bitmap to be traced by potrace -# This deals with a very common edge case: when an image has a transparent -# background and a white foreground, the alpha blend strips all information, so in -# that case we choose the inverted version -def compute_bitmap(img_arr, foreground_area_threshold=0.01): +def compute_bitmap( + img_arr: np.ndarray, + foreground_area_threshold: float = 0.01, +) -> np.ndarray: + """Select a foreground mask, falling back to a solid mask when it is empty.""" # compute bitmap bitmap = threshold(img_arr) foreground_area = np.sum(bitmap) diff --git a/vectorizing/solvers/color/ColorSolver.py b/vectorizing/solvers/color/ColorSolver.py index 5d07010..545c833 100644 --- a/vectorizing/solvers/color/ColorSolver.py +++ b/vectorizing/solvers/color/ColorSolver.py @@ -1,14 +1,21 @@ +"""Trace editable color layers with bounded overlap at shared edges.""" + import numpy as np import potrace +from pathops import Path +from PIL import Image from vectorizing.geometry.potrace import potrace_path_to_compound_path +from vectorizing.server.timer import Timer from vectorizing.solvers.color.bitmaps import add_bitmap_rims, create_bitmaps from vectorizing.solvers.color.quantize import quantize from vectorizing.util.limit_size import limit_size class ColorSolver: - def __init__(self, img, color_count, timer): + """Resize and quantize an image, then trace its ordered color layers.""" + + def __init__(self, img: Image.Image, color_count: int | None, timer: Timer) -> None: color_count = color_count or ColorSolver.DEFAULT_COLOR_COUNT color_count = max(color_count, ColorSolver.MIN_COLOR_COUNT) color_count = min(color_count, ColorSolver.MAX_COLOR_COUNT) @@ -21,7 +28,8 @@ def __init__(self, img, color_count, timer): self.timer = timer - def solve(self): + def solve(self) -> list[list[Path] | list[np.ndarray] | int]: + """Return paths, colors, width and height as a list.""" self.timer.start_timer("Quantization") labels, colors, has_background = quantize(self.img_arr, self.color_count) self.timer.end_timer() diff --git a/vectorizing/solvers/color/quantize.py b/vectorizing/solvers/color/quantize.py index a3bd234..109210a 100644 --- a/vectorizing/solvers/color/quantize.py +++ b/vectorizing/solvers/color/quantize.py @@ -1,146 +1,108 @@ +"""Quantize RGB colors while keeping transparent background pixels separate.""" + import cv2 import numpy as np -from PIL import Image -from faiss import Kmeans import scipy.ndimage as ndi +from faiss import Kmeans +from PIL import Image from skimage.measure import label BILATERAL_FILTER_DIAMETER = 7 BILATERAL_FILTER_SIGMA = 50 -# Applies a strong bilateral filter to the image. -# This greatly improves results for clip art. -# NOTE: photographs migh lose quite some detail due to this. + def bilateral_filter( - img_arr, - d = BILATERAL_FILTER_DIAMETER, - s = BILATERAL_FILTER_SIGMA -): + img_arr: np.ndarray, + d: int = BILATERAL_FILTER_DIAMETER, + s: float = BILATERAL_FILTER_SIGMA, +) -> np.ndarray: + """Smooth clip-art colors while preserving edges; photos may lose detail.""" return cv2.bilateralFilter(img_arr, d, s, s) -# First erodes a matrix, then dilates it -def dilate(matrix): - kernel = np.ones((2, 2)) - return cv2.dilate(matrix, kernel) -# Fills holes (defined as cells that are zero) in a matrix -# By using the closes value found in the matrix that is not zero. -# See: https://docs.scipy.org/doc/scipy/reference/generated/scipy.ndimage.distance_transform_edt.html -def fill_holes(matrix): +def fill_holes(matrix: np.ndarray) -> np.ndarray: + """Replace zero cells with their nearest nonzero value via a distance transform.""" closest = ndi.distance_transform_edt( - np.logical_not(matrix), - return_distances = False, - return_indices = True + np.logical_not(matrix), + return_distances=False, + return_indices=True, ) matrix = np.where(matrix != 0, matrix, matrix[closest[0], closest[1]]) return matrix -def enhance(img_arr, labels, colors): - dims = img_arr.shape[:2] - # 0 will be reserved for "unwritten" cells, or holes +def enhance( + img_arr: np.ndarray, + labels: np.ndarray, + colors: np.ndarray, +) -> tuple[np.ndarray, np.ndarray]: + """Remove boundary-dominated components with O(K*N) work and O(N) image storage.""" + # Reserve zero for holes, preserving the existing input-dtype shift behavior. labels = labels + 1 - - clusters = [ - # Even though we don't allow it, technically the max - # value in labels (before adding 1) is 255, so it's safer - # to convert to uint16 to avoid overflows - np.where(labels == idx + 1, idx + 1, 0).astype(np.uint16) - for idx, _ in enumerate(colors) - ] - - # For each cluster, store a matrix of its connected components - # NOTE: see https://scikit-image.org/docs/stable/api/skimage.measure.html#skimage.measure.label - original_connected_components_list = [ - label(cluster) + 1 - for cluster in clusters - ] - - # Bincounts for each connected components matrix - # Meaning bincount[i] = num_appearances(matrix, i) - # It's used later to determine what should be erased - connected_components_bincounts = [ - np.bincount(connected_components.flatten()) - for connected_components in original_connected_components_list - ] - - # Make a copy of the connected components - # These will be modified and compared with the originals later - connected_components_list = [ - np.array(item, copy = True) - for item in original_connected_components_list - ] - - for x, cluster_x in enumerate(clusters): - dilated_cluster_x = dilate(cluster_x) - - for y, cluster_y in enumerate(clusters): - if x == y: - continue - - # The overlap between dilated_cluster_x and cluster_y, - # meaning, the part of cluster_y that dilated_cluster_x covers - overlap = np.logical_and(dilated_cluster_x, cluster_y) - - # Remove overlap from connected components - connected_components_list[y] = np.where( - overlap, - 0, - connected_components_list[y] - ) - - for x, connected_components in enumerate(connected_components_list): - original_bincount = connected_components_bincounts[x] - count = original_bincount.shape[0] - - new_bincount = np.bincount( - connected_components.flatten(), - minlength = count # Force same length + valid = (labels > 0) & (labels <= len(colors)) + + # OpenCV's 2x2 dilation (default anchor/border) reaches a pixel from above, + # left, and upper-left. Mark overlap from any OTHER palette color once, + # without treating missing/out-of-palette neighbors as a dilated cluster. + boundary = np.zeros(labels.shape, dtype=bool) + boundary[1:, :] |= valid[:-1, :] & (labels[1:, :] != labels[:-1, :]) + boundary[:, 1:] |= valid[:, :-1] & (labels[:, 1:] != labels[:, :-1]) + boundary[1:, 1:] |= valid[:-1, :-1] & (labels[1:, 1:] != labels[:-1, :-1]) + + # The boundary is fixed; reuse its complement for every color's area count. + interior = ~boundary + del boundary, valid + cleaned = np.zeros(img_arr.shape[:2], dtype=np.uint16 if len(colors) else np.uint8) + # One scan finds every color's box, including all disconnected components. + # max_label excludes invalid labels; zero would instead infer the maximum. + regions = ndi.find_objects(labels, max_label=len(colors)) if len(colors) else [] + for index, region in enumerate(regions): + if region is None: + continue + cluster = labels[region] == index + 1 + # Keep full (8-neighbor) connectivity, including diagonal contacts. + components = label(cluster, connectivity=2) + original_counts = np.bincount(components.ravel()) + interior_counts = np.bincount( + # Slice the global mask: other colors outside the box still influence it. + components[interior[region]], + minlength=len(original_counts), ) - areas_ratio = np.divide( - new_bincount.astype(np.float32), - original_bincount.astype(np.float32), - out = np.ones((count, ), dtype = np.float32), - where = original_bincount != 0 + interior_counts.astype(np.float32), + original_counts.astype(np.float32), + out=np.ones(len(original_counts), dtype=np.float32), + where=original_counts != 0, ) + # Retain whole components, using the same float32, inclusive <=0.1 cutoff. + cleaned[region][cluster & (areas_ratio[components] > 0.1)] = index + 1 + # Release this map before allocating the next color's component map. + del components - # Update cluster - # Connected components that lost more than 0.9 of their original area - # are completely removed, others are kept - clusters[x] = np.where( - areas_ratio[original_connected_components_list[x]] <= 0.1, - 0, - clusters[x] - ) + return fill_holes(cleaned) - 1, colors - # Update labels with new cluster data - labels = np.zeros(dims).astype(np.uint8) - for cluster in clusters: - labels = np.where(labels == 0, cluster, labels) - - # Fill holes - labels = fill_holes(labels) - return labels - 1, colors # Try to get the cluster of pixels that represent a transparent background # If there is no transparent background, return None # This is needed because RGBA quantization is very volatile, and sometimes # opaque colors get assigned to the same clusters as highly transparent ones. # So we focus on the solid image + transparent background case for now. -def get_background_cluster(img_arr, t = 0.5): +def get_background_cluster(img_arr: np.ndarray, t: float = 0.5) -> np.ndarray | None: + """Return an alpha mask if any raw alpha is below t, otherwise return None.""" r, g, b, a = cv2.split(img_arr) - + has_transparent_background = np.any(a < t) if has_transparent_background: a = np.where(a / 255 < t, 1, 0) return np.reshape(a, img_arr.shape[:2]) - + return None + # Write the background cluster on top of a matrix of clusters (labels) # The background cluster has priority -def write_background_cluster(labels, bg_cluster): +def write_background_cluster(labels: np.ndarray, bg_cluster: np.ndarray) -> np.ndarray: + """Shift labels by one, reserving zero for the overriding background mask.""" labels = labels + 1 labels = np.where(bg_cluster > 0, 0, labels) return labels @@ -161,31 +123,37 @@ def get_initial_centroids(img_arr: np.ndarray, color_count: int) -> np.ndarray: return np.unique(palette[used_indices], axis=0) -def kmeans(img_arr, init_centroids): +def kmeans( + img_arr: np.ndarray, + init_centroids: np.ndarray, +) -> tuple[np.ndarray, np.ndarray]: + """Cluster RGB pixels with FAISS and return uint8 labels and centroids.""" channel_count = img_arr.shape[-1] data = np.reshape(img_arr, (-1, channel_count)).astype(np.float32) init_centroids = init_centroids.astype(np.float32) - km = Kmeans(channel_count, init_centroids.shape[0], niter = 100) - km.train(data, init_centroids = init_centroids) + km = Kmeans(channel_count, init_centroids.shape[0], niter=100) + km.train(data, init_centroids=init_centroids) _, labels = km.index.search(data, 1) return labels.astype(np.uint8), km.centroids.astype(np.uint8) - -def quantize(img_arr, color_count): + + +def quantize( + img_arr: np.ndarray, + color_count: int, +) -> tuple[np.ndarray, np.ndarray, bool]: + """Return cleaned labels, palette colors and whether a background was isolated.""" channel_count = img_arr.shape[-1] - + background_cluster = None if channel_count == 4: background_cluster = get_background_cluster(img_arr) img = Image.fromarray(img_arr) - img = img.convert('RGB') + img = img.convert("RGB") img_arr = np.asarray(img) img_arr = bilateral_filter(img_arr) - labels, colors = kmeans( - img_arr, - get_initial_centroids(img_arr, color_count) - ) + labels, colors = kmeans(img_arr, get_initial_centroids(img_arr, color_count)) labels = np.reshape(labels, img_arr.shape[:-1]) @@ -197,4 +165,4 @@ def quantize(img_arr, color_count): colors = colors.astype(np.uint8) labels, colors = enhance(img_arr, labels, colors) - return labels, colors, has_background \ No newline at end of file + return labels, colors, has_background diff --git a/vectorizing/svg/markup.py b/vectorizing/svg/markup.py index e44766a..643e7b6 100644 --- a/vectorizing/svg/markup.py +++ b/vectorizing/svg/markup.py @@ -1,62 +1,35 @@ -def truncate(number): - """ - Truncates a number to two decimal places. - We use this to limit the amount of decimal places used - in the final generated SVG markup (reduces file sizes). +"""Serialize traced paths without strokes or changes to their paint order.""" - Parameters: - number: The number to truncate. +from collections.abc import Iterable, Sequence - Returns: - The truncated number. - """ - return "{:.2f}".format(number) +import numpy as np +from pathops import Path -def to_SVG_color_string(color): - """ - Converts a color (represented as a tuple) to a valid - SVG color string. +def truncate(number: float) -> str: + """Format a coordinate with two decimal places to limit SVG size.""" + return f"{number:.2f}" - Parameters: - color: The input color. - Returns: - The SVG color string. - - Example: - to_SVG_color_string([20, 30, 40]) => rgb(20, 30, 40) - to_SVG_color_string([20, 30, 40, 0.5]) => rgba(20, 30, 40, 0.5) - - """ +def to_SVG_color_string(color: Sequence[float] | np.ndarray) -> str: + """Format RGB or RGBA components without scaling or rounding their values.""" func = "rgb" if len(color) == 3 else "rgba" tuple = "".join( [ f"{item}," if i != len(color) - 1 else f"{item}" for i, item in enumerate(color) - ] + ], ) return f"{func}({tuple})" -def generate_SVG_markup(compound_paths, colors, width, height): - """ - Generates SVG markup from a list of compound paths, - their respective colors, and the document dimensions. - - Parameters: - compound_paths: The compound path list (SKPath). - colors: Each compound path's color. - width: The width of the document. - height: The height of the document. - - Returns: - The SVG markup. - - Note: - compound paths are assumed to be made out of - only lines and cubic beziers. - """ +def generate_SVG_markup( + compound_paths: Iterable[Path], + colors: Iterable[Sequence[float] | np.ndarray], + width: int, + height: int, +) -> str: + """Serialize line/cubic paths in paint order, skipping empty geometry.""" paths_markup = [] for compound_path, color in zip(compound_paths, colors): @@ -89,7 +62,8 @@ def generate_SVG_markup(compound_paths, colors, width, height): paths_markup = "\n".join(paths_markup) return ( - f'\n' + f'\n' f"\n{paths_markup}\n\n" f"" ) diff --git a/vectorizing/tests/__init__.py b/vectorizing/tests/__init__.py index e69de29..2f3e1c1 100644 --- a/vectorizing/tests/__init__.py +++ b/vectorizing/tests/__init__.py @@ -0,0 +1 @@ +"""Regression checks for image vectorization and storage.""" diff --git a/vectorizing/tests/color_reference.py b/vectorizing/tests/color_reference.py new file mode 100644 index 0000000..c67b9e8 --- /dev/null +++ b/vectorizing/tests/color_reference.py @@ -0,0 +1,105 @@ +"""Frozen pre-optimization algorithms used as independent regression oracles. + +Keep these intentionally slow implementations separate from production helpers: +sharing optimized code would let a regression affect both sides of a comparison. +Cleanup shares only the unchanged fill_holes step so tests can bypass it in both +paths to inspect component removal before nearest-neighbor filling hides holes. +""" + +import cv2 +import numpy as np +from PIL import Image +from skimage.measure import label + +from vectorizing.solvers.color import quantize as color_quantize + + +def legacy_initial_centroids(img_arr: np.ndarray, color_count: int) -> np.ndarray: + """Preserve the RGB-pixel centroid algorithm from before PR 71.""" + img = ( + Image.fromarray(img_arr) + .quantize(color_count, method=Image.Quantize.FASTOCTREE) + .convert("RGB") + ) + pixels = np.asarray(img) + return np.unique(pixels.reshape(-1, pixels.shape[-1]), axis=0).astype(np.uint8) + + +def legacy_create_bitmaps( + labels: np.ndarray, + colors: np.ndarray, + has_background: bool, +) -> tuple[list[np.ndarray], list[np.ndarray]]: + """Preserve the pairwise bitmap accumulation from before PR 73.""" + bitmaps = [ + np.where(labels == index, 1, 0).astype(np.uint32) + for index in range(len(colors)) + ] + if has_background: + bitmaps = bitmaps[1:] + colors = colors[1:] + zipped = list(zip(bitmaps, colors)) + zipped = [[bitmap, color] for bitmap, color in zipped if np.sum(bitmap) > 0] + bitmaps = [bitmap for bitmap, _ in zipped] + colors = [color for _, color in zipped] + for x in range(len(bitmaps)): + bitmap_x = bitmaps[x] + for y in range(x + 1, len(bitmaps)): + bitmap_x += bitmaps[y] + bitmaps[x] = bitmap_x + return bitmaps, colors + + +def legacy_enhance( + img_arr: np.ndarray, + labels: np.ndarray, + colors: np.ndarray, +) -> tuple[np.ndarray, np.ndarray]: + """Preserve the pairwise cleanup from main at f69c69b, before PR 74.""" + dims = img_arr.shape[:2] + labels = labels + 1 + clusters = [ + np.where(labels == idx + 1, idx + 1, 0).astype(np.uint16) + for idx, _ in enumerate(colors) + ] + # Preserve integer labeling, full connectivity and shifted component IDs. + # The optimized path must match this, not redefine the expected behavior. + original_connected_components_list = [label(cluster) + 1 for cluster in clusters] + connected_components_bincounts = [ + np.bincount(connected_components.flatten()) + for connected_components in original_connected_components_list + ] + connected_components_list = [ + np.array(item, copy=True) for item in original_connected_components_list + ] + # Keep the expensive pairwise dilation scans: independence is the point. + for x, cluster_x in enumerate(clusters): + dilated_cluster_x = cv2.dilate(cluster_x, np.ones((2, 2))) + for y, cluster_y in enumerate(clusters): + if x == y: + continue + overlap = np.logical_and(dilated_cluster_x, cluster_y) + connected_components_list[y] = np.where( + overlap, + 0, + connected_components_list[y], + ) + for x, connected_components in enumerate(connected_components_list): + original_bincount = connected_components_bincounts[x] + count = original_bincount.shape[0] + new_bincount = np.bincount(connected_components.flatten(), minlength=count) + areas_ratio = np.divide( + new_bincount.astype(np.float32), + original_bincount.astype(np.float32), + out=np.ones((count,), dtype=np.float32), + where=original_bincount != 0, + ) + clusters[x] = np.where( + areas_ratio[original_connected_components_list[x]] <= 0.1, + 0, + clusters[x], + ) + labels = np.zeros(dims).astype(np.uint8) + for cluster in clusters: + labels = np.where(labels == 0, cluster, labels) + return color_quantize.fill_holes(labels) - 1, colors diff --git a/vectorizing/tests/config.py b/vectorizing/tests/config.py index 871fff8..beeb78b 100644 --- a/vectorizing/tests/config.py +++ b/vectorizing/tests/config.py @@ -1,57 +1,36 @@ +"""Define image fixtures and solver settings for the integration gallery.""" + TESTS = { - '1px.jpg': [ - { - 'id': '1px_binary', - 'solver': 0 - }, - { - 'id': '1px_color', - 'solver': 1, - 'color_count': 9 - } + "1px.jpg": [ + {"id": "1px_binary", "solver": 0}, + {"id": "1px_color", "solver": 1, "color_count": 9}, ], - 'black_rectangle.png': [ + "black_rectangle.png": [ { - 'id': 'black_rectangle_binary', - 'solver': 0, + "id": "black_rectangle_binary", + "solver": 0, }, { - 'id': 'black_rectangle_color', - 'solver': 1, - 'color_count': 5, + "id": "black_rectangle_color", + "solver": 1, + "color_count": 5, }, ], - 'bubbles.png': [ - { - 'id': 'bubbles', - 'solver': 1, - 'color_count': 5 - } - ], - 'geo_logo.png': [ + "bubbles.png": [{"id": "bubbles", "solver": 1, "color_count": 5}], + "geo_logo.png": [ { - 'id': 'geo_logo', - 'solver': 0, + "id": "geo_logo", + "solver": 0, }, ], - 'shapes.png': [ - { - 'id': 'shapes', - 'solver': 1, - 'color_count': 4 - }, - ], - 'shapes_2.png': [ - { - 'id': 'shapes_2', - 'solver': 1, - 'color_count': 7 - } + "shapes.png": [ + {"id": "shapes", "solver": 1, "color_count": 4}, ], - 'white_text.png': [ + "shapes_2.png": [{"id": "shapes_2", "solver": 1, "color_count": 7}], + "white_text.png": [ { - 'id': 'white_text', - 'solver': 0, + "id": "white_text", + "solver": 0, }, ], -} \ No newline at end of file +} diff --git a/vectorizing/tests/conftest.py b/vectorizing/tests/conftest.py index 993f15c..64bafcd 100644 --- a/vectorizing/tests/conftest.py +++ b/vectorizing/tests/conftest.py @@ -1,12 +1,16 @@ +"""Provide an HTTP client isolated from the application's S3 bucket.""" + import os import pytest +from flask.testing import FlaskClient import vectorizing @pytest.fixture() -def client(monkeypatch): +def client(monkeypatch: pytest.MonkeyPatch) -> FlaskClient: + """Route test uploads to a required, separate bucket and enable Flask testing.""" test_bucket = os.getenv("S3_TEST_BUCKET") if not test_bucket: pytest.fail("S3_TEST_BUCKET must be set") diff --git a/vectorizing/tests/test.py b/vectorizing/tests/test.py deleted file mode 100644 index 6dc3dd2..0000000 --- a/vectorizing/tests/test.py +++ /dev/null @@ -1,344 +0,0 @@ -import os -from io import BytesIO -from pathlib import Path -from unittest.mock import Mock - -import numpy as np -import pytest -from cairosvg import svg2png -from PIL import Image - -from vectorizing.geometry.bounds import compound_paths_bounds -from vectorizing.server.s3 import get_s3_client -from vectorizing.server.timer import Timer -from vectorizing.solvers.color import quantize as color_quantize -from vectorizing.solvers.color.bitmaps import create_bitmaps -from vectorizing.solvers.color.ColorSolver import ColorSolver -from vectorizing.svg.markup import generate_SVG_markup -from vectorizing.tests import testutil -from vectorizing.tests.config import TESTS -from vectorizing.tests.testutil import ( - MAX_IMAGE_DIFFERENCE, - RESULTS_FOLDER_PATH, - compute_img_difference, - get_image_url, - get_markup, - write_img_difference, -) - - -def test(client): - """ - Tests all images inside /images directory using the TESTS object - in config.py - """ - - results_path = Path(RESULTS_FOLDER_PATH) - - # Create results folder if non-existent - if not results_path.exists(): - results_path.mkdir() - - # Cache test results to allow all tests to run - diffs = [] - - for img_name, options_list in TESTS.items(): - for options in options_list: - id = options.get("id") - output_name = f"{id}.png" - - # Vectorized SVG markup - markup = get_markup(client, img_name, options) - - # Rasterized SVG - png = svg2png(bytestring=markup) - img = Image.open(BytesIO(png)).convert("RGB") - - # Store result if it doesn't exist yet - results_output_path = Path(RESULTS_FOLDER_PATH / output_name) - if not results_output_path.exists(): - img.save(results_output_path) - - else: - # Difference between result and expected result - difference = compute_img_difference( - img, - RESULTS_FOLDER_PATH / output_name, - ) - - # Write image difference to diff_output - if difference >= MAX_IMAGE_DIFFERENCE: - write_img_difference( - img, - RESULTS_FOLDER_PATH / output_name, - output_name, - ) - - diffs.append(difference) - - assert all([diff <= MAX_IMAGE_DIFFERENCE for diff in diffs]) - - -def test_write_img_difference_handles_rgba_baselines(tmp_path, monkeypatch): - baseline = tmp_path / "baseline.png" - output = tmp_path / "difference.png" - Image.new("RGBA", (2, 2), (0, 0, 0, 0)).save(baseline) - monkeypatch.setattr(testutil, "DIFF_OUTPUT_FOLDER_PATH", tmp_path) - - testutil.write_img_difference( - Image.new("RGB", (2, 2), (255, 255, 255)), - baseline, - output.name, - ) - - assert output.exists() - - -def test_color_solver_preserves_empty_transparent_image(): - image = Image.open(Path(__file__).parent / "images" / "empty.png") - paths, colors, width, height = ColorSolver(image, 6, Timer()).solve() - markup = generate_SVG_markup(paths, colors, width, height) - rendered = Image.open(BytesIO(svg2png(bytestring=markup))).convert("RGBA") - - assert rendered.getchannel("A").getextrema() == (0, 0) - - -def legacy_initial_centroids(img_arr: np.ndarray, color_count: int) -> np.ndarray: - """Freeze the original RGB-pixel algorithm as an independent regression oracle.""" - img = ( - Image.fromarray(img_arr) - .quantize(color_count, method=Image.Quantize.FASTOCTREE) - .convert("RGB") - ) - pixels = np.asarray(img) - return np.unique(pixels.reshape(-1, pixels.shape[-1]), axis=0).astype(np.uint8) - - -@pytest.mark.parametrize( - "variant", - ["noise", "strided", "one-pixel", "solid", "sparse"], -) -@pytest.mark.parametrize("color_count", [2, 6, 16, 64]) -@pytest.mark.parametrize("seed", range(5)) -def test_initial_centroids_match_original( - color_count: int, - seed: int, - variant: str, -) -> None: - """Preserve exact dtype, values and ordering on random and sparse RGB inputs.""" - rng = np.random.default_rng(seed) - noise = rng.integers(0, 256, size=(63, 67, 3), dtype=np.uint8) - pixels = { - "noise": noise, - "strided": noise[::2, ::2], - "one-pixel": noise[:1, :1], - "solid": np.full_like(noise, (175, 80, 230)), # Ignore unused black entries. - "sparse": (noise // 128) * 128, - }[variant] - expected = legacy_initial_centroids(pixels, color_count) - actual = color_quantize.get_initial_centroids(pixels, color_count) - np.testing.assert_equal(actual.dtype, expected.dtype) - np.testing.assert_equal(actual.dtype, np.dtype(np.uint8)) - np.testing.assert_array_equal(actual, expected) - - -def test_initial_centroids_ignore_unused_and_duplicate_palette_entries( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Keep only used RGB colors, deduplicated and sorted rather than palette order.""" - indexed = Image.new("P", (3, 1)) - indexed.putdata([7, 3, 5]) - palette = np.zeros((256, 3), dtype=np.uint8) - palette[1] = (255, 0, 0) # Unused nonblack entry. - palette[3] = palette[7] = (90, 10, 20) # Duplicate used RGB entries. - palette[5] = (10, 20, 30) - indexed.putpalette(palette.ravel().tolist()) - with monkeypatch.context() as patch: - patch.setattr(Image.Image, "quantize", lambda *args, **kwargs: indexed) - pixels = np.zeros((1, 3, 3), dtype=np.uint8) - actual = color_quantize.get_initial_centroids(pixels, 64) - expected = legacy_initial_centroids(pixels, 64) - np.testing.assert_array_equal(actual, expected) - np.testing.assert_array_equal(actual, [[10, 20, 30], [90, 10, 20]]) - - -def legacy_create_bitmaps( - labels: np.ndarray, - colors: np.ndarray, - has_background: bool, -) -> tuple[list[np.ndarray], list[np.ndarray]]: - """Freeze the original pairwise accumulation as an independent regression oracle.""" - bitmaps = [ - np.where(labels == index, 1, 0).astype(np.uint32) - for index in range(len(colors)) - ] - if has_background: - bitmaps = bitmaps[1:] - colors = colors[1:] - zipped = list(zip(bitmaps, colors)) - zipped = [[bitmap, color] for bitmap, color in zipped if np.sum(bitmap) > 0] - bitmaps = [bitmap for bitmap, _ in zipped] - colors = [color for _, color in zipped] - for x in range(len(bitmaps)): - bitmap_x = bitmaps[x] - for y in range(x + 1, len(bitmaps)): - bitmap_x += bitmaps[y] - bitmaps[x] = bitmap_x - return bitmaps, colors - - -@pytest.mark.parametrize( - "variant", - ["noise", "strided", "one-row", "one-column", "solid", "background", "sparse"], -) -@pytest.mark.parametrize("color_count", [1, 2, 6, 16, 64, 65]) -@pytest.mark.parametrize("has_background", [False, True]) -@pytest.mark.parametrize("seed", range(3)) -def test_bitmap_layering_matches_original( - color_count: int, - has_background: bool, - seed: int, - variant: str, -) -> None: - """Preserve masks, dtype, color order and read-only inputs across label patterns.""" - rng = np.random.default_rng(seed) - noise = rng.integers(0, color_count, size=(19, 23), dtype=np.uint16) - colors = rng.integers(0, 256, size=(color_count, 4), dtype=np.uint8) - colors.setflags(write=False) - labels = { - "noise": noise, - "strided": noise[::2, ::2], - "one-row": noise[:1, :], - "one-column": noise[:, :1], - "solid": np.full_like(noise, color_count - 1), - "background": np.zeros_like(noise), - "sparse": noise % 2, - }[variant] - labels.setflags(write=False) - expected_bitmaps, expected_colors = legacy_create_bitmaps( - labels, - colors, - has_background, - ) - actual_bitmaps, actual_colors = create_bitmaps(labels, colors, has_background) - np.testing.assert_array_equal(actual_colors, expected_colors) - np.testing.assert_equal(len(actual_bitmaps), len(expected_bitmaps)) - for actual, expected in zip(actual_bitmaps, expected_bitmaps): - np.testing.assert_equal(actual.dtype, np.dtype(np.uint32)) - np.testing.assert_array_equal(actual, expected) - - -@pytest.mark.parametrize("has_background", [False, True]) -@pytest.mark.parametrize("color_count", [0, 1, 6]) -def test_bitmap_layering_handles_empty_inputs( - has_background: bool, - color_count: int, -) -> None: - """Return no layers for empty labels or an empty palette.""" - colors = np.zeros((color_count, 3), dtype=np.uint8) - labels = np.empty((0, 3), dtype=np.uint8) - np.testing.assert_equal(create_bitmaps(labels, colors, has_background), ([], [])) - np.testing.assert_equal( - create_bitmaps(np.zeros((2, 2), dtype=np.uint8), colors[:0], has_background), - ([], []), - ) - - -def test_bitmap_layering_preserves_overlap_and_transparency() -> None: - """Keep suffix unions in palette order without including transparent pixels.""" - labels = np.array([[0, 1, 3, 4]], dtype=np.uint8) - colors = np.arange(20, dtype=np.uint8).reshape(5, 4) - # Label zero is transparent; label two is unused between visible layers. - bitmaps, visible_colors = create_bitmaps(labels, colors, True) - expected = np.array( - [[[0, 1, 1, 1]], [[0, 0, 1, 1]], [[0, 0, 0, 1]]], - dtype=np.uint32, - ) - np.testing.assert_array_equal(bitmaps, expected) - np.testing.assert_array_equal(visible_colors, colors[[1, 3, 4]]) - # The in-place accumulation must not make different layers share storage. - bitmaps[0][0, 0] = 1 - np.testing.assert_array_equal(bitmaps[1:], expected[1:]) - np.testing.assert_array_equal(labels, [[0, 1, 3, 4]]) - - -@pytest.mark.parametrize("optimization", ["centroids", "bitmaps"]) -@pytest.mark.parametrize( - "image_name, color_count", - [ - ("bubbles.png", 5), - ("shapes_2.png", 7), - ("aftermath.png", 2), - ("aftermath.png", 6), - ("aftermath.png", 16), - ("aftermath.png", 64), - ("empty.png", 6), - ("1px.jpg", 9), - ], -) -def test_color_optimizations_preserve_vectorization( - image_name: str, - color_count: int, - optimization: str, - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Compare final SVG, colors, dimensions and bounds against the old algorithm.""" - with Image.open(Path(__file__).parent / "images" / image_name) as image: - actual = ColorSolver(image, color_count, Timer()).solve() - target, legacy = { - "centroids": ( - "vectorizing.solvers.color.quantize.get_initial_centroids", - legacy_initial_centroids, - ), - "bitmaps": ( - "vectorizing.solvers.color.ColorSolver.create_bitmaps", - legacy_create_bitmaps, - ), - }[optimization] - # Patch each lookup site and verify the legacy implementation was used. - reference = Mock(wraps=legacy) - monkeypatch.setattr(target, reference) - expected = ColorSolver(image, color_count, Timer()).solve() - reference.assert_called_once() - - actual_paths, actual_colors, actual_width, actual_height = actual - expected_paths, expected_colors, expected_width, expected_height = expected - np.testing.assert_array_equal(actual_colors, expected_colors) - np.testing.assert_equal( - (actual_width, actual_height), - (expected_width, expected_height), - ) - np.testing.assert_equal( - compound_paths_bounds(actual_paths), - compound_paths_bounds(expected_paths), - ) - # Byte-identical SVG is stricter than a raster tolerance and includes transparency. - np.testing.assert_equal( - generate_SVG_markup(*actual), - generate_SVG_markup(*expected), - ) - - -def test_uploads_markup_to_s3(client): - response = client.post( - "/", - json={ - "url": get_image_url("black_rectangle.png"), - "solver": 0, - "raw": False, - }, - ) - - assert response.status_code == 200 - payload = response.get_json() - test_bucket = os.environ["S3_TEST_BUCKET"] - s3 = get_s3_client() - try: - stored = s3.get_object( - Bucket=test_bucket, - Key=payload["objectId"], - ) - assert stored["ContentType"] == "image/svg+xml" - assert stored["Body"].read().startswith(b" None: + """Keep rejected inputs false and return the established 400 JSON response.""" + assert vectorizing.validate_args(payload) is False + response = client.post("/", json=payload) + assert response.status_code == 400 + assert response.get_json() == {"success": False, "error": "INVALID_PARAMETERS"} + + +@pytest.mark.parametrize( + "payload", + [ + {"url": "image"}, + {"url": None}, + {"url": "image", "solver": True}, + {"url": "image", "crop_box": []}, + {"url": "image", "crop_box": [True, 0, 3, 4]}, + ], +) +def test_argument_validation_preserves_accepted_values( + payload: dict[str, object], +) -> None: + """Preserve presence-only URL checks, falsey crop boxes and integer booleans.""" + result = vectorizing.validate_args(payload) + assert isinstance(result, SimpleNamespace) + assert vars(result) == { + "url": payload["url"], + "solver": payload.get("solver", 0), + "crop_box": payload.get("crop_box"), + "raw": None, + "color_count": None, + } + + +@pytest.mark.parametrize("solver", [0, 1]) +def test_raw_requests_crop_before_vectorizing_and_do_not_upload( + solver: int, + client: FlaskClient, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep raw responses as SVG bytes and preserve crop ordering for both solvers.""" + image = Image.new("RGB", (8, 8), (40, 50, 60)) + load = Mock(return_value=image) + binary = Mock(wraps=vectorizing.process_binary) + color = Mock(wraps=vectorizing.process_color) + upload = Mock() + monkeypatch.setattr(vectorizing, "try_read_image_from_url", load) + monkeypatch.setattr(vectorizing, "process_binary", binary) + monkeypatch.setattr(vectorizing, "process_color", color) + monkeypatch.setattr(vectorizing, "upload_markup", upload) + response = client.post( + "/", + json={ + "url": "image", + "solver": solver, + "color_count": 2, + "crop_box": [0, 1, 4, 5], + "raw": True, + }, + ) + assert response.status_code == 200 + assert response.mimetype == "image/svg+xml" + assert response.data.startswith(b" None: + """Return the established 500 payload when image processing raises an error.""" + monkeypatch.setattr( + vectorizing, + "try_read_image_from_url", + Mock(side_effect=ValueError("bad image")), + ) + response = client.post("/", json={"url": "image"}) + assert response.status_code == 500 + assert response.get_json() == {"success": False, "error": "INTERNAL_SERVER_ERROR"} + + +def test_health_response(client: FlaskClient) -> None: + """Keep the health endpoint's status and JSON payload stable.""" + response = client.get("/health") + assert response.status_code == 200 + assert response.get_json() == {"success": True} diff --git a/vectorizing/tests/test_color_bitmaps.py b/vectorizing/tests/test_color_bitmaps.py new file mode 100644 index 0000000..c5d1ec4 --- /dev/null +++ b/vectorizing/tests/test_color_bitmaps.py @@ -0,0 +1,85 @@ +"""Overlapping bitmap regressions, including transparent and unused colors. + +Masks must be cumulative in palette order without sharing writable storage. +""" + +import numpy as np +import pytest + +from vectorizing.solvers.color.bitmaps import create_bitmaps +from vectorizing.tests.color_reference import legacy_create_bitmaps + + +@pytest.mark.parametrize( + "variant", + ["noise", "strided", "one-row", "one-column", "solid", "background", "sparse"], +) +@pytest.mark.parametrize("color_count", [1, 2, 6, 16, 64, 65]) +@pytest.mark.parametrize("has_background", [False, True]) +@pytest.mark.parametrize("seed", range(3)) +def test_bitmap_layering_matches_original( + color_count: int, + has_background: bool, + seed: int, + variant: str, +) -> None: + """Preserve masks, dtype, color order and read-only inputs across label patterns.""" + rng = np.random.default_rng(seed) + noise = rng.integers(0, color_count, size=(19, 23), dtype=np.uint16) + colors = rng.integers(0, 256, size=(color_count, 4), dtype=np.uint8) + colors.setflags(write=False) + labels = { + "noise": noise, + "strided": noise[::2, ::2], + "one-row": noise[:1, :], + "one-column": noise[:, :1], + "solid": np.full_like(noise, color_count - 1), + "background": np.zeros_like(noise), + "sparse": noise % 2, + }[variant] + labels.setflags(write=False) + expected_bitmaps, expected_colors = legacy_create_bitmaps( + labels, + colors, + has_background, + ) + actual_bitmaps, actual_colors = create_bitmaps(labels, colors, has_background) + np.testing.assert_array_equal(actual_colors, expected_colors) + np.testing.assert_equal(len(actual_bitmaps), len(expected_bitmaps)) + for actual, expected in zip(actual_bitmaps, expected_bitmaps): + np.testing.assert_equal(actual.dtype, np.dtype(np.uint32)) + np.testing.assert_array_equal(actual, expected) + + +@pytest.mark.parametrize("has_background", [False, True]) +@pytest.mark.parametrize("color_count", [0, 1, 6]) +def test_bitmap_layering_handles_empty_inputs( + has_background: bool, + color_count: int, +) -> None: + """Return no layers for empty labels or an empty palette.""" + colors = np.zeros((color_count, 3), dtype=np.uint8) + labels = np.empty((0, 3), dtype=np.uint8) + np.testing.assert_equal(create_bitmaps(labels, colors, has_background), ([], [])) + np.testing.assert_equal( + create_bitmaps(np.zeros((2, 2), dtype=np.uint8), colors[:0], has_background), + ([], []), + ) + + +def test_bitmap_layering_preserves_overlap_and_transparency() -> None: + """Keep suffix unions in palette order without including transparent pixels.""" + labels = np.array([[0, 1, 3, 4]], dtype=np.uint8) + colors = np.arange(20, dtype=np.uint8).reshape(5, 4) + # Label zero is transparent; label two is unused between visible layers. + bitmaps, visible_colors = create_bitmaps(labels, colors, True) + expected = np.array( + [[[0, 1, 1, 1]], [[0, 0, 1, 1]], [[0, 0, 0, 1]]], + dtype=np.uint32, + ) + np.testing.assert_array_equal(bitmaps, expected) + np.testing.assert_array_equal(visible_colors, colors[[1, 3, 4]]) + # The in-place accumulation must not make different layers share storage. + bitmaps[0][0, 0] = 1 + np.testing.assert_array_equal(bitmaps[1:], expected[1:]) + np.testing.assert_array_equal(labels, [[0, 1, 3, 4]]) diff --git a/vectorizing/tests/test_color_centroids.py b/vectorizing/tests/test_color_centroids.py new file mode 100644 index 0000000..faa63b9 --- /dev/null +++ b/vectorizing/tests/test_color_centroids.py @@ -0,0 +1,59 @@ +"""Used-palette centroid regressions: exact values, dtype and ordering. + +Centroid ordering changes K-means results, so set equality is not sufficient. +""" + +import numpy as np +import pytest +from PIL import Image + +from vectorizing.solvers.color import quantize as color_quantize +from vectorizing.tests.color_reference import legacy_initial_centroids + + +@pytest.mark.parametrize( + "variant", + ["noise", "strided", "one-pixel", "solid", "sparse"], +) +@pytest.mark.parametrize("color_count", [2, 6, 16, 64]) +@pytest.mark.parametrize("seed", range(5)) +def test_initial_centroids_match_original( + color_count: int, + seed: int, + variant: str, +) -> None: + """Preserve exact dtype, values and ordering on random and sparse RGB inputs.""" + rng = np.random.default_rng(seed) + noise = rng.integers(0, 256, size=(63, 67, 3), dtype=np.uint8) + pixels = { + "noise": noise, + "strided": noise[::2, ::2], + "one-pixel": noise[:1, :1], + "solid": np.full_like(noise, (175, 80, 230)), # Ignore unused black entries. + "sparse": (noise // 128) * 128, + }[variant] + expected = legacy_initial_centroids(pixels, color_count) + actual = color_quantize.get_initial_centroids(pixels, color_count) + np.testing.assert_equal(actual.dtype, expected.dtype) + np.testing.assert_equal(actual.dtype, np.dtype(np.uint8)) + np.testing.assert_array_equal(actual, expected) + + +def test_initial_centroids_ignore_unused_and_duplicate_palette_entries( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep only used RGB colors, deduplicated and sorted rather than palette order.""" + indexed = Image.new("P", (3, 1)) + indexed.putdata([7, 3, 5]) + palette = np.zeros((256, 3), dtype=np.uint8) + palette[1] = (255, 0, 0) # Unused nonblack entry. + palette[3] = palette[7] = (90, 10, 20) # Duplicate used RGB entries. + palette[5] = (10, 20, 30) + indexed.putpalette(palette.ravel().tolist()) + with monkeypatch.context() as patch: + patch.setattr(Image.Image, "quantize", lambda *args, **kwargs: indexed) + pixels = np.zeros((1, 3, 3), dtype=np.uint8) + actual = color_quantize.get_initial_centroids(pixels, 64) + expected = legacy_initial_centroids(pixels, 64) + np.testing.assert_array_equal(actual, expected) + np.testing.assert_array_equal(actual, [[10, 20, 30], [90, 10, 20]]) diff --git a/vectorizing/tests/test_color_cleanup.py b/vectorizing/tests/test_color_cleanup.py new file mode 100644 index 0000000..adc6576 --- /dev/null +++ b/vectorizing/tests/test_color_cleanup.py @@ -0,0 +1,224 @@ +"""Cluster cleanup parity, morphology edge cases and component-map lifetime. + +Unfilled comparisons expose the actual keep/remove decisions: hole filling can +otherwise repaint a wrongly removed component and hide the regression. +""" + +import weakref + +import numpy as np +import pytest + +from vectorizing.solvers.color import quantize as color_quantize +from vectorizing.tests.color_reference import legacy_enhance + +# After the final label shift, an unfilled zero wraps to uint16's maximum. +# This sentinel is used only by tests bypassing fill_holes with a nonempty palette. +UNFILLED_LABEL = np.iinfo(np.uint16).max + + +@pytest.mark.parametrize("before_fill", [False, True], ids=["filled", "unfilled"]) +@pytest.mark.parametrize( + "variant", + [ + "noise", + "strided", + "one-pixel", + "one-row", + "one-column", + "solid", + "sparse", + "checkerboard", + ], +) +# Transparent artwork can add a background entry to the 64 requested colors. +@pytest.mark.parametrize("color_count", [1, 2, 6, 16, 64, 65]) +@pytest.mark.parametrize("seed", range(3)) +def test_cluster_cleanup_matches_original( + before_fill: bool, + variant: str, + color_count: int, + seed: int, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Preserve component decisions and filled labels without mutating inputs.""" + rng = np.random.default_rng(seed) + noise = rng.integers(0, color_count, (19, 23), dtype=np.uint8) + labels = { + "noise": noise, + "strided": noise[::2, ::2], + "one-pixel": noise[:1, :1], + "one-row": noise[:1, :], + "one-column": noise[:, :1], + "solid": np.full_like(noise, color_count - 1), + "sparse": noise % 2, + "checkerboard": ( + np.indices(noise.shape).sum(axis=0) % min(color_count, 2) + ).astype(np.uint8), + }[variant] + colors = rng.integers(0, 256, (color_count, 4), dtype=np.uint8) + pixels = np.zeros((*labels.shape, 3), dtype=np.uint8) + labels.setflags(write=False) + colors.setflags(write=False) + pixels.setflags(write=False) + original_labels = labels.copy() + if before_fill: + # Compare removal decisions, not just the colors painted over their holes. + monkeypatch.setattr(color_quantize, "fill_holes", lambda matrix: matrix) + expected, expected_colors = legacy_enhance(pixels, labels, colors) + actual, actual_colors = color_quantize.enhance(pixels, labels, colors) + np.testing.assert_array_equal(actual, expected) + np.testing.assert_equal(actual.dtype, expected.dtype) + np.testing.assert_array_equal(actual_colors, expected_colors) + np.testing.assert_equal(actual_colors is colors, True) + np.testing.assert_array_equal(labels, original_labels) + + +@pytest.mark.parametrize("area", [9, 10, 11]) +def test_cluster_cleanup_preserves_inclusive_area_threshold( + area: int, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep a component with 1/9 interior area, but remove 1/10 and 1/11.""" + labels = np.zeros((8, 16), dtype=np.uint8) + # A 2x2 core has one interior pixel at (3, 3). Extending only its top row + # changes the component's total area without adding any interior pixels. + labels[2, 2:area] = 1 + labels[3, 2:4] = 1 + colors = np.zeros((2, 3), dtype=np.uint8) + pixels = np.zeros((*labels.shape, 3), dtype=np.uint8) + monkeypatch.setattr(color_quantize, "fill_holes", lambda matrix: matrix) + expected = labels.astype(np.uint16) + if area >= 10: + expected[labels == 1] = UNFILLED_LABEL + for cleaner in (legacy_enhance, color_quantize.enhance): + actual, _ = cleaner(pixels, labels, colors) + np.testing.assert_array_equal(actual, expected) + + +def test_cluster_cleanup_preserves_diagonal_connectivity( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Treat a 2x2 core and its diagonal thin tail as one removable component.""" + labels = np.zeros((8, 16), dtype=np.uint8) + # Seven tail pixels touch the 4-pixel core only at a corner. With full + # connectivity the combined 11-pixel component has 1/11 interior area; + # with 4-neighbor connectivity the core would incorrectly survive at 1/4. + labels[2:4, 2:4] = 1 + labels[4, 4:11] = 1 + colors = np.zeros((2, 3), dtype=np.uint8) + pixels = np.zeros((*labels.shape, 3), dtype=np.uint8) + monkeypatch.setattr(color_quantize, "fill_holes", lambda matrix: matrix) + expected = labels.astype(np.uint16) + expected[labels == 1] = UNFILLED_LABEL + for cleaner in (legacy_enhance, color_quantize.enhance): + actual, _ = cleaner(pixels, labels, colors) + np.testing.assert_array_equal(actual, expected) + + +def test_cluster_cleanup_preserves_dilation_anchor_and_border( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The 2x2 default anchor spares only the isolated top-left corner pixel.""" + labels = np.zeros((5, 5), dtype=np.uint8) + labels[::4, ::4] = 1 + colors = np.zeros((2, 3), dtype=np.uint8) + pixels = np.zeros((*labels.shape, 3), dtype=np.uint8) + monkeypatch.setattr(color_quantize, "fill_holes", lambda matrix: matrix) + expected = labels.astype(np.uint16) + expected[labels == 1] = UNFILLED_LABEL + # No upper/left source exists at (0, 0); outside the image is not a color. + expected[0, 0] = 1 + for cleaner in (legacy_enhance, color_quantize.enhance): + actual, _ = cleaner(pixels, labels, colors) + np.testing.assert_array_equal(actual, expected) + + +@pytest.mark.parametrize("before_fill", [False, True], ids=["filled", "unfilled"]) +@pytest.mark.parametrize("color_count", [0, 1, 3, 256]) +@pytest.mark.parametrize("dtype", [np.uint8, np.uint16, np.int16]) +def test_cluster_cleanup_preserves_unassigned_labels_and_dtype( + color_count: int, + dtype: type[np.generic], + before_fill: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Ignore unassigned neighbors and retain existing label-shift overflow behavior.""" + rng = np.random.default_rng(42) + # 255 + 1 wraps only for uint8; do not silently fix that old behavior here. + # Small palettes also leave labels whose pixels must not dilate into others. + labels = rng.choice([0, 1, 2, 3, 254, 255], size=(7, 9)).astype(dtype) + if dtype == np.int16: + labels[1, 1:3] = [-1, -2] + if before_fill: + monkeypatch.setattr(color_quantize, "fill_holes", lambda matrix: matrix) + colors = np.zeros((color_count, 3), dtype=np.uint8) + pixels = np.zeros((*labels.shape, 3), dtype=np.uint8) + expected, _ = legacy_enhance(pixels, labels, colors) + actual, actual_colors = color_quantize.enhance(pixels, labels, colors) + np.testing.assert_array_equal(actual, expected) + np.testing.assert_equal(actual.dtype, expected.dtype) + np.testing.assert_equal(actual_colors is colors, True) + + +@pytest.mark.parametrize("before_fill", [False, True], ids=["filled", "unfilled"]) +@pytest.mark.parametrize("transpose", [False, True]) +def test_cluster_cleanup_crops_labeling_without_losing_neighbor_influence( + before_fill: bool, + transpose: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Crop sparse colors while preserving disjoint islands and outside influences.""" + labels = np.zeros((13, 17), dtype=np.uint8) + labels[2:4, 3:5] = 1 + labels[8, 11] = 1 # A disconnected pixel of the same color must be removed. + labels[0, 14:17] = 2 # A single-row crop touching the image border. + labels[9, 2] = labels[10, 3] = 3 # Diagonal contact within a tight crop. + if transpose: + labels = labels.T + colors = np.zeros((6, 3), dtype=np.uint8) # Two palette entries are unused. + pixels = np.zeros((*labels.shape, 3), dtype=np.uint8) + if before_fill: + monkeypatch.setattr(color_quantize, "fill_holes", lambda matrix: matrix) + expected, _ = legacy_enhance(pixels, labels, colors) + original_label = color_quantize.label + shapes = [] + + def record_label(cluster: np.ndarray, connectivity: int) -> np.ndarray: + shapes.append(cluster.shape) + return original_label(cluster, connectivity=connectivity) + + monkeypatch.setattr(color_quantize, "label", record_label) + actual, _ = color_quantize.enhance(pixels, labels, colors) + np.testing.assert_array_equal(actual, expected) + # The background spans the image; each other color gets one minimal box, + # including all of its disconnected pieces, and unused entries get no pass. + expected_shapes = [(13, 17), (7, 9), (1, 3), (2, 2)] + if transpose: + expected_shapes = [(width, height) for height, width in expected_shapes] + np.testing.assert_array_equal(shapes, expected_shapes) + + +def test_cluster_cleanup_releases_each_component_map( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Release the old map before labeling the next color, not only on return.""" + original_label = color_quantize.label + maps: list[weakref.ReferenceType[np.ndarray]] = [] + + def record_label(cluster: np.ndarray, connectivity: int) -> np.ndarray: + # Weak references observe lifetime without keeping the maps alive. + # Checking before allocation catches even a temporary two-map overlap. + np.testing.assert_equal(all(ref() is None for ref in maps), True) + components = original_label(cluster, connectivity=connectivity) + maps.append(weakref.ref(components)) + return components + + monkeypatch.setattr(color_quantize, "label", record_label) + labels = (np.indices((8, 9)).sum(axis=0) % 4).astype(np.uint8) + pixels = np.zeros((*labels.shape, 3), dtype=np.uint8) + colors = np.zeros((4, 3), dtype=np.uint8) + color_quantize.enhance(pixels, labels, colors) + # All colors are present: a skipped labeler must not make this pass vacuously. + np.testing.assert_equal(len(maps), 4) + np.testing.assert_equal(all(ref() is None for ref in maps), True) diff --git a/vectorizing/tests/test_color_equivalence.py b/vectorizing/tests/test_color_equivalence.py new file mode 100644 index 0000000..2be8255 --- /dev/null +++ b/vectorizing/tests/test_color_equivalence.py @@ -0,0 +1,83 @@ +"""Full solver regressions comparing exact SVG bytes, colors and geometry. + +Replace one optimization at a time with its frozen reference while retaining the +others. This catches interactions that isolated mask/centroid/cleanup tests miss. +""" + +from pathlib import Path +from unittest.mock import Mock + +import numpy as np +import pytest +from PIL import Image + +from vectorizing.geometry.bounds import compound_paths_bounds +from vectorizing.server.timer import Timer +from vectorizing.solvers.color.ColorSolver import ColorSolver +from vectorizing.svg.markup import generate_SVG_markup +from vectorizing.tests.color_reference import ( + legacy_create_bitmaps, + legacy_enhance, + legacy_initial_centroids, +) + + +@pytest.mark.parametrize("optimization", ["centroids", "bitmaps", "cleanup"]) +@pytest.mark.parametrize( + "image_name, color_count", + [ + ("bubbles.png", 5), + ("shapes_2.png", 7), + ("aftermath.png", 2), + ("aftermath.png", 6), + ("aftermath.png", 16), + ("aftermath.png", 64), + ("empty.png", 6), + ("1px.jpg", 9), + ], +) +def test_color_optimizations_preserve_vectorization( + image_name: str, + color_count: int, + optimization: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Compare final SVG, colors, dimensions and bounds against the old algorithm.""" + with Image.open(Path(__file__).parent / "images" / image_name) as image: + actual = ColorSolver(image, color_count, Timer()).solve() + target, legacy = { + "centroids": ( + "vectorizing.solvers.color.quantize.get_initial_centroids", + legacy_initial_centroids, + ), + "bitmaps": ( + "vectorizing.solvers.color.ColorSolver.create_bitmaps", + legacy_create_bitmaps, + ), + "cleanup": ( + "vectorizing.solvers.color.quantize.enhance", + legacy_enhance, + ), + }[optimization] + # Patch each lookup site and verify the legacy implementation was used. + reference = Mock(wraps=legacy) + monkeypatch.setattr(target, reference) + expected = ColorSolver(image, color_count, Timer()).solve() + reference.assert_called_once() + + actual_paths, actual_colors, actual_width, actual_height = actual + expected_paths, expected_colors, expected_width, expected_height = expected + np.testing.assert_array_equal(actual_colors, expected_colors) + np.testing.assert_equal( + (actual_width, actual_height), + (expected_width, expected_height), + ) + np.testing.assert_equal( + compound_paths_bounds(actual_paths), + compound_paths_bounds(expected_paths), + ) + # Byte-identical SVG is stricter than a raster tolerance and includes transparency. + np.testing.assert_equal( + generate_SVG_markup(*actual), + generate_SVG_markup(*expected), + ) diff --git a/vectorizing/tests/test_developer_tools.py b/vectorizing/tests/test_developer_tools.py new file mode 100644 index 0000000..2df8f3a --- /dev/null +++ b/vectorizing/tests/test_developer_tools.py @@ -0,0 +1,406 @@ +"""Exercise offline reports, failure exits and benchmark measurement contracts.""" + +import copy +import json +import os +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from PIL import Image + +from scripts import _vectorization as shared +from scripts import benchmark_vectorization as bench +from scripts import vectorization_outputs as outputs + +SVG = ( + '' + '' +) +DETAILS = { + "input": { + "image": "1px.jpg", + "sha256": "fixture", + "solver": "binary", + "color_count": None, + }, + "colors": [[255, 0, 0, 1]], + "colors_dtype": "int64", + "dimensions": [8, 8], + "bounds": {"left": 0, "top": 0, "right": 8, "bottom": 8, "width": 8, "height": 8}, +} + + +@pytest.fixture +def fake_solver(monkeypatch: pytest.MonkeyPatch) -> None: + """Supply small deterministic outputs without exercising numerical algorithms.""" + monkeypatch.setattr( + shared, + "prepare_case", + lambda *_: lambda: (SVG, copy.deepcopy(DETAILS)), + ) + monkeypatch.setattr(shared, "environment", lambda: {"python": "fixture"}) + + +def save_capture(folder: Path, *names: str) -> dict[str, object]: + """Capture fixture outputs through the public command interface.""" + args = ["capture", str(folder)] + for name in names or ("1px_binary",): + args += ["--case", name] + assert outputs.main(args) == 0 + return json.loads((folder / "outputs.json").read_text()) + + +def test_capture_and_compare(tmp_path: Path, fake_solver: None) -> None: + """Artifacts contain actual SVG/pixel hashes and compare across metadata labels.""" + before, after = tmp_path / "before", tmp_path / "after" + report = save_capture(before) + save_capture(after) + assert (before / "1px_binary.svg").read_bytes() == SVG.encode() + record = report["cases"]["1px_binary"] + assert record["rgba_shape"] == [8, 8, 4] + assert record["rgba_sha256"] == outputs.pixels(before / "1px_binary.png")[0] + assert record["svg_sha256"] == shared.file_hash(before / "1px_binary.svg") + report["label"] = "different label" + report["environment"] = {"python": "other runtime"} + report["source_sha256"] = "different source" + (after / "outputs.json").write_text(json.dumps(report)) + assert outputs.main(["compare", str(before), str(after)]) == 0 + + +@pytest.mark.parametrize( + "field", + ["colors", "colors_dtype", "dimensions", "bounds", "input"], +) +def test_changed_metadata_fails( + tmp_path: Path, + fake_solver: None, + field: str, +) -> None: + """Equal rendered pixels cannot hide changed solver metadata or input identity.""" + before, after = tmp_path / "before", tmp_path / "after" + save_capture(before) + report = save_capture(after) + report["cases"]["1px_binary"][field] = "changed" + (after / "outputs.json").write_text(json.dumps(report)) + assert outputs.main(["compare", str(before), str(after)]) == 1 + + +def test_changed_svg_fails(tmp_path: Path, fake_solver: None) -> None: + """Different SVG bytes fail even when both versions render identically.""" + before, after = tmp_path / "before", tmp_path / "after" + save_capture(before) + report = save_capture(after) + path = after / "1px_binary.svg" + path.write_bytes(path.read_bytes() + b"\n") + report["cases"]["1px_binary"]["svg_sha256"] = shared.file_hash(path) + (after / "outputs.json").write_text(json.dumps(report)) + assert outputs.main(["compare", str(before), str(after)]) == 1 + + +def test_changed_pixels_fail(tmp_path: Path, fake_solver: None) -> None: + """Raster differences fail even when the SVG and solver metadata match.""" + before, after = tmp_path / "before", tmp_path / "after" + save_capture(before) + report = save_capture(after) + Image.new("RGBA", (8, 8), "blue").save(after / "1px_binary.png") + report["cases"]["1px_binary"]["rgba_sha256"] = outputs.pixels( + after / "1px_binary.png", + )[0] + (after / "outputs.json").write_text(json.dumps(report)) + assert outputs.main(["compare", str(before), str(after)]) == 1 + + +def test_changed_case_sets_fail(tmp_path: Path, fake_solver: None) -> None: + """Neither missing nor extra cases are silently skipped.""" + before, after = tmp_path / "before", tmp_path / "after" + save_capture(before) + save_capture(after, "1px_binary", "empty-color") + assert outputs.main(["compare", str(before), str(after)]) == 1 + assert outputs.main(["compare", str(after), str(before)]) == 1 + + +@pytest.mark.parametrize( + "damage", + ["svg", "png", "missing", "empty", "format", "fields", "nan", "name"], +) +def test_invalid_captures_fail( + tmp_path: Path, + fake_solver: None, + damage: str, +) -> None: + """Corrupted, incomplete and unsupported captures cannot report equality.""" + report = save_capture(tmp_path) + if damage == "svg": + (tmp_path / "1px_binary.svg").write_text("changed") + elif damage == "png": + Image.new("RGBA", (8, 8), "blue").save(tmp_path / "1px_binary.png") + elif damage == "missing": + (tmp_path / "1px_binary.png").unlink() + elif damage == "empty": + report["cases"] = {} + elif damage == "format": + report["schema_version"] = 100 + elif damage == "fields": + del report["cases"]["1px_binary"]["bounds"] + elif damage == "nan": + report["cases"]["1px_binary"]["bounds"] = float("nan") + else: + report["cases"] = {"../escape": report["cases"]["1px_binary"]} + (tmp_path / "outputs.json").write_text(json.dumps(report)) + with pytest.raises(SystemExit) as error: + outputs.main(["compare", str(tmp_path), str(tmp_path)]) + assert error.value.code == 2 + + +def test_refuse_overwrites(tmp_path: Path, fake_solver: None) -> None: + """Existing captures and benchmark reports survive accidental repeated commands.""" + save_capture(tmp_path) + original = (tmp_path / "outputs.json").read_bytes() + for command, args in [ + (outputs.main, ["capture", str(tmp_path), "--case", "1px_binary"]), + (bench.main, ["--output", str(tmp_path / "outputs.json")]), + ]: + with pytest.raises(SystemExit) as error: + command(args) + assert error.value.code == 2 + assert (tmp_path / "outputs.json").read_bytes() == original + + +@pytest.mark.parametrize("kind", ["capture", "benchmark"]) +def test_racing_writer_cannot_overwrite( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + kind: str, +) -> None: + """Files created after the initial existence check are also protected.""" + path = tmp_path / ("1px_binary.svg" if kind == "capture" else "report.json") + + def other_capture(*_: object) -> object: + path.write_text("another writer") + return lambda: (SVG, DETAILS) + + def other_benchmark(*_: object) -> tuple[dict[str, object], bool]: + path.write_text("another writer") + return {}, True + + monkeypatch.setattr(shared, "prepare_case", other_capture) + monkeypatch.setattr(bench, "benchmark", other_benchmark) + with pytest.raises(SystemExit) as error: + if kind == "capture": + outputs.main(["capture", str(tmp_path), "--case", "1px_binary"]) + else: + bench.main(["--output", str(path)]) + assert error.value.code == 2 + assert path.read_text() == "another writer" + + +def test_failed_capture_has_no_manifest( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failed run cannot leave a manifest advertising a complete baseline.""" + monkeypatch.setattr(shared, "prepare_case", Mock(side_effect=ValueError("failed"))) + with pytest.raises(SystemExit): + outputs.main(["capture", str(tmp_path), "--case", "1px_binary"]) + assert not (tmp_path / "outputs.json").exists() + + +@pytest.mark.parametrize("platform,value", [("linux", 3072), ("darwin", 3145728)]) +def test_rss_units(monkeypatch: pytest.MonkeyPatch, platform: str, value: int) -> None: + """Linux KiB and macOS byte counters both become MiB.""" + resource = pytest.importorskip("resource") + monkeypatch.setattr(bench.sys, "platform", platform) + monkeypatch.setattr( + resource, + "getrusage", + lambda _: SimpleNamespace(ru_maxrss=value), + ) + assert bench.peak_rss_mib() == 3 + + +def test_alternation_medians_and_equality(monkeypatch: pytest.MonkeyPatch) -> None: + """Paired samples alternate order and report medians rather than best runs.""" + order = [] + times = {"baseline": iter([9, 1, 5]), "candidate": iter([3, 1, 2])} + + def measure(name: str, root: Path, warmups: int) -> dict[str, object]: + order.append(root.name) + assert name == "bubbles" and warmups == 2 + return { + "elapsed_ms": next(times[root.name]), + "peak_rss_mib": 12, + "output": {"same": True}, + "consistent": True, + } + + monkeypatch.setattr(bench, "run_sample", measure) + cases, equal = bench.benchmark( + ["bubbles"], + {name: Path(name) for name in times}, + repeats=3, + warmups=2, + ) + assert order == [ + "baseline", + "candidate", + "candidate", + "baseline", + "baseline", + "candidate", + ] + assert equal + assert cases["bubbles"]["medians"]["baseline"]["elapsed_ms"] == 5 + assert cases["bubbles"]["medians"]["candidate"]["elapsed_ms"] == 2 + + +@pytest.mark.parametrize("inconsistent", [False, True]) +def test_benchmark_mismatches_save_failing_report( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + inconsistent: bool, +) -> None: + """Cross-version changes and within-process nondeterminism both fail the command.""" + + def measure(name: str, root: Path, warmups: int) -> dict[str, object]: + return { + "elapsed_ms": 1, + "peak_rss_mib": 2, + "consistent": not inconsistent, + "output": {"svg": "same" if inconsistent else str(root)}, + } + + monkeypatch.setattr(bench, "run_sample", measure) + path = tmp_path / "report.json" + assert ( + bench.main( + [ + "--output", + str(path), + "--baseline-root", + str(tmp_path), + "--case", + "bubbles", + "--repeats", + "1", + ], + ) + == 1 + ) + assert json.loads(path.read_text())["outputs_equal"] is False + + +def test_warmups_and_instability( + fake_solver: None, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Warmups execute before measurement and their changed output is not ignored.""" + run = Mock(side_effect=[(SVG, DETAILS), (SVG + "\n", DETAILS), (SVG, DETAILS)]) + monkeypatch.setattr(shared, "prepare_case", lambda *_: run) + monkeypatch.setattr(bench, "peak_rss_mib", lambda: 1.0) + monkeypatch.setattr( + shared, + "environment", + lambda: {"completed_runs": run.call_count}, + ) + row = bench.sample("1px_binary", shared.ROOT, warmups=2) + assert run.call_count == 3 + assert row["environment"]["completed_runs"] == 3 + assert row["consistent"] is False + assert row["elapsed_ms"] >= 0 + + +def test_instability_between_samples_fails(monkeypatch: pytest.MonkeyPatch) -> None: + """Separately stable processes must also agree with one another.""" + rows = [ + { + "elapsed_ms": 1, + "peak_rss_mib": 2, + "consistent": True, + "output": {"svg": value}, + } + for value in ["first", "second"] + ] + monkeypatch.setattr(bench, "run_sample", Mock(side_effect=rows)) + _, equal = bench.benchmark(["bubbles"], {"candidate": shared.ROOT}, 2, 0) + assert not equal + + +def test_worker_failure(monkeypatch: pytest.MonkeyPatch) -> None: + """A failed native worker propagates its error instead of yielding a timing.""" + process = Mock( + return_value=SimpleNamespace(returncode=7, stdout="", stderr="worker crashed"), + ) + monkeypatch.setattr(bench.subprocess, "run", process) + with pytest.raises(RuntimeError, match="worker crashed"): + bench.run_sample("bubbles", shared.ROOT, 1) + command = process.call_args.args[0] + assert command[0] == bench.sys.executable + assert Path(command[1]).is_absolute() + assert process.call_args.kwargs.get("shell", False) is False + + +@pytest.mark.parametrize( + "flag,value", + [("--repeats", "0"), ("--repeats", "-1"), ("--warmups", "-1")], +) +def test_invalid_sample_counts(flag: str, value: str) -> None: + """Invalid counts fail before any solver or worker runs.""" + with pytest.raises(SystemExit) as error: + bench.main([flag, value]) + assert error.value.code == 2 + + +@pytest.mark.skipif( + bench.sys.platform != "darwin" and not bench.sys.platform.startswith("linux"), + reason="Peak RSS measurement requires Linux or macOS", +) +def test_real_workers_are_offline_and_isolated( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Fresh workers resolve fixtures outside the repo cwd without AWS settings.""" + monkeypatch.chdir(tmp_path) + for name in ["PORT", "S3_BUCKET", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"]: + monkeypatch.delenv(name, raising=False) + first = bench.run_sample("1px_binary", shared.ROOT, 0) + second = bench.run_sample("1px_binary", shared.ROOT, 0) + assert first["pid"] != second["pid"] != os.getpid() + assert first["output"] == second["output"] + assert first["peak_rss_mib"] > 0 + assert first["environment"]["versions"]["numpy"] + + +def test_empty_bounds_are_valid_json() -> None: + """An empty color result retains infinite bounds as JSON-safe strings.""" + svg, details = shared.prepare_case("empty-color", shared.ROOT)() + assert details["colors"] == [] + assert details["bounds"]["left"] == "inf" + assert details["bounds"]["width"] == "-inf" + json.dumps(shared.fingerprint(svg, details), allow_nan=False) + + +def test_wrong_import_root_fails( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An already imported checkout cannot be mistaken for another source root.""" + monkeypatch.setattr(shared.sys, "path", list(shared.sys.path)) + (tmp_path / "vectorizing").mkdir() + (tmp_path / "vectorizing" / "__init__.py").write_text("") + with pytest.raises(ValueError, match="different checkout"): + shared.prepare_case("1px_binary", tmp_path) + + +def test_source_fingerprint(tmp_path: Path) -> None: + """Runtime edits change provenance; test-file edits do not.""" + folder = tmp_path / "vectorizing" + (folder / "tests").mkdir(parents=True) + code = folder / "solver.py" + code.write_text("first") + before = shared.source_hash(tmp_path) + (folder / "tests" / "test_solver.py").write_text("test") + assert shared.source_hash(tmp_path) == before + code.write_text("second") + assert shared.source_hash(tmp_path) != before diff --git a/vectorizing/tests/test_image_reading.py b/vectorizing/tests/test_image_reading.py new file mode 100644 index 0000000..cf4a009 --- /dev/null +++ b/vectorizing/tests/test_image_reading.py @@ -0,0 +1,116 @@ +"""Preserve image modes, request options and public loading-error behavior.""" + +from io import BytesIO +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from PIL import Image + +from vectorizing.util import read + + +@pytest.mark.parametrize( + "mode, expected_mode", + [ + ("RGB", "RGB"), + ("RGBA", "RGBA"), + ("1", "RGB"), + ("L", "RGB"), + ("P", "RGBA"), + ("PA", "RGBA"), + ], +) +def test_supported_image_modes(mode: str, expected_mode: str) -> None: + """Keep RGB/RGBA identity and preserve pixels when converting other modes.""" + image = Image.new(mode, (8, 8)) + actual = read.convert_RGB_A(image) + assert actual.mode == expected_mode + assert actual.tobytes() == image.convert(expected_mode).tobytes() + if mode in {"RGB", "RGBA"}: + assert actual is image + + +@pytest.mark.parametrize("reader", ["path", "url"]) +@pytest.mark.parametrize( + "exception_type", + [OSError, ValueError, MemoryError, KeyboardInterrupt, SystemExit], +) +def test_loading_failures_keep_the_public_wrapper( + reader: str, + exception_type: type[BaseException], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Retain catch-all wrapping, original context and the public exception args.""" + cause = exception_type("loading failed") + if reader == "path": + monkeypatch.setattr(read.Image, "open", Mock(side_effect=cause)) + loader, error_type = read.try_read_image_from_path, read.PathReadError + message = "Failed to read image from provided path." + else: + monkeypatch.setattr(read.requests, "get", Mock(side_effect=cause)) + loader, error_type = read.try_read_image_from_url, read.URLReadError + message = "Failed to read image from provided URL." + with pytest.raises(error_type) as caught: + loader("source") + assert caught.value.args == (caught.value, message) + assert caught.value.__context__ is cause + assert not caught.value.__suppress_context__ + + +@pytest.mark.parametrize("reader", ["path", "url"]) +def test_unsupported_modes_are_not_wrapped_as_loading_errors( + reader: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Leave mode conversion outside the image-loading exception wrapper.""" + monkeypatch.setattr( + read.Image, + "open", + Mock(return_value=Image.new("CMYK", (8, 8))), + ) + monkeypatch.setattr( + read.requests, + "get", + Mock(return_value=SimpleNamespace(content=b"")), + ) + loader = ( + read.try_read_image_from_path + if reader == "path" + else read.try_read_image_from_url + ) + with pytest.raises(read.ImageFormatError): + loader("source") + + +@pytest.mark.parametrize("status_code", [200, 404]) +def test_url_reader_preserves_request_and_decoding_contract( + status_code: int, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Send the service user agent and decode images regardless of HTTP status.""" + image = Image.new("RGBA", (8, 8), (20, 30, 40, 128)) + encoded = BytesIO() + image.save(encoded, format="PNG") + get = Mock( + return_value=SimpleNamespace( + content=encoded.getvalue(), + status_code=status_code, + ), + ) + monkeypatch.setattr(read.requests, "get", get) + url = "https://example.invalid/image.png?signature=fixture" + actual = read.try_read_image_from_url(url) + get.assert_called_once_with(url, headers={"User-Agent": "KittlVectorizing/1.0.0"}) + assert actual.mode == "RGBA" + assert actual.tobytes() == image.tobytes() + + +def test_path_reader_accepts_path_objects(tmp_path: Path) -> None: + """Read Path objects and normalize grayscale pixels to RGB.""" + path = tmp_path / "gray.png" + Image.new("L", (8, 8), 64).save(path) + actual = read.try_read_image_from_path(path) + assert actual.mode == "RGB" + assert actual.getpixel((0, 0)) == (64, 64, 64) diff --git a/vectorizing/tests/test_integration.py b/vectorizing/tests/test_integration.py new file mode 100644 index 0000000..8cb8231 --- /dev/null +++ b/vectorizing/tests/test_integration.py @@ -0,0 +1,133 @@ +"""HTTP/S3 integration, raster diagnostics and transparent SVG checks. + +The persisted raster gallery is a tolerant smoke test and creates missing +baselines. Exact before/after output checks live in test_color_equivalence.py. +Layer editability and anti-aliased edges are covered by the color layer suites. +""" + +import os +from io import BytesIO +from pathlib import Path + +import pytest +from cairosvg import svg2png +from flask.testing import FlaskClient +from PIL import Image + +from vectorizing.server.s3 import get_s3_client +from vectorizing.server.timer import Timer +from vectorizing.solvers.color.ColorSolver import ColorSolver +from vectorizing.svg.markup import generate_SVG_markup +from vectorizing.tests import testutil +from vectorizing.tests.config import TESTS +from vectorizing.tests.testutil import ( + MAX_IMAGE_DIFFERENCE, + RESULTS_FOLDER_PATH, + compute_img_difference, + get_image_url, + get_markup, + write_img_difference, +) + + +def test(client: FlaskClient) -> None: + """Render the configured gallery and compare or seed its local PNG baselines.""" + results_path = Path(RESULTS_FOLDER_PATH) + + # Create results folder if non-existent + if not results_path.exists(): + results_path.mkdir() + + # Cache test results to allow all tests to run + diffs = [] + + for img_name, options_list in TESTS.items(): + for options in options_list: + id = options.get("id") + output_name = f"{id}.png" + + # Vectorized SVG markup + markup = get_markup(client, img_name, options) + + # Rasterized SVG + png = svg2png(bytestring=markup) + img = Image.open(BytesIO(png)).convert("RGB") + + # Store result if it doesn't exist yet + results_output_path = Path(RESULTS_FOLDER_PATH / output_name) + if not results_output_path.exists(): + img.save(results_output_path) + + else: + # Difference between result and expected result + difference = compute_img_difference( + img, + RESULTS_FOLDER_PATH / output_name, + ) + + # Write image difference to diff_output + if difference >= MAX_IMAGE_DIFFERENCE: + write_img_difference( + img, + RESULTS_FOLDER_PATH / output_name, + output_name, + ) + + diffs.append(difference) + + assert all([diff <= MAX_IMAGE_DIFFERENCE for diff in diffs]) + + +def test_write_img_difference_handles_rgba_baselines( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Normalize transparent baseline pixels before computing RGB differences.""" + baseline = tmp_path / "baseline.png" + output = tmp_path / "difference.png" + Image.new("RGBA", (2, 2), (0, 0, 0, 0)).save(baseline) + monkeypatch.setattr(testutil, "DIFF_OUTPUT_FOLDER_PATH", tmp_path) + + testutil.write_img_difference( + Image.new("RGB", (2, 2), (255, 255, 255)), + baseline, + output.name, + ) + + assert output.exists() + + +def test_color_solver_preserves_empty_transparent_image() -> None: + """Keep every rendered pixel transparent for an empty input image.""" + image = Image.open(Path(__file__).parent / "images" / "empty.png") + paths, colors, width, height = ColorSolver(image, 6, Timer()).solve() + markup = generate_SVG_markup(paths, colors, width, height) + rendered = Image.open(BytesIO(svg2png(bytestring=markup))).convert("RGBA") + + assert rendered.getchannel("A").getextrema() == (0, 0) + + +def test_uploads_markup_to_s3(client: FlaskClient) -> None: + """Store the generated SVG with its media type and return the object's key.""" + response = client.post( + "/", + json={ + "url": get_image_url("black_rectangle.png"), + "solver": 0, + "raw": False, + }, + ) + + assert response.status_code == 200 + payload = response.get_json() + test_bucket = os.environ["S3_TEST_BUCKET"] + s3 = get_s3_client() + try: + stored = s3.get_object( + Bucket=test_bucket, + Key=payload["objectId"], + ) + assert stored["ContentType"] == "image/svg+xml" + assert stored["Body"].read().startswith(b" None: + """Retain Python's two-decimal formatting rather than truncating coordinates.""" + assert truncate(number) == expected + + +@pytest.mark.parametrize( + "color, expected", + [ + ([20, 30, 40], "rgb(20,30,40)"), + ([20, 30, 40, 0.5], "rgba(20,30,40,0.5)"), + ([20, 30, 40, 0], "rgba(20,30,40,0)"), + (np.array([20, 30, 40, 0.5]), "rgba(20.0,30.0,40.0,0.5)"), + ], +) +def test_color_formatting(color: Sequence[float] | np.ndarray, expected: str) -> None: + """Preserve component representations without scaling alpha or adding spaces.""" + assert to_SVG_color_string(color) == expected + + +def test_svg_serialization_preserves_exact_markup() -> None: + """Keep paint order, empty-path filtering, cubic syntax and document whitespace.""" + curve = Path() + curve.moveTo(0, 0) + curve.lineTo(10, 0) + curve.cubicTo(10, 1, 9, 2, 0, 0) + curve.close() + overlay = Path() + overlay.moveTo(1, 1) + overlay.lineTo(2, 2) + overlay.close() + actual = generate_SVG_markup( + [Path(), curve, overlay], + [[255, 255, 255], [20, 30, 40], [50, 60, 70, 0.5]], + 40, + 20, + ) + assert actual == ( + '\n' + '\n\n' + '\n' + "\n" + ) + + +def test_empty_svg_serialization() -> None: + """Keep the empty document's dimensions and blank group unchanged.""" + assert generate_SVG_markup([], [], 8, 9) == ( + '\n\n\n\n' + ) diff --git a/vectorizing/tests/testutil.py b/vectorizing/tests/testutil.py index 11ab5a6..6539a24 100644 --- a/vectorizing/tests/testutil.py +++ b/vectorizing/tests/testutil.py @@ -1,7 +1,10 @@ +"""Support the image gallery with S3 fixtures and RGB difference images.""" + from pathlib import Path import cv2 import numpy as np +from flask.testing import FlaskClient from PIL import Image from sewar.full_ref import uqi @@ -22,22 +25,8 @@ DIFF_OUTPUT_FOLDER_PATH = TESTS_FOLDER_PATH / "diff_output" -def get_image_url(img_name): - """ - Gets an image URL. Uploads it to the S3 testing bucket, - if not there already - - Parameters - ---------- - img_name : string - The name of the image. It is used as key for the upload. - - Returns - ------- - str - The URL of the image in the S3 bucket - """ - +def get_image_url(img_name: str) -> str | None: + """Return a fixture's S3 URL, uploading the fixture if it is not present.""" object_url = get_object_url(img_name, S3_TEST_BUCKET) if object_url: @@ -46,24 +35,12 @@ def get_image_url(img_name): return upload_file(IMAGES_FOLDER_PATH / img_name, S3_TEST_BUCKET, img_name) -def get_markup(client, img_name, request_params): - """ - Vectorizes an image and returns the SVG markup - - Parameters - ---------- - client : Flask - A flask test client - img_name: string - The name of the image - request_params: - Parameters for the vectorize request - - Returns - ------- - str - SVG markup - """ +def get_markup( + client: FlaskClient, + img_name: str, + request_params: dict[str, object], +) -> bytes: + """Vectorize an S3 fixture through the HTTP API and return raw response bytes.""" image_url = get_image_url(img_name) request_params = { @@ -75,26 +52,12 @@ def get_markup(client, img_name, request_params): return client.post("/", json=request_params).data -def compute_img_difference(img, expected_img_path, small_image_test_factor=0.1): - """ - Computes difference between two images using UQI - - Parameters - ---------- - img : PIL.Image - An image - expected_img_path: - A path to a baseline image to compare [img] with - - Both images are assumed to have the same size - - Returns - ------- - float - A number between 0 and 1. - 0 means equal - 1 means completely different - """ +def compute_img_difference( + img: Image.Image, + expected_img_path: Path, + small_image_test_factor: float = 0.1, +) -> float: + """Compare equal-sized RGB images with UQI or a small-image pixel threshold.""" expected_img = try_read_image_from_path(expected_img_path).convert("RGB") expected_img_arr = np.asarray(expected_img).astype(np.uint8) img_arr = np.asarray(img.convert("RGB")).astype(np.uint8) @@ -104,8 +67,7 @@ def compute_img_difference(img, expected_img_path, small_image_test_factor=0.1): if px_count <= MIN_UQI_PIXEL_COUNT: # For very small images, uqi sometimes returns nan - # We then use a simple metric for these cases, we also don't particularly care too - # much about such small images + # Use a pixel-count threshold instead of UQI for these small images. diff = img_arr - expected_img_arr norm = np.linalg.norm(diff, axis=2) @@ -128,26 +90,8 @@ def compute_img_difference(img, expected_img_path, small_image_test_factor=0.1): return 1 - image_quality_index -def convert_to_RGBGray(img_arr): - """ - Converts an image's colors to grayscale, and returns it - in RGB format. This is useful because to write image diffs, - we display the baseline image in gray to be able to have an - always-contrasting highlight color. - - This highlight color shouldn't be a shade of gray though, - so a luminance image is not enough. RGB is needed. - - Parameters - ---------- - img_arr : np.array - An image array - - Returns - ------- - np.array - The converted image - """ +def convert_to_RGBGray(img_arr: np.ndarray) -> np.ndarray: + """Return RGB grayscale pixels, compositing RGBA inputs over white.""" if img_arr.shape[-1] == 4: img_arr = alpha_blend(img_arr) img_arr = cv2.cvtColor(img_arr, cv2.COLOR_RGB2GRAY) @@ -155,19 +99,12 @@ def convert_to_RGBGray(img_arr): return img_arr -def write_img_difference(predicted_img, expected_img_path, output_name): - """ - Writes difference between two images to /diff_output - - Parameters - ---------- - predicted_img : PIL.Image - The predicted image - expected_img_path: string - A path to the expected, or baseline, image - output_name: - The name of the file to be placed in diff_output - """ +def write_img_difference( + predicted_img: Image.Image, + expected_img_path: Path, + output_name: str, +) -> None: + """Write a grayscale baseline with red highlights proportional to pixel error.""" expected_img = try_read_image_from_path(expected_img_path).convert("RGB") predicted_img_arr = np.asarray(predicted_img.convert("RGB")).astype(np.uint8) diff --git a/vectorizing/util/limit_size.py b/vectorizing/util/limit_size.py index 586c6a5..1823e11 100644 --- a/vectorizing/util/limit_size.py +++ b/vectorizing/util/limit_size.py @@ -1,11 +1,16 @@ +"""Bound image area while preserving the source aspect ratio.""" + import math + from PIL import Image # Size restrictions for image MIN_PIXEL_COUNT = 64 -MAX_PIXEL_COUNT = 1024 ** 2 +MAX_PIXEL_COUNT = 1024**2 + -def resize_to_target(img, target_pixel_count): +def resize_to_target(img: Image.Image, target_pixel_count: int) -> Image.Image: + """Resize with Lanczos to rounded dimensions near the requested pixel count.""" width = img.size[0] height = img.size[1] @@ -14,15 +19,17 @@ def resize_to_target(img, target_pixel_count): k = math.sqrt(target_pixel_count / (width * height)) target_width = round(k * width) target_height = round(k * height) - return img.resize((target_width, target_height), resample = Image.Resampling.LANCZOS) + return img.resize((target_width, target_height), resample=Image.Resampling.LANCZOS) -def limit_size(img): + +def limit_size(img: Image.Image) -> Image.Image: + """Resize images outside the area limits; otherwise return the original image.""" pixel_count = img.size[0] * img.size[1] if pixel_count < MIN_PIXEL_COUNT: return resize_to_target(img, MIN_PIXEL_COUNT) - + if pixel_count > MAX_PIXEL_COUNT: return resize_to_target(img, MAX_PIXEL_COUNT) - - return img \ No newline at end of file + + return img diff --git a/vectorizing/util/read.py b/vectorizing/util/read.py index 54edea1..7a84dfa 100644 --- a/vectorizing/util/read.py +++ b/vectorizing/util/read.py @@ -1,52 +1,73 @@ -import math +"""Read supported raster formats and normalize their color modes.""" + +from io import BytesIO +from os import PathLike + import requests -import numpy as np from PIL import Image -from io import BytesIO MIN_PIXEL_COUNT = 64 -MAX_PIXEL_COUNT = 1024 ** 2 -SMALL_PIXEL_COUNT = 512 ** 2 +MAX_PIXEL_COUNT = 1024**2 +SMALL_PIXEL_COUNT = 512**2 + + +class URLReadError(Exception): + """Wrap failures while downloading or opening an image URL.""" + + def __init__(self) -> None: + super().__init__(self, "Failed to read image from provided URL.") + -class URLReadError (Exception): - def __init__ (self): - super().__init__(self, 'Failed to read image from provided URL.') +class PathReadError(Exception): + """Wrap failures while opening a local image.""" -class PathReadError (Exception): - def __init__ (self): - super().__init__(self, 'Failed to read image from provided path.') + def __init__(self) -> None: + super().__init__(self, "Failed to read image from provided path.") -class ImageFormatError (Exception): - def __init__(self): - super().__init__(self, 'Image format not supported.') -def convert_RGB_A(img): - if img.mode == 'RGB' or img.mode == 'RGBA': +class ImageFormatError(Exception): + """Signal an unsupported image color mode.""" + + def __init__(self) -> None: + super().__init__(self, "Image format not supported.") + + +def convert_RGB_A(img: Image.Image) -> Image.Image: + """Keep RGB/RGBA images unchanged and convert supported grayscale/palette modes.""" + if img.mode == "RGB" or img.mode == "RGBA": return img - + map = { - '1': 'RGB', - 'L': 'RGB', - 'P': 'RGBA', - 'PA': 'RGBA', + "1": "RGB", + "L": "RGB", + "P": "RGBA", + "PA": "RGBA", } - if not img.mode in map: + if img.mode not in map: raise ImageFormatError() - + return img.convert(map.get(img.mode)) -def try_read_image_from_path(path): + +def try_read_image_from_path(path: str | PathLike[str]) -> Image.Image: + """Open an image, wrapping all loading exceptions including interrupts.""" try: img = Image.open(path) - except: + except BaseException: # noqa: B036 - All loading failures use the public wrapper. raise PathReadError() return convert_RGB_A(img) -def try_read_image_from_url(url): + +def try_read_image_from_url(url: str) -> Image.Image: + """Download without a time limit, wrapping all loading exceptions.""" try: - resp = requests.get(url, headers={"User-Agent": "KittlVectorizing/1.0.0"}) + # Image downloads wait without a timeout. + resp = requests.get( # noqa: S113 + url, + headers={"User-Agent": "KittlVectorizing/1.0.0"}, + ) img = Image.open(BytesIO(resp.content)) - except: + except BaseException: # noqa: B036 - All loading failures use the public wrapper. raise URLReadError() - return convert_RGB_A(img) \ No newline at end of file + return convert_RGB_A(img)