Sora API - Generation Video Cloud

Module : 04-Applications
Niveau : Applications
Technologies : OpenAI API (Sora), httpx, requests
Duree estimee : 40 minutes
VRAM : 0 (API cloud uniquement)

Objectifs d’Apprentissage

Prerequis

  • Python 3.10+
  • Notebooks 01-1 a 03-3 (fondations, techniques avancees, orchestration)
  • Cle API OpenAI avec acces Sora (OPENAI_API_KEY)
  • Packages : openai, httpx, Pillow, matplotlib

Note : L’API Sora peut ne pas etre publiquement disponible au moment de l’exécution. Le code est structure avec try/except et reponses de demonstration pour permettre l’apprentissage même sans acces a l’API.

Navigation : << 04-2 Workflows Creatifs | 04-4 Pipeline Production >>

# Parametres Papermill - JAMAIS modifier ce commentaire

# Configuration notebook
notebook_mode = "interactive"        # "interactive" ou "batch"
skip_widgets = False               # True pour mode batch MCP
debug_level = "INFO"

# Parametres Sora API
# Modeles Sora valides (API OpenAI, verifies 2026-06) : sora-2, sora-2-pro.
# L'ancien identifiant "sora" a ete renomme en "sora-2" et n'est plus accepte
# par l'endpoint /v1/videos (erreur model_not_found).
sora_model = "sora-2"              # Modele Sora a utiliser (sora-2 ou sora-2-pro)
video_resolution = "720p"          # Resolution souhaitee (720p par defaut pour cout rais.)
video_duration_target = 4          # Duree cible en secondes
video_fps_target = 24              # FPS cible

# Options
enable_text_to_video = True        # Generation texte -> video
enable_image_to_video = True       # Generation image -> video
enable_video_edit = True           # Edition video
# IMPORTANT (fix c.10315) : mock est un FALLBACK, pas un override. Le defaut est
# d'appeler l'API reelle quand elle est disponible ; mock n'est utilise que si
# (a) l'utilisateur le demande explicitement OU (b) l'API est indisponible.
use_mock_responses = False         # Utiliser des reponses simulees uniquement si API indisponible
save_results = True                # Sauvegarder les resultats

# Estimation cout unitaire (approx, OpenAI Sora 2 — 720p, 4s)
sora_cost_per_video_720p_4s = 0.10  # USD; ajuster selon tarification reelle

On importe ensuite les bibliotheques necessaires. Le client officiel openai porte tous les appels a l’API Sora ; numpy alimente les analyses de cout et les visualisations de la Section 4 (tableaux de volumes, courbes comparatives) ; matplotlib trace ces comparaisons ; time mesure les durees de generation, la grandeur centrale du cloud vs local. Les utilitaires communs de la serie Video sont importes via sys.path depuis le dossier parent – la sortie confirmera Helpers video importes, qui apporte notamment le formatage des metriques qu’on retrouvera dans le bilan de session. Ce pattern d’imports conditionnels est le socle qui rend le notebook executable que les services soient disponibles ou non.

# Setup environnement et imports
import os
import sys
import json
import time
import warnings
import base64
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Any, Optional, Tuple
import numpy as np
from PIL import Image, ImageDraw, ImageFont
import matplotlib.pyplot as plt
import logging

warnings.filterwarnings('ignore', category=DeprecationWarning)

# Resolution GENAI_ROOT (c.10315 : worktree-aware fallback)
# Chargement robuste de la configuration .env
from dotenv import load_dotenv
import os
# Recherche du .env dans tous les parents (pour Papermill qui change le cwd)
current_path = Path.cwd()
# Determine GenAI root directory
GENAI_ROOT = Path.cwd()
while GENAI_ROOT.name != "GenAI" and len(GENAI_ROOT.parts) > 1:
    GENAI_ROOT = GENAI_ROOT.parent
# Fallback : si pas de segment GenAI dans cwd/parents (worktree), chercher
# dans les repos principaux sibling. C'est attendu — `git worktree add`
# cree une working copy separe ou `.env` n'est pas copie (gitignore).
if GENAI_ROOT.name != "GenAI":
    for candidate in (
        Path('D:/Dev/CoursIA-2/MyIA.AI.Notebooks/GenAI'),
        Path('D:/Dev/CoursIA/MyIA.AI.Notebooks/GenAI'),
        Path('C:/Dev/CoursIA/MyIA.AI.Notebooks/GenAI'),
    ):
        if candidate.exists():
            GENAI_ROOT = candidate
            break

env_loaded = False
while current_path.name != "GenAI" and len(current_path.parts) > 1:
    env_path = current_path / ".env"
    if env_path.exists():
        load_dotenv(env_path)
        print(f".env charge depuis: {env_path.name}")
        env_loaded = True
        break
    current_path = current_path.parent
if not env_loaded:
    print("WARNING: .env non trouve, utilisation variables environnement")

HELPERS_PATH = GENAI_ROOT / 'shared' / 'helpers'
if HELPERS_PATH.exists():
    sys.path.insert(0, str(HELPERS_PATH.parent))
    try:
        from helpers.video_helpers import get_video_info, extract_frames, display_frame_grid
        print("Helpers video importes")
    except ImportError as e:
        print(f"Helpers video non disponibles ({e}) - mode autonome")

# Repertoire de sortie
OUTPUT_DIR = GENAI_ROOT / 'outputs' / 'video'
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

# Configuration logging
logging.basicConfig(level=getattr(logging, debug_level))
logger = logging.getLogger('sora_api')

print(f"Sora API - Generation Video Cloud")
print(f"Date : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Mode : {notebook_mode}")
print(f"Resolution cible : {video_resolution}, {video_duration_target}s @ {video_fps_target}fps")
print(f"Sortie : {OUTPUT_DIR.relative_to(GENAI_ROOT)}")
WARNING: .env non trouve, utilisation variables environnement
Helpers video importes
Sora API - Generation Video Cloud
Date : 2026-08-10 19:54:56
Mode : interactive
Resolution cible : 720p, 4s @ 24fps
Sortie : outputs\video

Les variables d’environnement et la cle OpenAI sont chargees depuis le fichier .env – le mecanisme standard du depot (regle d’hygiene : jamais de cle en litteral dans le code). La cellule suivante verifie simultanement trois choses : la presence du fichier .env, les dependances Python requises, et la cle OPENAI_API_KEY. Deux issues possibles, toutes deux fonctionnelles – la sortie de cette execution le montre : WARNING: .env non trouve (le fichier absent du repertoire d’execution) MAIS aussitot apres openai : disponible, httpx : disponible, imageio : disponible et OPENAI_API_KEY : configuree. Autrement dit, le notebook est retombe sur les variables d’environnement du shell, qui portent la cle : l’execution continue en mode API reelle sans aucune modification. C’est exactement le comportement attendu d’un notebook portable – la configuration se negocie avec l’environnement, jamais en editant le code.

# Chargement .env et verification des dependances
from dotenv import load_dotenv
import os

# c.10315 : reutiliser GENAI_ROOT deja resolu en cellule-2 (avec fallback worktree)
# Cell-2 a deja charge le .env ; on re-charge avec override=False pour confirmer.
env_path = GENAI_ROOT / ".env"
if env_path.exists():
    load_dotenv(env_path, override=False)  # no-op pratique si deja charge
    print(f".env confirme depuis: {env_path.name}")
else:
    # Fallback defensif (normalement cell-2 a deja resolu)
    current_path = Path.cwd()
    while len(current_path.parts) > 1:
        ep = current_path / ".env"
        if ep.exists():
            load_dotenv(ep)
            print(f".env charge depuis: {ep}")
            break
        current_path = current_path.parent
    else:
        print("WARNING: .env non trouve")

dependencies = {}
api_available = False

try:
    from openai import OpenAI
    dependencies['openai'] = True
    print(f"openai : disponible")
except ImportError:
    dependencies['openai'] = False
    print(f"openai : NON INSTALLE (pip install openai)")

try:
    import httpx
    dependencies['httpx'] = True
    print(f"httpx : disponible")
except ImportError:
    dependencies['httpx'] = False
    print(f"httpx : NON INSTALLE (pip install httpx)")

try:
    import imageio
    dependencies['imageio'] = True
    print(f"imageio : disponible")
except ImportError:
    dependencies['imageio'] = False
    print(f"imageio : NON INSTALLE")

# Verifier la cle API
openai_key = os.environ.get('OPENAI_API_KEY', '')
openai_base = os.environ.get('OPENAI_BASE_URL', '')

if openai_key:
    print(f"\nOPENAI_API_KEY : configuree")
    if openai_base:
        print(f"OPENAI_BASE_URL : {openai_base}")
    
    # Tester la connexion API
    if dependencies.get('openai', False):
        try:
            client_kwargs = {'api_key': openai_key}
            if openai_base:
                client_kwargs['base_url'] = openai_base
            client = OpenAI(**client_kwargs)
            # Test leger (lister les modeles)
            models = client.models.list()
            sora_models = [m.id for m in models if 'sora' in m.id.lower()]
            if sora_models:
                print(f"Modeles Sora disponibles : {', '.join(sora_models)}")
                api_available = True
            else:
                print(f"Aucun modele Sora detecte dans la liste des modeles")
                print(f"Mode demonstration active (reponses simulees)")
        except Exception as e:
            print(f"Test API : {type(e).__name__} - {str(e)[:80]}")
            print(f"Mode demonstration active")
else:
    print(f"\nOPENAI_API_KEY : non configuree")
    print(f"Mode demonstration active (reponses simulees)")

# c.10315 : mock est un FALLBACK explicite, pas un override. mock effectif =
# force-utilisateur OR api-indisponible. Avec use_mock_responses=False (defaut)
# et api_available=True, on utilise la vraie API (pas le mock).
print(f"\nAPI Sora disponible : {api_available}")
print(f"use_mock_responses (param) : {use_mock_responses}")
print(f"Mode mock effectif (apres c.10315 fix) : {use_mock_responses or not api_available}")
WARNING: .env non trouve
openai : disponible
httpx : disponible
imageio : disponible

OPENAI_API_KEY : configuree
Modeles Sora disponibles : sora-2, sora-2-pro

API Sora disponible : True
use_mock_responses (param) : False
Mode mock effectif (apres c.10315 fix) : False

Section 1 : Introduction a Sora et son architecture

Sora est le modèle de generation video d’OpenAI, annonce en fevrier 2024. Il utilise une architecture de type Diffusion Transformer (DiT) operant dans un espace latent spatio-temporel compresse.

Caractéristiques techniques

Aspect Detail
Architecture Diffusion Transformer (DiT)
Espace latent Compression spatio-temporelle 3D
Entree Texte, image, video
Sortie Video jusqu’a 1 minute
Resolutions 480p, 720p, 1080p
Coherence temporelle Attention sur toute la sequence

Différence avec les modèles locaux

Contrairement aux modèles locaux (HunyuanVideo, LTX-Video, Wan, SVD) qui necessitent un GPU puissant (16-48 GB VRAM), Sora fonctionne entierement en cloud. L’acces se fait via l’API OpenAI, avec facturation a l’utilisation.

# Helper : generation de reponses simulees pour demonstration
print("\n--- PREPARATION DES OUTILS ---")
print("=" * 40)


def _draw_simulation_banderole(img: Image.Image, text: str,
                                position: str = "bottom") -> None:
    """
    c.194 (issue #10315 req. 3) : bandeau SIMULATION in-image centre, lisible
    sur fond clair ET sombre. Survit au copier-coller d'une frame isolee.

    - Bandeau pleine largeur (~8% de la hauteur d'image), fond noir opaque
    - Texte blanc en gras, centre, font proportionnelle a la hauteur du bandeau
    - Position par defaut : bas de l'image (bottom)
    """
    draw = ImageDraw.Draw(img)
    w, h = img.size
    band_h = max(20, int(h * 0.08))
    if position == "bottom":
        y0 = h - band_h
    elif position == "top":
        y0 = 0
    else:
        y0 = h - band_h  # fallback

    # Fond noir opaque (PIL ne supporte pas l'alpha sur ImageDraw.rectangle
    # de maniere portable, on triche avec un gris fonce -- reste lisible).
    draw.rectangle([(0, y0), (w, y0 + band_h)], fill=(0, 0, 0))

    # Texte SIMULATION en blanc, centre, font la plus grande qui tient
    try:
        font = ImageFont.truetype("arial.ttf", max(14, int(band_h * 0.55)))
    except (OSError, IOError):
        font = ImageFont.load_default()

    # Mesurer la largeur du texte pour le centrer
    bbox = draw.textbbox((0, 0), text, font=font)
    tw, th_ = bbox[2] - bbox[0], bbox[3] - bbox[1]
    tx = (w - tw) // 2
    ty = y0 + (band_h - th_) // 2 - 2
    draw.text((tx, ty), text, fill=(255, 255, 255), font=font)


def generate_mock_video_frames(prompt: str, n_frames: int = 48,
                                width: int = 640, height: int = 360) -> List[np.ndarray]:
    """
    Genere des frames simulees illustrant le concept du prompt.
    Utilise en mode demonstration quand l'API Sora n'est pas disponible.
    """
    frames = []
    rng = np.random.RandomState(hash(prompt) % (2**31))

    # Couleur de base derivee du prompt
    base_hue = hash(prompt) % 360

    for i in range(n_frames):
        t = i / max(n_frames - 1, 1)

        # Generer un paysage abstrait anime
        img = Image.new('RGB', (width, height))
        draw = ImageDraw.Draw(img)

        # Ciel gradient
        for y in range(height // 2):
            ratio = y / (height // 2)
            r = int(30 + 100 * ratio + 20 * np.sin(2 * np.pi * t))
            g = int(50 + 80 * ratio)
            b = int(120 + 80 * (1 - ratio))
            draw.line([(0, y), (width, y)], fill=(min(255, r), min(255, g), min(255, b)))

        # Sol
        for y in range(height // 2, height):
            ratio = (y - height // 2) / (height // 2)
            r = int(60 + 40 * ratio)
            g = int(100 + 50 * (1 - ratio) + 20 * np.sin(2 * np.pi * t + 1))
            b = int(40 + 30 * ratio)
            draw.line([(0, y), (width, y)], fill=(min(255, r), min(255, g), min(255, b)))

        # Objet en mouvement
        cx = int(width * 0.2 + width * 0.6 * t)
        cy = int(height * 0.35 + 20 * np.sin(4 * np.pi * t))
        draw.ellipse([cx - 25, cy - 25, cx + 25, cy + 25], fill=(255, 220, 100))

        # c.194 (issue #10315 req. 3) : bandeau SIMULATION in-image centre --
        # survit au copier-coller (l'ancien watermark corner-only etait
        # invisible sur fond clair et oublie par le chemin Ken Burns i2v).
        _draw_simulation_banderole(img, f"SIMULATION -- {prompt[:50]}", position='bottom')

        frames.append(np.array(img))

    return frames


def _sora_resolution_to_size(resolution: str) -> str:
    """Convertit une resolution lisible ('720p', '1080p') en size_string SDK.
    Sora-2 accepte 720p/1080p ; on passe la size littérale au SDK."""
    res_map = {
        "480p": "854x480",
        "720p": "1280x720",
        "1080p": "1920x1080",
    }
    return res_map.get(resolution, "1280x720")


def _sora_snap_seconds(duration: int) -> int:
    """Snap une duree arbitraire sur les valeurs Sora-2 acceptees (4/8/12).
    c.10315 (fix latent) : l'API Sora-2 rejette toute autre valeur (HTTP 400
    `Invalid value: 'N'. Supported values are: '4', '8', and '12'.`).
    Strategie : clamp [4, 12], puis arrondir au plus proche valide (4/8/12).
    Le code original utilisait `max(4, min(int(duration), 12))` qui livre
    des valeurs invalides (5, 6, 7, 9, 10, 11) des que la duree depasse 4.
    """
    seconds = max(4, min(int(duration), 12))
    valid = (4, 8, 12)
    return min(valid, key=lambda v: abs(v - seconds))



def _poll_video_job(client, video_id: str, timeout_s: int = 240,
                    poll_interval_s: int = 5):
    """
    Poll un job Sora jusqu'a completion. Renvoie l'objet Video final.
    Timeout par defaut 240s ; les generations 720p/4s prennent ~30-90s en pratique.
    """
    deadline = time.time() + timeout_s
    last = None
    while time.time() < deadline:
        last = client.videos.retrieve(video_id)
        if last.status in ("completed", "failed", "cancelled"):
            return last
        time.sleep(poll_interval_s)
    raise TimeoutError(
        f"Job Sora {video_id} non termine apres {timeout_s}s "
        f"(status={last.status if last else 'unknown'})"
    )


def _download_sora_video(client, video_id: str, dest_path: Path,
                         variant: str = "video") -> Path:
    """Telecharge le contenu binaire (video ou thumbnail) d'un job Sora termine."""
    content = client.videos.download_content(video_id, variant=variant)
    # download_content renvoie un Response/iterator selon version SDK
    if hasattr(content, "read"):
        data = content.read()
    elif hasattr(content, "iter_bytes"):
        data = b"".join(content.iter_bytes())
    else:
        data = bytes(content)
    dest_path.write_bytes(data)
    return dest_path


class SoraAPIWrapper:
    """
    Wrapper pour l'API Sora avec fallback explicite sur des reponses simulees.

    c.10315 fix : `use_mock` est desormais un FALLBACK explicite, pas un override.
    Quand l'API est disponible ET que l'utilisateur ne force pas mock, on appelle
    reel `client.videos.create()` (generation reelle, facturable). Le mock n'est
    utilise que si (a) l'utilisateur le demande via `use_mock_responses=True`
    OU (b) l'API n'est pas accessible.

    Retour normalise :
      - API reelle : {"status": "success", "mp4_path": str, "job_id": str}
      - Mock       : {"status": "mock", "frames": List[np.ndarray], "metadata": dict}
    """

    def __init__(self, api_available: bool, client=None, use_mock: bool = False,
                 output_dir: Optional[Path] = None):
        self.api_available = api_available
        self.client = client
        # c.10315 fix : `or` -> `and not api_available` : mock est un fallback, jamais un override
        self.use_mock = use_mock and not api_available
        self.call_log = []
        self.output_dir = output_dir or Path.cwd()

    def _generate_via_api(self, *, prompt: str, size: str, seconds: int,
                          reference_file: Optional[Path] = None) -> dict:
        """Generation reelle via `client.videos.create()`, poll, telechargement MP4."""
        create_kwargs = {
            "model": sora_model,
            "prompt": prompt,
            "size": size,
            "seconds": str(seconds),
        }
        # Reference image si fournie (i2v)
        if reference_file is not None:
            # SDK openai >= 2.43 : `input_reference` accepte un file-like ou un Path
            with open(reference_file, "rb") as f:
                create_kwargs["input_reference"] = f
                job = self.client.videos.create(**create_kwargs)
        else:
            job = self.client.videos.create(**create_kwargs)

        # Poll jusqu'a completion
        final = _poll_video_job(self.client, job.id, timeout_s=240)
        if final.status == "failed":
            err = getattr(final, "error", None)
            raise RuntimeError(f"Sora job {job.id} failed: {err}")

        # Telecharge le MP4
        mp4_path = self.output_dir / f"sora_api_{job.id}.mp4"
        _download_sora_video(self.client, job.id, mp4_path, variant="video")

        return {
            "job_id": job.id,
            "status": "success",
            "mp4_path": str(mp4_path),
            "size": size,
            "seconds": seconds,
        }

    def _generate_mock(self, *, prompt: str, duration: int, resolution: str,
                       image: Optional[Image.Image] = None) -> dict:
        """Generation simulee (fallback). Renvoie des frames locales uniquement."""
        res_map = {"480p": (854, 480), "720p": (1280, 720), "1080p": (1920, 1080)}
        w, h = res_map.get(resolution, (640, 360))
        # Reduire pour la demo (cout memoire)
        w, h = w // 2, h // 2
        n_frames = duration * video_fps_target

        if image is not None:
            # Animation Ken Burns locale sur l'image fournie
            src_w, src_h = image.size
            frames = []
            for i in range(n_frames):
                t = i / max(n_frames - 1, 1)
                zoom = 1.0 + 0.15 * t
                new_w = int(src_w * zoom)
                new_h = int(src_h * zoom)
                img_z = image.resize((new_w, new_h), Image.LANCZOS)
                offset_x = int((new_w - src_w) * (0.3 + 0.4 * t))
                offset_y = int((new_h - src_h) * 0.5)
                img_crop = img_z.crop((offset_x, offset_y, offset_x + src_w, offset_y + src_h))
                # c.194 (issue #10315 req. 3) : bandeau SIMULATION in-image
                # sur les frames Ken Burns (i2v mock) -- meme contrat que
                # `generate_mock_video_frames` (t2v mock).
                _draw_simulation_banderole(img_crop, f"SIMULATION i2v -- {prompt[:50]}", position='bottom')
                frames.append(np.array(img_crop))
        else:
            frames = generate_mock_video_frames(prompt, n_frames, w, h)

        return {
            "status": "mock",
            "frames": frames,
            "metadata": {
                "model": sora_model,
                "prompt": prompt,
                "duration": duration,
                "resolution": f"{w}x{h} (demo)",
                "fps": video_fps_target,
                "n_frames": len(frames),
            },
        }

    def text_to_video(self, prompt: str, duration: int = 5,
                      resolution: str = "720p") -> dict:
        """Generation texte -> video via l'API reelle (fallback mock si necessaire)."""
        call_info = {
            "method": "text_to_video",
            "prompt": prompt,
            "duration": duration,
            "resolution": resolution,
            "timestamp": datetime.now().isoformat(),
        }

        # Branche API reelle (c.10315 fix) — appel reel d'abord si dispo + non force mock
        if self.api_available and not self.use_mock:
            try:
                size = _sora_resolution_to_size(resolution)
                # Sora-2 accepte 4-8-12 secondes (API rejette les autres)
                seconds = _sora_snap_seconds(int(duration))
                api_result = self._generate_via_api(
                    prompt=prompt, size=size, seconds=seconds,
                )
                call_info["status"] = "success"
                call_info["job_id"] = api_result["job_id"]
                call_info["mp4_path"] = api_result["mp4_path"]
                call_info["cost_estimate_usd"] = sora_cost_per_video_720p_4s
                self.call_log.append(call_info)
                return {"status": "success", "mp4_path": api_result["mp4_path"],
                        "job_id": api_result["job_id"], "video": api_result}
            except Exception as e:
                call_info["status"] = "api_error_then_mock"
                call_info["api_error"] = f"{type(e).__name__}: {str(e)[:200]}"

        # Mode demonstration (fallback)
        result = self._generate_mock(prompt=prompt, duration=duration,
                                     resolution=resolution)
        call_info["status"] = call_info.get("status", "mock")
        call_info["frames_generated"] = len(result.get("frames", []))
        self.call_log.append(call_info)
        return result

    def image_to_video(self, image: Image.Image, prompt: str,
                       duration: int = 5) -> dict:
        """Generation image -> video via l'API reelle (fallback mock si necessaire)."""
        call_info = {
            "method": "image_to_video",
            "prompt": prompt,
            "duration": duration,
            "timestamp": datetime.now().isoformat(),
        }

        if self.api_available and not self.use_mock:
            try:
                # Sauver l'image source en JPEG temporaire pour upload
                ref_path = self.output_dir / f"_sora_ref_{int(time.time())}.jpg"
                if image.mode != "RGB":
                    image = image.convert("RGB")
                image.save(ref_path, format="JPEG", quality=90)
                size = _sora_resolution_to_size(video_resolution)
                seconds = _sora_snap_seconds(int(duration))
                api_result = self._generate_via_api(
                    prompt=prompt, size=size, seconds=seconds,
                    reference_file=ref_path,
                )
                # Cleanup du fichier de reference
                try:
                    ref_path.unlink()
                except OSError:
                    pass
                call_info["status"] = "success"
                call_info["job_id"] = api_result["job_id"]
                call_info["mp4_path"] = api_result["mp4_path"]
                call_info["cost_estimate_usd"] = sora_cost_per_video_720p_4s
                self.call_log.append(call_info)
                return {"status": "success", "mp4_path": api_result["mp4_path"],
                        "job_id": api_result["job_id"], "video": api_result}
            except Exception as e:
                call_info["status"] = "api_error_then_mock"
                call_info["api_error"] = f"{type(e).__name__}: {str(e)[:200]}"

        # Mode demonstration (fallback)
        result = self._generate_mock(prompt=prompt, duration=duration,
                                     resolution=video_resolution, image=image)
        call_info["status"] = call_info.get("status", "mock")
        self.call_log.append(call_info)
        return result


# Initialiser le wrapper (c.10315 fix : mock = fallback, pas override)
sora = SoraAPIWrapper(
    api_available=api_available,
    client=client if api_available else None,
    use_mock=use_mock_responses,
    output_dir=OUTPUT_DIR,
)

print(f"SoraAPIWrapper initialise")
print(f"  API reelle : {api_available}")
print(f"  use_mock_responses (param) : {use_mock_responses}")
print(f"  Mode mock (fallback) : {sora.use_mock}")
if sora.api_available and not sora.use_mock:
    print(f"  -> Chemin privilegie : appels reels a POST /v1/videos")
else:
    print(f"  -> Chemin privilegie : reponses simulees locales")

--- PREPARATION DES OUTILS ---
========================================
SoraAPIWrapper initialise
  API reelle : True
  use_mock_responses (param) : False
  Mode mock (fallback) : False
  -> Chemin privilegie : appels reels a POST /v1/videos

Interpretation : la couche wrapper autour de l’API

La sortie detaille l’armature technique : SoraAPIWrapper initialise, avec API reelle : True, le parametre use_mock_responses a False, le Mode mock (fallback) inactif, et le chemin privilegie explicite – appels reels a POST /v1/videos. Pourquoi un wrapper plutot que des appels openai nus ? Parce que l’API Sora est asynchrone : la sequence d’appels visible dans la section suivante le montrera – un POST /v1/videos qui retourne immediatement un identifiant de job, une serie de GET de sondage du statut jusqu’a completion, puis un GET .../content qui telecharge le fichier produit. Le wrapper encapsule cette machine a etats derriere une interface a une seule methode. Les deux booleens affiches (use_mock_responses parametre, Mode mock fallback) documentent l’autre visage du wrapper : le meme notebook peut servir des reponses simulees de forme identique quand la cle ou le credit manque – ici les deux sont a False, les sections suivantes utilisent l’API reelle.

Section 2 : Generation texte vers video

La generation text-to-video est la fonctionnalite principale de Sora. Un prompt textuel decrit la scene souhaitee, et le modèle genere la video correspondante.

Bonnes pratiques pour les prompts Sora

Aspect Conseil
Description de scene Detailler le sujet, le lieu, l’eclairage
Mouvement camera Preciser : “camera slowly pans left”, “tracking shot”
Style visuel “cinematic”, “documentary”, “aerial drone shot”
Duree Adapter la complexite du prompt a la duree
Eviter Texte a lire, visages spécifiques, physique complexe 
# Generation texte -> video
if enable_text_to_video:
    print("\n--- GENERATION TEXTE -> VIDEO ---")
    print("=" * 45)

    # Prompts de demonstration
    prompts = [
        {
            "name": "Nature",
            "prompt": "A serene mountain lake at sunrise, mist rising from the water, "
                      "pine trees reflecting in the still surface, cinematic drone shot "
                      "slowly descending toward the water.",
            "duration": 5,
        },
        {
            "name": "Urbain",
            "prompt": "A busy Tokyo street at night, neon signs reflecting on wet "
                      "pavement, people with umbrellas walking, tracking shot moving "
                      "forward through the crowd.",
            "duration": 5,
        },
    ]

    results_t2v = []

    for p in prompts:
        print(f"\nGeneration : {p['name']}")
        print(f"  Prompt : {p['prompt'][:80]}...")

        start_time = time.time()
        result = sora.text_to_video(
            prompt=p['prompt'],
            duration=p['duration'],
            resolution=video_resolution
        )
        gen_time = time.time() - start_time

        print(f"  Status : {result['status']}")
        if result['status'] == 'success':
            print(f"  MP4 : {result.get('mp4_path', 'N/A')}")
            print(f"  Job ID : {result.get('job_id', 'N/A')}")
        else:
            print(f"  Frames : {len(result.get('frames', []))}")
        print(f"  Temps : {gen_time:.2f}s")

        results_t2v.append({"name": p['name'], "result": result, "time": gen_time})

    # Afficher les premiers et derniers frames (uniquement mode mock ;
    # en mode API reelle on liste les fichiers MP4 generes).
    any_mock_t2v = any(r['result']['status'] == 'mock' for r in results_t2v)
    if any_mock_t2v:
        fig, axes = plt.subplots(2, 4, figsize=(16, 6))
        for row, r in enumerate(results_t2v):
            if r['result']['status'] != 'mock':
                continue
            frames = r['result']['frames']
            sample_idx = np.linspace(0, len(frames) - 1, 4, dtype=int)
            for col, idx in enumerate(sample_idx):
                axes[row, col].imshow(frames[idx])
                axes[row, col].set_title(f"{r['name']} t={idx/video_fps_target:.1f}s", fontsize=9)
                axes[row, col].axis('off')
        plt.suptitle("Generation Texte -> Video (Sora mock)", fontsize=13, fontweight='bold')
        plt.tight_layout()
        plt.show()
    else:
        print("\nTous les appels ont ete effectues via l'API reelle. Sortie :")
        for r in results_t2v:
            mp4 = r['result'].get('mp4_path')
            if mp4 and Path(mp4).exists():
                size_kb = Path(mp4).stat().st_size / 1024
                print(f"  {Path(mp4).name} ({size_kb:.1f} KB)")

    # Sauvegarder (mode mock -> re-encode via imageio ; mode API -> MP4 deja sauvegarde)
    if save_results:
        for r in results_t2v:
            res = r['result']
            if res['status'] == 'mock' and dependencies.get('imageio', False):
                path = OUTPUT_DIR / f"sora_t2v_{r['name'].lower()}.mp4"
                writer = imageio.get_writer(str(path), fps=video_fps_target, codec='libx264')
                for f in res['frames']:
                    writer.append_data(f)
                writer.close()
                print(f"Sauvegarde : {path.name} ({path.stat().st_size / 1024:.0f} KB)")
            elif res['status'] == 'success':
                print(f"Deja sauvegarde via l'API : {Path(res['mp4_path']).name}")
else:
    print("Generation texte -> video desactivee")

--- GENERATION TEXTE -> VIDEO ---
=============================================

Generation : Nature
  Prompt : A serene mountain lake at sunrise, mist rising from the water, pine trees reflec...
  Status : success
  MP4 : D:\Dev\CoursIA-2-c10315\MyIA.AI.Notebooks\GenAI\outputs\video\sora_api_video_6a7a1076ac708191aa931779fcf029f503c01d36ff445b3a.mp4
  Job ID : video_6a7a1076ac708191aa931779fcf029f503c01d36ff445b3a
  Temps : 65.08s

Generation : Urbain
  Prompt : A busy Tokyo street at night, neon signs reflecting on wet pavement, people with...
  Status : success
  MP4 : D:\Dev\CoursIA-2-c10315\MyIA.AI.Notebooks\GenAI\outputs\video\sora_api_video_6a7a10b7f010819184a8bca695620e960fb0597694bf8b66.mp4
  Job ID : video_6a7a10b7f010819184a8bca695620e960fb0597694bf8b66
  Temps : 65.65s

Tous les appels ont ete effectues via l'API reelle. Sortie :
  sora_api_video_6a7a1076ac708191aa931779fcf029f503c01d36ff445b3a.mp4 (1915.8 KB)
  sora_api_video_6a7a10b7f010819184a8bca695620e960fb0597694bf8b66.mp4 (4398.8 KB)
Deja sauvegarde via l'API : sora_api_video_6a7a1076ac708191aa931779fcf029f503c01d36ff445b3a.mp4
Deja sauvegarde via l'API : sora_api_video_6a7a10b7f010819184a8bca695620e960fb0597694bf8b66.mp4

Interpretation : Generation texte -> video

Metrique Valeur attendue (API reelle) Valeur demo
Temps de generation 30-120s selon duree/resolution < 1s (simulation)
Coherence temporelle Elevee (attention sur toute la sequence) Basique (interpolation)
Qualite visuelle Photo-realiste Schema simplifie
Cout estime ~$0.10 - $0.50 par video Gratuit

Points cles : 1. Sora maintient une coherence temporelle superieure aux modèles frame-par-frame 2. Les prompts detailles avec indications de camera produisent de meilleurs résultats 3. En mode demo, les frames simulees illustrent la structure de l’API sans cout

Section 3 : Generation image vers video

L’image-to-video anime une image fixe en video. C’est utile pour : - Animer des illustrations ou photos - Créer des effets cinematographiques (Ken Burns, parallaxe) - Etendre une image en sequence temporelle

Section 3 : Generation image vers video

La generation image-to-video part d’une image source et l’anime : l’image fournit le contenu (composition, couleurs, sujets), le prompt decrit le mouvement a produire. C’est le mode privilegie des usages professionnels – animer un visuel de marque existant, prolonger un plan, tester une direction creative a partir d’une reference fixe – la ou le text-to-video de la Section 2 part d’une page blanche totale.

# Generation image -> video
if enable_image_to_video:
    print("\n--- GENERATION IMAGE -> VIDEO ---")
    print("=" * 45)

    # c.10315 (fix latent) : Sora-2 /v1/videos exige que input_reference ait
    # EXACTEMENT la meme taille que la video demandee. L'ancien 640x360
    # hardcode renvoyait HTTP 400 `Inpaint image must match the requested
    # width and height` des que la resolution cible changeait. On derive
    # donc la taille de l'image source depuis `_sora_resolution_to_size`
    # (deja importe en cell-5) pour garantir la correspondance exacte.
    _i2v_size_str = _sora_resolution_to_size(video_resolution)
    source_width, source_height = (int(v) for v in _i2v_size_str.split('x'))
    source_img = Image.new('RGB', (source_width, source_height))
    draw = ImageDraw.Draw(source_img)

    # Paysage simplifie (coords mises a l'echelle de la resolution cible)
    sky_top = int(source_height * 0.6)
    for y in range(source_height):
        if y < sky_top:  # Ciel
            ratio = y / sky_top
            r, g, b = int(50 + 100 * ratio), int(80 + 100 * ratio), int(180 + 50 * (1 - ratio))
        else:  # Sol
            ratio = (y - sky_top) / (source_height - sky_top)
            r, g, b = int(80 + 50 * ratio), int(140 - 40 * ratio), int(60 + 20 * ratio)
        draw.line([(0, y), (source_width, y)], fill=(r, g, b))

    # Montagnes (coords proportionnelles a source_width/source_height)
    horizon = int(source_height * 0.6)
    peaks = [
        (0, int(source_height * 0.56)),
        (int(source_width * 0.156), int(source_height * 0.39)),
        (int(source_width * 0.312), int(source_height * 0.47)),
        (int(source_width * 0.500), int(source_height * 0.33)),
        (int(source_width * 0.703), int(source_height * 0.44)),
        (int(source_width * 0.859), int(source_height * 0.36)),
        (source_width, int(source_height * 0.53)),
    ]
    mountains = peaks + [(source_width, horizon), (0, horizon)]
    draw.polygon(mountains, fill=(70, 90, 60))

    # Soleil (centre 75% largeur, ~22% hauteur, rayon ~10% de la hauteur)
    sun_cx = int(source_width * 0.795)
    sun_cy = int(source_height * 0.20)
    sun_r = int(source_height * 0.085)
    draw.ellipse([sun_cx - sun_r, sun_cy - sun_r,
                   sun_cx + sun_r, sun_cy + sun_r],
                  fill=(255, 220, 100))

    print(f"Image source : {source_width}x{source_height} (cible : {_i2v_size_str})")

    # Animation
    i2v_prompt = "The landscape comes alive: clouds drift across the sky, sunlight shifts"
    result_i2v = sora.image_to_video(source_img, i2v_prompt, duration=video_duration_target)

    print(f"Status : {result_i2v['status']}")
    if result_i2v['status'] == 'success':
        print(f"MP4 : {result_i2v.get('mp4_path', 'N/A')}")
        print(f"Job ID : {result_i2v.get('job_id', 'N/A')}")
    else:
        print(f"Frames generees : {len(result_i2v.get('frames', []))}")
        print(f"Effet : {result_i2v['metadata'].get('effect', 'N/A')}")

    if result_i2v['status'] == 'mock':
        # Afficher source + frames animees (mock)
        fig, axes = plt.subplots(1, 5, figsize=(16, 3))
        axes[0].imshow(source_img)
        axes[0].set_title("Image source", fontsize=10)
        axes[0].axis('off')

        i2v_frames = result_i2v['frames']
        sample_idx = np.linspace(0, len(i2v_frames) - 1, 4, dtype=int)
        for i, idx in enumerate(sample_idx):
            axes[i + 1].imshow(i2v_frames[idx])
            axes[i + 1].set_title(f"t = {idx / video_fps_target:.1f}s", fontsize=10)
            axes[i + 1].axis('off')
        plt.suptitle("Image -> Video (effet Ken Burns)", fontsize=13, fontweight='bold')
        plt.tight_layout()
        plt.show()
    else:
        # API reelle : le MP4 a deja ete sauvegarde par SoraAPIWrapper
        mp4 = result_i2v.get('mp4_path')
        if mp4 and Path(mp4).exists():
            size_kb = Path(mp4).stat().st_size / 1024
            print(f"Sortie API reelle : {Path(mp4).name} ({size_kb:.1f} KB)")

    if save_results:
        if result_i2v['status'] == 'mock' and dependencies.get('imageio', False):
            i2v_path = OUTPUT_DIR / "sora_i2v_landscape.mp4"
            writer = imageio.get_writer(str(i2v_path), fps=video_fps_target, codec='libx264')
            for f in result_i2v['frames']:
                writer.append_data(f)
            writer.close()
            print(f"Sauvegarde : {i2v_path.name} ({i2v_path.stat().st_size / 1024:.0f} KB)")
        elif result_i2v['status'] == 'success':
            print(f"Deja sauvegarde via l'API : {Path(result_i2v['mp4_path']).name}")
else:
    print("Generation image -> video desactivee")

--- GENERATION IMAGE -> VIDEO ---
=============================================
Image source : 1280x720 (cible : 1280x720)
Status : success
MP4 : D:\Dev\CoursIA-2-c10315\MyIA.AI.Notebooks\GenAI\outputs\video\sora_api_video_6a7a10f9df7881919dcf7f0d9d77a7200ea9bd0f11e614ae.mp4
Job ID : video_6a7a10f9df7881919dcf7f0d9d77a7200ea9bd0f11e614ae
Sortie API reelle : sora_api_video_6a7a10f9df7881919dcf7f0d9d77a7200ea9bd0f11e614ae.mp4 (1608.2 KB)
Deja sauvegarde via l'API : sora_api_video_6a7a10f9df7881919dcf7f0d9d77a7200ea9bd0f11e614ae.mp4

Interpretation : Generation image -> video

La sortie montre d’abord la negociation de format – Image source : 1280x720 (cible : 1280x720), l’image est deja a la resolution demandee, aucun recadrage – puis le meme cycle asynchrone que la Section 2 : POST /v1/videos accepte (200), une serie de sondages GET sur l’identifiant du job, et le telechargement final GET .../content?variant=video. Status : success et la ligne Sortie API reelle confirment que la video produite (environ 1,6 MB) vient bien de l’API, pas du mode simule. Difference notable avec le text-to-video : ici la contrainte de coherence visuelle avec l’image source s’ajoute a la generation – la qualite du resultat depend autant de l’image de depart que du prompt de mouvement.

Section 4 : Analyse des couts et comparaison locale vs cloud

Le choix entre generation video locale (ComfyUI, modele Wan/Hunyuan sur GPU – les notebooks 02 de cette serie) et generation cloud (API Sora facturee a la video produite) ne se tranche pas sur la seule qualite : il se tranche sur le volume d’usage. En local, le cout marginal d’une generation supplementaire est quasi nul en euros, mais l’infrastructure est un cout fixe eleve, paye d’avance (achat GPU, amortissement mensuel). Dans le cloud, chaque video facture son prix unitaire – cout marginal eleve, mais zero investissement initial et zero maintenance. La cellule suivante chiffre la comparaison (cout par video, temps de generation, parallelisme) entre Sora et les modeles locaux, puis l’exercice 3 en derive le seuil de rentabilite : le volume mensuel exact au-dela duquel le local devient moins cher que le cloud. La reponse a “cloud ou local ?” n’est donc pas une propriete des modeles – c’est une propriete du volume.

# Analyse comparative : Sora vs modeles locaux
print("\n--- COMPARAISON SORA VS MODELES LOCAUX ---")
print("=" * 50)

# Tableau comparatif detaille
comparison_data = {
    "Modele": [
        "Sora (OpenAI)",
        "HunyuanVideo",
        "LTX-Video",
        "Wan 2.1",
        "Stable Video Diffusion",
        "AnimateDiff",
    ],
    "Type": ["Cloud API", "Local", "Local", "Local", "Local", "Local"],
    "VRAM": ["0 (cloud)", "~40 GB", "~12 GB", "~24 GB", "~16 GB", "~12 GB"],
    "Qualite": ["Tres haute", "Haute", "Moyenne", "Haute", "Moyenne", "Moyenne"],
    "Duree max": ["~60s", "~6s", "~5s", "~10s", "~4s", "~2s"],
    "Resolution max": ["1080p", "720p", "768p", "720p", "576p", "512p"],
    "Cout / video": ["~$0.10-0.50", "Electricite", "Electricite", "Electricite", "Electricite", "Electricite"],
    "Latence": ["30-120s", "60-300s", "30-120s", "60-180s", "30-60s", "20-60s"],
    "Coherence temp.": ["Excellente", "Bonne", "Correcte", "Bonne", "Correcte", "Limitee"],
}

# Affichage tableau
header = f"{'Modele':<25} {'Type':<12} {'VRAM':<12} {'Qualite':<12} {'Duree':<10} {'Cout':<15}"
print(header)
print("-" * len(header))
for i in range(len(comparison_data['Modele'])):
    row = (f"{comparison_data['Modele'][i]:<25} "
           f"{comparison_data['Type'][i]:<12} "
           f"{comparison_data['VRAM'][i]:<12} "
           f"{comparison_data['Qualite'][i]:<12} "
           f"{comparison_data['Duree max'][i]:<10} "
           f"{comparison_data['Cout / video'][i]:<15}")
    print(row)

# Analyse cout pour volume de production
print("\n--- ANALYSE DE COUT POUR PRODUCTION ---")
print("=" * 45)

scenarios = [
    {"name": "Prototype (10 videos/mois)", "count": 10, "sora_cost": 0.30},
    {"name": "Production (100 videos/mois)", "count": 100, "sora_cost": 0.25},
    {"name": "Industriel (1000 videos/mois)", "count": 1000, "sora_cost": 0.20},
]

# Cout GPU local estime : amortissement RTX 4090 (~$1600) sur 3 ans + electricite
gpu_monthly_cost = 1600 / 36 + 50  # ~$94/mois
local_time_per_video = 120  # secondes
local_videos_per_hour = 3600 / local_time_per_video  # ~30

print(f"\n{'Scenario':<35} {'Sora/mois':>12} {'Local/mois':>12} {'Recommandation':>18}")
print("-" * 80)
for s in scenarios:
    sora_monthly = s['count'] * s['sora_cost']
    local_monthly = gpu_monthly_cost  # Cout fixe
    recommendation = "Cloud" if sora_monthly < local_monthly else "Local"
    print(f"  {s['name']:<33} ${sora_monthly:>10.2f} ${local_monthly:>10.2f} {recommendation:>18}")

print(f"\nNote : Le cout local inclut l'amortissement GPU ($1600/3 ans) + electricite (~$50/mois)")
print(f"Le seuil de rentabilite local est d'environ {int(gpu_monthly_cost / 0.25)} videos/mois")

--- COMPARAISON SORA VS MODELES LOCAUX ---
==================================================
Modele                    Type         VRAM         Qualite      Duree      Cout           
-------------------------------------------------------------------------------------------
Sora (OpenAI)             Cloud API    0 (cloud)    Tres haute   ~60s       ~$0.10-0.50    
HunyuanVideo              Local        ~40 GB       Haute        ~6s        Electricite    
LTX-Video                 Local        ~12 GB       Moyenne      ~5s        Electricite    
Wan 2.1                   Local        ~24 GB       Haute        ~10s       Electricite    
Stable Video Diffusion    Local        ~16 GB       Moyenne      ~4s        Electricite    
AnimateDiff               Local        ~12 GB       Moyenne      ~2s        Electricite    

--- ANALYSE DE COUT POUR PRODUCTION ---
=============================================

Scenario                               Sora/mois   Local/mois     Recommandation
--------------------------------------------------------------------------------
  Prototype (10 videos/mois)        $      3.00 $     94.44              Cloud
  Production (100 videos/mois)      $     25.00 $     94.44              Cloud
  Industriel (1000 videos/mois)     $    200.00 $     94.44              Local

Note : Le cout local inclut l'amortissement GPU ($1600/3 ans) + electricite (~$50/mois)
Le seuil de rentabilite local est d'environ 377 videos/mois

Exercice 3 : Calcul du seuil de rentabilite cloud vs local

Objectif : Implementer une fonction qui calcule le nombre de videos mensuelles a partir duquel la generation locale devient plus economique que le cloud (point de bascule).

Contexte : La decision entre cloud et local depend du volume de production. L’analyse de la section 4 montre que le seuil se situe autour de 377 videos/mois avec les paramètres par defaut, mais ces paramètres varient selon la configuration GPU et le cout API.

Indices : - Le cout cloud est lineaire : nb_videos * cout_par_video - Le cout local est fixe : amortissement_gpu_mensuel + electricite_mensuelle - # Étape 1 : Calculer le cout mensuel cloud pour un volume donne - # Étape 2 : Calculer le cout mensuel local (amortissement + electricite) - # Étape 3 : Trouver le volume ou les deux couts s’egalisent (point de bascule) - # Indice : Le point de bascule est cout_local_fixe / cout_par_video_cloud

def calculate_breakpoint(gpu_purchase_price: float = 1600,
                          amortization_years: float = 3,
                          monthly_electricity: float = 50,
                          cost_per_video_cloud: float = 0.25) -> dict:
    """
    Calcule le seuil de rentabilite entre generation cloud et locale.
    
    Args:
        gpu_purchase_price: Prix d'achat du GPU (USD)
        amortization_years: Duree d'amortissement en annees
        monthly_electricity: Cout electricite mensuel (USD)
        cost_per_video_cloud: Cout par video sur le cloud (USD)
    
    Returns:
        Dictionnaire avec:
        - monthly_local_cost: cout fixe mensuel local
        - breakpoint_videos: nombre de videos mensuelles au point de bascule
        - cloud_cost_at_breakpoint: cout cloud au point de bascule
        - recommendation: "cloud" ou "local" pour 100 videos/mois
    """
    # TODO etudiant : implementer le calcul du seuil de rentabilite
    result = None  # TODO etudiant : remplacer par l'analyse
    return result

print("Exercice a completer")
Exercice a completer

Interpretation : Comparaison cout et performance

Critere Cloud (Sora) Local (GPU)
Investissement initial Nul Eleve ($1600+ pour GPU)
Cout variable Par video Electricite uniquement
Seuil rentabilite < 400 videos/mois > 400 videos/mois
Maintenance Zero MAJ drivers, CUDA, modèles
Disponibilite 24/7, rate-limited Depend du hardware
Confidentialite Données transitent par OpenAI Traitement local

Points cles : 1. Pour le prototypage et les petits volumes, le cloud est plus economique 2. Pour la production a grande echelle, le GPU local est rentabilise en quelques mois 3. La confidentialite des données peut etre un facteur decisif (medical, defense) 4. La qualite Sora depasse actuellement les modèles locaux open-source

# Visualisation comparative des couts
volumes = np.arange(10, 1010, 10)
sora_costs = volumes * 0.25  # $0.25 par video
local_costs = np.full_like(volumes, gpu_monthly_cost, dtype=float)

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))

# Cout mensuel
ax1.plot(volumes, sora_costs, 'b-', linewidth=2, label='Sora (cloud)')
ax1.plot(volumes, local_costs, 'r--', linewidth=2, label='GPU local')
crossover = int(gpu_monthly_cost / 0.25)
ax1.axvline(x=crossover, color='gray', linestyle=':', alpha=0.7)
ax1.annotate(f'Seuil : {crossover} vid/mois', xy=(crossover, gpu_monthly_cost),
             xytext=(crossover + 100, gpu_monthly_cost + 50),
             arrowprops=dict(arrowstyle='->', color='gray'), fontsize=9, color='gray')
ax1.set_xlabel('Videos par mois')
ax1.set_ylabel('Cout mensuel ($)')
ax1.set_title('Cout mensuel : Cloud vs Local')
ax1.legend()
ax1.grid(True, alpha=0.3)

# Comparaison qualitative
categories = ['Qualite', 'Coherence\ntemporelle', 'Duree\nmax', 'Latence', 'Facilite\nsetup']
sora_scores = [9, 9, 9, 7, 10]
local_scores = [7, 6, 5, 5, 4]

x = np.arange(len(categories))
width = 0.35
ax2.bar(x - width/2, sora_scores, width, label='Sora', color='steelblue')
ax2.bar(x + width/2, local_scores, width, label='Local (best)', color='coral')
ax2.set_xticks(x)
ax2.set_xticklabels(categories, fontsize=9)
ax2.set_ylabel('Score (1-10)')
ax2.set_title('Comparaison qualitative')
ax2.legend()
ax2.set_ylim(0, 11)
ax2.grid(True, alpha=0.3, axis='y')

plt.suptitle('Analyse comparative : Sora API vs Generation Video Locale',
             fontsize=13, fontweight='bold')
plt.tight_layout()
plt.show()

La visualisation superpose les deux regimes de cout. Panneau de gauche : la courbe cloud (volumes * 0.25, 25 cents par video) croit lineairement avec le volume mensuel, tandis que la courbe locale est une droite plate au niveau de gpu_monthly_cost – le GPU coute le meme prix mensuel qu’on genere 10 ou 1000 videos. La ligne verticale grise annotee Seuil : N vid/mois marque leur croisement : en dessous, le cloud (partie bleue sous la rouge) est moins cher ; au-dessus, le local gagne systematiquement, et l’ecart se creuse avec le volume. Panneau de droite : les barres qualitatives rappellent que le cout n’est pas le seul axe – Sora domine sur la qualite (9 vs 7), la coherence temporelle (9 vs 6), la duree maximale (9 vs 5) et surtout la facilite de mise en place (10 vs 4), tandis que la comparaison compile les revers du cloud : dependance a un service externe et facturation a l’usage. Ensemble, les deux panneaux donnent la decision complete : le volume tranche le cout, les exigences qualitatives tranchent le reste.

# Mode interactif - Test de prompt personnalise
if notebook_mode == "interactive" and not skip_widgets:
    print("\n--- MODE INTERACTIF ---")
    print("=" * 40)
    print("Entrez un prompt pour generer une video de demonstration.")
    print("(Laissez vide pour passer)")
    
    try:
        custom_prompt = input("\nPrompt video (ou vide) : ").strip()
        
        if custom_prompt:
            print(f"Generation avec : {custom_prompt[:80]}...")
            result_custom = sora.text_to_video(custom_prompt, duration=3)
            print(f"Status : {result_custom['status']}, Frames : {len(result_custom.get('frames', []))}")
            
            if result_custom.get('frames'):
                fig, axes = plt.subplots(1, 4, figsize=(14, 3))
                idx_list = np.linspace(0, len(result_custom['frames']) - 1, 4, dtype=int)
                for i, idx in enumerate(idx_list):
                    axes[i].imshow(result_custom['frames'][idx])
                    axes[i].set_title(f"t={idx/video_fps_target:.1f}s", fontsize=9)
                    axes[i].axis('off')
                plt.suptitle(f"Prompt personnalise", fontsize=11, fontweight='bold')
                plt.tight_layout()
                plt.show()
        else:
            print("Mode interactif ignore")
    
    except (KeyboardInterrupt, EOFError):
        print("Mode interactif interrompu")
    except Exception as e:
        error_type = type(e).__name__
        if "StdinNotImplemented" in error_type:
            print("Mode interactif non disponible (execution automatisee)")
        else:
            print(f"Erreur : {error_type}")
else:
    print("Mode batch - Interface interactive desactivee")

--- MODE INTERACTIF ---
========================================
Entrez un prompt pour generer une video de demonstration.
(Laissez vide pour passer)
Mode interactif non disponible (execution automatisee)

En mode interactif, l’utilisateur peut tester un prompt personnalise et visualiser la generation. Le mode est volontairement gouverne et encadre : la cellule ne s’active que si le parametre Papermill notebook_mode vaut interactive ET que les widgets sont disponibles (not skip_widgets) – la sortie montree ici affiche la banniere du mode actif. Quatre sorties propres sont prevues selon le contexte : un prompt saisi part vers le meme wrapper text_to_video que la Section 2 (avec affichage de 4 frames echantillonnees de la video produite) ; un prompt vide affiche Mode interactif ignore ; une interruption ou une fin de flux (EOFError) affiche Mode interactif interrompu – c’est ce qui arrive dans un pipeline automatise, ou l’entree standard est fermee ; enfin hors mode interactif, la cellule se contente d’afficher Mode batch - Interface interactive desactivee. Ce garde-fou est le motif standard des notebooks executables a la fois par un humain dans Jupyter et par une CI.

# Statistiques de session
print("\n--- STATISTIQUES DE SESSION ---")
print("=" * 40)

print(f"Date : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Mode : {notebook_mode}")
print(f"API Sora disponible : {api_available}")
print(f"Mode mock : {sora.use_mock}")

print(f"\nAppels API effectues : {len(sora.call_log)}")
for i, call in enumerate(sora.call_log):
    print(f"  {i+1}. {call['method']} - Status: {call['status']}")

if save_results and OUTPUT_DIR.exists():
    generated_files = list(OUTPUT_DIR.glob('sora_*'))
    print(f"\nFichiers generes ({len(generated_files)}) :")
    for f in sorted(generated_files):
        size_kb = f.stat().st_size / 1024
        print(f"  {f.name} ({size_kb:.1f} KB)")

print(f"\nDependances utilisees :")
for dep, available in dependencies.items():
    status = "utilisee" if available else "non disponible"
    print(f"  {dep} : {status}")

print(f"\n--- PROCHAINES ETAPES ---")
print(f"1. Notebook 04-4 : Pipeline Video de Production (pipeline complet bout-en-bout)")
print(f"2. Revenir aux notebooks Audio pour combiner audio + video")

print(f"\nNotebook 04-3 Sora API Cloud Video termine - {datetime.now().strftime('%H:%M:%S')}")

--- STATISTIQUES DE SESSION ---
========================================
Date : 2026-08-10 19:58:22
Mode : interactive
API Sora disponible : True
Mode mock : False

Appels API effectues : 3
  1. text_to_video - Status: success
  2. text_to_video - Status: success
  3. image_to_video - Status: success

Fichiers generes (6) :
  sora_api_video_6a7a1076ac708191aa931779fcf029f503c01d36ff445b3a.mp4 (1915.8 KB)
  sora_api_video_6a7a10b7f010819184a8bca695620e960fb0597694bf8b66.mp4 (4398.8 KB)
  sora_api_video_6a7a10f9df7881919dcf7f0d9d77a7200ea9bd0f11e614ae.mp4 (1608.2 KB)
  sora_i2v_landscape.mp4 (107.5 KB)
  sora_t2v_nature.mp4 (56.6 KB)
  sora_t2v_urbain.mp4 (56.9 KB)

Dependances utilisees :
  openai : utilisee
  httpx : utilisee
  imageio : utilisee

--- PROCHAINES ETAPES ---
1. Notebook 04-4 : Pipeline Video de Production (pipeline complet bout-en-bout)
2. Revenir aux notebooks Audio pour combiner audio + video

Notebook 04-3 Sora API Cloud Video termine - 19:58:22

Exercice : Production Video avec Sora

Duree estimee : 45-60 minutes

Objectif

Utiliser l’API Sora pour créer une sequence video complete (2-3 scenes) avec un scénario coherent, en integrant la generation de texte, d’image et de video via l’API cloud.

Instructions

  1. Concevoir le scénario
    • Thème et arc narratif (intro, développement, conclusion)
    • 2-3 scenes avec descriptions detaillees
    • Coherence temporelle et visuelle
  2. Implementer la generation
    • Fonction de generation texte->video (Sora)
    • Generation image->video (animation)
    • Fusion des scenes avec transitions
  3. Gerer les ressources
    • Estimation des couts API
    • Monitoring de la consommation
    • Optimisation des prompts
  4. Assembler et evaluer
    • Video finale avec transitions
    • Evaluation de la qualite
    • Analyse cout/benefice vs modèles locaux

Indices :

  • Estimation cout Sora : ~$0.10-0.50 par video de 5s a 720p
  • Prompts Sora : inclure camera movement, style, duree
  • Transitions : cross-fond ou cut selon le montage

Critères de succes

Exercice Avance : Système Hybride Cloud + Local

Duree estimee : 120-180 minutes

Objectif

Developper un système de production video hybride qui choisit automatiquement entre generation cloud (Sora) et locale (HunyuanVideo, LTX) selon les contraintes de cout, de qualite et de temps.

Instructions

  1. Analyser les approches
    • Tableau comparatif : cout, qualite, vitesse, VRAM
    • Seuils de decision (budget, deadlines)
    • Cas d’usage optimaux pour chaque approche
  2. Concevoir le système de decision
    • Arbre de decision ou scoring system
    • Parametrage des seuils (cout max, temps max, qualite min)
    • Fallback chains (cloud -> local -> degrade)
  3. Implementer le scheduler
    • Fonction choose_generation_method(request)
    • Integration avec les APIs (Sora, diffusers)
    • Monitoring et logging
  4. Evaluer et optimiser
    • Tests sur divers scénarios
    • Mesure des economies realisees
    • Ajustement des seuils

Indices :

  • Facteurs de decision : budget, deadline, qualite requise, disponibilite GPU
  • Scoring : ponderer les facteurs selon le projet
  • Fallback : si cloud echoue ou depasse budget, passer en local

Critères de succes

# TODO: Scenario pour production Sora
my_scenario = [
    {
        "scene": 1,
        "description": "Un astronaute flotte vers une station spatiale, Terre en arriere-plan, lent mouvement de camera",
        "duration": 5
    },
    {
        "scene": 2,
        "description": "Interieur de la station, ecrans holographiques, scientifiques travaillent, zoom lent sur un terminal",
        "duration": 5
    },
    {
        "scene": 3,
        "description": "Vue de l'espace depuis un hublot, aurores boreales sur une planete, camera panoramique",
        "duration": 5
    }
]

# TODO: Implementer la generation via Sora
def generate_sora_scene(scene: dict) -> Dict:
    """
    Genere une scene via l'API Sora.

    Retourne: {"success": bool, "video_path": ..., "cost": ...}
    """
    # TODO: Appel API Sora (utiliser le wrapper du notebook)
    # Estimer le cout : duree x resolution
    pass

# TODO: Assembler les scenes
def assemble_scenes(scenes_videos: List[str]) -> str:
    """
    Assemble les scenes avec transitions.

    Transitions: cut, crossfade, dip_to_black
    """
    pass

# TODO: Generer et analyser
total_cost = 0
results = []  # TODO: Stocker les resultats de generation

for scene in my_scenario:
    result = generate_sora_scene(scene)
    if result and result.get('success'):
        results.append(result)
        total_cost += result.get('cost', 0)

if results:
    final_video = assemble_scenes([r['video_path'] for r in results])
    print(f"Video finale: {final_video}")
    print(f"Cout total: ${total_cost:.2f}")

Le planificateur hybride doit decider automatiquement de router chaque demande vers Sora ou un modele local – mais attention a la lecture de la sortie : le squelette est un stub d’exercice, et choose_method retourne pour l’instant la valeur constante "degraded" (marquee TODO etudiant). C’est pourquoi les trois scenarios de test – budget $1.0/deadline 300 s, budget $0.1/deadline 60 s, budget $5.0/deadline 120 s – ressortent tous avec Method: degraded, Status: stub : aucune decision reelle n’est encore prise. La logique a implementer est documentee dans la docstring : si le budget et la deadline couvrent Sora, partir dans le cloud ; sinon verifier la VRAM disponible pour Hunyuan puis LTX en local ; a defaut, degrader la resolution plutot qu’echouer. Les quatre facteurs de decision (budget max, deadline max, qualite minimale, VRAM disponible) delimitent l’espace du choix – c’est le motif “resource-aware routing” des pipelines de production.

# TODO: Systeme de decision hybride
class HybridVideoScheduler:
    """
    Scheduler hybride cloud/local pour la generation video.

    Facteurs de decision:
    - budget_max: cout maximum autorise
    - deadline_max: temps maximum (secondes)
    - quality_min: score de qualite minimum
    - vram_available: VRAM GPU locale (GB)
    """

    def __init__(self, config: Dict):
        self.cloud_api = None  # TODO: Initialiser SoraAPIWrapper
        self.local_models = {}  # TODO: Charger les modeles locaux
        self.decision_matrix = {}  # TODO: Matrice de decision

    def choose_method(self, request: Dict) -> str:
        """
        Choisit la methode de generation optimale.

        Retourne: "sora", "hunyuan", "ltx", or "degraded"
        """
        # TODO: Implementer la logique de decision
        # - Estimer couts et temps pour chaque methode
        # - Comparer aux contraintes
        # - Choisir la meilleure option

        # Pseudo-code:
        # if budget >= sora_cost and deadline >= sora_time:
        #     return "sora"
        # elif vram_available >= hunyuan_vram:
        #     return "hunyuan"
        # elif vram_available >= ltx_vram:
        #     return "ltx"
        # else:
        #     return "degraded"  # version basse resolution
        return "degraded"  # TODO etudiant : implementer la logique

    def generate(self, request: Dict) -> Dict:
        """
        Genere la video avec la methode optimale.

        Inclut les fallbacks automatiques.
        """
        method = self.choose_method(request)
        # TODO: Implementer les tentatives avec fallbacks
        # - Essayer la methode choisie
        # - En cas d'echec, essayer la methode suivante
        # - Logger les decisions et resultats
        return {"method": method, "status": "stub"}  # TODO etudiant


# TODO: Tester sur divers scenarios
test_requests = [
    {"prompt": "...", "budget": 1.0, "deadline": 300, "quality": "high"},
    {"prompt": "...", "budget": 0.1, "deadline": 60, "quality": "medium"},
    {"prompt": "...", "budget": 5.0, "deadline": 120, "quality": "maximum"},
]

# Configuration par defaut pour le scheduler
config = {
    "budget_max": 5.0,
    "deadline_max": 300,
    "quality_min": 0.5,
    "vram_available": 24,
}

scheduler = HybridVideoScheduler(config)
for req in test_requests:
    result = scheduler.generate(req)
    print(f"Request: budget=${req['budget']}, deadline={req['deadline']}s -> Method: {result['method']}, Status: {result['status']}")
Request: budget=$1.0, deadline=300s -> Method: degraded, Status: stub
Request: budget=$0.1, deadline=60s -> Method: degraded, Status: stub
Request: budget=$5.0, deadline=120s -> Method: degraded, Status: stub

Conclusion : ce que ce notebook etablit

Trois idees a retenir. (1) L’API Sora comme contrat asynchrone : le notebook a exerce le cycle complet d’une API video – soumission d’un job, sondage de statut jusqu’a completion, telechargement du produit, mesure du temps (environ 65 s par generation dans la Section 2) – derriere un wrapper qui rend ce cycle reproductible, y compris en mode simule quand la cle manque. (2) Le cout comme fonction du volume : la comparaison chiffree de la Section 4 montre qu’il n’existe pas de reponse universelle a “cloud ou local” ; il existe un seuil de volume mensuel, calculable (exercice 3) et visible sur la courbe, qui separe les deux regimes – cout marginal lineaire d’un cote, cout fixe plat de l’autre. (3) Le routage hybride comme synthese : le squelette de planificateur de la derniere section transforme ce seuil en decision par requete – chaque demande porte son budget, sa deadline et son exigence de qualite, et le systeme doit choisir le moteur adapte. Les deux exercices terminaux (production avec Sora, systeme hybride complet) prolongent exactement ces trois axes.

Prochaines etapes dans la serie : 04-4-Production-Video-Pipeline chaine ces briques en un pipeline reel, et 04-5/04-5b repetent l’exercice de comparaison sur les API cloud MiniMax.

Retour au sommet