from transformers import CLIPModel, CLIPProcessor
from PIL import Image
import torch
import requests

# Load CLIP
model     = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
model.eval()

# Load a real image
url = "https://upload.wikimedia.org/wikipedia/commons/thumb/4/43/Cute_dog.jpg/1200px-Cute_dog.jpg"
image = Image.open(requests.get(url, stream=True).raw)

# Candidate text descriptions
texts = [
    "a photo of a dog",
    "a photo of a cat",
    "a photo of a car",
    "a photo of a bird",
    "a photo of a dog running in a park",
]

# Process inputs
inputs = processor(
    text=texts,
    images=image,
    return_tensors="pt",
    padding=True
)

# Get embeddings and compute similarity
with torch.no_grad():
    outputs          = model(**inputs)
    image_embedding  = outputs.image_embeds    # (1, 512)
    text_embeddings  = outputs.text_embeds     # (5, 512)

    # Cosine similarity between image and each text
    image_norm  = image_embedding  / image_embedding.norm(dim=-1, keepdim=True)
    text_norms  = text_embeddings  / text_embeddings.norm(dim=-1, keepdim=True)
    similarities = (image_norm @ text_norms.T).squeeze()   # (5,)

    # Convert to probabilities (softmax over texts)
    probs = similarities.softmax(dim=0)

print("CLIP zero-shot classification results:")
for text, prob, sim in zip(texts, probs, similarities):
    print(f"  {prob*100:5.1f}%  (sim={sim:.3f})  |  '{text}'")

from transformers import CLIPModel
import torch

model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")

# Inspect the architecture
total_params = sum(p.numel() for p in model.parameters()) / 1e6
print(f"Total CLIP parameters: {total_params:.1f}M")

# Text encoder: Transformer
text_params   = sum(p.numel() for p in model.text_model.parameters()) / 1e6
print(f"Text encoder:  {text_params:.1f}M parameters")
print(f"  Architecture: 12-layer Transformer, 512-dim hidden, 512-dim output")

# Vision encoder: ViT (Vision Transformer)
vision_params = sum(p.numel() for p in model.vision_model.parameters()) / 1e6
print(f"Vision encoder: {vision_params:.1f}M parameters")
print(f"  Architecture: ViT-B/32 — 12-layer Transformer on 32×32 image patches")

# The embeddings are projected to a shared 512-dim space
print(f"\nShared embedding dimension: 512")
print("Both text and image embeddings are L2-normalized before similarity computation")

# Test with random inputs
dummy_text  = torch.randint(0, 1000, (2, 77))  # 2 text sequences, max 77 tokens
dummy_image = torch.randn(2, 3, 224, 224)       # 2 RGB images, 224x224

with torch.no_grad():
    t_emb = model.get_text_features(dummy_text)
    i_emb = model.get_image_features(dummy_image)

print(f"\nText embedding shape:  {t_emb.shape}")   # (2, 512)
print(f"Image embedding shape: {i_emb.shape}")   # (2, 512)

from transformers import CLIPModel, CLIPProcessor
from datasets import load_dataset
import torch

model     = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
model.eval()

def clip_zero_shot_classify(image, class_names, template="a photo of a {}"):
    """
    Zero-shot image classification with CLIP.
    class_names: list of class name strings
    template: how to turn a class name into a text prompt
    """
    # Create text prompts from class names
    text_prompts = [template.format(cn) for cn in class_names]

    inputs = processor(
        text=text_prompts,
        images=image,
        return_tensors="pt",
        padding=True,
    )

    with torch.no_grad():
        outputs = model(**inputs)
        image_emb = outputs.image_embeds / outputs.image_embeds.norm(dim=-1, keepdim=True)
        text_embs  = outputs.text_embeds  / outputs.text_embeds.norm(dim=-1, keepdim=True)
        logits     = (image_emb @ text_embs.T).squeeze() * model.logit_scale.exp()
        probs      = logits.softmax(dim=-1)

    results = {cn: prob.item() for cn, prob in zip(class_names, probs)}
    return dict(sorted(results.items(), key=lambda x: x[1], reverse=True))


# Compare zero-shot vs supervised on CIFAR-10
# CIFAR-10 classes
cifar10_classes = [
    "airplane", "automobile", "bird", "cat", "deer",
    "dog", "frog", "horse", "ship", "truck"
]

# Load a few CIFAR-10 test images
dataset = load_dataset("cifar10", split="test[:10]")

correct = 0
for sample in dataset:
    image = sample['img']
    true_label = cifar10_classes[sample['label']]
    predictions = clip_zero_shot_classify(image, cifar10_classes)
    predicted = max(predictions, key=predictions.get)
    is_correct = predicted == true_label
    correct += is_correct
    print(f"  True: {true_label:12s} | Predicted: {predicted:12s} | {'✓' if is_correct else '✗'}")

print(f"\nZero-shot accuracy: {correct}/10 = {correct*10}%")
print("CLIP ViT-B/32 on full CIFAR-10: ~65% zero-shot vs ~95% supervised ResNet-50")

import torch
from transformers import CLIPModel, CLIPProcessor
from PIL import Image
import numpy as np

def build_image_index(image_paths, model, processor):
    """Pre-compute and store embeddings for all images in a collection."""
    embeddings = []
    for path in image_paths:
        img    = Image.open(path).convert("RGB")
        inputs = processor(images=img, return_tensors="pt")
        with torch.no_grad():
            emb = model.get_image_features(**inputs)
        emb = emb / emb.norm(dim=-1, keepdim=True)
        embeddings.append(emb.squeeze().numpy())
    return np.array(embeddings)


def text_to_image_search(query, image_paths, image_embeddings, model, processor, top_k=3):
    """Find images that best match a text query."""
    # Embed the text query
    inputs = processor(text=[query], return_tensors="pt", padding=True)
    with torch.no_grad():
        text_emb = model.get_text_features(**inputs)
    text_emb = text_emb / text_emb.norm(dim=-1, keepdim=True)
    text_emb = text_emb.squeeze().numpy()

    # Cosine similarity
    sims = image_embeddings @ text_emb
    top_indices = sims.argsort()[-top_k:][::-1]

    return [(image_paths[i], float(sims[i])) for i in top_indices]


# Usage example (assuming you have a collection of images)
# image_paths = list(Path("./images").glob("*.jpg"))
# embeddings = build_image_index(image_paths, model, processor)
# results = text_to_image_search(
#     "a red car on a mountain road",
#     image_paths, embeddings, model, processor
# )
# for path, score in results:
#     print(f"  Score: {score:.3f}  |  {path}")

print("CLIP semantic search: find images matching natural language descriptions")
print("Applications: Google Photos (find 'beach sunset'), Unsplash (stock photo search)")

from transformers import LlavaForConditionalGeneration, AutoProcessor
from PIL import Image
import torch
import requests

# Load LLaVA-1.5 (7B model — needs ~16GB VRAM for fp16)
model_id = "llava-hf/llava-1.5-7b-hf"

# For memory efficiency on smaller GPUs, load in 4-bit
from transformers import BitsAndBytesConfig
quantization_config = BitsAndBytesConfig(load_in_4bit=True)

processor = AutoProcessor.from_pretrained(model_id)
model = LlavaForConditionalGeneration.from_pretrained(
    model_id,
    quantization_config=quantization_config,
    device_map="auto",
)

# Load an image
url   = "https://llava-vl.github.io/static/images/view.jpg"
image = Image.open(requests.get(url, stream=True).raw)

# Ask questions about the image
questions = [
    "What is happening in this image?",
    "How many people are in the image?",
    "What time of day does it appear to be?",
    "Describe the weather conditions.",
]

for question in questions:
    # LLaVA conversation format
    conversation = [
        {
            "role": "user",
            "content": [
                {"type": "image"},
                {"type": "text", "text": question},
            ],
        },
    ]

    prompt = processor.apply_chat_template(conversation, add_generation_prompt=True)
    inputs = processor(images=image, text=prompt, return_tensors="pt").to(model.device)

    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=128,
            do_sample=False,
        )

    # Decode only the generated tokens (not the prompt)
    generated = outputs[0][inputs["input_ids"].shape[1]:]
    answer    = processor.decode(generated, skip_special_tokens=True)
    print(f"Q: {question}")
    print(f"A: {answer}\n")

from openai import OpenAI
import base64
from pathlib import Path

client = OpenAI()  # requires OPENAI_API_KEY

def encode_image_base64(image_path):
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode('utf-8')


def gpt4v_query(image_path, question, detail="high"):
    """
    Query GPT-4V with an image and text question.
    detail: "low" (512 tokens, faster/cheaper) or "high" (up to 4096 tokens for resolution)
    """
    base64_image = encode_image_base64(image_path)

    response = client.chat.completions.create(
        model="gpt-4o",   # GPT-4o has best vision capabilities
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}",
                            "detail": detail,
                        }
                    },
                    {
                        "type": "text",
                        "text": question,
                    },
                ],
            }
        ],
        max_tokens=500,
    )
    return response.choices[0].message.content


# GPT-4V capabilities beyond LLaVA:
# 1. Read and interpret complex charts/graphs
# 2. Understand math equations written on whiteboards
# 3. Interpret X-ray or medical images
# 4. Read handwritten text
# 5. Understand diagrams and technical schematics
# 6. Code completion from a screenshot of code

print("GPT-4V: most capable commercial VLM")
print("LLaVA-1.5-13B: best open-source alternative, competitive for many tasks")

from transformers import Blip2ForConditionalGeneration, Blip2Processor
from PIL import Image
import requests
import torch

# Load BLIP-2 (OPT-2.7B backbone — smaller than 7B LLaVA)
processor = Blip2Processor.from_pretrained("Salesforce/blip2-opt-2.7b")
model = Blip2ForConditionalGeneration.from_pretrained(
    "Salesforce/blip2-opt-2.7b",
    torch_dtype=torch.float16,
    device_map="auto",
)
model.eval()

# Load a real image
url   = "https://storage.googleapis.com/sfr-vision-language-research/BLIP/demo.jpg"
image = Image.open(requests.get(url, stream=True).raw).convert('RGB')


# ── Task 1: Image Captioning (no question) ──
inputs = processor(images=image, return_tensors="pt").to(model.device, torch.float16)
with torch.no_grad():
    generated_ids = model.generate(**inputs, max_new_tokens=50)
caption = processor.batch_decode(generated_ids, skip_special_tokens=True)[0].strip()
print(f"Caption: {caption}")


# ── Task 2: Visual Question Answering ──
vqa_questions = [
    "What animals are in this photo?",
    "What is the weather like?",
    "How many people are visible?",
    "What is in the background?",
]

for question in vqa_questions:
    # Prepend question to guide generation
    inputs = processor(
        images=image,
        text=f"Question: {question} Answer:",
        return_tensors="pt"
    ).to(model.device, torch.float16)

    with torch.no_grad():
        generated_ids = model.generate(**inputs, max_new_tokens=30)
    answer = processor.batch_decode(generated_ids, skip_special_tokens=True)[0].strip()
    print(f"Q: {question}")
    print(f"A: {answer}\n")

import whisper
import torch

# Load Whisper (pip install openai-whisper)
# Model sizes: tiny (39M), base (74M), small (244M), medium (769M), large (1550M)
model = whisper.load_model("base")   # good balance of speed and accuracy

# Transcribe an audio file
def transcribe_audio(audio_path, language=None, task="transcribe"):
    """
    Transcribe or translate an audio file with Whisper.
    task: "transcribe" (keep original language) or "translate" (to English)
    language: force a specific language (e.g., "French") or None for auto-detection
    """
    options = {}
    if language:
        options['language'] = language
    if task == "translate":
        options['task'] = 'translate'

    result = model.transcribe(audio_path, **options)

    return {
        "text":     result["text"],
        "language": result.get("language", "detected"),
        "segments": result.get("segments", []),
    }

# Example transcription
result = transcribe_audio("meeting_recording.mp3")
print(f"Detected language: {result['language']}")
print(f"Transcript: {result['text']}")

# Word-level timestamps (with verbose_json)
result_detailed = model.transcribe(
    "meeting_recording.mp3",
    word_timestamps=True,
    verbose=False,
)
print("\nWord-level timestamps (first 5 words):")
if result_detailed.get("segments"):
    for segment in result_detailed["segments"][:2]:
        for word in segment.get("words", [])[:5]:
            print(f"  [{word['start']:.2f}s – {word['end']:.2f}s]: {word['word']}")

# Whisper via Hugging Face transformers (alternative API)
from transformers import WhisperForConditionalGeneration, WhisperProcessor
import torchaudio

processor = WhisperProcessor.from_pretrained("openai/whisper-small")
model_hf  = WhisperForConditionalGeneration.from_pretrained("openai/whisper-small")
model_hf.config.forced_decoder_ids = None

def transcribe_hf(audio_path):
    """Transcribe using Hugging Face transformers API."""
    # Load audio (resample to 16kHz — Whisper's required sample rate)
    waveform, sample_rate = torchaudio.load(audio_path)
    if sample_rate != 16000:
        resampler = torchaudio.transforms.Resample(sample_rate, 16000)
        waveform  = resampler(waveform)
    waveform = waveform.mean(dim=0)  # convert to mono

    # Extract log-Mel features
    inputs = processor(
        waveform.numpy(),
        sampling_rate=16000,
        return_tensors="pt",
    )

    # Generate transcription
    with torch.no_grad():
        predicted_ids = model_hf.generate(inputs.input_features)

    transcript = processor.batch_decode(predicted_ids, skip_special_tokens=True)
    return transcript[0]

# Whisper performance benchmarks:
print("\nWhisper Word Error Rate (WER) on LibriSpeech test-clean:")
print("  tiny:   WER = 9.8%   (fastest, use on-device)")
print("  base:   WER = 7.4%   (good for general use)")
print("  small:  WER = 5.5%   (recommended for most tasks)")
print("  medium: WER = 4.2%   (near-human on clean speech)")
print("  large:  WER = 3.0%   (human-level, ~5% on spontaneous speech)")

from transformers import (
    CLIPModel, CLIPProcessor,
    Blip2ForConditionalGeneration, Blip2Processor,
    WhisperForConditionalGeneration, WhisperProcessor,
)
from PIL import Image
import torch, numpy as np


class MultimodalProductCatalogue:
    """
    Automated product catalog pipeline:
    Input:  product photo + audio description
    Output: structured catalog entry + visual search embedding
    """

    def __init__(self):
        device = "cuda" if torch.cuda.is_available() else "cpu"
        self.device = device

        # Visual encoder for search (CLIP)
        self.clip_model = CLIPModel.from_pretrained(
            "openai/clip-vit-base-patch32"
        ).to(device).eval()
        self.clip_proc = CLIPProcessor.from_pretrained(
            "openai/clip-vit-base-patch32"
        )

        # Image captioning (BLIP-2)
        # Note: in production, use a 7B VLM for richer descriptions
        print("Loading BLIP-2 for image analysis...")

        # Speech recognition (Whisper)
        self.whisper = WhisperForConditionalGeneration.from_pretrained(
            "openai/whisper-base"
        ).to(device).eval()
        self.whisper_proc = WhisperProcessor.from_pretrained("openai/whisper-base")

        print("Multimodal pipeline ready")

    def process_product(self, image_path: str, audio_path: str) -> dict:
        """
        Complete pipeline: photo + audio → catalog entry + search embedding.
        """
        # Step 1: Transcribe audio description
        import torchaudio
        waveform, sr = torchaudio.load(audio_path)
        if sr != 16000:
            waveform = torchaudio.transforms.Resample(sr, 16000)(waveform)
        waveform = waveform.mean(dim=0)
        w_inputs = self.whisper_proc(
            waveform.numpy(), sampling_rate=16000, return_tensors="pt"
        ).to(self.device)
        with torch.no_grad():
            w_ids = self.whisper.generate(w_inputs.input_features)
        spoken_description = self.whisper_proc.batch_decode(
            w_ids, skip_special_tokens=True
        )[0]

        # Step 2: Analyze image with CLIP for attribute extraction
        image = Image.open(image_path).convert("RGB")
        attribute_queries = {
            "color":   ["red product", "blue product", "black product",
                          "white product", "green product", "yellow product"],
            "material": ["plastic item", "metal item", "wooden item",
                          "fabric item", "glass item", "leather item"],
            "size":     ["small compact item", "medium-sized item", "large bulky item"],
        }

        detected_attributes = {}
        for attr_name, options in attribute_queries.items():
            inputs = self.clip_proc(text=options, images=image,
                                    return_tensors="pt", padding=True).to(self.device)
            with torch.no_grad():
                logits = self.clip_model(**inputs).logits_per_image.squeeze()
            best_idx = logits.argmax().item()
            detected_attributes[attr_name] = options[best_idx].split()[0]  # first word

        # Step 3: Generate visual search embedding (CLIP image embedding)
        img_inputs = self.clip_proc(images=image, return_tensors="pt").to(self.device)
        with torch.no_grad():
            search_embedding = self.clip_model.get_image_features(**img_inputs)
        search_embedding = (
            search_embedding / search_embedding.norm(dim=-1, keepdim=True)
        ).squeeze().cpu().numpy()

        # Step 4: Assemble catalog entry
        catalogue_entry = {
            "audio_description": spoken_description,
            "visual_attributes": detected_attributes,
            "search_embedding":  search_embedding.tolist(),  # 512-dim CLIP embedding
            "embedding_dim":     len(search_embedding),
        }

        return catalogue_entry


# Illustrative demo output (simulating what the pipeline produces)
demo_output = {
    "audio_description": "This is a compact blue wireless speaker with a fabric cover.",
    "visual_attributes": {
        "color":   "blue",
        "material": "fabric",
        "size":     "small",
    },
    "embedding_dim": 512,
    "search_embedding": "[...512-dimensional CLIP embedding for visual search...]",
}

print("Multimodal catalog entry:")
for key, val in demo_output.items():
    if key != "search_embedding":
        print(f"  {key}: {val}")

# Example: multimodal reasoning with GPT-4o (API)
from openai import OpenAI
import base64

client = OpenAI()

def multimodal_reason(text_question, image_path=None, audio_path=None):
    """
    Query GPT-4o with text, and optionally image and audio.
    """
    messages = [{"role": "user", "content": []}]

    # Add image if provided
    if image_path:
        with open(image_path, "rb") as f:
            img_b64 = base64.b64encode(f.read()).decode()
        messages[0]["content"].append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"},
        })

    # Add audio if provided (GPT-4o audio preview)
    if audio_path:
        with open(audio_path, "rb") as f:
            audio_b64 = base64.b64encode(f.read()).decode()
        messages[0]["content"].append({
            "type": "input_audio",
            "input_audio": {
                "data": audio_b64,
                "format": "mp3",
            }
        })

    # Add text question
    messages[0]["content"].append({"type": "text", "text": text_question})

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=500,
    )
    return response.choices[0].message.content


# Example use cases:
use_cases = [
    "Describe the chart and identify the trend",         # chart image
    "What medication is shown in this prescription?",   # medical image
    "Transcribe and summarize this meeting clip",        # audio
    "What bug causes the error in this screenshot?",     # code/UI screenshot
    "What's the issue with this architectural diagram?", # diagram
]
for case in use_cases:
    print(f"  '{case}'")

# MusicGen: generate music from text description
# pip install audiocraft
from audiocraft.models import MusicGen
from audiocraft.data.audio import audio_write

model = MusicGen.get_pretrained('facebook/musicgen-small')
model.set_generation_params(duration=8)  # 8 seconds

# Generate music from description
descriptions = [
    "upbeat jazz piano with light drumming",
    "ambient electronic music with deep bass",
    "classical violin solo with reverb",
]

wav = model.generate(descriptions)
for i, one_wav in enumerate(wav):
    audio_write(f'music_{i}', one_wav.cpu(), model.sample_rate, strategy="loudness")

print("MusicGen: text → music generation")
print("AudioCraft: broader suite including sound effects")
print("Bark: text → speech with emotion, different voices, non-verbal sounds")