refactor(tools): group the sixty scripts by what they talk to
One directory with sixty scripts stopped being navigable, and the grouping is also the sandbox boundary:
Tools/unreal/ commandlets and the two map builders, with ue_common.py and ue_placement.py as the
shared imports (they need the engine)
Tools/blender/ builders, exporters, the FBX preset, the MPFB character pipeline, blender_screenshots
(they need the Blender Flatpak)
Tools/review/ the visual loop: render_screenshots.sh, judge_screenshot.py, hero_compare.py,
build_comparison_md.py, e2e_asset_check.py, hero_iterate.sh, visual_loop.py,
visual_targets.json
Tools/gen/ image and texture generation: the ComfyUI driver, deterministic texture writers,
reference fetching
Tools/input/ OS-level input injection
root _common.py, check-lfs.sh, addon_compat_probe.py, sibling_asset_audit.py, tests/
Every moved script now resolves the repository root as ../.. and puts the tools root plus its siblings on
sys.path, so 'import ue_common' still works from unreal/. The test suite bootstrap does the same for the five
folders. render_screenshots.sh takes the folder as a third argument to run_cmdlet, and the e2e driver,
hero_iterate.sh and the comparison generator were repointed.
51 README rows now link to their new location, the README gains a Layout section naming each folder's
sandbox, and the architecture notes and Blender README follow the moves.
Verified: 108 tests pass, prettier is clean, judge_screenshot.py and hero_compare.py run from their new
homes, and the e2e run was aborted mid-way (it had been rebuilding an empty DemoMap for 15 minutes) and is
being rerun against the new layout.
This commit is contained in:
1 parent
f681d83736
commit
4d27109668
168 files changed
+799
-407
No files matched your search
@@ -0,0 +1,417 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate image assets from the local ComfyUI instance.
|
||||
|
||||
The project needs textures it cannot photograph: skin detail, hair strands, fabric
|
||||
and surface grime. ComfyUI runs locally at `127.0.0.1:8188`, so this is a
|
||||
**human-triggered local tool**, not a network service — nothing in CI calls it, and
|
||||
no credential is involved.
|
||||
|
||||
# what is installed
|
||||
python3 Tools/gen/comfyui_generate.py models
|
||||
|
||||
# one image
|
||||
python3 Tools/gen/comfyui_generate.py txt2img \
|
||||
--prompt "seamless photoreal human skin texture, fine pores, neutral light" \
|
||||
--negative "text, watermark, hair, clothing" \
|
||||
--model juggernautXL_ragnarok.safetensors \
|
||||
--size 1024x1024 --seed 7 \
|
||||
--out Art/textures/generated/T_Character_SkinDetail.png
|
||||
|
||||
Outputs land in the given path and a JSON sidecar records the exact prompt, model,
|
||||
seed and workflow, so an asset can be regenerated or rejected later without
|
||||
guessing how it was made.
|
||||
|
||||
The server is a **Stability Matrix** install and nothing here starts it: if the tool cannot reach
|
||||
`127.0.0.1:8188`, start Stability Matrix, launch ComfyUI, and re-run. The install, its launch arguments and
|
||||
its model library are documented in `/home/zwitschi/StabilityMatrix/Data/README.md`.
|
||||
|
||||
Licence: the output is a generated image from a locally hosted model. `THIRD_PARTY
|
||||
NOTICES.md` must record the model and its licence before the texture ships; the
|
||||
sidecar carries what is needed to do that.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
HOST = os.environ.get("COMFYUI_HOST", "http://127.0.0.1:8188")
|
||||
DEFAULT_MODEL = "juggernautXL_ragnarok.safetensors"
|
||||
# A second checkpoint by default: one model's texture is a lottery ticket, two is a comparison. Both are
|
||||
# generated from the identical prompt and seed, then scored, and only the winner takes the output path.
|
||||
DEFAULT_ALT_MODEL = "epicrealismXL_pureFix.safetensors"
|
||||
|
||||
|
||||
def score_image(path):
|
||||
"""A deterministic, offline 'how much detail did this texture actually get' score.
|
||||
|
||||
Mean absolute Laplacian: high-frequency energy. Generated textures that are blurry or flat score low,
|
||||
which is the failure this comparison exists to catch. Pillow is used only here, so a host without it
|
||||
still generates and simply skips scoring.
|
||||
"""
|
||||
try:
|
||||
from PIL import Image, ImageFilter
|
||||
except ImportError: # pragma: no cover - host without Pillow
|
||||
return None
|
||||
with Image.open(path) as image:
|
||||
grey = image.convert("L").filter(ImageFilter.FIND_EDGES)
|
||||
pixels = list(grey.getdata())
|
||||
return sum(pixels) / max(1, len(pixels))
|
||||
|
||||
|
||||
# The four verified graphs in the Stability Matrix install
|
||||
# (/home/zwitschi/StabilityMatrix/Data/Workflows, documented in Workflows/texture_generation.md).
|
||||
# They are stored in ComfyUI's UI format, which /prompt cannot take, so the equivalent API graphs are
|
||||
# assembled here with exactly the settings those files use.
|
||||
TEXTURE_PROFILES = {
|
||||
"albedobase": dict(
|
||||
label="AlbedoBase XL v3.1 (SDXL) - game albedo, commercial use with credit",
|
||||
checkpoint="albedobaseXL_v31Large.safetensors",
|
||||
steps=30, cfg=6.0, sampler="dpmpp_2m", scheduler="karras",
|
||||
negative="shadows, perspective, vignette, dramatic lighting, blurry, low quality, text, watermark",
|
||||
negative_style="text", clip_type=None, vae=None,
|
||||
),
|
||||
"krea2": dict(
|
||||
label="Krea-2 Turbo NVFP4 - fast stochastic detail (stone, bark, foliage)",
|
||||
unet="krea2_turbo_nvfp4.safetensors",
|
||||
clip="qwen3vl_4b_int8_convrot.safetensors", clip_type="krea2",
|
||||
vae="texture_fix_vae_for_qwen_image_1_and_krea_2_bf16.safetensors",
|
||||
steps=8, cfg=1.0, sampler="euler", scheduler="simple",
|
||||
negative="", negative_style="zero_out",
|
||||
),
|
||||
"qwen21": dict(
|
||||
label="Qwen-Image 2.1 int8 - material sheets with readable labels (research licence)",
|
||||
unet="qwen_image_2.1_int8_convrot.safetensors",
|
||||
clip="qwen3vl_8b_w4a8.safetensors", clip_type="qwen_image",
|
||||
vae="texture_fix_vae_for_qwen_image_2.1_bf16.safetensors",
|
||||
steps=25, cfg=1.0, sampler="euler", scheduler="simple",
|
||||
negative="", negative_style="none",
|
||||
),
|
||||
}
|
||||
DEFAULT_TEXTURE_PROFILES = ("albedobase", "krea2")
|
||||
|
||||
TEXTURE_PROMPT = ("top-down orthographic game texture, flat even lighting, no shadows, no AO, "
|
||||
"no vignette, PBR albedo, material scan, neutral colours")
|
||||
|
||||
|
||||
def texture_workflow(profile: str, prompt: str, width: int, height: int, seed: int) -> dict:
|
||||
"""API graph for one of the documented texture workflows.
|
||||
|
||||
Text-to-image only: the graphs produce a colour/albedo sheet, and derived normal/roughness maps are
|
||||
made downstream. None of them guarantee wrap-around edges - ComfyUI 0.44 core has no seamless sampler -
|
||||
so a tiling check is still required.
|
||||
"""
|
||||
spec = TEXTURE_PROFILES[profile]
|
||||
graph = {}
|
||||
if "checkpoint" in spec:
|
||||
graph["1"] = {"class_type": "CheckpointLoaderSimple",
|
||||
"inputs": {"ckpt_name": spec["checkpoint"]}}
|
||||
model_ref, clip_ref = ["1", 0], ["1", 1]
|
||||
vae_ref = ["1", 2]
|
||||
else:
|
||||
graph["1"] = {"class_type": "UNETLoader",
|
||||
"inputs": {"unet_name": spec["unet"], "weight_dtype": "default"}}
|
||||
graph["2"] = {"class_type": "CLIPLoader",
|
||||
"inputs": {"clip_name": spec["clip"], "type": spec["clip_type"],
|
||||
"device": "default"}}
|
||||
graph["3"] = {"class_type": "VAELoader", "inputs": {"vae_name": spec["vae"]}}
|
||||
model_ref, clip_ref, vae_ref = ["1", 0], ["2", 0], ["3", 0]
|
||||
|
||||
encode = "TextEncodeQwenImage21" if profile == "qwen21" else "CLIPTextEncode"
|
||||
graph["4"] = {"class_type": encode, "inputs": {"text": prompt, "clip": clip_ref}}
|
||||
if spec["negative_style"] == "zero_out":
|
||||
graph["5"] = {"class_type": "ConditioningZeroOut", "inputs": {"conditioning": ["4", 0]}}
|
||||
negative_ref = ["5", 0]
|
||||
elif spec["negative_style"] == "text":
|
||||
graph["5"] = {"class_type": "CLIPTextEncode",
|
||||
"inputs": {"text": spec["negative"], "clip": clip_ref}}
|
||||
negative_ref = ["5", 0]
|
||||
else:
|
||||
negative_ref = ["4", 0]
|
||||
graph["6"] = {"class_type": "EmptyLatentImage",
|
||||
"inputs": {"width": width, "height": height, "batch_size": 1}}
|
||||
graph["7"] = {"class_type": "KSampler",
|
||||
"inputs": {"seed": seed, "steps": spec["steps"], "cfg": spec["cfg"],
|
||||
"sampler_name": spec["sampler"], "scheduler": spec["scheduler"],
|
||||
"denoise": 1.0, "model": model_ref, "positive": ["4", 0],
|
||||
"negative": negative_ref, "latent_image": ["6", 0]}}
|
||||
graph["8"] = {"class_type": "VAEDecode",
|
||||
"inputs": {"samples": ["7", 0], "vae": vae_ref}}
|
||||
graph["9"] = {"class_type": "SaveImage",
|
||||
"inputs": {"images": ["8", 0], "filename_prefix": f"texture_{profile}"}}
|
||||
return graph
|
||||
|
||||
|
||||
def api(path: str) -> dict:
|
||||
with urllib.request.urlopen(f"{HOST}{path}", timeout=30) as response:
|
||||
return json.loads(response.read().decode())
|
||||
|
||||
|
||||
def post(path: str, body: dict) -> dict:
|
||||
request = urllib.request.Request(
|
||||
f"{HOST}{path}", data=json.dumps(body).encode(), method="POST",
|
||||
headers={"Content-Type": "application/json"})
|
||||
with urllib.request.urlopen(request, timeout=60) as response:
|
||||
raw = response.read().decode()
|
||||
return json.loads(raw) if raw.strip() else {}
|
||||
|
||||
|
||||
def checkpoints() -> list[str]:
|
||||
info = api("/object_info/CheckpointLoaderSimple")
|
||||
return info["CheckpointLoaderSimple"]["input"]["required"]["ckpt_name"][0]
|
||||
|
||||
|
||||
def workflow(prompt: str, negative: str, model: str, width: int, height: int,
|
||||
seed: int, steps: int, cfg: float, sampler: str, scheduler: str) -> dict:
|
||||
"""A minimal SDXL/SD1.5 txt2img graph in ComfyUI's API format.
|
||||
|
||||
Node ids are strings and references are `[node_id, output_index]`, which is the
|
||||
format `/prompt` expects — the UI's workflow JSON is a different shape and cannot
|
||||
be posted directly.
|
||||
"""
|
||||
return {
|
||||
"1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": model}},
|
||||
"2": {"class_type": "CLIPTextEncode",
|
||||
"inputs": {"text": prompt, "clip": ["1", 1]}},
|
||||
"3": {"class_type": "CLIPTextEncode",
|
||||
"inputs": {"text": negative, "clip": ["1", 1]}},
|
||||
"4": {"class_type": "EmptyLatentImage",
|
||||
"inputs": {"width": width, "height": height, "batch_size": 1}},
|
||||
"5": {"class_type": "KSampler",
|
||||
"inputs": {"seed": seed, "steps": steps, "cfg": cfg,
|
||||
"sampler_name": sampler, "scheduler": scheduler,
|
||||
"denoise": 1.0, "model": ["1", 0],
|
||||
"positive": ["2", 0], "negative": ["3", 0],
|
||||
"latent_image": ["4", 0]}},
|
||||
"6": {"class_type": "VAEDecode", "inputs": {"samples": ["5", 0], "vae": ["1", 2]}},
|
||||
"7": {"class_type": "SaveImage",
|
||||
"inputs": {"images": ["6", 0], "filename_prefix": "gigline"}},
|
||||
}
|
||||
|
||||
|
||||
def wait_for_image(prompt_id: str, timeout: float = 300.0, poll: float = 1.5):
|
||||
"""Poll /history until the job has an image, then fetch its bytes."""
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
history = api(f"/history/{prompt_id}")
|
||||
entry = history.get(prompt_id)
|
||||
if entry and entry.get("outputs"):
|
||||
images = []
|
||||
for node in entry["outputs"].values():
|
||||
images.extend(node.get("images", []))
|
||||
if not images:
|
||||
return None
|
||||
image = images[0]
|
||||
query = urllib.parse.urlencode({
|
||||
"filename": image["filename"],
|
||||
"subfolder": image.get("subfolder", ""),
|
||||
"type": image.get("type", "output")})
|
||||
with urllib.request.urlopen(f"{HOST}/view?{query}", timeout=60) as response:
|
||||
return response.read()
|
||||
time.sleep(poll)
|
||||
return None
|
||||
|
||||
|
||||
def run_graph(graph, out_path, meta) -> int:
|
||||
"""Queue one API graph, wait for the image, write it plus a provenance sidecar."""
|
||||
queued = post("/prompt", {"prompt": graph})
|
||||
prompt_id = queued.get("prompt_id")
|
||||
if not prompt_id:
|
||||
print(f"error: ComfyUI rejected the prompt: {queued}", file=sys.stderr)
|
||||
return 1
|
||||
data = wait_for_image(prompt_id, timeout=meta["timeout"], poll=meta["poll"])
|
||||
if data is None:
|
||||
print("error: no image came back (timeout or empty job)", file=sys.stderr)
|
||||
return 1
|
||||
os.makedirs(os.path.dirname(os.path.abspath(out_path)) or ".", exist_ok=True)
|
||||
with open(out_path, "wb") as handle:
|
||||
handle.write(data)
|
||||
meta = dict(meta)
|
||||
meta.pop("timeout", None)
|
||||
meta.pop("poll", None)
|
||||
meta.update({"tool": "Tools/gen/comfyui_generate.py",
|
||||
"generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"workflow": graph, "comfyui_prompt_id": prompt_id,
|
||||
"note": "generated locally; record the model licence in "
|
||||
"THIRD_PARTY_NOTICES.md before shipping"})
|
||||
with open(os.path.splitext(out_path)[0] + ".json", "w") as handle:
|
||||
json.dump(meta, handle, indent=2)
|
||||
print(f"wrote {out_path} (+ sidecar)")
|
||||
return 0
|
||||
|
||||
|
||||
def texture(args) -> int:
|
||||
"""Generate a material sheet with the documented texture workflows, two profiles by default."""
|
||||
width, height = (int(v) for v in args.size.lower().split("x"))
|
||||
base_seed = args.seed if args.seed is not None else random.randrange(0, 2**31)
|
||||
prompt = args.prompt or f"{TEXTURE_PROMPT}, {args.material}"
|
||||
profiles = [p for p in (args.profile, args.alt_profile) if p]
|
||||
if not profiles:
|
||||
profiles = list(DEFAULT_TEXTURE_PROFILES)
|
||||
stem, ext = os.path.splitext(args.out)
|
||||
candidates = []
|
||||
for name in profiles:
|
||||
path = f"{stem}__{name}{ext}"
|
||||
graph = texture_workflow(name, prompt, width, height, base_seed)
|
||||
meta = {"prompt": prompt, "profile": name, "size": [width, height], "seed": base_seed,
|
||||
"settings": {k: TEXTURE_PROFILES[name][k] for k in
|
||||
("steps", "cfg", "sampler", "scheduler")},
|
||||
"licence": TEXTURE_PROFILES[name]["label"],
|
||||
"timeout": args.timeout, "poll": args.poll}
|
||||
if run_graph(graph, path, meta) != 0:
|
||||
print(f"profile {name} failed; continuing")
|
||||
continue
|
||||
candidates.append((name, path, score_image(path)))
|
||||
if not candidates:
|
||||
return 1
|
||||
scored = [c for c in candidates if c[2] is not None]
|
||||
winner = max(scored, key=lambda c: c[2]) if scored else candidates[0]
|
||||
for name, path, score in candidates:
|
||||
print(f" {name}: {'n/a' if score is None else round(score, 2)} {path}")
|
||||
os.replace(winner[1], args.out)
|
||||
if os.path.exists(winner[1] + ".json"):
|
||||
os.replace(winner[1] + ".json", args.out + ".json")
|
||||
print(f"chose {winner[0]} -> {args.out}")
|
||||
return 0
|
||||
|
||||
|
||||
def generate(args) -> int:
|
||||
width, height = (int(v) for v in args.size.lower().split("x"))
|
||||
seed = args.seed if args.seed is not None else random.randrange(0, 2**31)
|
||||
graph = workflow(args.prompt, args.negative, args.model, width, height, seed,
|
||||
args.steps, args.cfg, args.sampler, args.scheduler)
|
||||
queued = post("/prompt", {"prompt": graph})
|
||||
prompt_id = queued.get("prompt_id")
|
||||
if not prompt_id:
|
||||
print(f"error: ComfyUI rejected the prompt: {queued}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"queued {prompt_id} ({args.model}, {width}x{height}, seed {seed})")
|
||||
|
||||
data = wait_for_image(prompt_id, timeout=args.timeout, poll=args.poll)
|
||||
if data is None:
|
||||
print("error: no image came back (timeout or empty job)", file=sys.stderr)
|
||||
return 1
|
||||
os.makedirs(os.path.dirname(os.path.abspath(args.out)) or ".", exist_ok=True)
|
||||
with open(args.out, "wb") as handle:
|
||||
handle.write(data)
|
||||
sidecar = os.path.splitext(args.out)[0] + ".json"
|
||||
with open(sidecar, "w") as handle:
|
||||
json.dump({
|
||||
"tool": "Tools/gen/comfyui_generate.py",
|
||||
"generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"prompt": args.prompt,
|
||||
"negative": args.negative,
|
||||
"model": args.model,
|
||||
"size": [width, height],
|
||||
"seed": seed,
|
||||
"steps": args.steps,
|
||||
"cfg": args.cfg,
|
||||
"sampler": args.sampler,
|
||||
"scheduler": args.scheduler,
|
||||
"workflow": graph,
|
||||
"comfyui_prompt_id": prompt_id,
|
||||
"note": "generated locally; record the model licence in "
|
||||
"THIRD_PARTY_NOTICES.md before shipping",
|
||||
}, handle, indent=2)
|
||||
print(f"wrote {args.out} (+ sidecar)")
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
|
||||
def generate_pair(args) -> int:
|
||||
"""Generate the same prompt with two checkpoints and keep the better one at `--out`.
|
||||
|
||||
Both are written beside the output as `<out stem>__<model>.png` with their own sidecars, so the losing
|
||||
candidate is kept for inspection instead of thrown away. The winner is chosen by `score_image()`; when
|
||||
Pillow is missing or the scores tie, the primary model keeps the output and the run says so.
|
||||
"""
|
||||
import copy
|
||||
|
||||
stem, ext = os.path.splitext(args.out)
|
||||
candidates = []
|
||||
for label, model in (("primary", args.model), ("alternate", args.alt_model)):
|
||||
out = f"{stem}__{model.replace('/', '_')[:40]}{ext}"
|
||||
sub_args = copy.copy(args)
|
||||
sub_args.model = model
|
||||
sub_args.out = out
|
||||
if generate(sub_args) != 0:
|
||||
print(f"{label} model {model} failed; continuing with what worked")
|
||||
continue
|
||||
candidates.append((model, out, score_image(out)))
|
||||
if not candidates:
|
||||
return 1
|
||||
scored = [c for c in candidates if c[2] is not None]
|
||||
if not scored:
|
||||
print(f"no scoring available; keeping {candidates[0][1]} as the output")
|
||||
winner = candidates[0]
|
||||
else:
|
||||
winner = max(scored, key=lambda c: c[2])
|
||||
for model, path, score in candidates:
|
||||
print(f" {model}: {score if score is None else round(score, 2)} {path}")
|
||||
os.replace(winner[1], args.out)
|
||||
sidecar = winner[1] + ".json"
|
||||
if os.path.exists(sidecar):
|
||||
os.replace(sidecar, args.out + ".json")
|
||||
print(f"chose {winner[0]} -> {args.out}")
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
argv = sys.argv[1:] if argv is None else argv
|
||||
ap = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
sub.add_parser("models", help="list installed checkpoints")
|
||||
t = sub.add_parser("txt2img", help="generate one image")
|
||||
t.add_argument("--prompt", required=True)
|
||||
t.add_argument("--negative", default="text, watermark, signature, blurry, low quality")
|
||||
t.add_argument("--model", default=DEFAULT_MODEL)
|
||||
t.add_argument("--alt-model", default=DEFAULT_ALT_MODEL,
|
||||
help=f"second checkpoint generated from the same prompt and seed, then compared "
|
||||
f"(default {DEFAULT_ALT_MODEL}); pass an empty string to generate one image only")
|
||||
t.add_argument("--size", default="1024x1024")
|
||||
t.add_argument("--seed", type=int, default=None)
|
||||
t.add_argument("--steps", type=int, default=28)
|
||||
t.add_argument("--cfg", type=float, default=5.5)
|
||||
t.add_argument("--sampler", default="dpmpp_2m")
|
||||
t.add_argument("--scheduler", default="karras")
|
||||
t.add_argument("--out", required=True)
|
||||
t.add_argument("--timeout", type=float, default=300.0)
|
||||
t.add_argument("--poll", type=float, default=1.5)
|
||||
x = sub.add_parser("texture", help="material sheet via the documented texture workflows")
|
||||
x.add_argument("--material", required=True,
|
||||
help="what to draw, e.g. 'weathered grey limestone ashlar'")
|
||||
x.add_argument("--prompt", default="", help="full prompt, overriding the material shorthand")
|
||||
x.add_argument("--profile", default=DEFAULT_TEXTURE_PROFILES[0],
|
||||
choices=sorted(TEXTURE_PROFILES))
|
||||
x.add_argument("--alt-profile", default=DEFAULT_TEXTURE_PROFILES[1],
|
||||
choices=sorted(TEXTURE_PROFILES) + [""],
|
||||
help="second workflow, generated from the same prompt and seed, then compared")
|
||||
x.add_argument("--size", default="1024x1024")
|
||||
x.add_argument("--seed", type=int, default=None)
|
||||
x.add_argument("--out", required=True)
|
||||
x.add_argument("--timeout", type=float, default=300.0)
|
||||
x.add_argument("--poll", type=float, default=1.5)
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.cmd == "texture":
|
||||
return texture(args)
|
||||
if args.cmd == "models":
|
||||
for name in checkpoints():
|
||||
print(name)
|
||||
return 0
|
||||
if args.alt_model and args.alt_model != args.model:
|
||||
return generate_pair(args)
|
||||
return generate(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Executable
+263
@@ -0,0 +1,263 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fetch reference photographs from Wikimedia Commons.
|
||||
|
||||
Local, private development reference only. Images fetched here are **not**
|
||||
committed: the output directory is gitignored. The tool records provenance
|
||||
(author, licence, source URL) in a JSON sidecar next to the images so a local
|
||||
working copy stays traceable.
|
||||
|
||||
# dry run: list what the search returns, download nothing
|
||||
python3 Tools/gen/commons_fetch.py --query "Grauman's Chinese Theatre" --slug north-chinese --dry-run
|
||||
|
||||
# download up to 6 images at 1600 px wide
|
||||
python3 Tools/gen/commons_fetch.py --query "Grauman's Chinese Theatre" --slug north-chinese --limit 6
|
||||
|
||||
# fetch specific files by title
|
||||
python3 Tools/gen/commons_fetch.py --file "File:Grauman's Chinese Theatre, Hollywood.JPG" --slug north-chinese
|
||||
|
||||
Defaults: out dir `Art/reference/<slug>/`, 1600 px wide thumbnails, one request
|
||||
per second, JPEG/PNG only, at least 800 px wide on the original.
|
||||
|
||||
Requires only the standard library.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
API = "https://commons.wikimedia.org/w/api.php"
|
||||
UA = "gig-line-references/0.1 (local development reference fetch)"
|
||||
ALLOWED_MIME = {"image/jpeg", "image/png"}
|
||||
MIN_WIDTH = 800
|
||||
|
||||
|
||||
def api_get(params: dict, retries: int = 3) -> dict:
|
||||
query = urllib.parse.urlencode({**params, "format": "json", "formatversion": "2"})
|
||||
url = f"{API}?{query}"
|
||||
for attempt in range(retries):
|
||||
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=45) as resp:
|
||||
return json.loads(resp.read().decode("utf-8"))
|
||||
except Exception as exc: # noqa: BLE001 - report and retry, then fail
|
||||
if attempt == retries - 1:
|
||||
raise SystemExit(f"Commons API failed: {exc}")
|
||||
time.sleep(2 * (attempt + 1))
|
||||
return {}
|
||||
|
||||
|
||||
def strip_html(value: str) -> str:
|
||||
text = re.sub(r"<[^>]+>", " ", value or "")
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
|
||||
|
||||
def search_titles(query: str, limit: int) -> list[str]:
|
||||
data = api_get(
|
||||
{
|
||||
"action": "query",
|
||||
"generator": "search",
|
||||
"gsrsearch": query,
|
||||
"gsrnamespace": "6",
|
||||
"gsrlimit": str(limit),
|
||||
"prop": "imageinfo",
|
||||
"iiprop": "url|size|mime|extmetadata",
|
||||
}
|
||||
)
|
||||
pages = data.get("query", {}).get("pages", [])
|
||||
# generator search returns unordered; keep relevance order from the index list
|
||||
order = [p.get("pageid") for p in pages]
|
||||
titles = []
|
||||
for page in pages:
|
||||
info = (page.get("imageinfo") or [{}])[0]
|
||||
if info.get("mime") not in ALLOWED_MIME:
|
||||
continue
|
||||
if info.get("width", 0) < MIN_WIDTH:
|
||||
continue
|
||||
titles.append(page["title"])
|
||||
return titles, order
|
||||
|
||||
|
||||
def category_titles(category: str, limit: int) -> list[str]:
|
||||
data = api_get(
|
||||
{
|
||||
"action": "query",
|
||||
"list": "categorymembers",
|
||||
"cmtitle": category,
|
||||
"cmtype": "file",
|
||||
"cmlimit": str(max(limit, 50)),
|
||||
}
|
||||
)
|
||||
members = data.get("query", {}).get("categorymembers", [])
|
||||
return [m["title"] for m in members][:limit]
|
||||
|
||||
|
||||
def image_info(titles: list[str], thumb_width: int) -> list[dict]:
|
||||
data = api_get(
|
||||
{
|
||||
"action": "query",
|
||||
"titles": "|".join(titles),
|
||||
"prop": "imageinfo",
|
||||
"iiprop": "url|size|mime|extmetadata|sha1|user",
|
||||
"iiurlwidth": str(thumb_width),
|
||||
}
|
||||
)
|
||||
out = []
|
||||
for page in data.get("query", {}).get("pages", []):
|
||||
info = (page.get("imageinfo") or [{}])[0]
|
||||
if not info or info.get("mime") not in ALLOWED_MIME:
|
||||
continue
|
||||
meta = info.get("extmetadata", {})
|
||||
out.append(
|
||||
{
|
||||
"title": page["title"],
|
||||
"page_url": info.get("descriptionurl"),
|
||||
"file_url": info.get("url"),
|
||||
"download_url": info.get("thumburl") or info.get("url"),
|
||||
"width": info.get("width"),
|
||||
"height": info.get("height"),
|
||||
"mime": info.get("mime"),
|
||||
"uploader": info.get("user"),
|
||||
"sha1": info.get("sha1"),
|
||||
"artist": strip_html(meta.get("Artist", {}).get("value", "")),
|
||||
"credit": strip_html(meta.get("Credit", {}).get("value", "")),
|
||||
"licence": strip_html(
|
||||
meta.get("LicenseShortName", {}).get("value", "")
|
||||
or meta.get("UsageTerms", {}).get("value", "")
|
||||
),
|
||||
"licence_url": strip_html(meta.get("LicenseUrl", {}).get("value", "")),
|
||||
"attribution_required": strip_html(
|
||||
meta.get("AttributionRequired", {}).get("value", "")
|
||||
),
|
||||
"date": strip_html(
|
||||
meta.get("DateTimeOriginal", {}).get("value", "")
|
||||
or meta.get("DateTime", {}).get("value", "")
|
||||
),
|
||||
"description": strip_html(meta.get("ImageDescription", {}).get("value", ""))[:400],
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def safe_name(title: str, index: int) -> str:
|
||||
base = title.split(":", 1)[-1]
|
||||
base = re.sub(r"[^A-Za-z0-9._-]+", "-", base).strip("-").lower()
|
||||
ext = os.path.splitext(base)[1] or ".jpg"
|
||||
stem = os.path.splitext(base)[0][:60]
|
||||
return f"{index:02d}-{stem}{ext}"
|
||||
|
||||
|
||||
def download(url: str, path: str, retries: int = 3) -> str:
|
||||
for attempt in range(retries):
|
||||
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=90) as resp:
|
||||
payload = resp.read()
|
||||
with open(path, "wb") as handle:
|
||||
handle.write(payload)
|
||||
return hashlib.sha256(payload).hexdigest()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
if attempt == retries - 1:
|
||||
raise SystemExit(f"download failed for {url}: {exc}")
|
||||
time.sleep(2 * (attempt + 1))
|
||||
return ""
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--query", help="Commons search string")
|
||||
ap.add_argument("--category", help='Commons category, e.g. "Category:Hollywood Masonic Temple"')
|
||||
ap.add_argument("--file", action="append", default=[], help="exact File: title, repeatable")
|
||||
ap.add_argument("--slug", required=True, help="output folder name under the out dir")
|
||||
ap.add_argument("--out", default="Art/reference", help="base output directory")
|
||||
ap.add_argument("--limit", type=int, default=6)
|
||||
ap.add_argument("--width", type=int, default=1600, help="thumbnail width in px")
|
||||
ap.add_argument("--dry-run", action="store_true")
|
||||
ap.add_argument("--delay", type=float, default=2.0, help="seconds between downloads")
|
||||
args = ap.parse_args()
|
||||
|
||||
if not args.query and not args.file and not args.category:
|
||||
ap.error("need --query, --category, or --file")
|
||||
|
||||
titles = list(args.file)
|
||||
if args.category:
|
||||
found = category_titles(args.category, args.limit * 4)
|
||||
print(f"{len(found)} file(s) in {args.category}")
|
||||
titles += [t for t in found if t not in titles]
|
||||
if args.query:
|
||||
found, _ = search_titles(args.query, args.limit * 3)
|
||||
titles += [t for t in found if t not in titles]
|
||||
|
||||
if not titles:
|
||||
print("no candidate files matched")
|
||||
return 1
|
||||
|
||||
print(f"{len(titles)} candidate file(s)")
|
||||
if args.dry_run:
|
||||
for title in titles[: args.limit]:
|
||||
print(f" {title}")
|
||||
return 0
|
||||
|
||||
out_dir = os.path.join(args.out, args.slug)
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
|
||||
selected = image_info(titles[: args.limit], args.width)
|
||||
if not selected:
|
||||
print("no downloadable candidates")
|
||||
return 1
|
||||
|
||||
records = []
|
||||
for index, item in enumerate(selected, start=1):
|
||||
name = safe_name(item["title"], index)
|
||||
path = os.path.join(out_dir, name)
|
||||
print(f" [{index}/{len(selected)}] {item['title']}")
|
||||
try:
|
||||
digest = download(item["download_url"], path)
|
||||
except SystemExit as exc:
|
||||
# one throttled file must not discard the ones already fetched
|
||||
print(f" skipped: {exc}", flush=True)
|
||||
time.sleep(args.delay * 5)
|
||||
continue
|
||||
item.update(
|
||||
{
|
||||
"local_file": name,
|
||||
"sha256": digest,
|
||||
"bytes": os.path.getsize(path),
|
||||
}
|
||||
)
|
||||
records.append(item)
|
||||
time.sleep(args.delay)
|
||||
|
||||
sidecar = {
|
||||
"fetched_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
|
||||
"source": "Wikimedia Commons API",
|
||||
"api": API,
|
||||
"query": args.query,
|
||||
"category": args.category,
|
||||
"requested_files": args.file,
|
||||
"slug": args.slug,
|
||||
"thumb_width": args.width,
|
||||
"tool": "Tools/gen/commons_fetch.py",
|
||||
"note": "local development reference only; not committed to the repository",
|
||||
"images": records,
|
||||
}
|
||||
sidecar_path = os.path.join(out_dir, "sidecar.json")
|
||||
with open(sidecar_path, "w", encoding="utf-8") as handle:
|
||||
json.dump(sidecar, handle, indent=2, ensure_ascii=False)
|
||||
|
||||
print(f"\nwrote {len(records)} image(s) + sidecar.json to {out_dir}")
|
||||
for item in records:
|
||||
print(f" {item['local_file']} {item['licence'] or 'licence unknown'} {item['artist'][:40]}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,328 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate frontal building plates from the local reference photography.
|
||||
|
||||
Each study takes one photograph we already hold (`Art/reference/`, gitignored) as
|
||||
the geometric anchor and asks an image model to re-stage it as a clean **frontal
|
||||
view of the building** — the whole elevation, the entrance in its facade, or the
|
||||
building with its neighbours — evenly lit, no crowd, no staging, so the modelling
|
||||
pass has something measurable instead of a tourist photo.
|
||||
|
||||
Deliberately **not** close-up detail crops: a doorway or cornice crop cannot be
|
||||
measured for proportion, which is what the modelling pass actually needs.
|
||||
|
||||
python3 Tools/gen/facade_detail_concepts.py --list
|
||||
python3 Tools/gen/facade_detail_concepts.py --only door-recess,parapet-motto
|
||||
python3 Tools/gen/facade_detail_concepts.py --all
|
||||
python3 Tools/gen/facade_detail_concepts.py --all --model google/gemini-2.5-flash-image
|
||||
|
||||
Output: `Art/reference/facade-details/<study>.png` plus a `<study>.refgen.json`
|
||||
sidecar recording the model, the source photograph, the prompt and the SHA-256 of
|
||||
the result. `Art/reference/` is gitignored, so **neither the images nor the
|
||||
sidecars are committed** — they are local working material. The study list and
|
||||
the tool are the committed part.
|
||||
|
||||
Design notes
|
||||
------------
|
||||
* Standard library only: the OpenRouter chat-completions call is `urllib`, and
|
||||
the returned `data:` URL is decoded with `base64` and written straight to disk.
|
||||
* An existing output is never overwritten; `-v2`, `-v3` … are used instead, so
|
||||
re-running after a prompt edit keeps both studies comparable.
|
||||
* Generation costs money, so it is never automatic: `--list` is free, and every
|
||||
generating run prints the model and the study count first.
|
||||
* The token is read from `OPENROUTER_API_KEY` or the ignored `.env` and is never
|
||||
written to a sidecar or a log line.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
||||
REFERENCE_DIR = os.path.join(ROOT, "Art", "reference")
|
||||
OUT_DIR = os.path.join(REFERENCE_DIR, "facade-details")
|
||||
API_URL = "https://openrouter.ai/api/v1/chat/completions"
|
||||
DEFAULT_MODEL = "google/gemini-2.5-flash-image"
|
||||
|
||||
# The instruction appended to every study. Kept identical to the sibling
|
||||
# project's convention so the two sets stay comparable: the photograph fixes the
|
||||
# geometry, the prompt only restages light, weather, people and surface.
|
||||
ANCHOR_SUFFIX = (
|
||||
"Use the attached reference photograph as the source of the architecture: keep "
|
||||
"the building's proportions, storey count, column rhythm, band heights and "
|
||||
"signage placement exactly as photographed. You may REFRAME — step back across "
|
||||
"the street, zoom out or widen the field of view — so that the whole building "
|
||||
"fits in the frame from the pavement to the parapet; the reference is often "
|
||||
"cropped and the plate must not be. Keep the view exterior and frontal (or "
|
||||
"three-quarter where asked). Restage only lighting, weather, people and surface "
|
||||
"condition. Do not invent extra storeys, wings or ornaments, and do not change "
|
||||
"the architecture to fit the frame."
|
||||
)
|
||||
|
||||
# name -> (reference photo relative to Art/reference/, what the plate must show)
|
||||
#
|
||||
# Verified framing (vision pre-check, 2026-11-09): the model follows the reference
|
||||
# crop closely, so a plate is only as wide as its source photo. `building-entrance`
|
||||
# and `building-street-front` come back as full-elevation frontal views; the studies
|
||||
# anchored on the tightly cropped 2009 facade photo stay cropped, because the model
|
||||
# will not invent the parapet and sky that the photograph does not contain. A full
|
||||
# building plate for 6840 needs a full-elevation photograph, which the local set
|
||||
# does not have (see Docs/reference-notes-hollywood.md).
|
||||
#
|
||||
# Every study asks for a *frontal view of a building*, not a crop of one detail:
|
||||
# the whole elevation, the entrance in its facade, or the building together with
|
||||
# its neighbours. Close-up framing is deliberately absent — a detail crop cannot
|
||||
# be measured for proportion, which is what the modelling pass needs.
|
||||
STUDIES: dict[str, tuple[str, str]] = {
|
||||
"building-front": (
|
||||
"hollywood/el-capitan-facade-day-wikimedia-2009-10-27.jpg",
|
||||
"A frontal view of the complete building shown in the reference "
|
||||
"photograph: the whole elevation from the pavement to the parapet fits in "
|
||||
"the frame with clear sky above the roofline and pavement below the ground "
|
||||
"floor, shot straight on and centred. Nothing may be cut off at the top or "
|
||||
"bottom. Even daylight, no crowd, no vehicles.",
|
||||
),
|
||||
"building-entrance": (
|
||||
"el-capitan-entrance/01-hollywood-el-capitan-entrance.jpg",
|
||||
"A frontal view of the building in the reference photograph, framed so its "
|
||||
"street entrance is clearly visible as part of the facade — the doors and "
|
||||
"the wall around them in proportion, not a close-up of the doorway. "
|
||||
"Include the pavement in front. Even daylight, no crowd.",
|
||||
),
|
||||
"building-with-neighbours": (
|
||||
"hollywood/el-capitan-theatre-blade-sign-day-wikimedia-2008-10.jpg",
|
||||
"A straight-on view of the street wall in the reference photograph: the "
|
||||
"building and its neighbouring buildings either side, each at full height "
|
||||
"from pavement to parapet with sky above, the joins between them visible, "
|
||||
"nothing cut off. Shot from across the road, parallel to the frontage. "
|
||||
"Even daylight, no crowd.",
|
||||
),
|
||||
"building-street-front": (
|
||||
"hollywood/walk-of-fame-terrazzo-ovation-day-cc0-2019-06-20.jpg",
|
||||
"A frontal view along the pavement of the building in the reference "
|
||||
"photograph: the building's frontage seen from street level with its "
|
||||
"neighbours continuing either side and the pavement in the foreground. "
|
||||
"Straight on, even daylight, no crowd.",
|
||||
),
|
||||
"building-front-oblique": (
|
||||
"hollywood/el-capitan-facade-day-wikimedia-2009-10-27.jpg",
|
||||
"A three-quarter view of the complete building in the reference "
|
||||
"photograph: the frontal elevation and one side visible together so the "
|
||||
"building's depth and mass read, with its neighbouring building at one "
|
||||
"edge. The whole building stays in frame. Even daylight, no crowd.",
|
||||
),
|
||||
"theatre-front-with-neighbours": (
|
||||
"el-capitan-theater-detail/01-el-capitan-marquee.jpg",
|
||||
"A frontal view of the theatre in the reference photograph together with "
|
||||
"the buildings either side of it, the whole frontage and its full height "
|
||||
"in frame, shot from across the road. Even daylight, no crowd, marquee "
|
||||
"lettering blocked out as unreadable texture.",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def load_token() -> str:
|
||||
token = os.environ.get("OPENROUTER_API_KEY")
|
||||
if token:
|
||||
return token.strip()
|
||||
env_path = os.path.join(ROOT, ".env")
|
||||
if os.path.exists(env_path):
|
||||
with open(env_path, encoding="utf-8") as handle:
|
||||
for line in handle:
|
||||
line = line.strip()
|
||||
if line.startswith("OPENROUTER_API_KEY="):
|
||||
return line.split("=", 1)[1].strip().strip('"').strip("'")
|
||||
raise SystemExit("no OpenRouter key: set OPENROUTER_API_KEY or add it to .env")
|
||||
|
||||
|
||||
def build_prompt(study: str) -> str:
|
||||
"""The full prompt for a study: its request plus the fixed anchor suffix."""
|
||||
if study not in STUDIES:
|
||||
raise KeyError(f"unknown study {study!r}; known: {sorted(STUDIES)}")
|
||||
return f"{STUDIES[study][1]}\n\n{ANCHOR_SUFFIX}"
|
||||
|
||||
|
||||
def reference_path(study: str) -> str:
|
||||
return os.path.join(REFERENCE_DIR, STUDIES[study][0])
|
||||
|
||||
|
||||
def missing_references() -> list[str]:
|
||||
"""Studies whose source photograph is not present locally."""
|
||||
return [name for name in STUDIES if not os.path.isfile(reference_path(name))]
|
||||
|
||||
|
||||
def next_free_path(study: str, directory: str = OUT_DIR) -> str:
|
||||
"""`<study>.png`, or `<study>-v2.png`, `-v3` … if the name is taken."""
|
||||
base = os.path.join(directory, f"{study}.png")
|
||||
if not os.path.exists(base):
|
||||
return base
|
||||
index = 2
|
||||
while True:
|
||||
candidate = os.path.join(directory, f"{study}-v{index}.png")
|
||||
if not os.path.exists(candidate):
|
||||
return candidate
|
||||
index += 1
|
||||
|
||||
|
||||
def data_url(path: str) -> str:
|
||||
suffix = os.path.splitext(path)[1].lower()
|
||||
mime = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png"}.get(suffix)
|
||||
if not mime:
|
||||
raise SystemExit(f"unsupported reference type: {path}")
|
||||
with open(path, "rb") as handle:
|
||||
return f"data:{mime};base64,{base64.b64encode(handle.read()).decode('ascii')}"
|
||||
|
||||
|
||||
def decode_image(payload: dict) -> bytes:
|
||||
"""Extract the returned image bytes from a chat-completions payload."""
|
||||
message = (payload.get("choices") or [{}])[0].get("message", {})
|
||||
urls = [item.get("image_url", {}).get("url", "")
|
||||
for item in (message.get("images") or []) if isinstance(item, dict)]
|
||||
if not urls:
|
||||
match = re.search(r"data:image/[a-z]+;base64,[A-Za-z0-9+/=]+",
|
||||
str(message.get("content") or ""))
|
||||
if match:
|
||||
urls.append(match.group(0))
|
||||
if not urls or not urls[0]:
|
||||
raise SystemExit(f"model returned no image data: {json.dumps(payload)[:400]}")
|
||||
url = urls[0]
|
||||
if url.startswith("data:"):
|
||||
return base64.b64decode(url.split(",", 1)[1])
|
||||
with urllib.request.urlopen(urllib.request.Request(url), timeout=120) as response:
|
||||
return response.read()
|
||||
|
||||
|
||||
def generate(study: str, token: str, model: str, force: bool = False) -> str:
|
||||
"""Generate one study; returns the written path. Never overwrites by default."""
|
||||
path = next_free_path(study)
|
||||
if force and os.path.exists(os.path.join(OUT_DIR, f"{study}.png")):
|
||||
path = os.path.join(OUT_DIR, f"{study}.png")
|
||||
# Write to a temp name then replace, so a failed call cannot destroy the
|
||||
# previous result.
|
||||
target = path + ".new"
|
||||
else:
|
||||
target = path
|
||||
|
||||
body = {
|
||||
"model": model,
|
||||
"messages": [{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": build_prompt(study)},
|
||||
{"type": "image_url", "image_url": {"url": data_url(reference_path(study))}},
|
||||
],
|
||||
}],
|
||||
}
|
||||
request = urllib.request.Request(
|
||||
API_URL,
|
||||
data=json.dumps(body).encode("utf-8"),
|
||||
headers={
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
"HTTP-Referer": "https://git.allucanget.biz/zwitschi/gig-line",
|
||||
"X-Title": "gig-line facade details",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
|
||||
payload = None
|
||||
for attempt in range(3):
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=180) as response:
|
||||
payload = json.load(response)
|
||||
break
|
||||
except urllib.error.HTTPError as exc:
|
||||
if exc.code == 429 or exc.code >= 500:
|
||||
wait = 5 * (attempt + 1)
|
||||
sys.stderr.write(f" HTTP {exc.code}; retrying in {wait}s\n")
|
||||
time.sleep(wait)
|
||||
continue
|
||||
detail = exc.read().decode("utf-8", "replace")[:300]
|
||||
raise SystemExit(f"{study}: HTTP {exc.code} — {detail}")
|
||||
except urllib.error.URLError as exc:
|
||||
sys.stderr.write(f" network error ({exc.reason}); retrying\n")
|
||||
time.sleep(5 * (attempt + 1))
|
||||
if payload is None:
|
||||
raise SystemExit(f"{study}: no response after retries")
|
||||
|
||||
image = decode_image(payload)
|
||||
os.makedirs(OUT_DIR, exist_ok=True)
|
||||
with open(target, "wb") as handle:
|
||||
handle.write(image)
|
||||
if target != path:
|
||||
os.replace(target, path)
|
||||
|
||||
import hashlib
|
||||
sidecar = {
|
||||
"study": study,
|
||||
"generated": os.path.basename(path),
|
||||
"generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"model": model,
|
||||
"prompt": build_prompt(study),
|
||||
"reference": {
|
||||
"file": STUDIES[study][0],
|
||||
"note": "local reference photography; Art/reference is gitignored and never shipped",
|
||||
},
|
||||
"sha256": hashlib.sha256(image).hexdigest(),
|
||||
"bytes": len(image),
|
||||
"redistribution": "local development reference only; not committed, not shipped",
|
||||
}
|
||||
sidecar_path = os.path.splitext(path)[0] + ".refgen.json"
|
||||
with open(sidecar_path, "w", encoding="utf-8") as handle:
|
||||
json.dump(sidecar, handle, indent=2)
|
||||
if token in json.dumps(sidecar):
|
||||
raise SystemExit("refusing to write a sidecar containing the token")
|
||||
return path
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("--list", action="store_true", help="list studies and exit (free)")
|
||||
ap.add_argument("--all", action="store_true", help="generate every study")
|
||||
ap.add_argument("--only", help="comma-separated study names")
|
||||
ap.add_argument("--model", default=DEFAULT_MODEL)
|
||||
ap.add_argument("--force", action="store_true",
|
||||
help="replace the existing <study>.png instead of writing a -vN")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.list or not (args.all or args.only):
|
||||
missing = set(missing_references())
|
||||
print(f"{len(STUDIES)} studies; model {args.model}")
|
||||
for name, (ref, _prompt) in STUDIES.items():
|
||||
state = "MISSING reference" if name in missing else "ready"
|
||||
print(f" {name:22} {state:18} <- {ref}")
|
||||
if missing:
|
||||
print(f"\n{len(missing)} study(ies) cannot run without their reference photograph.")
|
||||
return 0
|
||||
|
||||
names = list(STUDIES) if args.all else [s.strip() for s in args.only.split(",") if s.strip()]
|
||||
unknown = [n for n in names if n not in STUDIES]
|
||||
if unknown:
|
||||
raise SystemExit(f"unknown study(ies): {unknown}; known: {sorted(STUDIES)}")
|
||||
blocked = [n for n in names if n in set(missing_references())]
|
||||
if blocked:
|
||||
raise SystemExit(f"missing reference photograph for: {blocked}")
|
||||
token = load_token()
|
||||
|
||||
print(f"generating {len(names)} study(ies) with {args.model} -> {OUT_DIR}")
|
||||
for index, name in enumerate(names, 1):
|
||||
print(f"[{index}/{len(names)}] {name}")
|
||||
try:
|
||||
path = generate(name, token, args.model, force=args.force)
|
||||
except SystemExit as exc:
|
||||
sys.stderr.write(f" FAILED: {exc}\n")
|
||||
continue
|
||||
print(f" wrote {os.path.relpath(path, ROOT)}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,168 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Procedural character textures: skin detail and hair strands.
|
||||
|
||||
The local ComfyUI instance was tried first for skin, and failed in a way worth
|
||||
recording: every "flat skin swatch" prompt produced an anatomical body part, twice
|
||||
out of three attempts, and the third produced a leather-scale surface. SDXL has no
|
||||
useful notion of "texture swatch without anatomy", and a character texture that
|
||||
needs manual rejection on every seed is worse than noise we control.
|
||||
|
||||
So skin detail is generated here instead: fine tileable noise with a low-frequency
|
||||
break-up, which is what the base colour actually needs (pores, not a pattern). Hair
|
||||
is generated as a strand sheet with alpha, for hair cards.
|
||||
|
||||
python3 Tools/gen/generate_character_textures.py --out Art/textures/generated
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import random
|
||||
import struct
|
||||
import zlib
|
||||
|
||||
SKIN = {
|
||||
"name": "T_Character_SkinDetail",
|
||||
"size": 512,
|
||||
# Base tone sits near the material's measured skin albedo so a multiply blend
|
||||
# keeps the character's colour rather than tinting it.
|
||||
"tint": (222, 178, 152),
|
||||
"detail": 9,
|
||||
}
|
||||
|
||||
HAIR = {
|
||||
"name": "T_Character_Hair",
|
||||
"size": 512,
|
||||
"base": (44, 32, 26),
|
||||
"highlight": (96, 74, 58),
|
||||
"strands": 190,
|
||||
}
|
||||
|
||||
|
||||
def write_png(path, width, height, pixels, alpha=True):
|
||||
"""Write RGBA or RGB pixels with zlib, no image library needed."""
|
||||
raw = bytearray()
|
||||
for y in range(height):
|
||||
raw.append(0)
|
||||
start = y * width * (4 if alpha else 3)
|
||||
step = 4 if alpha else 3
|
||||
for x in range(width):
|
||||
i = start + x * step
|
||||
raw += pixels[i:i + step]
|
||||
|
||||
def chunk(tag, data):
|
||||
return (struct.pack(">I", len(data)) + tag + data +
|
||||
struct.pack(">I", zlib.crc32(tag + data) & 0xffffffff))
|
||||
|
||||
colour_type = 6 if alpha else 2
|
||||
ihdr = struct.pack(">IIBBBBB", width, height, 8, colour_type, 0, 0, 0)
|
||||
png = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr)
|
||||
png += chunk(b"IDAT", zlib.compress(bytes(raw), 6)) + chunk(b"IEND", b"")
|
||||
with open(path, "wb") as handle:
|
||||
handle.write(png)
|
||||
|
||||
|
||||
def value_noise(size, cells, seed):
|
||||
"""Tileable value noise: a lattice that wraps, bilinearly interpolated."""
|
||||
rnd = random.Random(seed)
|
||||
lattice = [[rnd.random() for _ in range(cells)] for _ in range(cells)]
|
||||
out = [0.0] * (size * size)
|
||||
for y in range(size):
|
||||
fy = y / size * cells
|
||||
y0 = int(fy) % cells
|
||||
y1 = (y0 + 1) % cells
|
||||
ty = fy - int(fy)
|
||||
ty = ty * ty * (3 - 2 * ty)
|
||||
for x in range(size):
|
||||
fx = x / size * cells
|
||||
x0 = int(fx) % cells
|
||||
x1 = (x0 + 1) % cells
|
||||
tx = fx - int(fx)
|
||||
tx = tx * tx * (3 - 2 * tx)
|
||||
a = lattice[y0][x0] * (1 - tx) + lattice[y0][x1] * tx
|
||||
b = lattice[y1][x0] * (1 - tx) + lattice[y1][x1] * tx
|
||||
out[y * size + x] = a * (1 - ty) + b * ty
|
||||
return out
|
||||
|
||||
|
||||
def skin_texture(spec, out_dir):
|
||||
size, tint, detail = spec["size"], spec["tint"], spec["detail"]
|
||||
# Three octaves: fine pores, mid break-up, broad blotch. All tileable.
|
||||
fine = value_noise(size, 96, 7)
|
||||
mid = value_noise(size, 24, 13)
|
||||
broad = value_noise(size, 7, 21)
|
||||
px = bytearray(size * size * 3)
|
||||
for i in range(size * size):
|
||||
n = (fine[i] - 0.5) * detail * 1.0 + (mid[i] - 0.5) * detail * 0.6 + (broad[i] - 0.5) * detail * 0.4
|
||||
for c in range(3):
|
||||
v = tint[c] + n
|
||||
px[i * 3 + c] = max(0, min(255, int(v)))
|
||||
path = os.path.join(out_dir, spec["name"] + ".png")
|
||||
write_png(path, size, size, px, alpha=False)
|
||||
return path, {"octaves": {"fine": 96, "mid": 24, "broad": 7}, "detail": detail, "tint": tint}
|
||||
|
||||
|
||||
def hair_texture(spec, out_dir):
|
||||
size = spec["size"]
|
||||
base, hi = spec["base"], spec["highlight"]
|
||||
rnd = random.Random(5)
|
||||
px = bytearray(size * size * 4)
|
||||
# Vertical strands: each is a line with slight wander, alpha fading at the edges.
|
||||
strands = []
|
||||
for _ in range(spec["strands"]):
|
||||
strands.append({
|
||||
"x": rnd.uniform(0, size),
|
||||
"w": rnd.uniform(0.6, 2.2),
|
||||
"a": rnd.uniform(0.35, 0.95),
|
||||
"wander": rnd.uniform(2.0, 9.0),
|
||||
"phase": rnd.uniform(0, math.tau),
|
||||
"bright": rnd.random(),
|
||||
})
|
||||
for y in range(size):
|
||||
for x in range(size):
|
||||
i = (y * size + x) * 4
|
||||
alpha = 0.0
|
||||
lift = 0.0
|
||||
for s in strands:
|
||||
sx = (s["x"] + math.sin(y / size * math.tau + s["phase"]) * s["wander"]) % size
|
||||
d = abs(x - sx)
|
||||
d = min(d, size - d)
|
||||
if d < s["w"]:
|
||||
a = (1.0 - d / s["w"]) * s["a"]
|
||||
if a > alpha:
|
||||
alpha = a
|
||||
lift = s["bright"]
|
||||
if alpha <= 0.02:
|
||||
px[i:i + 4] = bytes((0, 0, 0, 0))
|
||||
continue
|
||||
shade = [base[c] + (hi[c] - base[c]) * lift for c in range(3)]
|
||||
px[i:i + 3] = bytes(max(0, min(255, int(v))) for v in shade)
|
||||
px[i + 3] = int(min(1.0, alpha) * 255)
|
||||
path = os.path.join(out_dir, spec["name"] + ".png")
|
||||
write_png(path, size, size, px, alpha=True)
|
||||
return path, {"strands": spec["strands"], "base": base, "highlight": hi}
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--out", default="Art/textures/generated")
|
||||
args = ap.parse_args()
|
||||
os.makedirs(args.out, exist_ok=True)
|
||||
results = {}
|
||||
for spec, maker in ((SKIN, skin_texture), (HAIR, hair_texture)):
|
||||
path, meta = maker(spec, args.out)
|
||||
sidecar = os.path.splitext(path)[0] + ".json"
|
||||
with open(sidecar, "w") as handle:
|
||||
json.dump({"tool": "Tools/gen/generate_character_textures.py",
|
||||
"asset": spec["name"], "size": spec["size"],
|
||||
"procedural": True, "meta": meta,
|
||||
"note": "generated, not photographed; no third-party licence"},
|
||||
handle, indent=2)
|
||||
results[spec["name"]] = path
|
||||
print(f"wrote {path} (+ sidecar)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,226 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate the facade texture maps the demo materials sample.
|
||||
|
||||
The materials started as procedural graphs only (noise + coursing), which reads
|
||||
as tinted stone but has no grain. These maps add the grain: an ashlar stone
|
||||
course pattern, a rendered stucco tooth, asphalt aggregate and concrete pores,
|
||||
each as base colour plus a roughness map so the specular breaks up.
|
||||
|
||||
Standard library only — the PNG encoder is the same pure-zlib writer used for the
|
||||
MPFB skin, so this runs on the host without numpy or Pillow.
|
||||
|
||||
python3 Tools/gen/generate_facade_textures.py # write to /tmp/gigline/textures
|
||||
python3 Tools/gen/generate_facade_textures.py --size 1024
|
||||
python3 Tools/gen/generate_facade_textures.py --out Art/blender/textures
|
||||
|
||||
Output names match the Unreal import destination `/Game/Hollywood/Textures`:
|
||||
|
||||
T_StoneAshlar_BC.png T_StoneAshlar_R.png
|
||||
T_Stucco_BC.png T_Stucco_R.png
|
||||
T_Asphalt_BC.png T_Asphalt_R.png
|
||||
T_Concrete_BC.png T_Concrete_R.png
|
||||
|
||||
Determinism: every pattern is a deterministic hash of the pixel coordinate, so a
|
||||
re-run reproduces the same files byte for byte (no RNG state, no seam drift).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import math
|
||||
import os
|
||||
import struct
|
||||
import zlib
|
||||
|
||||
# --- PNG -------------------------------------------------------------------
|
||||
|
||||
|
||||
def write_png(path, width, height, rgb_rows):
|
||||
"""Write 8-bit RGB rows (bytes) to a PNG. `rgb_rows` is one bytes per row."""
|
||||
raw = bytearray()
|
||||
for y in range(height):
|
||||
raw.append(0) # filter type 0
|
||||
raw += rgb_rows[y]
|
||||
|
||||
def chunk(tag, data):
|
||||
c = struct.pack(">I", len(data)) + tag + data
|
||||
c += struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
|
||||
return c
|
||||
|
||||
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # colour type 2 = RGB
|
||||
png = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr)
|
||||
png += chunk(b"IDAT", zlib.compress(bytes(raw), 6)) + chunk(b"IEND", b"")
|
||||
os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
|
||||
with open(path, "wb") as handle:
|
||||
handle.write(png)
|
||||
return path
|
||||
|
||||
|
||||
# --- deterministic value noise --------------------------------------------
|
||||
|
||||
|
||||
def _hash2(x, y, seed=0):
|
||||
"""Deterministic 0..1 hash of an integer cell. No RNG, fully reproducible."""
|
||||
n = (x * 374761393 + y * 668265263 + seed * 1442695040888963407) & 0xFFFFFFFF
|
||||
n = (n ^ (n >> 13)) * 1274126177 & 0xFFFFFFFF
|
||||
return ((n ^ (n >> 16)) & 0xFFFF) / 65535.0
|
||||
|
||||
|
||||
def value_noise(x, y, seed=0):
|
||||
"""Smooth bilinear value noise at continuous (x, y) in cell units."""
|
||||
xi, yi = math.floor(x), math.floor(y)
|
||||
fx, fy = x - xi, y - yi
|
||||
# smoothstep weights so cells blend without visible lattice lines
|
||||
sx = fx * fx * (3.0 - 2.0 * fx)
|
||||
sy = fy * fy * (3.0 - 2.0 * fy)
|
||||
n00 = _hash2(xi, yi, seed)
|
||||
n10 = _hash2(xi + 1, yi, seed)
|
||||
n01 = _hash2(xi, yi + 1, seed)
|
||||
n11 = _hash2(xi + 1, yi + 1, seed)
|
||||
return (n00 * (1 - sx) + n10 * sx) * (1 - sy) + (n01 * (1 - sx) + n11 * sx) * sy
|
||||
|
||||
|
||||
def fractal(x, y, octaves=4, seed=0):
|
||||
"""Fractal sum of value noise, normalised to 0..1."""
|
||||
total, amp, freq, norm = 0.0, 1.0, 1.0, 0.0
|
||||
for _ in range(octaves):
|
||||
total += value_noise(x * freq, y * freq, seed) * amp
|
||||
norm += amp
|
||||
amp *= 0.5
|
||||
freq *= 2.0
|
||||
return total / norm
|
||||
|
||||
|
||||
def clamp8(value):
|
||||
return max(0, min(255, int(value * 255.0 + 0.5)))
|
||||
|
||||
|
||||
# --- patterns --------------------------------------------------------------
|
||||
# Each returns (base_rgb_function, roughness_function) over u, v in 0..1.
|
||||
|
||||
|
||||
def stone_ashlar(courses=8, per_course=4):
|
||||
"""Coursed ashlar: horizontal beds, staggered vertical joints, stone grain."""
|
||||
course_h = 1.0 / courses
|
||||
|
||||
def block_uv(u, v):
|
||||
row = int(v / course_h)
|
||||
shift = 0.5 * (row % 2) / per_course # stagger alternate courses
|
||||
col = int(((u + shift) % 1.0) * per_course)
|
||||
return row, col, (v % course_h) / course_h, ((u + shift) % 1.0) * per_course - col
|
||||
|
||||
def base(u, v, px, py):
|
||||
row, col, fj, fi = block_uv(u, v)
|
||||
tone = 0.62 + 0.10 * _hash2(row, col, 11) # each block a little different
|
||||
grain = 0.06 * (fractal(px * 0.15, py * 0.15, 5, 3) - 0.5) * 2.0
|
||||
# joint: a dark band along the bed and the head of each block
|
||||
joint = 0.0
|
||||
if fj < 0.055 or fi < 0.055:
|
||||
joint = -0.22
|
||||
value = tone + grain + joint
|
||||
return (value * 1.02, value, value * 0.94) # faintly warm stone
|
||||
|
||||
def rough(u, v, px, py):
|
||||
row, col, fj, fi = block_uv(u, v)
|
||||
base_r = 0.66 + 0.10 * _hash2(row, col, 29)
|
||||
if fj < 0.055 or fi < 0.055:
|
||||
base_r = 0.86 # joints are matte
|
||||
return base_r + 0.06 * fractal(px * 0.2, py * 0.2, 3, 7)
|
||||
|
||||
return base, rough
|
||||
|
||||
|
||||
def stucco():
|
||||
def base(u, v, px, py):
|
||||
blot = fractal(px * 0.03, py * 0.03, 4, 21)
|
||||
tooth = fractal(px * 0.9, py * 0.9, 3, 22)
|
||||
value = 0.58 + 0.12 * (blot - 0.5) + 0.05 * (tooth - 0.5)
|
||||
return (value * 1.03, value, value * 0.96)
|
||||
|
||||
def rough(u, v, px, py):
|
||||
return 0.74 + 0.18 * fractal(px * 0.06, py * 0.06, 3, 31)
|
||||
|
||||
return base, rough
|
||||
|
||||
|
||||
def asphalt():
|
||||
def base(u, v, px, py):
|
||||
agg = fractal(px * 1.4, py * 1.4, 3, 41) # fine aggregate
|
||||
patch = fractal(px * 0.02, py * 0.02, 3, 42) # repair patches
|
||||
value = 0.055 + 0.05 * (patch - 0.5) + 0.06 * (agg - 0.5)
|
||||
return (value, value, value * 1.05) # cool bitumen
|
||||
|
||||
def rough(u, v, px, py):
|
||||
return 0.90 + 0.08 * fractal(px * 0.5, py * 0.5, 2, 43)
|
||||
|
||||
return base, rough
|
||||
|
||||
|
||||
def concrete():
|
||||
def base(u, v, px, py):
|
||||
pores = fractal(px * 0.8, py * 0.8, 3, 51)
|
||||
tone = fractal(px * 0.04, py * 0.04, 4, 52)
|
||||
value = 0.50 + 0.10 * (tone - 0.5) + 0.05 * (pores - 0.5)
|
||||
return (value * 1.01, value, value * 0.97)
|
||||
|
||||
def rough(u, v, px, py):
|
||||
return 0.84 + 0.10 * fractal(px * 0.15, py * 0.15, 3, 53)
|
||||
|
||||
return base, rough
|
||||
|
||||
|
||||
TEXTURES = {
|
||||
"T_StoneAshlar": stone_ashlar(),
|
||||
"T_Stucco": stucco(),
|
||||
"T_Asphalt": asphalt(),
|
||||
"T_Concrete": concrete(),
|
||||
}
|
||||
|
||||
|
||||
def render(name, pattern, size):
|
||||
"""Render one texture; returns {suffix: path} for _BC and _R."""
|
||||
base_fn, rough_fn = pattern
|
||||
paths = {}
|
||||
for suffix, fn in (("BC", base_fn), ("R", rough_fn)):
|
||||
# Wrap in a function so the inner loop stays tight.
|
||||
rows = []
|
||||
for y in range(size):
|
||||
v = y / size
|
||||
row = bytearray()
|
||||
for x in range(size):
|
||||
u = x / size
|
||||
value = fn(u, v, x / size * size / 64.0, y / size * size / 64.0)
|
||||
if isinstance(value, tuple):
|
||||
row += bytes((clamp8(value[0]), clamp8(value[1]), clamp8(value[2])))
|
||||
else:
|
||||
grey = clamp8(value)
|
||||
row += bytes((grey, grey, grey))
|
||||
rows.append(bytes(row))
|
||||
path = os.path.join(OUT_DIR, f"{name}_{suffix}.png")
|
||||
write_png(path, size, size, rows)
|
||||
paths[suffix] = path
|
||||
return paths
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
global OUT_DIR
|
||||
ap = argparse.ArgumentParser(description=__doc__,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("--out", default="/tmp/gigline/textures",
|
||||
help="output directory (default /tmp/gigline/textures)")
|
||||
ap.add_argument("--size", type=int, default=512, help="edge length in pixels")
|
||||
args = ap.parse_args(argv)
|
||||
OUT_DIR = args.out
|
||||
|
||||
print(f"writing {len(TEXTURES) * 2} maps at {args.size}² to {OUT_DIR}")
|
||||
for name, pattern in TEXTURES.items():
|
||||
paths = render(name, pattern, args.size)
|
||||
for suffix, path in paths.items():
|
||||
print(f" {os.path.basename(path):28} {os.path.getsize(path):>8} bytes")
|
||||
return 0
|
||||
|
||||
|
||||
OUT_DIR = "/tmp/gigline/textures"
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,146 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Download the main photograph from a web page by reading its Open Graph image.
|
||||
|
||||
For reference gathering where the source is a gallery or photo page rather than
|
||||
an API: the page's `og:image` (or `twitter:image`) is the image the site itself
|
||||
publishes as representative, so it is a defensible pick without scraping every
|
||||
`<img>` on the page.
|
||||
|
||||
python3 Tools/gen/page_image_fetch.py \
|
||||
--page "https://www.flickr.com/photos/…/in/…" --slug hero-masonic --name entry-01
|
||||
|
||||
Writes into `Art/reference/<slug>/` (gitignored) and appends provenance to
|
||||
`sidecar-web.json` there: page URL, image URL, timestamp, byte size, SHA-256.
|
||||
|
||||
Flickr note: the og:image is served from `live.staticflickr.com` at ~1024 px.
|
||||
The suffix is rewritten to the `_k` (2048 px) variant when the page shows one;
|
||||
if that 404s the original is used.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/130.0 Safari/537.36 reference-fetch/0.1")
|
||||
META_PATTERNS = [
|
||||
r'<meta[^>]+property=["\']og:image(?::url)?["\'][^>]+content=["\']([^"\']+)',
|
||||
r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+property=["\']og:image',
|
||||
r'<meta[^>]+name=["\']twitter:image["\'][^>]+content=["\']([^"\']+)',
|
||||
]
|
||||
|
||||
|
||||
def fetch(url, timeout=45, retries=3):
|
||||
for attempt in range(retries):
|
||||
req = urllib.request.Request(url, headers={"User-Agent": UA,
|
||||
"Accept": "*/*",
|
||||
"Accept-Language": "en-US,en;q=0.9"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
return resp.read(), resp.headers.get("Content-Type", "")
|
||||
except urllib.error.HTTPError as exc:
|
||||
if attempt == retries - 1:
|
||||
raise
|
||||
time.sleep(2 * (attempt + 1))
|
||||
except Exception:
|
||||
if attempt == retries - 1:
|
||||
raise
|
||||
time.sleep(2 * (attempt + 1))
|
||||
return b"", ""
|
||||
|
||||
|
||||
def og_image(html: str):
|
||||
for pattern in META_PATTERNS:
|
||||
match = re.search(pattern, html, flags=re.I)
|
||||
if match:
|
||||
return urllib.parse.unquote(match.group(1))
|
||||
return None
|
||||
|
||||
|
||||
def upgrade_flickr(url: str):
|
||||
"""Try the large-size suffix for a live.staticflickr.com URL."""
|
||||
if "staticflickr.com" not in url:
|
||||
return None
|
||||
candidate = re.sub(r"_[a-z]\.jpg$", "_k.jpg", url)
|
||||
return candidate if candidate != url else None
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--page", required=True)
|
||||
ap.add_argument("--slug", required=True)
|
||||
ap.add_argument("--out", default="Art/reference")
|
||||
ap.add_argument("--name", default=None, help="output file stem; defaults to a hash of the URL")
|
||||
ap.add_argument("--note", default="")
|
||||
args = ap.parse_args()
|
||||
|
||||
out_dir = os.path.join(args.out, args.slug)
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
|
||||
print(f"page: {args.page}")
|
||||
raw, ctype = fetch(args.page)
|
||||
if "html" not in ctype.lower():
|
||||
print(f" not HTML (content-type {ctype}); skipping")
|
||||
return 1
|
||||
html = raw.decode("utf-8", errors="replace")
|
||||
image_url = og_image(html)
|
||||
if not image_url:
|
||||
print(" no og:image found; skipping")
|
||||
return 1
|
||||
print(f" og:image: {image_url[:110]}")
|
||||
|
||||
upgraded = upgrade_flickr(image_url)
|
||||
if upgraded:
|
||||
try:
|
||||
blob, _ = fetch(upgraded)
|
||||
if len(blob) > 10000:
|
||||
image_url = upgraded
|
||||
print(f" upgraded to large: {upgraded[:110]}")
|
||||
except Exception as exc:
|
||||
print(f" large variant failed ({exc}); using original")
|
||||
|
||||
blob, ctype = fetch(image_url)
|
||||
if len(blob) < 5000:
|
||||
print(f" too small ({len(blob)} bytes); skipping")
|
||||
return 1
|
||||
|
||||
stem = args.name or hashlib.sha256(args.page.encode()).hexdigest()[:10]
|
||||
ext = ".jpg" if "jpeg" in ctype or "jpg" in ctype else ".png"
|
||||
filename = f"{stem}{ext}"
|
||||
path = os.path.join(out_dir, filename)
|
||||
with open(path, "wb") as handle:
|
||||
handle.write(blob)
|
||||
digest = hashlib.sha256(blob).hexdigest()
|
||||
print(f" saved {filename} ({len(blob)//1024} KB)")
|
||||
|
||||
sidecar_path = os.path.join(out_dir, "sidecar-web.json")
|
||||
data = {"source": "web page og:image", "tool": "Tools/gen/page_image_fetch.py",
|
||||
"note": "local development reference only; not committed", "images": []}
|
||||
if os.path.exists(sidecar_path):
|
||||
try:
|
||||
data = json.load(open(sidecar_path))
|
||||
except Exception:
|
||||
pass
|
||||
data["images"].append({
|
||||
"local_file": filename,
|
||||
"page_url": args.page,
|
||||
"image_url": image_url,
|
||||
"bytes": len(blob),
|
||||
"sha256": digest,
|
||||
"note": args.note,
|
||||
"fetched_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
})
|
||||
with open(sidecar_path, "w", encoding="utf-8") as handle:
|
||||
json.dump(data, handle, indent=2)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user