Spaces:
Sleeping
Sleeping
from fastapi import FastAPI, HTTPException, UploadFile, File | |
from fastapi.responses import JSONResponse, HTMLResponse, PlainTextResponse | |
from pydantic import BaseModel | |
from typing import List, Optional | |
import requests | |
import json | |
import base64 | |
from PIL import Image | |
import io | |
import os | |
import time | |
import uvicorn | |
app = FastAPI(title="Ollama Fashion Analyzer API", version="1.0.0") | |
class OllamaFashionAnalyzer: | |
def __init__(self, base_url=None): | |
"""Initialize Ollama client""" | |
self.base_url = base_url or os.getenv("OLLAMA_BASE_URL", "http://localhost:11434") | |
self.model = "llava:7b" # Using LLaVA for vision analysis | |
def encode_image_from_bytes(self, image_bytes): | |
"""Encode image bytes to base64 for Ollama""" | |
image = Image.open(io.BytesIO(image_bytes)) | |
# Convert to RGB if necessary | |
if image.mode != 'RGB': | |
image = image.convert('RGB') | |
# Convert to base64 | |
buffered = io.BytesIO() | |
image.save(buffered, format="JPEG") | |
img_str = base64.b64encode(buffered.getvalue()).decode() | |
return img_str | |
def analyze_clothing_from_bytes(self, image_bytes): | |
"""Detailed clothing analysis using Ollama from image bytes""" | |
# Encode image | |
image_b64 = self.encode_image_from_bytes(image_bytes) | |
# Fashion analysis prompt | |
prompt = """Analyze this clothing item in detail and provide information about: | |
1. GARMENT TYPE: What type of clothing is this? | |
2. COLORS: Primary and secondary colors | |
3. COLLAR/NECKLINE: Style of collar or neckline | |
4. SLEEVES: Sleeve type and length | |
5. PATTERN: Any patterns or designs | |
6. FIT: How does it fit (loose, fitted, etc.) | |
7. MATERIAL: Apparent fabric type | |
8. FEATURES: Buttons, pockets, zippers, etc. | |
9. STYLE: Fashion style category | |
10. OCCASION: Suitable occasions for wearing | |
Be specific and detailed in your analysis.""" | |
# Make request to Ollama | |
payload = { | |
"model": self.model, | |
"prompt": prompt, | |
"images": [image_b64], | |
"stream": False, | |
"options": { | |
"temperature": 0.2, | |
"num_predict": 500 | |
} | |
} | |
try: | |
response = requests.post( | |
f"{self.base_url}/api/generate", | |
json=payload, | |
timeout=120 # Increased timeout for vision models | |
) | |
response.raise_for_status() | |
result = response.json() | |
return result.get('response', 'No response received') | |
except requests.exceptions.RequestException as e: | |
return f"Error: {str(e)}" | |
# Initialize analyzer | |
analyzer = OllamaFashionAnalyzer() | |
# Request/Response models | |
class AnalysisResponse(BaseModel): | |
analysis: str | |
# API Endpoints | |
async def root(): | |
"""Main page with file upload interface""" | |
return """ | |
<!DOCTYPE html> | |
<html> | |
<head> | |
<title>Fashion Analyzer</title> | |
<style> | |
body { font-family: Arial, sans-serif; max-width: 800px; margin: 50px auto; padding: 20px; } | |
.upload-area { border: 2px dashed #ccc; padding: 50px; text-align: center; margin: 20px 0; } | |
.result { background: #f5f5f5; padding: 20px; margin: 20px 0; border-radius: 5px; } | |
</style> | |
</head> | |
<body> | |
<h1>π½ Fashion Analyzer</h1> | |
<p>Upload an image of clothing to get detailed fashion analysis</p> | |
<div class="upload-area"> | |
<input type="file" id="imageInput" accept="image/*" style="margin: 10px;"> | |
<br> | |
<button onclick="analyzeImage()" style="padding: 10px 20px; margin: 10px;">Analyze Fashion</button> | |
</div> | |
<div id="result" class="result" style="display: none;"> | |
<h3>Analysis Result:</h3> | |
<pre id="analysisText"></pre> | |
</div> | |
<script> | |
async function analyzeImage() { | |
const input = document.getElementById('imageInput'); | |
const file = input.files[0]; | |
if (!file) { | |
alert('Please select an image file'); | |
return; | |
} | |
const formData = new FormData(); | |
formData.append('file', file); | |
document.getElementById('analysisText').textContent = 'Analyzing... Please wait...'; | |
document.getElementById('result').style.display = 'block'; | |
try { | |
const response = await fetch('/analyze-image', { | |
method: 'POST', | |
body: formData | |
}); | |
const result = await response.json(); | |
document.getElementById('analysisText').textContent = result.analysis; | |
} catch (error) { | |
document.getElementById('analysisText').textContent = 'Error: ' + error.message; | |
} | |
} | |
</script> | |
</body> | |
</html> | |
""" | |
async def analyze_image(file: UploadFile = File(...)): | |
"""Analyze uploaded image""" | |
try: | |
# Read image bytes | |
image_bytes = await file.read() | |
# Analyze the clothing | |
analysis = analyzer.analyze_clothing_from_bytes(image_bytes) | |
return AnalysisResponse(analysis=analysis) | |
except Exception as e: | |
raise HTTPException(status_code=500, detail=f"Error analyzing image: {str(e)}") | |
async def health_check(): | |
"""Health check endpoint""" | |
try: | |
# Test Ollama connection | |
response = requests.get(f"{analyzer.base_url}/api/tags", timeout=5) | |
if response.status_code == 200: | |
return {"status": "healthy", "ollama": "connected"} | |
else: | |
return {"status": "unhealthy", "ollama": "disconnected"} | |
except: | |
return {"status": "unhealthy", "ollama": "disconnected"} | |
if __name__ == "__main__": | |
uvicorn.run(app, host="0.0.0.0", port=7860) |