Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Live probes for the Gemini media/batch/tuning domain. Every call is logged by scripts/live.py.3Raw (sanitized, base64 stripped) responses -> tmp-live/gemini-media/*.json. Budget target <= $0.10."""4from __future__ import annotations56import asyncio7import base648import io9import json10import struct11import sys12import time13import wave14from pathlib import Path1516ROOT = Path(__file__).resolve().parents[2]17sys.path.insert(0, str(ROOT))18from scripts import live # noqa: E4021920OUT = ROOT / "tmp-live" / "gemini-media"21OUT.mkdir(parents=True, exist_ok=True)22G = live.gemini_request23SUMMARY: dict = {}242526def strip_b64(obj, keep_len=True):27 """Replace inlineData.data / data blobs with a length marker."""28 if isinstance(obj, dict):29 out = {}30 for k, v in obj.items():31 if k == "data" and isinstance(v, str) and len(v) > 200:32 out[k] = f"<{len(base64.b64decode(v + '=' * (-len(v) % 4)))} bytes, base64 len {len(v)}>"33 else:34 out[k] = strip_b64(v)35 return out36 if isinstance(obj, list):37 return [strip_b64(x) for x in obj]38 return obj394041def save(name, st, body, hdrs=None, extra=None):42 rec = {"status": st, "headers": live.interesting_headers(hdrs or {}), "body": strip_b64(body) if not isinstance(body, bytes) else f"<{len(body)} bytes>"}43 if extra:44 rec.update(extra)45 live.save_sanitized(rec, OUT / f"{name}.json")46 print(f"[{name}] HTTP {st}", flush=True)47 return rec484950def first_media_part(body):51 parts = body.get("candidates", [{}])[0].get("content", {}).get("parts", [])52 for p in parts:53 if "inlineData" in p:54 return p55 return None565758# ---------------------------------------------------------------- 1. batches (create first so they run while we probe)59t0 = time.time()60batch_body = {"batch": {"display_name": "atlas-media-probe", "input_config": {"requests": {"requests": [61 {"request": {"contents": [{"parts": [{"text": "Reply with OK."}]}], "generationConfig": {"maxOutputTokens": 8}}, "metadata": {"key": "r1"}},62 {"request": {"contents": [{"parts": [{"text": "Reply with OK."}]}], "generationConfig": {"maxOutputTokens": 8}}, "metadata": {"key": "r2"}},63]}}}}64st, b, h = G("POST", "/v1beta/models/gemini-3.5-flash-lite:batchGenerateContent", batch_body, est_cost_usd=0.0001, note="inline batch 2 reqs")65save("batch-create", st, b, h)66batch_name = b.get("name") if isinstance(b, dict) else None67SUMMARY["batch_name"] = batch_name68SUMMARY["batch_created_at"] = t06970emb_body = {"batch": {"display_name": "atlas-embed-probe", "input_config": {"requests": {"requests": [71 {"request": {"content": {"parts": [{"text": "OK"}]}, "embedContentConfig": {"outputDimensionality": 8}}, "metadata": {"key": "e1"}},72]}}}}73st, b, h = G("POST", "/v1beta/models/gemini-embedding-2:asyncBatchEmbedContent", emb_body, est_cost_usd=0.00001, note="inline embed batch 1 req")74save("batch-embed-create", st, b, h)75emb_name = b.get("name") if isinstance(b, dict) else None76SUMMARY["embed_batch_name"] = emb_name7778if batch_name:79 st, b, h = G("PATCH", f"/v1beta/{batch_name}:updateGenerateContentBatch?updateMask=priority", {"priority": "1"}, note="patch priority")80 save("batch-patch-priority", st, b, h)81st, b, h = G("GET", "/v1beta/batches?pageSize=5", note="list batches")82save("batch-list", st, b, h)8384# ---------------------------------------------------------------- 2. native image generation85img_body = {"contents": [{"parts": [{"text": "a plain white square"}]}],86 "generationConfig": {"responseModalities": ["TEXT", "IMAGE"], "imageConfig": {"aspectRatio": "1:1", "imageSize": "1K"}}}87t = time.time()88st, b, h = G("POST", "/v1beta/models/gemini-3.1-flash-lite-image:generateContent", img_body, timeout=180, est_cost_usd=0.034, note="native image 1K white square")89rec = save("image-generate", st, b, h, {"elapsed_s": round(time.time() - t, 2)})90if st == 200:91 p = first_media_part(b)92 if p:93 raw = base64.b64decode(p["inlineData"]["data"])94 ext = p["inlineData"]["mimeType"].split("/")[-1]95 (OUT / f"image-generate.{ext}").write_bytes(raw)96 SUMMARY["image"] = {"mime": p["inlineData"]["mimeType"], "bytes": len(raw), "usage": b.get("usageMetadata")}97st, b, h = G("POST", "/v1beta/models/gemini-3.1-flash-lite-image:generateContent",98 {"contents": [{"parts": [{"text": "a plain white square"}]}], "generationConfig": {"responseModalities": ["IMAGE"], "imageConfig": {"imageSize": "1k"}}},99 note="invalid imageSize lowercase")100save("image-invalid-imagesize", st, b, h)101st, b, h = G("POST", "/v1beta/models/gemini-3.1-flash-lite-image:generateContent",102 {"contents": [{"parts": [{"text": "x"}]}], "generationConfig": {"responseModalities": ["IMAGE"], "imageConfig": {"imageSize": "4K"}}},103 note="4K on lite (unsupported per docs)")104save("image-4k-on-lite", st, b, h)105st, b, h = G("POST", "/v1beta/models/gemini-3.5-flash-lite:generateContent",106 {"contents": [{"parts": [{"text": "Reply with OK."}]}], "generationConfig": {"maxOutputTokens": 8, "imageConfig": {"aspectRatio": "1:1"}}},107 note="imageConfig on text model")108save("image-config-on-text-model", st, b, h)109110# ---------------------------------------------------------------- 3. Imagen predict (shut down) + predict on Gemini model111for m in ["imagen-4.0-fast-generate-001", "imagen-4.0-generate-001", "gemini-3.5-flash-lite"]:112 st, b, h = G("POST", f"/v1beta/models/{m}:predict", {"instances": [{"prompt": "a plain white square"}], "parameters": {"sampleCount": 1}}, note=f"predict {m}")113 save(f"predict-{m}", st, b, h)114115# ---------------------------------------------------------------- 4. Veo validation probes + operations + generatedFiles116st, b, h = G("POST", "/v1beta/models/veo-3.1-lite-generate-preview:predictLongRunning", {}, note="empty body")117save("veo-empty-body", st, b, h)118st, b, h = G("POST", "/v1beta/models/veo-3.1-lite-generate-preview:predictLongRunning", {"instances": [{}]}, note="empty instance")119save("veo-empty-instance", st, b, h)120st, b, h = G("POST", "/v1beta/models/veo-9-generate:predictLongRunning", {"instances": [{"prompt": "x"}]}, note="bogus model")121save("veo-bogus-model", st, b, h)122st, b, h = G("GET", "/v1beta/models/veo-3.1-lite-generate-preview/operations", note="list model operations")123save("operations-list-veo", st, b, h)124st, b, h = G("GET", "/v1beta/models/veo-3.1-lite-generate-preview/operations/bogus-op", note="bogus op")125save("operations-get-bogus", st, b, h)126st, b, h = G("GET", "/v1beta/generatedFiles?pageSize=10", note="list generatedFiles")127save("generatedfiles-list", st, b, h)128st, b, h = G("GET", "/v1beta/generatedFiles/bogus/operations/bogus", note="bogus generatedFiles op")129save("generatedfiles-op-bogus", st, b, h)130st, b, h = G("GET", "/v1beta/batches/bogus-batch", note="bogus batch")131save("batch-get-bogus", st, b, h)132133# ---------------------------------------------------------------- 5. TTS134for m in ["gemini-2.5-flash-preview-tts", "gemini-3.1-flash-tts-preview"]:135 body = {"contents": [{"parts": [{"text": "Say: OK"}]}],136 "generationConfig": {"responseModalities": ["AUDIO"], "speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}}}}137 t = time.time()138 st, b, h = G("POST", f"/v1beta/models/{m}:generateContent", body, timeout=120, est_cost_usd=0.001, note=f"tts OK {m}")139 rec = save(f"tts-{m}", st, b, h, {"elapsed_s": round(time.time() - t, 2)})140 if st == 200:141 p = first_media_part(b)142 if p:143 raw = base64.b64decode(p["inlineData"]["data"])144 mime = p["inlineData"]["mimeType"]145 SUMMARY[f"tts_{m}"] = {"mime": mime, "bytes": len(raw), "usage": b.get("usageMetadata"), "seconds_if_pcm16_24k_mono": round(len(raw) / 48000, 2)}146 with wave.open(str(OUT / f"tts-{m}.wav"), "wb") as w:147 w.setnchannels(1); w.setsampwidth(2); w.setframerate(24000); w.writeframes(raw)148 break # one paid success is enough149# invalid voice (free)150st, b, h = G("POST", "/v1beta/models/gemini-3.1-flash-tts-preview:generateContent",151 {"contents": [{"parts": [{"text": "Say: OK"}]}], "generationConfig": {"responseModalities": ["AUDIO"], "speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "NotAVoice"}}}}},152 note="invalid voice")153save("tts-invalid-voice", st, b, h)154155# ---------------------------------------------------------------- 6. transcription (0.5 s silence WAV)156buf = io.BytesIO()157with wave.open(buf, "wb") as w:158 w.setnchannels(1); w.setsampwidth(2); w.setframerate(16000); w.writeframes(b"\x00\x00" * 8000)159wav_b64 = base64.b64encode(buf.getvalue()).decode()160tr_body = {"contents": [{"parts": [{"inlineData": {"mimeType": "audio/wav", "data": wav_b64}}]}]}161st, b, h = G("POST", "/v1beta/models/gemini-3.5-transcribe:generateContent", tr_body, est_cost_usd=0.0001, note="transcribe 0.5s silence")162save("transcribe-silence", st, b, h)163SUMMARY["transcribe"] = {"status": st, "usage": b.get("usageMetadata") if isinstance(b, dict) else None}164tr_body2 = dict(tr_body, generationConfig={"audioTranscriptionConfig": {"mode": "SMART", "languageCodes": ["en-US"]}})165st, b, h = G("POST", "/v1beta/models/gemini-3.5-transcribe:generateContent", tr_body2, est_cost_usd=0.0001, note="transcribe SMART + audioTranscriptionConfig")166save("transcribe-smart-config", st, b, h)167st, b, h = G("POST", "/v1beta/models/gemini-3.5-transcribe-live:generateContent", tr_body, note="transcribe-live via generateContent (expect error)")168save("transcribe-live-unary", st, b, h)169170# ---------------------------------------------------------------- 7. Lyria clip ($0.04) + Lyria RealTime websocket171t = time.time()172st, b, h = G("POST", "/v1beta/models/lyria-3-clip-preview:generateContent",173 {"contents": [{"parts": [{"text": "A short instrumental acoustic guitar piece."}]}]}, timeout=300, est_cost_usd=0.04, note="lyria clip 30s")174rec = save("lyria-clip", st, b, h, {"elapsed_s": round(time.time() - t, 2)})175if st == 200:176 p = first_media_part(b)177 if p:178 raw = base64.b64decode(p["inlineData"]["data"])179 mime = p["inlineData"]["mimeType"]180 (OUT / ("lyria-clip." + ("mp3" if "mp3" in mime or "mpeg" in mime else "bin"))).write_bytes(raw)181 SUMMARY["lyria_clip"] = {"mime": mime, "bytes": len(raw), "usage": b.get("usageMetadata"),182 "part_kinds": [list(pp.keys()) for pp in b["candidates"][0]["content"]["parts"]]}183184185async def lyria_rt():186 import os187 import websockets188 events = []189 for version in ["v1beta", "v1alpha"]:190 uri = f"wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.{version}.GenerativeService.BidiGenerateMusic"191 try:192 async with websockets.connect(uri, additional_headers={"x-goog-api-key": os.environ["GEMINI_API_KEY"]}, max_size=None) as ws:193 await ws.send(json.dumps({"setup": {"model": "models/lyria-realtime-exp"}}))194 msg = json.loads(await asyncio.wait_for(ws.recv(), 20))195 events.append({"dir": "server", "keys": list(msg.keys()), "msg": strip_b64(msg)})196 await ws.send(json.dumps({"clientContent": {"weightedPrompts": [{"text": "minimal techno", "weight": 1.0}]}}))197 await ws.send(json.dumps({"musicGenerationConfig": {"bpm": 120, "temperature": 1.0}}))198 await ws.send(json.dumps({"playbackControl": "PLAY"}))199 chunks = 0200 while chunks < 2:201 raw = await asyncio.wait_for(ws.recv(), 30)202 msg = json.loads(raw)203 ev = {"dir": "server", "keys": list(msg.keys())}204 sc = msg.get("serverContent")205 if sc and sc.get("audioChunks"):206 chunks += 1207 ac = sc["audioChunks"][0]208 d = ac.get("data", "")209 ev["audioChunk"] = {"mimeType": ac.get("mimeType"), "bytes": len(base64.b64decode(d + "=" * (-len(d) % 4))), "sourceMetadata": ac.get("sourceMetadata"), "n_chunks_in_msg": len(sc["audioChunks"])}210 else:211 ev["msg"] = strip_b64(msg)212 events.append(ev)213 await ws.send(json.dumps({"playbackControl": "STOP"}))214 live.log_request("gemini", "WS", f"/ws/google.ai.generativelanguage.{version}.GenerativeService.BidiGenerateMusic", 101, 0.0, "lyria-realtime-exp setup+prompt, 2 chunks, stop")215 return {"version": version, "uri": uri, "events": events}216 except Exception as e: # noqa: BLE001217 events.append({"version": version, "error": live.mask(repr(e))[:500]})218 live.log_request("gemini", "WS", f"/ws/google.ai.generativelanguage.{version}.GenerativeService.BidiGenerateMusic", 0, 0.0, f"error {type(e).__name__}")219 return {"events": events}220221rt = asyncio.run(lyria_rt())222live.save_sanitized(rt, OUT / "lyria-realtime.json")223print("[lyria-realtime]", json.dumps(strip_b64(rt))[:600], flush=True)224225# ---------------------------------------------------------------- 8. tunedModels226st, b, h = G("GET", "/v1beta/tunedModels?pageSize=5", note="list tunedModels")227save("tunedmodels-list", st, b, h)228st, b, h = G("GET", "/v1beta/tunedModels/bogus-model", note="get bogus tunedModel")229save("tunedmodels-get-bogus", st, b, h)230st, b, h = G("GET", "/v1beta/tunedModels/bogus-model/permissions", note="list permissions bogus")231save("tunedmodels-permissions-bogus", st, b, h)232st, b, h = G("GET", "/v1beta/tunedModels/bogus-model/operations/bogus", note="bogus tuned op")233save("tunedmodels-op-bogus", st, b, h)234235# ---------------------------------------------------------------- 9. poll batches (<= 5 min total from creation)236timeline = []237final = {}238deadline = SUMMARY["batch_created_at"] + 300239names = [n for n in [batch_name, emb_name] if n]240while names and time.time() < deadline:241 for n in list(names):242 st, b, h = G("GET", f"/v1beta/{n}", note="poll")243 state = (b.get("metadata") or {}).get("state") if isinstance(b, dict) else None244 timeline.append({"t": round(time.time() - SUMMARY["batch_created_at"], 1), "name": n, "status": st, "done": b.get("done") if isinstance(b, dict) else None, "state": state})245 print(f" poll {n} t={timeline[-1]['t']}s state={state} done={timeline[-1]['done']}", flush=True)246 if isinstance(b, dict) and b.get("done"):247 final[n] = save(f"batch-final-{n.split('/')[-1]}", st, b, h)248 names.remove(n)249 if names:250 time.sleep(20)251for n in names: # not done: cancel, record, then delete252 st, b, h = G("POST", f"/v1beta/{n}:cancel", note="cancel (timeout)")253 save(f"batch-cancel-{n.split('/')[-1]}", st, b, h)254 time.sleep(3)255 st, b, h = G("GET", f"/v1beta/{n}", note="after cancel")256 final[n] = save(f"batch-after-cancel-{n.split('/')[-1]}", st, b, h)257for n in [batch_name, emb_name]:258 if n:259 st, b, h = G("DELETE", f"/v1beta/{n}", note="delete batch")260 save(f"batch-delete-{n.split('/')[-1]}", st, b, h)261 st, b, h = G("GET", f"/v1beta/{n}", note="get after delete")262 save(f"batch-get-after-delete-{n.split('/')[-1]}", st, b, h)263SUMMARY["batch_timeline"] = timeline264live.save_sanitized(SUMMARY, OUT / "SUMMARY.json")265print(json.dumps(strip_b64(SUMMARY), indent=1, default=str)[:4000])266