Immich on 6 GB RAM: import 500 photos with ML on and off

#!/usr/bin/env python3
"""Runs in the VM. bench.py <label> <ml:1|0> <n_images> <imgdir> <outdir>
Creates the admin, an API key, optionally disables machine learning through PUT /api/system-config, uploads 10 warm-up images
(not counted, they pull the ML models in on the ML run), waits for the queues, then uploads n images with 4 workers and polls
GET /api/queues every 2 s until every queue has been empty (active+waiting+delayed = 0) for 3 polls in a row.
The smart-search request is NOT made here (it loads the textual CLIP model and would add to the post-work memory window); run.sh runs search.py after the memory sampler has stopped."""
import sys, os, json, time, threading, urllib.request, urllib.error, glob, uuid
label, ml, n, imgdir, outdir = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]), sys.argv[4], sys.argv[5]
U = "http://127.0.0.1:8080/api"; os.makedirs(outdir, exist_ok=True)
def call(method, path, body=None, headers=None, raw=None):
    h = dict(headers or {}); data = raw
    if body is not None: data = json.dumps(body).encode(); h["Content-Type"] = "application/json"
    r = urllib.request.Request(U + path, data=data, method=method, headers=h)
    try:
        with urllib.request.urlopen(r, timeout=120) as x: t = x.read(); return x.status, (json.loads(t) if t else None)
    except urllib.error.HTTPError as e: return e.code, e.read().decode()[:300]
st, r = call("POST", "/auth/admin-sign-up", {"email": "admin@lab.invalid", "name": "Lab Admin", "password": "lab-only-not-a-secret"}); assert st in (200, 201), (st, r)
st, r = call("POST", "/auth/login", {"email": "admin@lab.invalid", "password": "lab-only-not-a-secret"}); tok = r["accessToken"]
st, r = call("POST", "/api-keys", {"name": "bench", "permissions": ["all"]}, {"Authorization": "Bearer " + tok}); KEY = {"x-api-key": r["secret"]}
call("PUT", "/users/me/onboarding", {"isOnboarded": True}, KEY)
st, cfg = call("GET", "/system-config", headers=KEY)
cfg["machineLearning"]["enabled"] = bool(ml); st, r = call("PUT", "/system-config", cfg, KEY); assert st == 200, (st, r)
st, cfg = call("GET", "/system-config", headers=KEY); print("machineLearning.enabled =", cfg["machineLearning"]["enabled"], flush=True)
def upload(path, i):
    b = uuid.uuid4().hex; name = os.path.basename(path); data = open(path, "rb").read()
    def f(k, v): return f'--{b}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()
    body = (f("deviceAssetId", f"{name}-{label}") + f("deviceId", "labgen") + f("fileCreatedAt", "2024-01-01T00:00:00.000Z") + f("fileModifiedAt", "2024-01-01T00:00:00.000Z")
            + f'--{b}\r\nContent-Disposition: form-data; name="assetData"; filename="{name}"\r\nContent-Type: image/jpeg\r\n\r\n'.encode() + data + f"\r\n--{b}--\r\n".encode())
    return call("POST", "/assets", headers={**KEY, "Content-Type": f"multipart/form-data; boundary={b}"}, raw=body)[0]
def busy():
    st, qs = call("GET", "/queues", headers=KEY); assert st == 200, (st, qs)
    return {q["name"]: q["statistics"]["active"] + q["statistics"]["waiting"] + q["statistics"]["delayed"] for q in qs}
def wait_idle(csv, quiet=3, timeout=7200):
    ok = 0; t0 = time.time(); first = True; clear = {}
    while time.time() - t0 < timeout:
        b = busy(); now = time.time()
        if first and csv is not None: csv.write("epoch," + ",".join(sorted(b)) + "\n"); first = False
        if csv is not None: csv.write(f"{now:.2f}," + ",".join(str(b[k]) for k in sorted(b)) + "\n"); csv.flush()
        ok = ok + 1 if sum(b.values()) == 0 else 0
        if ok >= quiet: return now
        time.sleep(2)
    raise SystemExit("queues never cleared")
files = sorted(glob.glob(f"{imgdir}/*.jpg")); assert len(files) >= n + 10, len(files)
res = {"label": label, "ml_enabled": bool(ml), "n": n}
t = time.time(); errs = sum(1 for i, p in enumerate(files[:10]) if upload(p, i) not in (200, 201))
w_end = wait_idle(None); res["warmup"] = {"images": 10, "errors": errs, "secs": w_end - t}; print("warmup", res["warmup"], flush=True)
open(f"{outdir}/MARK_warm_done", "w").write(str(time.time()))
csv = open(f"{outdir}/{label}-queues.csv", "w"); batch = files[10:10 + n]; it = iter(enumerate(batch)); lk = threading.Lock(); bad = [0]
def worker():
    while True:
        with lk: x = next(it, None)
        if x is None: return
        if upload(x[1], x[0]) not in (200, 201):
            with lk: bad[0] += 1
t0 = time.time(); th = [threading.Thread(target=worker) for _ in range(4)]; [x.start() for x in th]
poll = threading.Thread(target=lambda: None)
[x.join() for x in th]; t1 = time.time(); res["upload"] = {"t_start": t0, "t_end": t1, "secs": t1 - t0, "errors": bad[0]}; print("upload", res["upload"], flush=True)
t2 = wait_idle(csv); res["queues_clear"] = {"t_end": t2, "secs_from_upload_start": t2 - t0, "secs_from_upload_end": t2 - t1}; print("clear", res["queues_clear"], flush=True)
st, stats = call("GET", "/server/statistics", headers=KEY); res["server_statistics"] = stats
json.dump(res, open(f"{outdir}/{label}-bench.json", "w"), indent=1); print(json.dumps(res)[:600])