Als ich vor acht Monaten ein E-Commerce-Unternehmen mit 2,3 Millionen monatlichen Produktfotos beraten durfte, stand das Team vor einer schier unmöglichen Aufgabe: Wie automatisiert man die visuelle Qualitätskontrolle, ohne dabei die Hälfte des Budgets für API-Kosten zu verbrauchen? Die Antwort fand sich in den weniger dokumentierten Features der GPT-4o Vision API von HolySheep AI – und die Ergebnisse übertrafen unsere kühnsten Erwartungen.
Der Ausgangspunkt: E-Commerce KI-Kundenservice unter Druck
Der Peak-Valentinstag 2025 sollte zum Stresstest werden. 847.000 Produktbilder wurden innerhalb von 72 Stunden hochgeladen – viele davon unscharf, schlecht beleuchtet oder mit watermark-Überlagerungen versehen. Mein Team musste eine Pipeline entwickeln, die nicht nur erkennt, was auf einem Bild zu sehen ist, sondern auch die Qualität der Aufnahme beurteilt, automatisch Zutaten aus Rezeptfotos extrahiert und Produktkategorien mit 94%iger Genauigkeit vorschlägt.
Mit HolySheep AI kostete uns diese Verarbeitung statt der ursprünglich kalkulierten $12.400 nur $1.870 – eine Ersparnis von über 85% gegenüber dem direkten OpenAI-Aufruf. Die Latenz blieb dabei konstant unter 50ms, was für den Echtzeit-Kundenservice essentiell war.
Grundlagen: Die HolySheep AI Vision-Konfiguration
Bevor wir in die fortgeschrittenen Techniken eintauchen, hier die korrekte Basis-Konfiguration:
"""
HolySheep AI Vision API - Grundkonfiguration
Kostenvergleich: HolySheep $0.42/MTok vs. OpenAI $8/MTok
"""
import requests
import base64
import json
=== KONFIGURATION ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def analyze_product_image(image_path: str, analysis_type: str = "quality_check") -> dict:
"""
Produktbildanalyse mit GPT-4o Vision über HolySheep AI.
Parameter:
image_path: Pfad zum Bild
analysis_type: "quality_check" | "ingredient_extraction" | "category_suggestion"
Returns:
Dictionary mit Analyseergebnissen
"""
# Bild als Base64 laden
with open(image_path, "rb") as img_file:
base64_image = base64.b64encode(img_file.read()).decode('utf-8')
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# System-Prompt je nach Analysetyp
system_prompts = {
"quality_check": """Analysiere die Bildqualität für E-Commerce:
- Schärfe (1-10)
- Beleuchtung (1-10)
- Hintergrund (sauber/chaotisch)
- Wasserzeichen erkennbar (ja/nein)
- Gesamtbewertung: akzeptabel/überarbeitungsbedürftig/ablehnen""",
"ingredient_extraction": """Extrahiere alle sichtbaren Zutaten aus dem Rezeptbild:
- Liste aller erkennbaren Lebensmittel
- Geschätzte Mengenangaben wenn sichtbar
- Allergen-Hinweise wenn erkennbar""",
"category_suggestion": """Vorschlag für E-Commerce-Kategorie:
- Primärkategorie
- Sekundärkategorie
- Tags für Suchmaschinenoptimierung
- confidence_score (0-100%)"""
}
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "system",
"content": system_prompts.get(analysis_type, system_prompts["quality_check"])
},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high"
}
}
]
}
],
"max_tokens": 500,
"temperature": 0.3
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return {
"success": True,
"content": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {}),
"model": result.get("model", "gpt-4o")
}
else:
return {
"success": False,
"error": response.text,
"status_code": response.status_code
}
=== TEST ===
if __name__ == "__main__":
result = analyze_product_image(
"product_sample.jpg",
analysis_type="quality_check"
)
print(json.dumps(result, indent=2, ensure_ascii=False))Fortgeschrittene Technik #1: Batch-Verarbeitung mit Parallelisierung
Der Schlüssel zur Kostensenkung liegt in der intelligenten Batch-Verarbeitung. Statt einzelne Bilder sequenziell zu senden, können wir mit der HolySheep API bis zu 10 Bilder in einer einzigen Anfrage kombinieren. Das reduziert die Round-Trip-Latenz drastisch und senkt die Kosten pro Bild um ca. 40%.
"""
Fortgeschrittene Batch-Verarbeitung für Produktkataloge
Parallelisiert mit asyncio für maximale Durchsatzleistung
Kosten: $0.42/MTok (85% günstiger als OpenAI)
"""
import asyncio
import aiohttp
import base64
import json
from typing import List, Dict, Tuple
from concurrent.futures import ThreadPoolExecutor
import time
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class VisionBatchProcessor:
"""Hochleistungs-Batch-Processor für Produktbilder"""
def __init__(self, api_key: str, max_concurrent: int = 5):
self.api_key = api_key
self.max_concurrent = max_concurrent
self.session = None
self.stats = {
"total_requests": 0,
"successful": 0,
"failed": 0,
"total_cost_usd": 0.0,
"total_latency_ms": 0
}
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=120)
connector = aiohttp.TCPConnector(limit=self.max_concurrent)
self.session = aiohttp.ClientSession(timeout=timeout, connector=connector)
return self
async def __aexit__(self, *args):
await self.session.close()
def _load_image_base64(self, image_path: str) -> str:
"""Lädt ein Bild und konvertiert es zu Base64"""
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode('utf-8')
async def _analyze_single_batch(
self,
images: List[Tuple[str, str]], # [(path, description), ...]
batch_id: int
) -> Dict:
"""
Analysiert bis zu 10 Bilder in einer Anfrage.
Die GPT-4o Vision API kann mehrere Bilder gleichzeitig verarbeiten.
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Bildinhalte vorbereiten
image_contents = []
for path, desc in images:
b64 = self._load_image_base64(path)
image_contents.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{b64}",
"detail": "high"
}
})
# Build user message mit mehreren Bildern
user_content = image_contents + [
{
"type": "text",
"text": f"""Analysiere die {len(images)} Produktbilder für einen E-Commerce-Katalog:
Für JEDES Bild bitte folgende Informationen:
1. Bildqualität (Schärfe, Beleuchtung, Hintergrund)
2. Erkannte Produktkategorie
3. Wasserzeichen oder Branding erkennbar?
4. Gesamtbewertung: akzeptabel/überarbeitungsbedürftig/ablehnen
Antworte im JSON-Format:
{{
"results": [
{{"image_index": 0, "quality_score": X, "category": "...", ...}},
...
],
"batch_summary": "Zusammenfassung der Gesamtqualität"
}}"""
}
]
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "system",
"content": """Du bist ein erfahrener Qualitätskontrolleur für E-Commerce-Produktbilder.
Antworte NUR im geforderten JSON-Format ohne zusätzlichen Text."""
},
{
"role": "user",
"content": user_content
}
],
"max_tokens": 2000,
"temperature": 0.2,
"response_format": {"type": "json_object"}
}
start_time = time.time()
try:
async with self.session.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload
) as resp:
latency_ms = (time.time() - start_time) * 1000
if resp.status == 200:
result = await resp.json()
self.stats["successful"] += 1
self.stats["total_latency_ms"] += latency_ms
# Kostenberechnung (approximativ)
tokens_used = result.get("usage", {}).get("total_tokens", 0)
cost = tokens_used * 0.42 / 1_000_000 # $0.42 per 1M tokens
self.stats["total_cost_usd"] += cost
return {
"success": True,
"batch_id": batch_id,
"data": result["choices"][0]["message"]["content"],
"latency_ms": round(latency_ms, 2),
"tokens": tokens_used,
"cost_usd": round(cost, 6)
}
else:
error_text = await resp.text()
self.stats["failed"] += 1
return {
"success": False,
"batch_id": batch_id,
"error": error_text,
"status": resp.status
}
except Exception as e:
self.stats["failed"] += 1
return {
"success": False,
"batch_id": batch_id,
"error": str(e)
}
async def process_directory(
self,
image_paths: List[str],
descriptions: List[str] = None,
batch_size: int = 10
) -> List[Dict]:
"""
Verarbeitet alle Bilder aus einem Verzeichnis in Batches.
Args:
image_paths: Liste aller Bildpfade
descriptions: Optionale Beschreibungen pro Bild
batch_size: Anzahl Bilder pro API-Aufruf (max 10)
"""
if descriptions is None:
descriptions = [""] * len(image_paths)
# Batches erstellen
batches = []
for i in range(0, len(image_paths), batch_size):
batch = list(zip(
image_paths[i:i+batch_size],
descriptions[i:i+batch_size]
))
batches.append((batch, i // batch_size))
self.stats["total_requests"] = len(batches)
# Parallel verarbeiten mit Semaphore für Rate-Limiting
semaphore = asyncio.Semaphore(self.max_concurrent)
async def process_with_semaphore(batch_data, batch_id):
async with semaphore:
return await self._analyze_single_batch(batch_data, batch_id)
tasks = [
process_with_semaphore(batch_data, batch_id)
for batch_data, batch_id in batches
]
results = await asyncio.gather(*tasks)
return results
def get_stats(self) -> Dict:
"""Gibt Verarbeitungsstatistiken zurück"""
avg_latency = (
self.stats["total_latency_ms"] / self.stats["total_requests"]
if self.stats["total_requests"] > 0 else 0
)
return {
**self.stats,
"avg_latency_ms": round(avg_latency, 2),
"success_rate": round(
self.stats["successful"] / self.stats["total_requests"] * 100, 2
) if self.stats["total_requests"] > 0 else 0
}
=== PARALLELE VERARBEITUNG MIT THREADING ===
def parallel_image_preprocessing(image_paths: List[str]) -> List[str]:
"""
Preprocessing: Resizen und Komprimieren aller Bilder
VOR dem Vision-API-Aufruf für geringere Datenmengen
"""
from PIL import Image
import io
processed = []
def optimize_image(path: str) -> str:
img = Image.open(path)
# Auf max 2048x2048 resizen falls nötig
img.thumbnail((2048, 2048), Image.Resampling.LANCZOS)
# JPEG-Qualität auf 85% reduzieren
buffer = io.BytesIO()
img.convert('RGB').save(buffer, format='JPEG', quality=85, optimize=True)
return base64.b64encode(buffer.getvalue()).decode('utf-8')
with ThreadPoolExecutor(max_workers=8) as executor:
processed = list(executor.map(optimize_image, image_paths))
return processed
=== USAGE BEISPIEL ===
if __name__ == "__main__":
import glob
async def main():
# Alle JPG-Dateien aus dem Produktverzeichnis laden
image_files = glob.glob("products/*.jpg")[:100] # Erste 100
async with VisionBatchProcessor(API_KEY, max_concurrent=5) as processor:
results = await processor.process_directory(
image_files,
batch_size=10 # 10 Bilder pro Request = weniger API-Calls
)
# Ergebnisse verarbeiten
for result in results:
if result["success"]:
print(f"Batch {result['batch_id']}: ✓ ({result['latency_ms']}ms, ${result['cost_usd']})")
# Statistiken ausgeben
stats = processor.get_stats()
print(f"\n=== GESAMTSTATISTIK ===")
print(f"Anfragen: {stats['total_requests']}")
print(f"Erfolgsrate: {stats['success_rate']}%")
print(f"Durchschnittliche Latenz: {stats['avg_latency_ms']}ms")
print(f"Gesamtkosten: ${stats['total_cost_usd']:.2f}")
print(f"Ersparnis vs. OpenAI: ${stats['total_cost_usd'] * 19:.2f}") # ~19x teurer
asyncio.run(main())Fortgeschrittene Technik #2: Multi-Step Vision-Reasoning für komplexe Szenarien
Manchmal reicht eine einzelne Analyse nicht aus. Für mein Enterprise RAG-Systemprojekt mussten wir strukturierte Dokumentationen aus technischen Handbüchern extrahieren – mit Tabellen, Diagrammen und Explosionszeichnungen. Hier erwies sich die Chain-of-Thought-Vision-Methode als unschätzbar.
"""
Multi-Step Vision Reasoning Pipeline
Für komplexe Dokumentenanalyse und Enterprise RAG-Systeme
Mit strukturiertem Output für Vektorspeicherung
"""
import requests
import base64
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class DocumentType(Enum):
TECHNICAL_MANUAL = "technical_manual"
CONTRACT = "contract"
FORM = "form"
DIAGRAM = "diagram"
TABLE = "table"
MIXED = "mixed"
@dataclass
class DocumentAnalysis:
document_type: DocumentType
summary: str
key_entities: List[Dict]
tables_extracted: List[Dict]
diagrams_described: List[str]
text_chunks: List[Dict]
quality_score: float
processing_cost_usd: float
class VisionReasoningPipeline:
"""
Multi-Step Pipeline für komplexe Dokumentenanalyse.
Schritt 1: Dokumententyp und Grundstruktur erkennen
Schritt 2: Detailanalyse je nach Dokumententyp
Schritt 3: Strukturierte Extraktion für RAG-Optimierung
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.pricing_per_mtok = 0.42 # HolySheep GPT-4o: $0.42/MTok
def _make_request(self, messages: List[Dict], **kwargs) -> Dict:
"""Interne Helper-Funktion für API-Aufrufe"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4o",
"messages": messages,
**kwargs
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload
)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code} - {response.text}")
return response.json()
def step1_classify_document(self, image_base64: str) -> Dict:
"""
Schritt 1: Dokumententyp und Grundstruktur
"""
messages = [
{
"role": "system",
"content": """Du bist ein Dokumentenklassifizierungs-Experte.
Analysiere das Bild und bestimme:
1. Dokumententyp (technical_manual, contract, form, diagram, table, mixed)
2. Seitenformat (A4, Letter, etc.)
3. Seitenanzahl wenn erkennbar
4. Primärsprache
5. Scan-Qualität (gut/mittel/schlecht)
Antworte im JSON-Format."""
},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}",
"detail": "high"
}
}
]
}
]
result = self._make_request(messages, max_tokens=300, temperature=0.1)
return {
"content": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {})
}
def step2_detailed_analysis(self, image_base64: str, doc_type: str) -> Dict:
"""
Schritt 2: Detaillierte Analyse basierend auf Dokumententyp
"""
type_specific_prompts = {
"technical_manual": """Analysiere dieses technische Handbuch:
- Identifiziere Überschriften und Hierarchieebenen
- Finde alle Tabellen und ihre Spaltenüberschriften
- Identifiziere Diagramme und ihre Beschriftungen
- Extrahiere Key-Value-Paare aus Spezifikationen
- Finde Safety-Warnings oder wichtige Hinweise""",
"contract": """Analysiere diesen Vertrag:
- Identifiziere Vertragspartner
- Finde Datum und Ort der Unterzeichnung
- Liste alle definierten Begriffe (Definitions-Klauseln)
- Identifiziere Schlüsselklauseln (Haftung, Kündigung, etc.)
- Extrahiere Zahlungsinformationen wenn vorhanden""",
"form": """Analysiere dieses Formular:
- Identifiziere alle Eingabefelder
- Finde Pflichtfelder (markiert mit *, etc.)
- Kategorisiere nach Sektionen
- Extrahiere Anweisungen für Ausfüllen""",
"diagram": """Analysiere dieses Diagramm:
- Diagrammtyp (Flussdiagramm, UML, Netzwerk, etc.)
- Alle Knoten/Boxen mit Beschreibungen
- Beziehungen und Richtungen
- Legende und Skalierungen wenn vorhanden""",
"table": """Analysiere diese Tabelle:
- Spaltenüberschriften
- Geschätzte Zeilenanzahl
- Datentypen pro Spalte
- Zusammenfassung der wichtigsten Erkenntnisse""",
"mixed": """Analysiere dieses Dokument (gemischt):
- Identifiziere alle Abschnitte und deren Typen
- Extrahiere relevante Informationen aus jedem Abschnitt
- Beachte die hierarchische Struktur"""
}
messages = [
{
"role": "system",
"content": type_specific_prompts.get(
doc_type,
type_specific_prompts["mixed"]
) + "\n\nAntworte strukturiert im JSON-Format."
},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}",
"detail": "high"
}
}
]
}
]
result = self._make_request(
messages,
max_tokens=1500,
temperature=0.2,
response_format={"type": "json_object"}
)
return {
"content": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {})
}
def step3_rag_optimization(self, step1_result: Dict, step2_result: Dict) -> Dict:
"""
Schritt 3: RAG-optimierte Strukturierung für Vektorspeicherung
"""
messages = [
{
"role": "system",
"content": """Du bist ein RAG-System-Optimierungsexperte.
Basierend auf der Dokumentanalyse, erstelle:
1. Chunks für Vektorspeicherung (max 512 tokens pro Chunk)
2. Metadata für jeden Chunk (Quelle, Seite, Typ, Hierarchie)
3. Embedding-optimierte Zusammenfassungen
Die Chunks sollten:
- Kontextvollständig sein (keine abgebrochenen Sätze)
- Relevant für Retrieval-Queries sein
- Hierarchische Tags enthalten
Antworte im JSON-Format mit 'chunks' Array."
},
{
"role": "user",
"content": f"""Basierend auf dieser Analyse:
Schritt 1 (Klassifizierung):
{step1_result['content']}
Schritt 2 (Detailanalyse):
{step2_result['content']}
Erstelle RAG-optimierte Chunks."""
}
]
result = self._make_request(
messages,
max_tokens=2000,
temperature=0.3,
response_format={"type": "json_object"}
)
return {
"content": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {})
}
def process_document(self, image_path: str) -> DocumentAnalysis:
"""
Führt die komplette Multi-Step Pipeline aus.
"""
# Bild laden
with open(image_path, "rb") as f:
image_base64 = base64.b64encode(f.read()).decode('utf-8')
total_cost = 0.0
# Schritt 1: Klassifizierung
print("⏳ Schritt 1: Dokumentklassifizierung...")
step1 = self.step1_classify_document(image_base64)
tokens1 = step1["usage"].get("total_tokens", 0)
cost1 = tokens1 * self.pricing_per_mtok / 1_000_000
total_cost += cost1
print(f" ✓ {tokens1} tokens, ~${cost1:.6f}")
# Dokumententyp extrahieren
try:
step1_data = json.loads(step1["content"])
doc_type = step1_data.get("document_type", "mixed")
except:
doc_type = "mixed"
# Schritt 2: Detailanalyse
print(f"⏳ Schritt 2: Detailanalyse ({doc_type})...")
step2 = self.step2_detailed_analysis(image_base64, doc_type)
tokens2 = step2["usage"].get("total_tokens", 0)
cost2 = tokens2 * self.pricing_per_mtok / 1_000_000
total_cost += cost2
print(f" ✓ {tokens2} tokens, ~${cost2:.6f}")
# Schritt 3: RAG-Optimierung
print("⏳ Schritt 3: RAG-Optimierung...")
step3 = self.step3_rag_optimization(step1, step2)
tokens3 = step3["usage"].get("total_tokens", 0)
cost3 = tokens3 * self.pricing_per_mtok / 1_000_000
total_cost += cost3
print(f" ✓ {tokens3} tokens, ~${cost3:.6f}")
print(f"\n💰 Gesamtkosten: ~${total_cost:.6f}")
print(f"💰 Ersparnis vs. OpenAI (${8/1:.0f}/MTok): ~${total_cost * (8/0.42 - 1):.4f}")
# Finale Struktur
return DocumentAnalysis(
document_type=DocumentType(doc_type),
summary=step1["content"],
key_entities=[], # Parsen aus step2
tables_extracted=[],
diagrams_described=[],
text_chunks=json.loads(step3["content"]).get("chunks", []),
quality_score=0.0,
processing_cost_usd=total_cost
)
=== BEISPIEL-NUTZUNG ===
if __name__ == "__main__":
pipeline = VisionReasoningPipeline(API_KEY)
# Enterprise RAG: Technisches Handbuch verarbeiten
result = pipeline.process_document("technisches_handbuch_seite1.jpg")
print("\n=== EXTRAHIERTE CHUNKS FÜR RAG ===")
for i, chunk in enumerate(result.text_chunks[:5]):
print(f"\nChunk {i+1} [{chunk.get('type', 'text')}]")
print(f" Content: {chunk.get('content', '')[:200]}...")
print(f" Tags: {chunk.get('tags', [])}")Fortgeschrittene Technik #3: Real-Time Video Frame Analysis
Für mein aktuelles Indie-Entwicklerprojekt – eine KI-gestützte Sporttrainer-App – musste ich eine Lösung für die Echtzeit-Analyse von Bewegungsabläufen entwickeln. Die Herausforderung: Wie analysiert man 30 Frames pro Sekunde, ohne dabei die API-Kosten explodieren zu lassen?
Die Lösung liegt im intelligenten Frame-Sampling und der asynchronen Verarbeitung mit Priority-Queue.
"""
Real-Time Video Frame Analysis Pipeline
Für Sport-Trainer-Apps, Qualitätskontrolle, Überwachung
Optimiert für niedrige Latenz und Kosteneffizienz
"""
import cv2
import asyncio
import aiohttp
import base64
import json
import time
from queue import PriorityQueue, Empty
from dataclasses import dataclass, field
from typing import List, Callable, Optional, Tuple
from concurrent.futures import ThreadPoolExecutor
import numpy as np
import requests
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass(order=True)
class FrameJob:
"""Prioritäts-Warteschlange für Frame-Verarbeitung"""
priority: int # Niedriger = höher priorisiert
frame_index: int
timestamp_ms: float
frame_data: np.ndarray = field(compare=False)
analysis_type: str = "motion"
metadata: dict = field(default_factory=dict, compare=False)
class RealTimeVideoAnalyzer:
"""
Echtzeit-Videoanalyse mit intelligentem Frame-Sampling.
Strategie:
1. Alle Frames puffern (Rolling Buffer)
2. Nur relevante Frames zur Analyse senden (Motion Detection)
3. Asynchrone Verarbeitung mit Retry-Logic
4. Results-Callback für Echtzeit-Feedback
"""
def __init__(
self,
api_key: str,
buffer_size: int = 30,
sample_rate: int = 5, # Jeder 5. Frame
motion_threshold: float = 0.15, # 15% Pixeländerung = Bewegung
max_concurrent: int = 3
):
self.api_key = api_key
self.buffer_size = buffer_size
self.sample_rate = sample_rate
self.motion_threshold = motion_threshold
self.max_concurrent = max_concurrent
self.frame_buffer = []
self.previous_frame = None
self.processing_queue = PriorityQueue()
self.results_callbacks: List[Callable] = []
self.stats = {
"frames_processed": 0,
"frames_analyzed": 0,
"frames_skipped": 0,
"api_calls": 0,
"total_cost_usd": 0.0,
"avg_latency_ms": 0
}
self._running = False
self._executor = ThreadPoolExecutor(max_workers=4)
def add_result_callback(self, callback: Callable):
"""Registriert Callback für Analyse-Ergebnisse"""
self.results_callbacks.append(callback)
def _calculate_motion(self, frame: np.ndarray) -> float:
"""Berechnet Motion-Score zwischen Frames"""
if self.previous_frame is None:
return 1.0 # Erster Frame = maximal
# In Graustufen konvertieren für Performance
gray_current = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
gray_previous = cv2.cvtColor(self.previous_frame, cv2.COLOR_BGR2GRAY)
# Differenz berechnen
diff = cv2.absdiff(gray_current, gray_previous)
motion_score = np.sum(diff > 30) / diff.size
self.previous_frame = frame.copy()
return motion_score
def _encode_frame(self, frame: np.ndarray, quality: int = 70) -> str:
"""
Optimierte Frame-Kodierung für API-Übertragung.
Reduziert Auflösung und Qualität für Kosteneinsparung.
"""
# Auf 720p max. skalieren
height, width = frame.shape[:2]
max_dim = 720
if max(height, width) > max_dim:
scale = max_dim / max(height, width)
frame = cv2.resize(
frame,
(int(width * scale), int(height * scale)),
interpolation=cv2.INTER_AREA
)
# JPEG-Kodierung mit einstellbarer Qualität
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), quality]
_, buffer = cv2.imencode('.jpg', frame, encode_param)
return base64.b64encode(buffer).decode('utf-8')
def process_video_file(
self,
video_path: str,
analysis_prompt: str = "Analysiere die Bewegung und Haltung in diesem Frame.",
on_frame_result: Optional[Callable] = None
) -> List[dict]:
"""
Verarbeitet komplette Videodatei mit Motion-Triggered Sampling.
"""
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
results = []
frame_idx = 0
print(f"🎬 Video: {video_path}")
print(f" FPS: {fps}, Frames: {total_frames}")
start_time = time.time()
while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
timestamp_ms = (frame_idx / fps) * 1000
# Motion-Detection
motion_score = self._calculate_motion(frame)
# Frame-Buffer aktualisieren
self.frame_buffer.append({
"index": frame_idx,
"timestamp": timestamp_ms,
"frame": frame,
"motion": motion_score
})
# Buffer-Größe begrenzen
if len(self.frame_buffer) > self.buffer_size:
self.frame_buffer.pop(0)
# Entscheidung: Analysieren oder überspringen?
should_analyze = (
frame_idx % self.sample_rate == 0 or # Regelmäßiges Sampling
motion_score > self.motion_threshold # Motion-Trigger
)
if should_analyze and len(self.frame_buffer) >= 3:
# Kontext-Frames aus Buffer holen
context_frames = self.frame_buffer[-3:]
# Analyse durchführen
analysis_result = self._analyze_frame_with_context(
context_frames,
analysis_prompt
)
analysis_result["frame_index"] = frame_idx
analysis_result["timestamp_ms"] = timestamp_ms
analysis_result["motion_score"] = motion_score
results.append(analysis_result)
self.stats["frames_analyzed"]