Apps/ImageApi
1
0 Fork 0
Code Issues 7 Pull Requests Releases Wiki Activity
Files
2d6663c12c9f2114fed2330341da17d032bfd124
ImageApi/dev/e2e/gen_fixtures.py
T
Cameron CordesandClaude Opus 5.5 6135034d09 feat(versions): e2e fixtures, check_versions, docs 
Fixtures gain an XMP-linked version (renamed file), a name-linked version
and an edit with a reference photo; e2e.sh counts with include_versions and
runs check_versions.py. CLAUDE.md documents the feature.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-30 13:53:21 -04:00

606 lines
25 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Generate the synthetic e2e media library.
Every image is rendered with its own relative path, date and scenario
burned into the pixels, so an agent looking at a screenshot can tell
exactly which fixture is on screen. EXIF (dates, GPS, camera) is stamped
in a single exiftool batch. A `manifest.json` next to the media records
what was generated and what the server should derive from it, so tests
can assert against it instead of hardcoding counts.
Runs on native Linux with ffmpeg / exiftool, or in WSL driving the Windows
binaries (ffmpeg.exe / exiftool.exe), in which case every path handed to
them goes through `wslpath -w`.
Usage: gen_fixtures.py <out_dir> [--import DIR]
<out_dir>/media/ the pristine library (never served directly)
<out_dir>/manifest.json what's in it
"""
from __future__ import annotations
import argparse
import datetime as dt
import json
import os
import shutil
import struct
import subprocess
import sys
import zlib
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path
from zoneinfo import ZoneInfo
def _find_ffmpeg() -> str | None:
"""The first ffmpeg that can burn labels in (has `drawtext`, i.e. was
built with freetype). E2E_FFMPEG overrides. Minimal builds earlier on
PATH (e.g. a custom one in ~/bin) often lack it."""
if os.environ.get("E2E_FFMPEG"):
return os.environ["E2E_FFMPEG"]
candidates = [shutil.which("ffmpeg.exe"), shutil.which("ffmpeg"), "/usr/bin/ffmpeg"]
for cand in dict.fromkeys(c for c in candidates if c and os.path.exists(c)):
try:
filters = subprocess.run(
[cand, "-hide_banner", "-filters"], capture_output=True, text=True
).stdout
except OSError:
continue
if " drawtext " in filters:
return cand
return None
FFMPEG = _find_ffmpeg()
EXIFTOOL = shutil.which("exiftool.exe") or shutil.which("exiftool")
def _font() -> str:
"""A monospace font for ffmpeg's drawtext, as a filter-escaped path."""
if FFMPEG and FFMPEG.endswith(".exe"):
# Windows ffmpeg needs an explicit font; the colon after the drive
# letter must be escaped inside a filter argument.
return "C\\:/Windows/Fonts/consola.ttf"
try:
found = subprocess.check_output(
["fc-match", "-f", "%{file}", "monospace"], text=True
).strip()
if found:
return found
except (OSError, subprocess.CalledProcessError):
pass
return "/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf"
FONT = _font()
# The owner's display zone and the server zone the e2e env pins (TZ=...).
# Naive fixture datetimes without GPS are walls in this zone.
HOME_TZ = ZoneInfo("America/New_York")
PLACES = {
"santa_monica": (34.0100, -118.4960),
"paris": (48.8584, 2.2945),
"yosemite": (37.7456, -119.5936),
"tokyo": (35.6595, 139.7005),
}
# IANA zone of each place — what the server's GPS rung of the offset cascade
# resolves, so the expected true instant is the wall read in this zone.
PLACE_ZONES = {
"santa_monica": "America/Los_Angeles",
"paris": "Europe/Paris",
"yosemite": "America/Los_Angeles",
"tokyo": "Asia/Tokyo",
}
PALETTE = [
"0x1f4e79", "0x7a3e9d", "0x2e7d32", "0xb71c1c", "0xe65100",
"0x006064", "0x4e342e", "0x283593", "0x827717", "0xad1457",
]
@dataclass
class Item:
rel: str # path relative to the library root, forward slashes
kind: str # "image" | "video"
size: tuple[int, int] = (1600, 1200)
# Capture wall clock: EXIF DateTimeOriginal as-is; for videos, the wall in
# HOME_TZ, stamped as UTC creation_time the way phones write containers.
taken: dt.datetime | None = None
gps: str | None = None # key into PLACES
camera: tuple[str, str] | None = None # (Make, Model)
mtime: dt.datetime | None = None # pinned file mtime (undated fallback)
duration: int = 0 # video seconds
tags: list[str] = field(default_factory=list)
scenario: str = ""
copy_of: str | None = None # byte-identical duplicate of another item
# Photo versions (ImageApi #191): this PNG is a restoration / upscale / edit
# of another item. `xmp` embeds the producer's XMP link (DerivedFrom = the
# original's blake3, Ingredients = `references`), as the ComfyUI node does;
# without it only the `<name>.<kind>.png` file-name convention links it.
version_of: str | None = None
version_kind: str = "restored"
xmp: bool = False
references: list[str] = field(default_factory=list) # item paths (provenance only)
def label_date(self) -> str:
if self.taken:
return self.taken.strftime("%Y-%m-%d %H:%M")
return "no EXIF date"
def win(p: Path) -> str:
"""`p` as the external tools see it: a Windows path for WSL-driven
.exe tools, unchanged for native ones."""
if not (FFMPEG or "").endswith(".exe"):
return str(p)
return subprocess.check_output(["wslpath", "-w", str(p)], text=True).strip()
def build_items(today: dt.date) -> list[Item]:
items: list[Item] = []
canon = ("Canon", "Canon EOS R5")
iphone = ("Apple", "iPhone 12")
pixel = ("Google", "Pixel 7")
# Dated album with GPS — map pins, date grouping, tag search.
for i in range(6):
items.append(Item(
f"2019/2019-07-04 Beach Day/beach_{i + 1:02d}.jpg", "image",
taken=dt.datetime(2019, 7, 4, 10 + i, 15), gps="santa_monica",
camera=canon, tags=["beach", "summer"], scenario="dated album + GPS",
))
# Multi-day trip; mix of orientations.
for i in range(8):
items.append(Item(
f"2021/Paris Trip/IMG_{1000 + i}.jpg", "image",
size=(1200, 1600) if i % 3 == 0 else (1600, 1200),
taken=dt.datetime(2021, 5, 12 + i // 3, 9 + i, 30), gps="paris",
camera=iphone, tags=["travel", "paris"], scenario="trip, mixed orientation",
))
for i in range(5):
items.append(Item(
f"2022/Hiking/hike_{i + 1:02d}.jpg", "image",
taken=dt.datetime(2022, 8, 20, 7 + i, 0), gps="yosemite",
camera=pixel, tags=["outdoors"], scenario="album with a video",
))
items.append(Item(
"2022/Hiking/summit.mp4", "video", size=(1280, 720), duration=6,
taken=dt.datetime(2022, 8, 20, 12, 30), tags=["outdoors"],
scenario="landscape video (HLS)",
))
# No GPS, tagged people-ish album.
for i in range(6):
items.append(Item(
f"2023/Birthday/party_{i + 1:02d}.jpg", "image",
taken=dt.datetime(2023, 11, 3, 18, 5 * i), camera=iphone,
tags=["family"], scenario="dated, no GPS",
))
# Enough items to force scrolling / pagination in the grid.
start = dt.datetime(2020, 1, 3, 12, 0)
for i in range(60):
items.append(Item(
f"Grid Stress/photo_{i + 1:03d}.jpg", "image", size=(800, 600),
taken=start + dt.timedelta(days=6 * i), camera=pixel,
scenario="bulk grid / scroll",
))
# Dates that only exist in the filename (the memories.rs regexes).
items += [
Item("Screenshots/Screenshot_20220315-101500.png", "image", size=(1080, 2340),
scenario="date from filename (YYYYMMDD-HHMMSS), PNG, no EXIF"),
Item("Screenshots/IMG_20210704_123000.jpg", "image",
scenario="date from filename (YYYYMMDD_HHMMSS), no EXIF"),
Item("Screenshots/2020-02-29_08-00-00.jpg", "image",
scenario="date from filename (leap day), no EXIF"),
]
# No date anywhere: falls back to the pinned mtime.
for i in range(3):
items.append(Item(
f"Undated/scan_{i + 1:02d}.jpg", "image",
mtime=dt.datetime(2015, 3, 1 + i, 12, 0),
scenario="no EXIF, no filename date (mtime fallback)",
))
# "On this day" — same month/day as generation date in earlier years.
for year in (2018, 2020, 2024):
taken = _same_day(today, year, 14)
items.append(Item(
f"On This Day/otd_{year}.jpg", "image", taken=taken, gps="tokyo",
camera=canon, scenario=f"memories: on this day ({year})",
))
# Excluded via EXCLUDED_DIRS=/private — never indexed, must not appear anywhere.
items.append(Item(
"private/otd_hidden.jpg", "image", taken=_same_day(today, 2019, 9),
scenario="EXCLUDED_DIRS: never indexed or shown",
))
# Portrait video, filename-dated.
items.append(Item(
"Videos/VID_20230610_150000.mp4", "video", size=(720, 1280), duration=4,
taken=dt.datetime(2023, 6, 10, 15, 0), scenario="portrait video",
))
# Byte-identical duplicate for duplicate detection.
items.append(Item(
"Duplicates/IMG_1000 copy.jpg", "image", copy_of="2021/Paris Trip/IMG_1000.jpg",
scenario="exact duplicate of 2021/Paris Trip/IMG_1000.jpg",
))
# Photo versions (#191). Originals are dated, GPS-tagged JPEGs so the
# version's copied date / GPS is checkable; every case lives in its own
# folder with the producer's `aiedits/` subfolder.
canon_v = ("Canon", "Canon EOS R5")
for folder, stem, when in (
("Xmp Linked", "scan_0042", dt.datetime(2004, 6, 12, 14, 30)),
("Name Linked", "family_shot", dt.datetime(2008, 12, 24, 9, 15)),
("Reference Photo", "subject", dt.datetime(2010, 3, 7, 16, 45)),
):
items.append(Item(
f"Versions/{folder}/{stem}.jpg", "image", taken=when, gps="paris",
camera=canon_v, tags=["versions"], scenario=f"versions: original ({folder})",
))
# Linked by the XMP chunk alone: the name says nothing about the original.
items.append(Item(
"Versions/Xmp Linked/aiedits/enhanced_final.png", "image", version_of=
"Versions/Xmp Linked/scan_0042.jpg", version_kind="restored", xmp=True,
scenario="version linked by XMP only (renamed file)",
))
# Linked by the file-name convention alone: no XMP in the file.
items.append(Item(
"Versions/Name Linked/aiedits/family_shot.jpg.upscale.png", "image", version_of=
"Versions/Name Linked/family_shot.jpg", version_kind="upscale",
scenario="version linked by file name only (no XMP)",
))
# An edit that also fed a second photo to the model: the other photo is a
# reference (provenance), never hidden or grouped.
items.append(Item(
"Versions/Reference Photo/other_view.jpg", "image",
taken=dt.datetime(2010, 3, 7, 16, 50), camera=canon_v,
scenario="reference photo of a version: stays an ordinary photo",
))
items.append(Item(
"Versions/Reference Photo/aiedits/subject.jpg.edit.png", "image", version_of=
"Versions/Reference Photo/subject.jpg", version_kind="edit", xmp=True,
references=["Versions/Reference Photo/other_view.jpg"],
scenario="version with a reference photo (XMP Ingredients)",
))
return items
def expected_wall_clock(it: Item) -> str:
"""The wall clock the server should report (`wall_clock`): the filename
date for the filename class (it outranks EXIF), else the capture wall,
else the pinned mtime rendered in HOME_TZ."""
import re
name = it.rel.rsplit("/", 1)[-1]
m = (re.search(r"(\d{4})(\d{2})(\d{2})[_-](\d{2})(\d{2})(\d{2})", name)
or re.search(r"(\d{4})-(\d{2})-(\d{2})_(\d{2})-(\d{2})-(\d{2})", name))
if m:
wall = dt.datetime(*map(int, m.groups()))
else:
wall = it.taken or it.mtime or dt.datetime(2024, 1, 1)
return wall.strftime("%Y-%m-%dT%H:%M:%S")
def expected_true_instant(it: Item) -> int:
"""The true instant the server should report: the expected wall read in
the zone the camera was in (GPS place, else HOME_TZ — the server's
assumed zone). Filename walls outrank EXIF, but still take GPS evidence."""
wall = dt.datetime.fromisoformat(expected_wall_clock(it))
zone = ZoneInfo(PLACE_ZONES[it.gps]) if it.gps else HOME_TZ
return int(wall.replace(tzinfo=zone).timestamp())
def _same_day(today: dt.date, year: int, hour: int) -> dt.datetime:
# Feb 29 in a non-leap year would be invalid; fall back to Feb 28.
try:
return dt.datetime(year, today.month, today.day, hour, 0)
except ValueError:
return dt.datetime(year, today.month, 28, hour, 0)
# ── photo versions (#191) ───────────────────────────────────────────────
try:
import blake3 as _blake3
def blake3_hex(data: bytes) -> str:
return _blake3.blake3(data).hexdigest()
except ImportError: # keep the generator dependency-free: reference BLAKE3
_IV = [0x6A09E667, 0xBB67AE85, 0x3C6EF372, 0xA54FF53A,
0x510E527F, 0x9B05688C, 0x1F83D9AB, 0x5BE0CD19]
_PERM = [2, 6, 3, 10, 7, 0, 4, 13, 1, 11, 12, 5, 9, 14, 15, 8]
_CHUNK_START, _CHUNK_END, _PARENT, _ROOT = 1, 2, 4, 8
_M = 0xFFFFFFFF
def _compress(cv, block, counter, block_len, flags):
v = list(cv) + _IV[:4] + [counter & _M, counter >> 32, block_len, flags]
m = list(block)
def g(a, b, c, d, mx, my):
v[a] = (v[a] + v[b] + mx) & _M
v[d] = ((v[d] ^ v[a]) >> 16 | (v[d] ^ v[a]) << 16) & _M
v[c] = (v[c] + v[d]) & _M
v[b] = ((v[b] ^ v[c]) >> 12 | (v[b] ^ v[c]) << 20) & _M
v[a] = (v[a] + v[b] + my) & _M
v[d] = ((v[d] ^ v[a]) >> 8 | (v[d] ^ v[a]) << 24) & _M
v[c] = (v[c] + v[d]) & _M
v[b] = ((v[b] ^ v[c]) >> 7 | (v[b] ^ v[c]) << 25) & _M
for r in range(7):
g(0, 4, 8, 12, m[0], m[1]); g(1, 5, 9, 13, m[2], m[3])
g(2, 6, 10, 14, m[4], m[5]); g(3, 7, 11, 15, m[6], m[7])
g(0, 5, 10, 15, m[8], m[9]); g(1, 6, 11, 12, m[10], m[11])
g(2, 7, 8, 13, m[12], m[13]); g(3, 4, 9, 14, m[14], m[15])
if r < 6:
m = [m[i] for i in _PERM]
return [v[i] ^ v[i + 8] for i in range(8)] + [v[i + 8] ^ cv[i] for i in range(8)]
def _words(block):
return struct.unpack("<16I", block.ljust(64, b"\0"))
def _chunk_output(chunk, counter):
"""(cv_in, block_words, counter, block_len, flags) of the chunk's last block,
plus the chaining value entering it."""
cv = _IV
blocks = [chunk[i:i + 64] for i in range(0, len(chunk), 64)] or [b""]
for n, blk in enumerate(blocks[:-1]):
flags = _CHUNK_START if n == 0 else 0
cv = _compress(cv, _words(blk), counter, 64, flags)[:8]
last = blocks[-1]
flags = _CHUNK_END | (_CHUNK_START if len(blocks) == 1 else 0)
return cv, _words(last), counter, len(last), flags
def blake3_hex(data: bytes) -> str:
"""Plain 32-byte BLAKE3 of `data`, lowercase hex (the reference algorithm)."""
chunks = [data[i:i + 1024] for i in range(0, len(data), 1024)] or [b""]
def subtree(lo, hi, root):
n = hi - lo
if n == 1:
cv, w, ctr, ln, fl = _chunk_output(chunks[lo], lo)
return _compress(cv, w, ctr, ln, fl | (_ROOT if root else 0))[:8]
split = 1 << ((n - 1).bit_length() - 1) # largest power of two below n
left = subtree(lo, lo + split, False)
right = subtree(lo + split, hi, False)
flags = _PARENT | (_ROOT if root else 0)
return _compress(_IV, left + right, 0, 64, flags)[:8]
return struct.pack("<8I", *subtree(0, len(chunks), True)).hex()
XMP_KEY = "XML:com.adobe.xmp"
IMAGEAPI_NS = "https://git.crkc.xyz/Apps/ImageApi/ns/versions/1.0/"
def _resource_ref(file_hash: str, name: str) -> str:
return (f"<stRef:documentID>blake3:{file_hash}</stRef:documentID>"
f"<stRef:filePath>{name}</stRef:filePath>")
def build_xmp(source_hash: str, source_name: str, kind: str,
references: list[tuple[str, str]]) -> str:
"""The packet comfyui-imageapi-versions writes (versioning.build_xmp)."""
ingredients = ""
if references:
lis = "".join(
f'<rdf:li rdf:parseType="Resource">{_resource_ref(h, n)}</rdf:li>'
for h, n in references
)
ingredients = f"<xmpMM:Ingredients><rdf:Bag>{lis}</rdf:Bag></xmpMM:Ingredients>"
return (
'<?xpacket begin="\ufeff" id="W5M0MpCehiHzreSzNTczkc9d"?>'
'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
'<rdf:Description rdf:about=""'
' xmlns:xmpMM="http://ns.adobe.com/xap/1.0/mm/"'
' xmlns:stRef="http://ns.adobe.com/xap/1.0/sType/ResourceRef#"'
f' xmlns:imageapi="{IMAGEAPI_NS}">'
f'<xmpMM:DerivedFrom rdf:parseType="Resource">{_resource_ref(source_hash, source_name)}</xmpMM:DerivedFrom>'
f"{ingredients}<imageapi:versionKind>{kind}</imageapi:versionKind>"
"</rdf:Description></rdf:RDF></x:xmpmeta>"
'<?xpacket end="w"?>'
)
def embed_xmp(png: Path, xmp: str) -> None:
"""Insert the XMP as an uncompressed iTXt chunk right after IHDR."""
data = png.read_bytes()
assert data[:8] == b"\x89PNG\r\n\x1a\n", f"{png} is not a PNG"
body = XMP_KEY.encode() + b"\0\0\0\0\0" + xmp.encode("utf-8")
chunk = struct.pack(">I", len(body)) + b"iTXt" + body
chunk += struct.pack(">I", zlib.crc32(b"iTXt" + body) & 0xFFFFFFFF)
ihdr_end = 8 + 4 + 4 + 13 + 4
png.write_bytes(data[:ihdr_end] + chunk + data[ihdr_end:])
def link_versions(items: list["Item"], media: Path) -> None:
"""Write the XMP link into each `xmp` version. Runs after EXIF stamping:
the link is the original's hash over its final bytes."""
by_path = {it.rel: it for it in items}
for it in items:
if not (it.version_of and it.xmp):
continue
original = media / it.version_of
refs = [(blake3_hex((media / r).read_bytes()), Path(r).name) for r in it.references]
embed_xmp(media / it.rel,
build_xmp(blake3_hex(original.read_bytes()), original.name,
it.version_kind, refs))
assert by_path[it.version_of].kind == "image"
def render(item: Item, idx: int, media: Path) -> None:
out = media / item.rel
out.parent.mkdir(parents=True, exist_ok=True)
w, h = item.size
color = PALETTE[idx % len(PALETTE)]
lines = [item.rel, item.label_date(), item.scenario]
# Consolas glyphs are ~0.55em wide; shrink so the longest line fits.
longest = max(len(line) for line in lines)
fs = max(12, min(min(w, h) // 22, int(w * 0.92 / (longest * 0.55))))
text = ",".join(
f"drawtext=fontfile='{FONT}':text='{_esc(line)}':fontcolor=white:"
f"fontsize={fs}:x=(w-text_w)/2:y=(h/2)+({n - 1}*{int(fs * 1.6)})"
f":box=1:boxcolor=black@0.45:boxborderw=8"
for n, line in enumerate(lines)
)
if item.kind == "image":
cmd = [
FFMPEG, "-hide_banner", "-loglevel", "error", "-y",
"-f", "lavfi", "-i",
f"gradients=s={w}x{h}:c0={color}:c1=0x101010:seed={idx}:duration=1",
"-vf", text, "-frames:v", "1",
]
cmd += ["-q:v", "3"] if out.suffix.lower() == ".jpg" else []
else:
cmd = [
FFMPEG, "-hide_banner", "-loglevel", "error", "-y",
"-f", "lavfi", "-i", f"testsrc2=s={w}x{h}:r=30:d={item.duration}",
"-f", "lavfi", "-i", f"sine=frequency={300 + idx * 20}:d={item.duration}",
"-vf", text, "-c:v", "libx264", "-pix_fmt", "yuv420p",
"-c:a", "aac", "-shortest",
]
if item.taken:
utc = item.taken.replace(tzinfo=HOME_TZ).astimezone(dt.timezone.utc)
cmd += ["-metadata", f"creation_time={utc.strftime('%Y-%m-%dT%H:%M:%S')}Z"]
subprocess.run(cmd + [win(out)], check=True)
def _esc(s: str) -> str:
# drawtext text escaping: backslash, colon, quote, percent, comma.
for ch in ("\\", ":", "'", "%", ","):
s = s.replace(ch, "\\" + ch)
return s
def stamp_exif(items: list[Item], media: Path, work: Path) -> None:
args: list[str] = []
for it in items:
if it.kind != "image" or it.copy_of or not (it.taken or it.gps or it.camera):
continue
if not it.rel.lower().endswith(".jpg"):
continue
if it.taken:
stamp = it.taken.strftime("%Y:%m:%d %H:%M:%S")
args += [f"-DateTimeOriginal={stamp}", f"-CreateDate={stamp}"]
if it.camera:
args += [f"-Make={it.camera[0]}", f"-Model={it.camera[1]}"]
if it.gps:
lat, lon = PLACES[it.gps]
args += [
f"-GPSLatitude={abs(lat)}", f"-GPSLatitudeRef={'N' if lat >= 0 else 'S'}",
f"-GPSLongitude={abs(lon)}", f"-GPSLongitudeRef={'E' if lon >= 0 else 'W'}",
]
args += [win(media / it.rel), "-execute"]
argfile = work / "exif_args.txt"
argfile.write_text("\n".join(args) + "\n", encoding="utf-8")
subprocess.run(
[EXIFTOOL, "-@", win(argfile), "-common_args", "-q", "-overwrite_original",
"-charset", "filename=utf8"],
check=True,
)
argfile.unlink()
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("out_dir", type=Path)
ap.add_argument("--import", dest="import_dir", type=Path,
help="also copy this folder (read-only source) into media/imported/")
a = ap.parse_args()
if not FFMPEG or not EXIFTOOL:
print(
"ffmpeg(.exe) with the drawtext filter (built with freetype) and "
"exiftool(.exe) must be on PATH; set E2E_FFMPEG to pick an ffmpeg",
file=sys.stderr,
)
return 1
out: Path = a.out_dir.resolve()
media = out / "media"
if media.exists():
shutil.rmtree(media)
media.mkdir(parents=True)
(media / "Empty Album").mkdir() # edge case: folder with no media
today = dt.date.today()
items = build_items(today)
originals = [(i, it) for i, it in enumerate(items) if not it.copy_of]
with ThreadPoolExecutor(max_workers=8) as pool:
list(pool.map(lambda p: render(p[1], p[0], media), originals))
stamp_exif(items, media, out)
for it in items:
if it.copy_of:
(media / it.rel).parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(media / it.copy_of, media / it.rel)
link_versions(items, media)
# Pin mtimes last: exiftool rewrites (and so touches) the files.
for it in items:
when = it.mtime or it.taken or dt.datetime(2024, 1, 1)
ts = when.replace(tzinfo=HOME_TZ).timestamp() # not WSL's local zone
os.utime(media / it.rel, (ts, ts))
imported = 0
if a.import_dir:
dest = media / "imported"
shutil.copytree(a.import_dir, dest) # copy: the source is never touched
imported = sum(1 for p in dest.rglob("*") if p.is_file())
by_path = {it.rel: it for it in items}
tags: dict[str, list[str]] = {}
for it in items:
for t in it.tags:
tags.setdefault(t, []).append(it.rel)
manifest = {
"generated_on": today.isoformat(),
"note": "'On This Day' fixtures match generated_on's month/day; regenerate "
"(e2e.sh fixtures --force) on another day to keep memories populated.",
"counts": {
"images": sum(1 for it in items if it.kind == "image"),
"videos": sum(1 for it in items if it.kind == "video"),
"total": len(items),
# What the server should index: EXCLUDED_DIRS is skipped at ingest.
"indexed": sum(1 for it in items if not it.rel.startswith("private/")) + imported,
"imported_files": imported,
},
"excluded_dirs": ["/private"],
"places": {k: {"lat": v[0], "lon": v[1]} for k, v in PLACES.items()},
"tags": tags,
# Photo versions (#191): hidden behind `original` in listings; the
# reference photos stay visible. `source` is how ImageApi links it.
"versions": [
{
"original": it.version_of, "version": it.rel, "kind": it.version_kind,
"source": "xmp" if it.xmp else "name", "references": it.references,
}
for it in items if it.version_of
],
"items": [
{
"path": it.rel, "kind": it.kind, "scenario": it.scenario,
"width": it.size[0], "height": it.size[1],
"taken": it.taken.isoformat() if it.taken else None,
"gps": it.gps, "camera": it.camera[1] if it.camera else None,
"tags": it.tags, "duplicate_of": it.copy_of,
"version_of": it.version_of,
# A duplicate carries its source's bytes, EXIF included; a
# version takes its original's date and GPS (ImageApi copies
# them onto the version row).
"wall_clock": expected_wall_clock(by_path.get(it.copy_of or it.version_of, it)),
"true_instant": expected_true_instant(by_path.get(it.copy_of or it.version_of, it)),
}
for it in items
],
}
(out / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
print(f"generated {len(items)} fixtures ({imported} imported) in {media}")
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in new issue View Git Blame Copy Permalink