Setting the file. One moment.
Image Gen · Imagegen · openai/skills · Skills Docs
Back to the top of the page def _validate_model
— line 129
This file
Number 40.6
Position 6 of 9
Type Python
Size 31 KB
Lines 926 scripts/ image_gen.py
Python · 926 lines · 31 KB
import
os
16 from pathlib import Path
17 import re
18 import sys
19 import time
20 from typing import Any, Dict, Iterable, List, Optional, Tuple
21
22 from io import BytesIO
23
24 DEFAULT_MODEL = "gpt-image-1.5"
25 DEFAULT_SIZE = "1024x1024"
26 DEFAULT_QUALITY = "auto"
27 DEFAULT_OUTPUT_FORMAT = "png"
28 DEFAULT_CONCURRENCY = 5
29 DEFAULT_DOWNSCALE_SUFFIX = "-web"
30 DEFAULT_OUTPUT_PATH = "output/imagegen/output.png"
31 GPT_IMAGE_MODEL_PREFIX = "gpt-image-"
32
33 ALLOWED_SIZES = { "1024x1024" , "1536x1024" , "1024x1536" , "auto" }
34 ALLOWED_QUALITIES = { "low" , "medium" , "high" , "auto" }
35 ALLOWED_BACKGROUNDS = { "transparent" , "opaque" , "auto" , None }
36 ALLOWED_INPUT_FIDELITIES = { "low" , "high" , None }
37
38 MAX_IMAGE_BYTES = 50 * 1024 * 1024
39 MAX_BATCH_JOBS = 500
40
41
42 def _die (message: str , code: int = 1 ) -> None :
43 print ( f "Error: { message } " , file = sys.stderr)
44 raise SystemExit (code)
45
46
47 def _warn (message: str ) -> None :
48 print ( f "Warning: { message } " , file = sys.stderr)
49
50
51 def _dependency_hint (package: str , * , upgrade: bool = False ) -> str :
52 command = f "uv pip install { '-U ' if upgrade else '' }{ package } "
53 return (
54 "Activate the repo-selected environment first, then install it with "
55 f "` { command } `. If this repo uses a local virtualenv, start with "
56 "`source .venv/bin/activate`; otherwise use this repo's configured shared fallback "
57 "environment. If your project declares dependencies, prefer that project's normal "
58 "`uv sync` flow."
59 )
60
61
62 def _ensure_api_key (dry_run: bool ) -> None :
63 if os.getenv( "OPENAI_API_KEY" ):
64 print ( "OPENAI_API_KEY is set." , file = sys.stderr)
65 return
66 if dry_run:
67 _warn( "OPENAI_API_KEY is not set; dry-run only." )
68 return
69 _die( "OPENAI_API_KEY is not set. Export it before running." )
70
71
72 def _read_prompt (prompt: Optional[ str ], prompt_file: Optional[ str ]) -> str :
73 if prompt and prompt_file:
74 _die( "Use --prompt or --prompt-file, not both." )
75 if prompt_file:
76 path = Path(prompt_file)
77 if not path.exists():
78 _die( f "Prompt file not found: { path } " )
79 return path.read_text( encoding = "utf-8" ).strip()
80 if prompt:
81 return prompt.strip()
82 _die( "Missing prompt. Use --prompt or --prompt-file." )
83 return "" # unreachable
84
85
86 def _check_image_paths (paths: Iterable[ str ]) -> List[Path]:
87 resolved: List[Path] = []
88 for raw in paths:
89 path = Path(raw)
90 if not path.exists():
91 _die( f "Image file not found: { path } " )
92 if path.stat().st_size > MAX_IMAGE_BYTES :
93 _warn( f "Image exceeds 50MB limit: { path } " )
94 resolved.append(path)
95 return resolved
96
97
98 def _normalize_output_format (fmt: Optional[ str ]) -> str :
99 if not fmt:
100 return DEFAULT_OUTPUT_FORMAT
101 fmt = fmt.lower()
102 if fmt not in { "png" , "jpeg" , "jpg" , "webp" }:
103 _die( "output-format must be png, jpeg, jpg, or webp." )
104 return "jpeg" if fmt == "jpg" else fmt
105
106
107 def _validate_size (size: str ) -> None :
108 if size not in ALLOWED_SIZES :
109 _die(
110 "size must be one of 1024x1024, 1536x1024, 1024x1536, or auto for GPT image models."
111 )
112
113
114 def _validate_quality (quality: str ) -> None :
115 if quality not in ALLOWED_QUALITIES :
116 _die( "quality must be one of low, medium, high, or auto." )
117
118
119 def _validate_background (background: Optional[ str ]) -> None :
120 if background not in ALLOWED_BACKGROUNDS :
121 _die( "background must be one of transparent, opaque, or auto." )
122
123
124 def _validate_input_fidelity (input_fidelity: Optional[ str ]) -> None :
125 if input_fidelity not in ALLOWED_INPUT_FIDELITIES :
126 _die( "input-fidelity must be one of low or high." )
127
128
129 def _validate_model (model: str ) -> None :
130 if not model.startswith( GPT_IMAGE_MODEL_PREFIX ):
131 _die(
132 "model must be a GPT Image model (for example gpt-image-1.5, gpt-image-1, or gpt-image-1-mini)."
133 )
134
135
136 def _validate_transparency (background: Optional[ str ], output_format: str ) -> None :
137 if background == "transparent" and output_format not in { "png" , "webp" }:
138 _die( "transparent background requires output-format png or webp." )
139
140
141 def _validate_generate_payload (payload: Dict[ str , Any]) -> None :
142 _validate_model( str (payload.get( "model" , DEFAULT_MODEL )))
143 n = int (payload.get( "n" , 1 ))
144 if n < 1 or n > 10 :
145 _die( "n must be between 1 and 10" )
146 size = str (payload.get( "size" , DEFAULT_SIZE ))
147 quality = str (payload.get( "quality" , DEFAULT_QUALITY ))
148 background = payload.get( "background" )
149 _validate_size(size)
150 _validate_quality(quality)
151 _validate_background(background)
152 oc = payload.get( "output_compression" )
153 if oc is not None and not ( 0 <= int (oc) <= 100 ):
154 _die( "output_compression must be between 0 and 100" )
155
156
157 def _build_output_paths (
158 out: str ,
159 output_format: str ,
160 count: int ,
161 out_dir: Optional[ str ],
162 ) -> List[Path]:
163 ext = "." + output_format
164
165 if out_dir:
166 out_base = Path(out_dir)
167 out_base.mkdir( parents = True , exist_ok = True )
168 return [out_base / f "image_ { i }{ ext } " for i in range ( 1 , count + 1 )]
169
170 out_path = Path(out)
171 if out_path.exists() and out_path.is_dir():
172 out_path.mkdir( parents = True , exist_ok = True )
173 return [out_path / f "image_ { i }{ ext } " for i in range ( 1 , count + 1 )]
174
175 if out_path.suffix == "" :
176 out_path = out_path.with_suffix(ext)
177 elif output_format and out_path.suffix.lstrip( "." ).lower() != output_format:
178 _warn(
179 f "Output extension { out_path.suffix } does not match output-format { output_format } ."
180 )
181
182 if count == 1 :
183 return [out_path]
184
185 return [
186 out_path.with_name( f " { out_path.stem } - { i }{ out_path.suffix } " )
187 for i in range ( 1 , count + 1 )
188 ]
189
190
191 def _augment_prompt (args: argparse.Namespace, prompt: str ) -> str :
192 fields = _fields_from_args(args)
193 return _augment_prompt_fields(args.augment, prompt, fields)
194
195
196 def _augment_prompt_fields (augment: bool , prompt: str , fields: Dict[ str , Optional[ str ]]) -> str :
197 if not augment:
198 return prompt
199
200 sections: List[ str ] = []
201 if fields.get( "use_case" ):
202 sections.append( f "Use case: { fields[ 'use_case' ] } " )
203 sections.append( f "Primary request: { prompt } " )
204 if fields.get( "scene" ):
205 sections.append( f "Scene/background: { fields[ 'scene' ] } " )
206 if fields.get( "subject" ):
207 sections.append( f "Subject: { fields[ 'subject' ] } " )
208 if fields.get( "style" ):
209 sections.append( f "Style/medium: { fields[ 'style' ] } " )
210 if fields.get( "composition" ):
211 sections.append( f "Composition/framing: { fields[ 'composition' ] } " )
212 if fields.get( "lighting" ):
213 sections.append( f "Lighting/mood: { fields[ 'lighting' ] } " )
214 if fields.get( "palette" ):
215 sections.append( f "Color palette: { fields[ 'palette' ] } " )
216 if fields.get( "materials" ):
217 sections.append( f "Materials/textures: { fields[ 'materials' ] } " )
218 if fields.get( "text" ):
219 sections.append( f "Text (verbatim): \"{ fields[ 'text' ] }\" " )
220 if fields.get( "constraints" ):
221 sections.append( f "Constraints: { fields[ 'constraints' ] } " )
222 if fields.get( "negative" ):
223 sections.append( f "Avoid: { fields[ 'negative' ] } " )
224
225 return " \n " .join(sections)
226
227
228 def _fields_from_args (args: argparse.Namespace) -> Dict[ str , Optional[ str ]]:
229 return {
230 "use_case" : getattr (args, "use_case" , None ),
231 "scene" : getattr (args, "scene" , None ),
232 "subject" : getattr (args, "subject" , None ),
233 "style" : getattr (args, "style" , None ),
234 "composition" : getattr (args, "composition" , None ),
235 "lighting" : getattr (args, "lighting" , None ),
236 "palette" : getattr (args, "palette" , None ),
237 "materials" : getattr (args, "materials" , None ),
238 "text" : getattr (args, "text" , None ),
239 "constraints" : getattr (args, "constraints" , None ),
240 "negative" : getattr (args, "negative" , None ),
241 }
242
243
244 def _print_request (payload: dict ) -> None :
245 print (json.dumps(payload, indent = 2 , sort_keys = True ))
246
247
248 def _decode_and_write (images: List[ str ], outputs: List[Path], force: bool ) -> None :
249 for idx, image_b64 in enumerate (images):
250 if idx >= len (outputs):
251 break
252 out_path = outputs[idx]
253 if out_path.exists() and not force:
254 _die( f "Output already exists: { out_path } (use --force to overwrite)" )
255 out_path.parent.mkdir( parents = True , exist_ok = True )
256 out_path.write_bytes(base64.b64decode(image_b64))
257 print ( f "Wrote { out_path } " )
258
259
260 def _derive_downscale_path (path: Path, suffix: str ) -> Path:
261 if suffix and not suffix.startswith( "-" ) and not suffix.startswith( "_" ):
262 suffix = "-" + suffix
263 return path.with_name( f " { path.stem }{ suffix }{ path.suffix } " )
264
265
266 def _downscale_image_bytes (image_bytes: bytes , * , max_dim: int , output_format: str ) -> bytes :
267 try :
268 from PIL import Image
269 except Exception :
270 _die( f "Downscaling requires Pillow. { _dependency_hint( 'pillow' ) } " )
271
272 if max_dim < 1 :
273 _die( "--downscale-max-dim must be >= 1" )
274
275 with Image.open(BytesIO(image_bytes)) as img:
276 img.load()
277 w, h = img.size
278 scale = min ( 1.0 , float (max_dim) / float ( max (w, h)))
279 target = ( max ( 1 , int ( round (w * scale))), max ( 1 , int ( round (h * scale))))
280
281 resized = img if target == (w, h) else img.resize(target, Image.Resampling. LANCZOS )
282
283 fmt = output_format.lower()
284 if fmt == "jpg" :
285 fmt = "jpeg"
286
287 if fmt == "jpeg" :
288 if resized.mode in ( "RGBA" , "LA" ) or ( "transparency" in getattr (resized, "info" , {})):
289 bg = Image.new( "RGB" , resized.size, ( 255 , 255 , 255 ))
290 bg.paste(resized.convert( "RGBA" ), mask = resized.convert( "RGBA" ).split()[ - 1 ])
291 resized = bg
292 else :
293 resized = resized.convert( "RGB" )
294
295 out = BytesIO()
296 resized.save(out, format = fmt.upper())
297 return out.getvalue()
298
299
300 def _decode_write_and_downscale (
301 images: List[ str ],
302 outputs: List[Path],
303 * ,
304 force: bool ,
305 downscale_max_dim: Optional[ int ],
306 downscale_suffix: str ,
307 output_format: str ,
308 ) -> None :
309 for idx, image_b64 in enumerate (images):
310 if idx >= len (outputs):
311 break
312 out_path = outputs[idx]
313 if out_path.exists() and not force:
314 _die( f "Output already exists: { out_path } (use --force to overwrite)" )
315 out_path.parent.mkdir( parents = True , exist_ok = True )
316
317 raw = base64.b64decode(image_b64)
318 out_path.write_bytes(raw)
319 print ( f "Wrote { out_path } " )
320
321 if downscale_max_dim is None :
322 continue
323
324 derived = _derive_downscale_path(out_path, downscale_suffix)
325 if derived.exists() and not force:
326 _die( f "Output already exists: { derived } (use --force to overwrite)" )
327 derived.parent.mkdir( parents = True , exist_ok = True )
328 resized = _downscale_image_bytes(raw, max_dim = downscale_max_dim, output_format = output_format)
329 derived.write_bytes(resized)
330 print ( f "Wrote { derived } " )
331
332
333 def _create_client ():
334 try :
335 from openai import OpenAI
336 except ImportError :
337 _die( f "openai SDK not installed in the active environment. { _dependency_hint( 'openai' ) } " )
338 return OpenAI()
339
340
341 def _create_async_client ():
342 try :
343 from openai import AsyncOpenAI
344 except ImportError :
345 try :
346 import openai as _openai # noqa: F401
347 except ImportError :
348 _die(
349 f "openai SDK not installed in the active environment. { _dependency_hint( 'openai' ) } "
350 )
351 _die(
352 "AsyncOpenAI not available in this openai SDK version. "
353 f " { _dependency_hint( 'openai' , upgrade = True ) } "
354 )
355 return AsyncOpenAI()
356
357
358 def _slugify (value: str ) -> str :
359 value = value.strip().lower()
360 value = re.sub( r " [ ^a-z0-9 ] + " , "-" , value)
361 value = re.sub( r "- {2,} " , "-" , value).strip( "-" )
362 return value[: 60 ] if value else "job"
363
364
365 def _normalize_job (job: Any, idx: int ) -> Dict[ str , Any]:
366 if isinstance (job, str ):
367 prompt = job.strip()
368 if not prompt:
369 _die( f "Empty prompt at job { idx } " )
370 return { "prompt" : prompt}
371 if isinstance (job, dict ):
372 if "prompt" not in job or not str (job[ "prompt" ]).strip():
373 _die( f "Missing prompt for job { idx } " )
374 return job
375 _die( f "Invalid job at index { idx } : expected string or object." )
376 return {} # unreachable
377
378
379 def _read_jobs_jsonl (path: str ) -> List[Dict[ str , Any]]:
380 p = Path(path)
381 if not p.exists():
382 _die( f "Input file not found: { p } " )
383 jobs: List[Dict[ str , Any]] = []
384 for line_no, raw in enumerate (p.read_text( encoding = "utf-8" ).splitlines(), start = 1 ):
385 line = raw.strip()
386 if not line or line.startswith( "#" ):
387 continue
388 try :
389 item: Any
390 if line.startswith( "{" ):
391 item = json.loads(line)
392 else :
393 item = line
394 jobs.append(_normalize_job(item, idx = line_no))
395 except json.JSONDecodeError as exc:
396 _die( f "Invalid JSON on line { line_no } : { exc } " )
397 if not jobs:
398 _die( "No jobs found in input file." )
399 if len (jobs) > MAX_BATCH_JOBS :
400 _die( f "Too many jobs ( { len (jobs) } ). Max is { MAX_BATCH_JOBS } ." )
401 return jobs
402
403
404 def _merge_non_null (dst: Dict[ str , Any], src: Dict[ str , Any]) -> Dict[ str , Any]:
405 merged = dict (dst)
406 for k, v in src.items():
407 if v is not None :
408 merged[k] = v
409 return merged
410
411
412 def _job_output_paths (
413 * ,
414 out_dir: Path,
415 output_format: str ,
416 idx: int ,
417 prompt: str ,
418 n: int ,
419 explicit_out: Optional[ str ],
420 ) -> List[Path]:
421 out_dir.mkdir( parents = True , exist_ok = True )
422 ext = "." + output_format
423
424 if explicit_out:
425 base = Path(explicit_out)
426 if base.suffix == "" :
427 base = base.with_suffix(ext)
428 elif base.suffix.lstrip( "." ).lower() != output_format:
429 _warn(
430 f "Job { idx } : output extension { base.suffix } does not match output-format { output_format } ."
431 )
432 base = out_dir / base.name
433 else :
434 slug = _slugify(prompt[: 80 ])
435 base = out_dir / f " { idx :03d} - { slug }{ ext } "
436
437 if n == 1 :
438 return [base]
439 return [
440 base.with_name( f " { base.stem } - { i }{ base.suffix } " )
441 for i in range ( 1 , n + 1 )
442 ]
443
444
445 def _extract_retry_after_seconds (exc: Exception ) -> Optional[ float ]:
446 # Best-effort: openai SDK errors vary by version. Prefer a conservative fallback.
447 for attr in ( "retry_after" , "retry_after_seconds" ):
448 val = getattr (exc, attr, None )
449 if isinstance (val, ( int , float )) and val >= 0 :
450 return float (val)
451 msg = str (exc)
452 m = re.search( r "retry [ - ] after [ := ] + ([ 0-9 ] + (?: \\ .[ 0-9 ] + ) ? ) " , msg, re. IGNORECASE )
453 if m:
454 try :
455 return float (m.group( 1 ))
456 except Exception :
457 return None
458 return None
459
460
461 def _is_rate_limit_error (exc: Exception ) -> bool :
462 name = exc. __class__ . __name__ .lower()
463 if "ratelimit" in name or "rate_limit" in name:
464 return True
465 msg = str (exc).lower()
466 return "429" in msg or "rate limit" in msg or "too many requests" in msg
467
468
469 def _is_transient_error (exc: Exception ) -> bool :
470 if _is_rate_limit_error(exc):
471 return True
472 name = exc. __class__ . __name__ .lower()
473 if "timeout" in name or "timedout" in name or "tempor" in name:
474 return True
475 msg = str (exc).lower()
476 return "timeout" in msg or "timed out" in msg or "connection reset" in msg
477
478
479 async def _generate_one_with_retries (
480 client: Any,
481 payload: Dict[ str , Any],
482 * ,
483 attempts: int ,
484 job_label: str ,
485 ) -> Any:
486 last_exc: Optional[ Exception ] = None
487 for attempt in range ( 1 , attempts + 1 ):
488 try :
489 return await client.images.generate( ** payload)
490 except Exception as exc:
491 last_exc = exc
492 if not _is_transient_error(exc):
493 raise
494 if attempt == attempts:
495 raise
496 sleep_s = _extract_retry_after_seconds(exc)
497 if sleep_s is None :
498 sleep_s = min ( 60.0 , 2.0 ** attempt)
499 print (
500 f " { job_label } attempt { attempt } / { attempts } failed ( { exc. __class__ . __name__ } ); retrying in { sleep_s :.1f} s" ,
501 file = sys.stderr,
502 )
503 await asyncio.sleep(sleep_s)
504 raise last_exc or RuntimeError ( "unknown error" )
505
506
507 async def _run_generate_batch (args: argparse.Namespace) -> int :
508 jobs = _read_jobs_jsonl(args.input)
509 out_dir = Path(args.out_dir)
510
511 base_fields = _fields_from_args(args)
512 base_payload = {
513 "model" : args.model,
514 "n" : args.n,
515 "size" : args.size,
516 "quality" : args.quality,
517 "background" : args.background,
518 "output_format" : args.output_format,
519 "output_compression" : args.output_compression,
520 "moderation" : args.moderation,
521 }
522
523 if args.dry_run:
524 for i, job in enumerate (jobs, start = 1 ):
525 prompt = str (job[ "prompt" ]).strip()
526 fields = _merge_non_null(base_fields, job.get( "fields" , {}))
527 # Allow flat job keys as well (use_case, scene, etc.)
528 fields = _merge_non_null(fields, {k: job.get(k) for k in base_fields.keys()})
529 augmented = _augment_prompt_fields(args.augment, prompt, fields)
530
531 job_payload = dict (base_payload)
532 job_payload[ "prompt" ] = augmented
533 job_payload = _merge_non_null(job_payload, {k: job.get(k) for k in base_payload.keys()})
534 job_payload = {k: v for k, v in job_payload.items() if v is not None }
535
536 _validate_generate_payload(job_payload)
537 effective_output_format = _normalize_output_format(job_payload.get( "output_format" ))
538 _validate_transparency(job_payload.get( "background" ), effective_output_format)
539 job_payload[ "output_format" ] = effective_output_format
540
541 n = int (job_payload.get( "n" , 1 ))
542 outputs = _job_output_paths(
543 out_dir = out_dir,
544 output_format = effective_output_format,
545 idx = i,
546 prompt = prompt,
547 n = n,
548 explicit_out = job.get( "out" ),
549 )
550 downscaled = None
551 if args.downscale_max_dim is not None :
552 downscaled = [
553 str (_derive_downscale_path(p, args.downscale_suffix)) for p in outputs
554 ]
555 _print_request(
556 {
557 "endpoint" : "/v1/images/generations" ,
558 "job" : i,
559 "outputs" : [ str (p) for p in outputs],
560 "outputs_downscaled" : downscaled,
561 ** job_payload,
562 }
563 )
564 return 0
565
566 client = _create_async_client()
567 sem = asyncio.Semaphore(args.concurrency)
568
569 any_failed = False
570
571 async def run_job (i: int , job: Dict[ str , Any]) -> Tuple[ int , Optional[ str ]]:
572 nonlocal any_failed
573 prompt = str (job[ "prompt" ]).strip()
574 job_label = f "[job { i } / { len (jobs) } ]"
575
576 fields = _merge_non_null(base_fields, job.get( "fields" , {}))
577 fields = _merge_non_null(fields, {k: job.get(k) for k in base_fields.keys()})
578 augmented = _augment_prompt_fields(args.augment, prompt, fields)
579
580 payload = dict (base_payload)
581 payload[ "prompt" ] = augmented
582 payload = _merge_non_null(payload, {k: job.get(k) for k in base_payload.keys()})
583 payload = {k: v for k, v in payload.items() if v is not None }
584
585 n = int (payload.get( "n" , 1 ))
586 _validate_generate_payload(payload)
587 effective_output_format = _normalize_output_format(payload.get( "output_format" ))
588 _validate_transparency(payload.get( "background" ), effective_output_format)
589 payload[ "output_format" ] = effective_output_format
590 outputs = _job_output_paths(
591 out_dir = out_dir,
592 output_format = effective_output_format,
593 idx = i,
594 prompt = prompt,
595 n = n,
596 explicit_out = job.get( "out" ),
597 )
598 try :
599 async with sem:
600 print ( f " { job_label } starting" , file = sys.stderr)
601 started = time.time()
602 result = await _generate_one_with_retries(
603 client,
604 payload,
605 attempts = args.max_attempts,
606 job_label = job_label,
607 )
608 elapsed = time.time() - started
609 print ( f " { job_label } completed in { elapsed :.1f} s" , file = sys.stderr)
610 images = [item.b64_json for item in result.data]
611 _decode_write_and_downscale(
612 images,
613 outputs,
614 force = args.force,
615 downscale_max_dim = args.downscale_max_dim,
616 downscale_suffix = args.downscale_suffix,
617 output_format = effective_output_format,
618 )
619 return i, None
620 except Exception as exc:
621 any_failed = True
622 print ( f " { job_label } failed: { exc } " , file = sys.stderr)
623 if args.fail_fast:
624 raise
625 return i, str (exc)
626
627 tasks = [asyncio.create_task(run_job(i, job)) for i, job in enumerate (jobs, start = 1 )]
628
629 try :
630 await asyncio.gather( * tasks)
631 except Exception :
632 for t in tasks:
633 if not t.done():
634 t.cancel()
635 raise
636
637 return 1 if any_failed else 0
638
639
640 def _generate_batch (args: argparse.Namespace) -> None :
641 exit_code = asyncio.run(_run_generate_batch(args))
642 if exit_code:
643 raise SystemExit (exit_code)
644
645
646 def _generate (args: argparse.Namespace) -> None :
647 prompt = _read_prompt(args.prompt, args.prompt_file)
648 prompt = _augment_prompt(args, prompt)
649
650 payload = {
651 "model" : args.model,
652 "prompt" : prompt,
653 "n" : args.n,
654 "size" : args.size,
655 "quality" : args.quality,
656 "background" : args.background,
657 "output_format" : args.output_format,
658 "output_compression" : args.output_compression,
659 "moderation" : args.moderation,
660 }
661 payload = {k: v for k, v in payload.items() if v is not None }
662
663 output_format = _normalize_output_format(args.output_format)
664 _validate_transparency(args.background, output_format)
665 payload[ "output_format" ] = output_format
666 output_paths = _build_output_paths(args.out, output_format, args.n, args.out_dir)
667 downscaled = None
668 if args.downscale_max_dim is not None :
669 downscaled = [ str (_derive_downscale_path(p, args.downscale_suffix)) for p in output_paths]
670
671 if args.dry_run:
672 _print_request(
673 {
674 "endpoint" : "/v1/images/generations" ,
675 "outputs" : [ str (p) for p in output_paths],
676 "outputs_downscaled" : downscaled,
677 ** payload,
678 }
679 )
680 return
681
682 print (
683 "Calling Image API (generation). This can take up to a couple of minutes." ,
684 file = sys.stderr,
685 )
686 started = time.time()
687 client = _create_client()
688 result = client.images.generate( ** payload)
689 elapsed = time.time() - started
690 print ( f "Generation completed in { elapsed :.1f} s." , file = sys.stderr)
691
692 images = [item.b64_json for item in result.data]
693 _decode_write_and_downscale(
694 images,
695 output_paths,
696 force = args.force,
697 downscale_max_dim = args.downscale_max_dim,
698 downscale_suffix = args.downscale_suffix,
699 output_format = output_format,
700 )
701
702
703 def _edit (args: argparse.Namespace) -> None :
704 prompt = _read_prompt(args.prompt, args.prompt_file)
705 prompt = _augment_prompt(args, prompt)
706
707 image_paths = _check_image_paths(args.image)
708 mask_path = Path(args.mask) if args.mask else None
709 if mask_path:
710 if not mask_path.exists():
711 _die( f "Mask file not found: { mask_path } " )
712 if mask_path.suffix.lower() != ".png" :
713 _warn( f "Mask should be a PNG with an alpha channel: { mask_path } " )
714 if mask_path.stat().st_size > MAX_IMAGE_BYTES :
715 _warn( f "Mask exceeds 50MB limit: { mask_path } " )
716
717 payload = {
718 "model" : args.model,
719 "prompt" : prompt,
720 "n" : args.n,
721 "size" : args.size,
722 "quality" : args.quality,
723 "background" : args.background,
724 "output_format" : args.output_format,
725 "output_compression" : args.output_compression,
726 "input_fidelity" : args.input_fidelity,
727 "moderation" : args.moderation,
728 }
729 payload = {k: v for k, v in payload.items() if v is not None }
730
731 output_format = _normalize_output_format(args.output_format)
732 _validate_transparency(args.background, output_format)
733 payload[ "output_format" ] = output_format
734 _validate_input_fidelity(args.input_fidelity)
735 output_paths = _build_output_paths(args.out, output_format, args.n, args.out_dir)
736 downscaled = None
737 if args.downscale_max_dim is not None :
738 downscaled = [ str (_derive_downscale_path(p, args.downscale_suffix)) for p in output_paths]
739
740 if args.dry_run:
741 payload_preview = dict (payload)
742 payload_preview[ "image" ] = [ str (p) for p in image_paths]
743 if mask_path:
744 payload_preview[ "mask" ] = str (mask_path)
745 _print_request(
746 {
747 "endpoint" : "/v1/images/edits" ,
748 "outputs" : [ str (p) for p in output_paths],
749 "outputs_downscaled" : downscaled,
750 ** payload_preview,
751 }
752 )
753 return
754
755 print (
756 f "Calling Image API (edit) with { len (image_paths) } image(s)." ,
757 file = sys.stderr,
758 )
759 started = time.time()
760 client = _create_client()
761
762 with _open_files(image_paths) as image_files, _open_mask(mask_path) as mask_file:
763 request = dict (payload)
764 request[ "image" ] = image_files if len (image_files) > 1 else image_files[ 0 ]
765 if mask_file is not None :
766 request[ "mask" ] = mask_file
767 result = client.images.edit( ** request)
768
769 elapsed = time.time() - started
770 print ( f "Edit completed in { elapsed :.1f} s." , file = sys.stderr)
771 images = [item.b64_json for item in result.data]
772 _decode_write_and_downscale(
773 images,
774 output_paths,
775 force = args.force,
776 downscale_max_dim = args.downscale_max_dim,
777 downscale_suffix = args.downscale_suffix,
778 output_format = output_format,
779 )
780
781
782 def _open_files (paths: List[Path]):
783 return _FileBundle(paths)
784
785
786 def _open_mask (mask_path: Optional[Path]):
787 if mask_path is None :
788 return _NullContext()
789 return _SingleFile(mask_path)
790
791
792 class _NullContext :
793 def __enter__ (self):
794 return None
795
796 def __exit__ (self, exc_type, exc, tb):
797 return False
798
799
800 class _SingleFile :
801 def __init__ (self, path: Path):
802 self ._path = path
803 self ._handle = None
804
805 def __enter__ (self):
806 self ._handle = self ._path.open( "rb" )
807 return self ._handle
808
809 def __exit__ (self, exc_type, exc, tb):
810 if self ._handle:
811 try :
812 self ._handle.close()
813 except Exception :
814 pass
815 return False
816
817
818 class _FileBundle :
819 def __init__ (self, paths: List[Path]):
820 self ._paths = paths
821 self ._handles: List[ object ] = []
822
823 def __enter__ (self):
824 self ._handles = [p.open( "rb" ) for p in self ._paths]
825 return self ._handles
826
827 def __exit__ (self, exc_type, exc, tb):
828 for handle in self ._handles:
829 try :
830 handle.close()
831 except Exception :
832 pass
833 return False
834
835
836 def _add_shared_args (parser: argparse.ArgumentParser) -> None :
837 parser.add_argument( "--model" , default = DEFAULT_MODEL )
838 parser.add_argument( "--prompt" )
839 parser.add_argument( "--prompt-file" )
840 parser.add_argument( "--n" , type = int , default = 1 )
841 parser.add_argument( "--size" , default = DEFAULT_SIZE )
842 parser.add_argument( "--quality" , default = DEFAULT_QUALITY )
843 parser.add_argument( "--background" )
844 parser.add_argument( "--output-format" )
845 parser.add_argument( "--output-compression" , type = int )
846 parser.add_argument( "--moderation" )
847 parser.add_argument( "--out" , default = DEFAULT_OUTPUT_PATH )
848 parser.add_argument( "--out-dir" )
849 parser.add_argument( "--force" , action = "store_true" )
850 parser.add_argument( "--dry-run" , action = "store_true" )
851 parser.add_argument( "--augment" , dest = "augment" , action = "store_true" )
852 parser.add_argument( "--no-augment" , dest = "augment" , action = "store_false" )
853 parser.set_defaults( augment = True )
854
855 # Prompt augmentation hints
856 parser.add_argument( "--use-case" )
857 parser.add_argument( "--scene" )
858 parser.add_argument( "--subject" )
859 parser.add_argument( "--style" )
860 parser.add_argument( "--composition" )
861 parser.add_argument( "--lighting" )
862 parser.add_argument( "--palette" )
863 parser.add_argument( "--materials" )
864 parser.add_argument( "--text" )
865 parser.add_argument( "--constraints" )
866 parser.add_argument( "--negative" )
867
868 # Post-processing (optional): generate an additional downscaled copy for fast web loading.
869 parser.add_argument( "--downscale-max-dim" , type = int )
870 parser.add_argument( "--downscale-suffix" , default = DEFAULT_DOWNSCALE_SUFFIX )
871
872
873 def main () -> int :
874 parser = argparse.ArgumentParser(
875 description = "Fallback CLI for explicit image generation or editing via GPT Image models"
876 )
877 subparsers = parser.add_subparsers( dest = "command" , required = True )
878
879 gen_parser = subparsers.add_parser( "generate" , help = "Create a new image" )
880 _add_shared_args(gen_parser)
881 gen_parser.set_defaults( func = _generate)
882
883 batch_parser = subparsers.add_parser(
884 "generate-batch" ,
885 help = "Generate multiple prompts concurrently (JSONL input)" ,
886 )
887 _add_shared_args(batch_parser)
888 batch_parser.add_argument( "--input" , required = True , help = "Path to JSONL file (one job per line)" )
889 batch_parser.add_argument( "--concurrency" , type = int , default = DEFAULT_CONCURRENCY )
890 batch_parser.add_argument( "--max-attempts" , type = int , default = 3 )
891 batch_parser.add_argument( "--fail-fast" , action = "store_true" )
892 batch_parser.set_defaults( func = _generate_batch)
893
894 edit_parser = subparsers.add_parser( "edit" , help = "Edit an existing image" )
895 _add_shared_args(edit_parser)
896 edit_parser.add_argument( "--image" , action = "append" , required = True )
897 edit_parser.add_argument( "--mask" )
898 edit_parser.add_argument( "--input-fidelity" )
899 edit_parser.set_defaults( func = _edit)
900
901 args = parser.parse_args()
902 if args.n < 1 or args.n > 10 :
903 _die( "--n must be between 1 and 10" )
904 if getattr (args, "concurrency" , 1 ) < 1 or getattr (args, "concurrency" , 1 ) > 25 :
905 _die( "--concurrency must be between 1 and 25" )
906 if getattr (args, "max_attempts" , 3 ) < 1 or getattr (args, "max_attempts" , 3 ) > 10 :
907 _die( "--max-attempts must be between 1 and 10" )
908 if args.output_compression is not None and not ( 0 <= args.output_compression <= 100 ):
909 _die( "--output-compression must be between 0 and 100" )
910 if args.command == "generate-batch" and not args.out_dir:
911 _die( "generate-batch requires --out-dir" )
912 if getattr (args, "downscale_max_dim" , None ) is not None and args.downscale_max_dim < 1 :
913 _die( "--downscale-max-dim must be >= 1" )
914
915 _validate_size(args.size)
916 _validate_quality(args.quality)
917 _validate_background(args.background)
918 _validate_model(args.model)
919 _ensure_api_key(args.dry_run)
920
921 args.func(args)
922 return 0
923
924
925 if __name__ == "__main__" :
926 raise SystemExit (main())
