hayro-jpeg2000 0.3.0

A memory-safe, pure-Rust JPEG 2000 decoder.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

#!/usr/bin/env python3
import argparse
import json
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

SCRIPT_DIR = Path(__file__).resolve().parent
TEST_INPUTS_DIR = SCRIPT_DIR / "test-inputs"
REMOTE_BASE = "https://hayro-assets.dev/jpeg2000"
MANIFESTS = [
    ("serenity", SCRIPT_DIR / "manifest_serenity.json"),
    ("openjpeg", SCRIPT_DIR / "manifest_openjpeg.json"),
    ("custom", SCRIPT_DIR / "manifest_custom.json"),
]


def load_manifest(path: Path) -> list[dict]:
    if not path.exists():
        return []
    raw_entries = json.loads(path.read_text())
    entries: list[dict] = []
    for item in raw_entries:
        if isinstance(item, str):
            entries.append({"id": item, "path": item, "render": True})
            continue

        entry = dict(item)
        entry.setdefault("render", True)
        entry.setdefault(
            "path", entry.get("path") or entry.get("file") or entry.get("id")
        )
        entries.append(entry)
    return entries


def download_file(namespace: str, asset_path: str, *, force: bool) -> tuple[bool, str]:
    target_dir = TEST_INPUTS_DIR / namespace
    target_dir.mkdir(parents=True, exist_ok=True)
    destination = target_dir / asset_path
    was_cached = destination.exists()

    if was_cached and not force:
        return True, "cached"

    url = f"{REMOTE_BASE}/{namespace}/{asset_path}"
    request = Request(url, headers={"User-Agent": "hayro-jpeg2000-sync/1.0"})
    try:
        with urlopen(request, timeout=60) as response:
            data = response.read()
    except (HTTPError, URLError) as exc:
        return False, str(exc)

    temp_path = destination.with_suffix(destination.suffix + ".tmp")
    temp_path.write_bytes(data)
    temp_path.replace(destination)

    if was_cached:
        return True, "updated"
    return True, "downloaded"


def main() -> None:
    parser = argparse.ArgumentParser(description="Download jpeg2000 test inputs")
    parser.add_argument(
        "--force", action="store_true", help="redownload files even if cached"
    )
    args = parser.parse_args()

    TEST_INPUTS_DIR.mkdir(exist_ok=True)

    failures: list[tuple[str, str]] = []
    total = 0
    for namespace, manifest_path in MANIFESTS:
        entries = load_manifest(manifest_path)
        for entry in entries:
            total += 1
            asset_path = entry["path"]
            label = f"{namespace}/{asset_path}"
            success, status = download_file(namespace, asset_path, force=args.force)
            print(f"[{status}] {label}")
            if not success:
                failures.append((label, status))

    if failures:
        print("\nFailed downloads:")
        for label, message in failures:
            print(f"- {label}: {message}")
    else:
        if total:
            print("\nAll test inputs are ready.")
        else:
            print("No manifest entries were found.")


if __name__ == "__main__":
    main()