Luma Photon

POST https://api.lumalabs.ai/dream-machine/v1/generations/image
Authorization: Bearer $LUMAAI_API_KEY
Content-Type: application/json

GET    /dream-machine/v1/generations/{id}
GET    /dream-machine/v1/generations?limit=100&offset=0
DELETE /dream-machine/v1/generations/{id}
GET    /dream-machine/v1/credits

{
  "model": "photon-flash-1",
  "prompt": "Use the reference only for the cobalt ceramic shape and three-quarter camera angle. Place that object on warm white paper under soft overhead light; editorial product photograph, open negative space on the left.",
  "aspect_ratio": "16:9",
  "format": "png",
  "image_ref": [
    {"url": "https://assets.example.com/signed/product-reference.png", "weight": 0.8}
  ],
  "sync": false
}

{
  "model": "photon-1",
  "prompt": "Replace only the white daisies with yellow sunflowers. Preserve the vase geometry, tabletop, camera angle, crop, lighting direction, shadows, and background.",
  "aspect_ratio": "4:3",
  "format": "png",
  "modify_image_ref": {
    "url": "https://assets.example.com/signed/approved-vase.png",
    "weight": 1.0
  },
  "sync": false
}

import hashlib
import ipaddress
import json
import math
import os
import pathlib
import random
import socket
import tempfile
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
import urllib.error
import urllib.parse
import urllib.request
import uuid

from PIL import Image, UnidentifiedImageError

API_ROOT = "https://api.lumalabs.ai/dream-machine/v1"
CREATE_URL = f"{API_ROOT}/generations/image"
OUT = pathlib.Path("luma-photon-output")
MAX_JSON_BYTES = 2 * 1024 * 1024
MAX_IMAGE_BYTES = 30 * 1024 * 1024
MAX_IMAGE_PIXELS = 20_000_000
MAX_POLL_SECONDS = 600
Image.MAX_IMAGE_PIXELS = MAX_IMAGE_PIXELS


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


api_opener = urllib.request.build_opener(NoRedirect())


class LumaHTTPError(RuntimeError):
    def __init__(self, code, detail, retry_after=None):
        super().__init__(f"Luma HTTP {code}: {detail}")
        self.code = code
        self.retry_after = retry_after


def host_set(name, defaults=()):
    values = {value.strip().lower() for value in os.getenv(name, "").split(",") if value.strip()}
    return values | set(defaults)


def redact_url(url):
    parsed = urllib.parse.urlsplit(url)
    return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))


def validate_url_shape(url, allowed_hosts):
    parsed = urllib.parse.urlsplit(url)
    if (
        parsed.scheme != "https"
        or parsed.hostname is None
        or parsed.hostname.lower() not in allowed_hosts
        or parsed.username
        or parsed.password
        or parsed.fragment
        or parsed.port not in (None, 443)
    ):
        raise ValueError("URL must be HTTPS on an approved exact host without credentials, fragment, or non-443 port")


def validate_public_dns(url, allowed_hosts):
    validate_url_shape(url, allowed_hosts)
    parsed = urllib.parse.urlsplit(url)
    addresses = {info[4][0] for info in socket.getaddrinfo(parsed.hostname, 443)}
    if not addresses or any(not ipaddress.ip_address(value).is_global for value in addresses):
        raise ValueError("URL host did not resolve exclusively to public addresses")


def bounded_json(request, expected_status, timeout):
    try:
        # Never follow an API redirect with the bearer token attached. The fixed
        # api.lumalabs.ai origin must answer directly or the request fails closed.
        with api_opener.open(request, timeout=timeout) as response:
            if response.status != expected_status:
                raise RuntimeError(f"Unexpected HTTP status {response.status}")
            raw = response.read(MAX_JSON_BYTES + 1)
    except urllib.error.HTTPError as exc:
        try:
            retry_after = exc.headers.get("Retry-After")
            detail = exc.read(64 * 1024).decode("utf-8", "replace")
        finally:
            exc.close()
        raise LumaHTTPError(exc.code, detail, retry_after) from exc
    if len(raw) > MAX_JSON_BYTES:
        raise RuntimeError("Luma JSON response exceeded configured cap")
    try:
        value = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RuntimeError("Luma response was not valid bounded JSON") from exc
    if not isinstance(value, dict):
        raise RuntimeError("Luma response was not a JSON object")
    return value


def api_request(method, path, key, payload=None, expected_status=200, timeout=90):
    body = None if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8")
    headers = {"Authorization": f"Bearer {key}", "Accept": "application/json"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    request = urllib.request.Request(
        f"{API_ROOT}{path}", data=body, method=method, headers=headers
    )
    return bounded_json(request, expected_status, timeout)


def safe_credit_balance(key):
    try:
        value = api_request("GET", "/credits", key)
        return float(value["credit_balance"])
    except (KeyError, TypeError, ValueError, RuntimeError, urllib.error.URLError) as exc:
        print({"credit_balance_warning": str(exc)})
        return None


def atomic_write(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".luma-manifest-")
    try:
        with os.fdopen(fd, "wb") as output:
            fd = None
            output.write(data)
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary, path)
    except Exception:
        if fd is not None:
            os.close(fd)
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass
        raise


def read_local_json(path):
    if path.stat().st_size > MAX_JSON_BYTES:
        raise ValueError("Local ledger exceeded JSON cap")
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError("Local ledger was not a JSON object")
    return value


def retry_after_seconds(value):
    if not value:
        return 0.0
    try:
        seconds = float(value)
    except ValueError:
        try:
            moment = parsedate_to_datetime(value)
            if moment.tzinfo is None:
                moment = moment.replace(tzinfo=timezone.utc)
            seconds = (moment - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            return 0.0
    return seconds if math.isfinite(seconds) and seconds > 0 else 0.0


def get_generation_with_retry(generation_id, key, deadline, initial_wait=0.0):
    wait = max(0.0, initial_wait)
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("generation GET deadline exceeded")
        if wait:
            sleep_for = wait + random.random()
            if sleep_for >= remaining:
                raise TimeoutError("required generation GET delay exceeds remaining deadline")
            time.sleep(sleep_for)
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError("generation GET deadline exceeded after wait")
        try:
            return api_request("GET", f"/generations/{generation_id}", key,
                               timeout=max(0.001, min(90.0, remaining)))
        except LumaHTTPError as exc:
            if exc.code not in {429, 500, 502, 503, 504}:
                raise
            provider_wait = retry_after_seconds(exc.retry_after)
            wait = max(min(max(wait, 2.0) * 1.6, 30.0), provider_wait, 2.0)
        except (urllib.error.URLError, TimeoutError):
            wait = min(max(wait, 2.0) * 1.6, 30.0)


def download_image(url, allowed_hosts, directory, requested_format):
    opener = urllib.request.build_opener(NoRedirect())
    current = url
    response = None
    for _ in range(6):
        validate_public_dns(current, allowed_hosts)
        # Deliberately no LUMAAI_API_KEY header on any asset hop.
        request = urllib.request.Request(
            current, headers={"Accept": "image/jpeg,image/png"}
        )
        try:
            response = opener.open(request, timeout=90)
            break
        except urllib.error.HTTPError as exc:
            if exc.code in {301, 302, 303, 307, 308}:
                location = exc.headers.get("Location")
                exc.close()
                if not location:
                    raise ValueError("Artifact redirect omitted Location")
                current = urllib.parse.urljoin(current, location)
                continue
            try:
                detail = exc.read(64 * 1024).decode("utf-8", "replace")
            finally:
                exc.close()
            raise RuntimeError(f"Artifact HTTP {exc.code}: {detail}") from exc
    else:
        raise ValueError("Too many artifact redirects")

    directory.mkdir(parents=True, exist_ok=True)
    fd = None
    temporary = None
    total = 0
    prefix = b""
    hasher = hashlib.sha256()
    try:
        with response:
            mime = response.headers.get_content_type()
            if mime not in {"image/jpeg", "image/png"}:
                raise ValueError(f"Unexpected artifact MIME: {mime}")
            declared = response.headers.get("Content-Length")
            if declared:
                try:
                    if int(declared) > MAX_IMAGE_BYTES:
                        raise ValueError("Artifact exceeds declared-size cap")
                except ValueError as exc:
                    raise ValueError("Invalid or excessive Content-Length") from exc
            fd, temporary = tempfile.mkstemp(dir=directory, prefix=".luma-image-")
            with os.fdopen(fd, "wb") as output:
                fd = None
                while True:
                    chunk = response.read(64 * 1024)
                    if not chunk:
                        break
                    total += len(chunk)
                    if total > MAX_IMAGE_BYTES:
                        raise ValueError("Artifact exceeded streaming byte cap")
                    prefix += chunk[: max(0, 12 - len(prefix))]
                    hasher.update(chunk)
                    output.write(chunk)
                output.flush()
                os.fsync(output.fileno())

        if prefix.startswith(b"\xff\xd8\xff"):
            expected_mime, expected_format, suffix = "image/jpeg", "JPEG", ".jpg"
        elif prefix.startswith(b"\x89PNG\r\n\x1a\n"):
            expected_mime, expected_format, suffix = "image/png", "PNG", ".png"
        else:
            raise ValueError("Artifact signature is neither JPEG nor PNG")
        if mime != expected_mime:
            raise ValueError("Artifact MIME/signature mismatch")
        requested_mime = {"jpg": "image/jpeg", "png": "image/png"}[requested_format]
        if mime != requested_mime:
            raise ValueError("Artifact MIME did not match the requested output format")

        try:
            with Image.open(temporary) as probe:
                dimensions = probe.size
                if dimensions[0] * dimensions[1] > MAX_IMAGE_PIXELS:
                    raise ValueError(f"Artifact exceeds pixel cap: {dimensions}")
                decoded_format = probe.format
                probe.verify()
            with Image.open(temporary) as probe:
                probe.load()
        except (UnidentifiedImageError, OSError, Image.DecompressionBombError) as exc:
            raise ValueError("Artifact failed bounded full decode") from exc
        if decoded_format != expected_format:
            raise ValueError("Artifact decoder/signature mismatch")

        destination = directory / f"image{suffix}"
        os.replace(temporary, destination)
        temporary = None
        return {
            "path": str(destination),
            "bytes": total,
            "mime_type": mime,
            "dimensions": list(dimensions),
            "sha256": hasher.hexdigest(),
        }
    except Exception:
        if fd is not None:
            os.close(fd)
        try:
            if temporary is not None:
                os.unlink(temporary)
        except FileNotFoundError:
            pass
        raise


model = os.getenv("LUMA_MODEL", "photon-flash-1")
mode = os.getenv("LUMA_MODE", "text")
if model not in {"photon-1", "photon-flash-1"}:
    raise SystemExit("LUMA_MODEL must be photon-1 or photon-flash-1")
if mode not in {"text", "image-reference", "style-reference", "character-reference", "modify"}:
    raise SystemExit("Unsupported LUMA_MODE")

prompt = (
    "16:9 editorial product photograph of a cobalt ceramic tea set on warm white "
    "paper at dawn, tabletop camera height, soft overhead light, restrained cool "
    "shadows, clean negative space on the left, realistic glaze and contact shadows. "
    "No added lettering or logos."
)
payload = {
    "model": model,
    "prompt": prompt,
    "aspect_ratio": "16:9",
    "format": "png",
    "sync": False,
}

reference_urls = []
if mode != "text":
    try:
        reference_urls = json.loads(os.getenv("LUMA_REFERENCE_URLS_JSON", "[]"))
    except json.JSONDecodeError as exc:
        raise SystemExit("LUMA_REFERENCE_URLS_JSON must be a JSON array") from exc
    if not isinstance(reference_urls, list) or not all(isinstance(url, str) for url in reference_urls):
        raise SystemExit("LUMA_REFERENCE_URLS_JSON must contain only URL strings")
    expected = (1, 4) if mode in {"image-reference", "character-reference"} else (1, 1)
    if not expected[0] <= len(reference_urls) <= expected[1]:
        raise SystemExit(f"{mode} requires {expected[0]}..{expected[1]} reference URLs")
    input_hosts = host_set("LUMA_INPUT_HOSTS")
    if not input_hosts:
        raise SystemExit("Set LUMA_INPUT_HOSTS to the exact owned source hosts")
    for url in reference_urls:
        validate_url_shape(url, input_hosts)
    weight = float(os.getenv("LUMA_REFERENCE_WEIGHT", "0.8"))
    if not math.isfinite(weight):
        raise SystemExit("LUMA_REFERENCE_WEIGHT must be finite")
    if mode == "image-reference":
        payload["image_ref"] = [{"url": url, "weight": weight} for url in reference_urls]
    elif mode == "style-reference":
        payload["style_ref"] = [{"url": reference_urls[0], "weight": weight}]
    elif mode == "character-reference":
        payload["character_ref"] = {"identity0": {"images": reference_urls}}
    else:
        payload["modify_image_ref"] = {"url": reference_urls[0], "weight": weight}

body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
request_sha256 = hashlib.sha256(body).hexdigest()
client_job_key = os.getenv("LUMA_CLIENT_JOB_KEY", f"local-{request_sha256[:20]}")
display_payload = json.loads(json.dumps(payload))
for key in ("image_ref", "style_ref"):
    for value in display_payload.get(key, []):
        value["url"] = redact_url(value["url"])
if display_payload.get("character_ref"):
    display_payload["character_ref"]["identity0"]["images"] = [
        redact_url(url) for url in reference_urls
    ]
if display_payload.get("modify_image_ref"):
    display_payload["modify_image_ref"]["url"] = redact_url(reference_urls[0])

public_price_snapshot = {"photon-1": 0.015, "photon-flash-1": 0.004}[model]
print(json.dumps({
    "dry_run": os.getenv("SEND_LUMA_REQUEST") != "1",
    "client_job_key": client_job_key,
    "request": display_payload,
    "request_sha256": request_sha256,
    "conservative_public_unit_price_usd_snapshot_2026_07_09": public_price_snapshot,
    "price_requires_dashboard_verification": True,
}, indent=2))

if os.getenv("SEND_LUMA_REQUEST") != "1":
    raise SystemExit("Dry run only; set SEND_LUMA_REQUEST=1 after approving the paid call")
key = os.getenv("LUMAAI_API_KEY")
if not key:
    raise SystemExit("Set LUMAAI_API_KEY before the paid call")

job_token = hashlib.sha256(client_job_key.encode("utf-8")).hexdigest()[:20]
job_dir = OUT / job_token
ledger_path = job_dir / "job.json"
manifest_path = job_dir / "manifest.json"


def save_job(state, generation_id=None, failure_reason=None, note=None):
    value = {
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "model": model,
        "state": state,
        "generation_id": generation_id,
        "failure_reason": failure_reason,
        "note": note,
        "updated_unix": time.time(),
        "reference_url_sha256": [
            hashlib.sha256(url.encode("utf-8")).hexdigest() for url in reference_urls
        ],
    }
    atomic_write(ledger_path, json.dumps(value, indent=2).encode("utf-8"))


def claim_new_job():
    value = {
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "model": model,
        "state": "submitting",
        "generation_id": None,
        "failure_reason": None,
        "note": "Exclusive local claim created before any paid call",
        "updated_unix": time.time(),
        "reference_url_sha256": [
            hashlib.sha256(url.encode("utf-8")).hexdigest() for url in reference_urls
        ],
    }
    job_dir.mkdir(parents=True, exist_ok=True)
    fd = os.open(ledger_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as output:
            fd = None
            json.dump(value, output, indent=2)
            output.flush(); os.fsync(output.fileno())
    finally:
        if fd is not None: os.close(fd)

if reference_urls:
    for url in reference_urls:
        validate_public_dns(url, input_hosts)

if manifest_path.exists():
    print(json.dumps(read_local_json(manifest_path), indent=2))
    raise SystemExit("This client job is already complete; choose a new LUMA_CLIENT_JOB_KEY for a new paid job")

credit_before = None
deadline = time.monotonic() + MAX_POLL_SECONDS
if ledger_path.exists():
    existing = read_local_json(ledger_path)
    if existing.get("client_job_key") != client_job_key or existing.get("request_sha256") != request_sha256:
        raise SystemExit("Existing ledger does not match this client job/request")
    generation_id = existing.get("generation_id")
    if not generation_id:
        raise SystemExit(
            "Existing create has no confirmed generation ID; treat it as ambiguous and reconcile. "
            "Use a new explicit LUMA_CLIENT_JOB_KEY only after an operator approves a new paid job."
        )
    try: uuid.UUID(str(generation_id))
    except (ValueError, AttributeError) as exc:
        raise SystemExit("Existing ledger contains an invalid generation UUID") from exc
    try:
        generation = get_generation_with_retry(generation_id, key, deadline)
    except TimeoutError as exc:
        raise SystemExit("Resume GET deadline exceeded; keep the ID and reconcile later") from exc
else:
    try:
        claim_new_job()
    except FileExistsError as exc:
        raise SystemExit(
            "Another worker claimed this client job key; do not submit. Retry later to resume its ledger."
        ) from exc
    credit_before = safe_credit_balance(key)
    try:
        generation = api_request(
            "POST", "/generations/image", key, payload, expected_status=201, timeout=180
        )
    except (urllib.error.URLError, TimeoutError) as exc:
        save_job("ambiguous_create", note="Do not replay; reconcile recent generations and credits")
        raise SystemExit(
            "Ambiguous create outcome: do not replay; reconcile recent generations and credits"
        ) from exc
    generation_id = generation.get("id")
    try:
        uuid.UUID(str(generation_id))
    except (ValueError, AttributeError) as exc:
        save_job("invalid_create_response", note="Create returned no valid generation UUID")
        raise SystemExit("Create response lacked a valid generation UUID") from exc
    save_job("accepted", generation_id)

delay = 2.0
while True:
    state = generation.get("state")
    save_job(state, generation_id, generation.get("failure_reason"))
    if state == "completed":
        break
    if state == "failed":
        raise SystemExit(f"Luma generation failed: {generation.get('failure_reason')}")
    if state not in {"queued", "dreaming"}:
        raise SystemExit(f"Unexpected generation state: {state!r}")
    remaining = deadline - time.monotonic()
    if remaining <= 0:
        save_job(state, generation_id, note="Polling deadline exceeded; reconcile later")
        raise SystemExit("Polling deadline exceeded; keep the ID and reconcile later")
    try:
        generation = get_generation_with_retry(generation_id, key, deadline, initial_wait=delay)
        delay = min(delay * 1.6, 10.0)
    except TimeoutError as exc:
        save_job(state, generation_id, note="Polling/retry deadline exceeded; reconcile later")
        raise SystemExit("Polling deadline exceeded; keep the ID and reconcile later") from exc

asset_url = (generation.get("assets") or {}).get("image")
if not asset_url:
    raise SystemExit("Completed generation has no assets.image URL")
asset_hosts = host_set("LUMA_ASSET_HOSTS", {"storage.cdn-luma.com"})
artifact = download_image(asset_url, asset_hosts, job_dir, payload["format"])
credit_after = safe_credit_balance(key)
delta_cents = None
if credit_before is not None and credit_after is not None:
    delta_cents = credit_before - credit_after

manifest = {
    "generation_id": generation_id,
    "state": generation.get("state"),
    "model_requested": model,
    "model_reported": generation.get("model"),
    "client_job_key": client_job_key,
    "request_sha256": request_sha256,
    "prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
    "reference_url_sha256": [hashlib.sha256(url.encode("utf-8")).hexdigest() for url in reference_urls],
    "public_unit_price_snapshot_usd": public_price_snapshot,
    "credit_balance_before_cents": credit_before,
    "credit_balance_after_cents": credit_after,
    "observed_balance_delta_cents": delta_cents,
    "artifact": artifact,
    "review_status": "pending",
    "downstream_ai_notice": "required for every API user",
    "public_ai_disclosure_review": "required when output identifies or resembles a person",
}
atomic_write(manifest_path, json.dumps(manifest, indent=2).encode("utf-8"))
print(json.dumps(manifest, indent=2))