Update app.py

app.py

@@ -1,170 +1,60 @@
 import os
-import pathlib
-import tempfile
 from collections.abc import Iterator
 from threading import Thread
 
-import av
 import gradio as gr
 import spaces
 import torch
-from gradio.utils import get_upload_folder
 from transformers import AutoModelForImageTextToText, AutoProcessor
 from transformers.generation.streamers import TextIteratorStreamer
 
+# --- Model- und Prozessor-Setup ---
+# Lädt das Modell und den Prozessor von Hugging Face.
+# device_map="auto" weist das Modell automatisch der verfügbaren Hardware (GPU/CPU) zu.
+# torch_dtype=torch.bfloat16 verwendet eine speichereffizientere Datenart für schnellere Berechnungen.
 model_id = "google/gemma-3n-E4B-it"
 
 processor = AutoProcessor.from_pretrained(model_id)
 model = AutoModelForImageTextToText.from_pretrained(model_id, device_map="auto", torch_dtype=torch.bfloat16)
 
-IMAGE_FILE_TYPES = (".jpg", ".jpeg", ".png", ".webp")
-VIDEO_FILE_TYPES = (".mp4", ".mov", ".webm")
-AUDIO_FILE_TYPES = (".mp3", ".wav")
-
-GRADIO_TEMP_DIR = get_upload_folder()
-
-TARGET_FPS = int(os.getenv("TARGET_FPS", "3"))
-MAX_FRAMES = int(os.getenv("MAX_FRAMES", "30"))
+# --- Konstanten ---
+# Definiert die maximal zulässige Anzahl von Eingabe-Tokens, um Speicherfehler zu vermeiden.
 MAX_INPUT_TOKENS = int(os.getenv("MAX_INPUT_TOKENS", "10_000"))
 
 
-def get_file_type(path: str) -> str:
-    if path.endswith(IMAGE_FILE_TYPES):
-        return "image"
-    if path.endswith(VIDEO_FILE_TYPES):
-        return "video"
-    if path.endswith(AUDIO_FILE_TYPES):
-        return "audio"
-    error_message = f"Unsupported file type: {path}"
-    raise ValueError(error_message)
-
-
-def count_files_in_new_message(paths: list[str]) -> tuple[int, int]:
-    video_count = 0
-    non_video_count = 0
-    for path in paths:
-        if path.endswith(VIDEO_FILE_TYPES):
-            video_count += 1
-        else:
-            non_video_count += 1
-    return video_count, non_video_count
-
-
-def validate_media_constraints(message: dict) -> bool:
-    video_count, non_video_count = count_files_in_new_message(message["files"])
-    if video_count > 1:
-        gr.Warning("Only one video is supported.")
-        return False
-    if video_count == 1 and non_video_count > 0:
-        gr.Warning("Mixing images and videos is not allowed.")
-        return False
-    return True
-
-
-def extract_frames_to_tempdir(
-    video_path: str,
-    target_fps: float,
-    max_frames: int | None = None,
-    parent_dir: str | None = None,
-    prefix: str = "frames_",
-) -> str:
-    temp_dir = tempfile.mkdtemp(prefix=prefix, dir=parent_dir)
-
-    container = av.open(video_path)
-    video_stream = container.streams.video[0]
-
-    if video_stream.duration is None or video_stream.time_base is None:
-        raise ValueError("video_stream is missing duration or time_base")
-
-    time_base = video_stream.time_base
-    duration = float(video_stream.duration * time_base)
-    interval = 1.0 / target_fps
-
-    total_frames = int(duration * target_fps)
-    if max_frames is not None:
-        total_frames = min(total_frames, max_frames)
-
-    target_times = [i * interval for i in range(total_frames)]
-    target_index = 0
-
-    for frame in container.decode(video=0):
-        if frame.pts is None:
-            continue
-
-        timestamp = float(frame.pts * time_base)
-
-        if target_index < len(target_times) and abs(timestamp - target_times[target_index]) < (interval / 2):
-            frame_path = pathlib.Path(temp_dir) / f"frame_{target_index:04d}.jpg"
-            frame.to_image().save(frame_path)
-            target_index += 1
-
-            if max_frames is not None and target_index >= max_frames:
-                break
-
-    container.close()
-    return temp_dir
-
-
-def process_new_user_message(message: dict) -> list[dict]:
-    if not message["files"]:
-        return [{"type": "text", "text": message["text"]}]
-
-    file_types = [get_file_type(path) for path in message["files"]]
-
-    if len(file_types) == 1 and file_types[0] == "video":
-        gr.Info(f"Video will be processed at {TARGET_FPS} FPS, max {MAX_FRAMES} frames in this Space.")
-
-        temp_dir = extract_frames_to_tempdir(
-            message["files"][0],
-            target_fps=TARGET_FPS,
-            max_frames=MAX_FRAMES,
-            parent_dir=GRADIO_TEMP_DIR,
-        )
-        paths = sorted(pathlib.Path(temp_dir).glob("*.jpg"))
-        return [
-            {"type": "text", "text": message["text"]},
-            *[{"type": "image", "image": path.as_posix()} for path in paths],
-        ]
-
-    return [
-        {"type": "text", "text": message["text"]},
-        *[{"type": file_type, file_type: path} for path, file_type in zip(message["files"], file_types, strict=True)],
-    ]
-
-
-def process_history(history: list[dict]) -> list[dict]:
-    messages = []
-    current_user_content: list[dict] = []
-    for item in history:
-        if item["role"] == "assistant":
-            if current_user_content:
-                messages.append({"role": "user", "content": current_user_content})
-                current_user_content = []
-            messages.append({"role": "assistant", "content": [{"type": "text", "text": item["content"]}]})
-        else:
-            content = item["content"]
-            if isinstance(content, str):
-                current_user_content.append({"type": "text", "text": content})
-            else:
-                filepath = content[0]
-                file_type = get_file_type(filepath)
-                current_user_content.append({"type": file_type, file_type: filepath})
-    return messages
-
-
 @spaces.GPU(duration=120)
 @torch.inference_mode()
-def generate(message: dict, history: list[dict], system_prompt: str = "", max_new_tokens: int = 512) -> Iterator[str]:
-    if not validate_media_constraints(message):
+def transcribe_german(audio_input: str | None) -> Iterator[str]:
+    """
+    Diese Funktion nimmt einen Audiopfad entgegen, transkribiert ihn mit dem Gemma-Modell
+    ins Deutsche und gibt die Transkription als Text-Stream zurück.
+
+    Args:
+        audio_input: Der Dateipfad zur Audiodatei, die transkribiert werden soll.
+                     Kann None sein, wenn keine Eingabe erfolgt.
+
+    Yields:
+        Einen String-Iterator, der das Transkript Stück für Stück ausgibt.
+    """
+    if audio_input is None:
+        gr.Warning("Bitte stellen Sie eine Audiodatei zur Verfügung oder nehmen Sie Audio auf.")
         yield ""
         return
 
-    messages = []
-    if system_prompt:
-        messages.append({"role": "system", "content": [{"type": "text", "text": system_prompt}]})
-    messages.extend(process_history(history))
-    messages.append({"role": "user", "content": process_new_user_message(message)})
+    # Die Eingabe für das Modell vorbereiten.
+    # Wir geben dem Modell eine explizite Anweisung ("Transkribiere dies auf Deutsch.")
+    # und die Audiodatei.
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "Transkribiere dies auf Deutsch."},
+                {"type": "audio", "audio": audio_input},
+            ],
+        }
+    ]
 
+    # Das Chat-Template des Prozessors verwenden, um die Eingabe korrekt zu formatieren.
     inputs = processor.apply_chat_template(
         messages,
         add_generation_prompt=True,
@@ -172,87 +62,69 @@ def generate(message: dict, history: list[dict], system_prompt: str = "", max_ne
         return_dict=True,
         return_tensors="pt",
     )
+
+    # Überprüfen, ob die Eingabe die maximale Token-Länge überschreitet.
     n_tokens = inputs["input_ids"].shape[1]
     if n_tokens > MAX_INPUT_TOKENS:
         gr.Warning(
-            f"Input too long. Max {MAX_INPUT_TOKENS} tokens. Got {n_tokens} tokens. This limit is set to avoid CUDA out-of-memory errors in this Space."
+            f"Eingabe zu lang. Max. {MAX_INPUT_TOKENS} Tokens. Habe {n_tokens} Tokens erhalten. "
+            "Dieses Limit dient dazu, CUDA-Speicherfehler in diesem Space zu vermeiden."
         )
         yield ""
         return
 
+    # Die formatierten Eingaben auf das Gerät des Modells verschieben.
     inputs = inputs.to(device=model.device, dtype=torch.bfloat16)
 
+    # Einen Streamer einrichten, um die Ausgabe des Modells in Echtzeit zu erhalten.
     streamer = TextIteratorStreamer(processor, timeout=30.0, skip_prompt=True, skip_special_tokens=True)
+    
+    # Die Argumente für die Generierung definieren.
     generate_kwargs = dict(
         inputs,
         streamer=streamer,
-        max_new_tokens=max_new_tokens,
+        max_new_tokens=700,  # Ausreichend für die meisten Transkriptionen
         do_sample=False,
-        disable_compile=True,
+        disable_compile=True, # Kompilierung für schnellere Inferenz deaktivieren (kann bei manchen Setups zu Problemen führen)
     )
+    
+    # Die Generierung in einem separaten Thread starten, damit die UI nicht blockiert wird.
     t = Thread(target=model.generate, kwargs=generate_kwargs)
     t.start()
 
+    # Den generierten Text durchlaufen und schrittweise ausgeben.
     output = ""
     for delta in streamer:
         output += delta
         yield output
 
 
-examples = [
-    [
-        {
-            "text": "What is the capital of France?",
-            "files": [],
-        }
-    ],
-    [
-        {
-            "text": "Describe this image in detail.",
-            "files": ["assets/cat.jpeg"],
-        }
-    ],
-    [
-        {
-            "text": "Transcribe the following speech segment in English.",
-            "files": ["assets/speech.wav"],
-        }
+# --- Gradio UI Setup ---
+# Erstellt die Benutzeroberfläche mit Gradio.
+demo = gr.Interface(
+    fn=transcribe_german,
+    inputs=[
+        gr.Audio(sources=["upload", "microphone"], type="filepath", label="Audio hochladen oder aufnehmen")
     ],
-    [
-        {
-            "text": "Transcribe the following speech segment in English.",
-            "files": ["assets/speech2.wav"],
-        }
+    outputs=[
+        gr.Textbox(label="Deutsches Transkript", interactive=False)
     ],
-    [
-        {
-            "text": "Describe this video",
-            "files": ["assets/holding_phone.mp4"],
+    title="Audio-zu-Text Transkription (Deutsch)",
+    description="Laden Sie eine Audiodatei hoch oder nehmen Sie Audio mit Ihrem Mikrofon auf, um ein deutsches Transkript zu erhalten. Das Modell ist `google/gemma-3n-E4B-it`.",
+    theme="soft",
+    css="""
+        .gradio-container {
+            font-family: 'Inter', sans-serif;
+            max-width: 800px;
+            margin-left: auto;
+            margin-right: auto;
         }
-    ],
-]
-
-demo = gr.ChatInterface(
-    fn=generate,
-    type="messages",
-    textbox=gr.MultimodalTextbox(
-        file_types=list(IMAGE_FILE_TYPES + VIDEO_FILE_TYPES + AUDIO_FILE_TYPES),
-        file_count="multiple",
-        autofocus=True,
-    ),
-    multimodal=True,
-    additional_inputs=[
-        gr.Textbox(label="System Prompt", value="You are a helpful assistant."),
-        gr.Slider(label="Max New Tokens", minimum=100, maximum=2000, step=10, value=700),
-    ],
-    stop_btn=False,
-    title="Gemma 3n E4B it",
-    examples=examples,
-    run_examples_on_click=False,
-    cache_examples=False,
-    css_paths="style.css",
-    delete_cache=(1800, 1800),
+        footer {display: none !important}
+    """,
+    allow_flagging="never",
 )
 
+# --- App starten ---
 if __name__ == "__main__":
     demo.launch()
+