import subprocess import torch from PIL import Image import requests from io import BytesIO from transformers import AutoProcessor, AutoModelForCausalLM import os import threading import time import urllib.parse from fastapi import FastAPI, UploadFile, File, HTTPException, Form from fastapi.responses import JSONResponse app = FastAPI( title="Florence-2 Image Captioning Server", description="Auto-captions images from middleware server using Florence-2" ) import threading import time import urllib.parse # Attempt to install flash-attn try: subprocess.run('pip install flash-attn --no-build-isolation timm einops', env={'FLASH_ATTENTION_SKIP_CUDA_BUILD': "TRUE"}, check=True, shell=True) except subprocess.CalledProcessError as e: print(f"Error installing flash-attn: {e}") print("Continuing without flash-attn.") # Determine the device to use device = "cuda" if torch.cuda.is_available() else "cpu" # Load Florence-2-large model and processor try: vision_language_model = AutoModelForCausalLM.from_pretrained('microsoft/Florence-2-large', trust_remote_code=True).to(device).eval() vision_language_processor = AutoProcessor.from_pretrained('microsoft/Florence-2-large', trust_remote_code=True) print("✓ Florence-2-large model loaded successfully") except Exception as e: print(f"Error loading Florence-2-large model: {e}") vision_language_model = None vision_language_processor = None def load_image_from_url(image_url): """Load an image from a URL.""" try: response = requests.get(image_url, timeout=30) response.raise_for_status() image = Image.open(BytesIO(response.content)) return image.convert('RGB') except Exception as e: raise ValueError(f"Error loading image from URL: {e}") def process_image_description(model, processor, image): """Process an image and generate description using the specified model.""" if not isinstance(image, Image.Image): image = Image.fromarray(image) inputs = processor(text="", images=image, return_tensors="pt").to(device) with torch.no_grad(): generated_ids = model.generate( input_ids=inputs["input_ids"], pixel_values=inputs["pixel_values"], max_new_tokens=1024, early_stopping=False, do_sample=False, num_beams=3, ) generated_text = processor.batch_decode(generated_ids, skip_special_tokens=False)[0] processed_description = processor.post_process_generation( generated_text, task="", image_size=(image.width, image.height) ) image_description = processed_description[""] return image_description def describe_image(uploaded_image, model_choice): """Generate description from uploaded image.""" if uploaded_image is None: return "Please upload an image." if vision_language_model is None: return "Florence-2-large model failed to load." model = vision_language_model processor = vision_language_processor try: return process_image_description(model, processor, uploaded_image) except Exception as e: return f"Error generating caption: {str(e)}" def describe_image_from_url(image_url, model_choice=None): """Generate description from image URL.""" try: if not image_url: return {"error": "image_url is required"} if vision_language_model is None: return {"error": "Florence-2-large model not available"} # Load image from URL image = load_image_from_url(image_url) # Use the loaded large model model = vision_language_model processor = vision_language_processor # Generate caption caption = process_image_description(model, processor, image) return { "status": "success", "model": model_choice, "caption": caption, "image_size": {"width": image.width, "height": image.height} } except Exception as e: return {"error": f"Error processing image: {str(e)}"} IMAGE_SERVER_BASE = os.getenv("IMAGE_SERVER_BASE", " ") DATA_COLLECTION_BASE = os.getenv("DATA_COLLECTION_BASE", "https://fred808-flow.hf.space") REQUESTER_ID = os.getenv("FLO_REQUESTER_ID", f"florence-2-{os.getpid()}") MODEL_CHOICE = "Florence-2-large" # Always use large model def sanitize_name(name: str, max_len: int = 200) -> str: """Sanitize a filename while preserving extension.""" import re name = str(name).strip() # replace spaces with underscores name = re.sub(r"\s+", "_", name) # remove any characters not alphanumeric, dot, dash, or underscore name = re.sub(r"[^A-Za-z0-9_.-]", "", name) if len(name) > max_len: base, ext = os.path.splitext(name) name = base[: max_len - len(ext)] + ext return name or "file" def _build_download_url(course: str, video: str, frame: str) -> str: """Build download URL with proper encoding of all path segments.""" # The middleware /download endpoint expects the 'file' parameter to be # a path relative to the course folder (e.g. "video_name/frame.jpg"). # Frames live under a "{base_course}_frames" folder. base_course = course if not base_course.endswith("_frames"): course_dir = f"{base_course}_frames" else: course_dir = base_course base_course = course_dir[:-7] # strip _frames for consistency # Sanitize and encode path segments safe_course = sanitize_name(course_dir) safe_video = sanitize_name(video) safe_frame = sanitize_name(frame) file_param = f"{safe_video}/{safe_frame}" url = f"{IMAGE_SERVER_BASE.rstrip('/')}/download?course={urllib.parse.quote(safe_course, safe='')}&file={urllib.parse.quote(file_param, safe='')}" print(f"[BACKGROUND] Built URL: {url}") return url, safe_frame def _download_bytes(url: str, timeout: int = 30, chunk_size=32768): try: print(f"[BACKGROUND] Starting download: {url}") response = requests.get(url, timeout=timeout, stream=True) response.raise_for_status() content = BytesIO() total_size = int(response.headers.get('content-length', 0)) print(f"[BACKGROUND] Total size: {total_size} bytes") bytes_read = 0 for chunk in response.iter_content(chunk_size=chunk_size): if chunk: content.write(chunk) bytes_read += len(chunk) if total_size: print(f"\rDownloading: {bytes_read}/{total_size} bytes ({(bytes_read/total_size)*100:.1f}%)", end="", flush=True) print() # New line after progress print(f"[BACKGROUND] Download complete: {bytes_read} bytes") return content.getvalue(), response.headers.get('content-type') except Exception as e: print(f"[BACKGROUND] download failed {url}: {e}") return None, None def _post_submit(caption: str, image_name: str, course: str, image_url: str, image_bytes: bytes): submit_url = f"{DATA_COLLECTION_BASE.rstrip('/')}/submit" files = {'image': (image_name, image_bytes, 'application/octet-stream')} data = {'caption': caption, 'image_name': image_name, 'course': course, 'image_url': image_url} print(f"[BACKGROUND] Submitting to {submit_url}") print(f"[BACKGROUND] Image name: {image_name}") print(f"[BACKGROUND] Course: {course}") print(f"[BACKGROUND] Caption length: {len(caption)} chars") try: r = requests.post(submit_url, data=data, files=files, timeout=30) print(f"[BACKGROUND] Submit response status: {r.status}") try: resp = r.json() print(f"[BACKGROUND] Submit response JSON: {resp}") return r.status_code, resp except Exception: print(f"[BACKGROUND] Submit response text: {r.text}") return r.status_code, r.text except Exception as e: print(f"[BACKGROUND] Submit POST failed: {e}") return None, None def _release_frame(course: str, video: str, frame: str): try: release_url = f"{IMAGE_SERVER_BASE.rstrip('/')}/middleware/release/frame/{urllib.parse.quote(course, safe='')}/{urllib.parse.quote(video, safe='')}/{urllib.parse.quote(frame, safe='')}" requests.post(release_url, params={"requester_id": REQUESTER_ID}, timeout=10) except Exception as e: print(f"[BACKGROUND] release frame failed: {e}") def _release_course(course: str): try: release_url = f"{IMAGE_SERVER_BASE.rstrip('/')}/middleware/release/course/{urllib.parse.quote(course, safe='')}" requests.post(release_url, params={"requester_id": REQUESTER_ID}, timeout=10) except Exception as e: print(f"[BACKGROUND] release course failed: {e}") # Background worker implementation def background_worker(): """Background worker that processes images from the middleware server.""" print("[BACKGROUND] Starting worker, waiting for model...") # Wait for model to be ready waited = 0 while waited < 120: if vision_language_model is not None: break time.sleep(1) waited += 1 if waited >= 120: print("[BACKGROUND] Model not available after timeout") return print(f"[BACKGROUND] Model {MODEL_CHOICE} ready, starting processing loop") while True: try: # Get next course courses_url = f"{IMAGE_SERVER_BASE}/courses" print(f"[BACKGROUND] Fetching courses from {courses_url}") try: r = requests.get(courses_url, timeout=15) r.raise_for_status() courses_data = r.json() if not courses_data.get('courses'): print("[BACKGROUND] No courses found, waiting...") time.sleep(3) continue # Get first course course_entry = courses_data['courses'][0] if isinstance(course_entry, dict): course = course_entry.get('course_folder') else: course = str(course_entry) if not course: print("[BACKGROUND] Invalid course entry") time.sleep(2) continue print(f"[BACKGROUND] Processing course: {course}") # Get images list images_url = f"{IMAGE_SERVER_BASE}/images/{urllib.parse.quote(course, safe='')}" r = requests.get(images_url, timeout=15) r.raise_for_status() images_data = r.json() if isinstance(images_data, dict): image_list = images_data.get('images', []) else: image_list = images_data if not image_list: print(f"[BACKGROUND] No images found for course {course}") time.sleep(2) continue print(f"[BACKGROUND] Found {len(image_list)} images") # Process images for img_entry in image_list: try: # Extract filename and metadata if isinstance(img_entry, dict): filename = img_entry.get('filename') if not filename: continue else: filename = str(img_entry) # Download image download_url = f"{IMAGE_SERVER_BASE}/images/{urllib.parse.quote(course, safe='')}/{urllib.parse.quote(filename, safe='')}" print(f"[BACKGROUND] Downloading {download_url}") img_bytes, _ = _download_bytes(download_url) if not img_bytes: print(f"[BACKGROUND] Failed to download {filename}") continue # Process with Florence try: pil_img = Image.open(BytesIO(img_bytes)).convert('RGB') model = vision_language_model processor = vision_language_processor print(f"[BACKGROUND] Generating caption for {filename}") caption = process_image_description(model, processor, pil_img) print(f"[BACKGROUND] Generated caption for {filename}:") print("-" * 40) print(caption) print("-" * 40) # Submit result print(f"[BACKGROUND] Submitting caption to {DATA_COLLECTION_BASE}/submit") status, resp = _post_submit(caption, filename, course, download_url, img_bytes) if status and status < 400: print(f"[BACKGROUND] Successfully submitted {filename} (status={status})") if resp: print(f"[BACKGROUND] Response: {resp}") else: print(f"[BACKGROUND] Failed to submit {filename}: status={status}, response={resp}") except Exception as e: print(f"[BACKGROUND] Error processing {filename}: {e}") continue finally: # Clean up if 'pil_img' in locals(): del pil_img if 'img_bytes' in locals(): del img_bytes time.sleep(0.5) # Small delay between images except Exception as e: print(f"[BACKGROUND] Error in image loop: {e}") continue print(f"[BACKGROUND] Completed course {course}") time.sleep(1) except Exception as e: print(f"[BACKGROUND] Error in course loop: {e}") time.sleep(5) continue except Exception as e: print(f"[BACKGROUND] Main loop error: {e}") time.sleep(5) def _start_worker_thread(): """Start the background worker thread.""" t = threading.Thread(target=background_worker, daemon=True) t.start() return t # FastAPI endpoints for status/health @app.get("/") async def root(): return { "name": "Florence-2 Image Captioning Server", "status": "running", "model": "Florence-2-large", "model_loaded": vision_language_model is not None, "device": device } @app.get("/health") async def health(): return { "status": "healthy", "model": "Florence-2-large", "model_loaded": vision_language_model is not None, "device": device, "model_choice": MODEL_CHOICE } @app.get("/analyze") async def analyze_get(image_url: str = None, model_choice: str = None): """Analyze an image by URL. Usage: /analyze?image_url=https://...&model_choice=Florence-2-base""" try: mc = model_choice or MODEL_CHOICE if image_url: result = describe_image_from_url(image_url, mc) if isinstance(result, dict) and result.get("status") == "success": return JSONResponse(content={"success": True, "caption": result.get("caption"), "image_size": result.get("image_size")}) else: return JSONResponse(status_code=400, content={"success": False, "error": result}) else: raise HTTPException(status_code=400, detail="image_url query parameter is required") except HTTPException: raise except Exception as e: return JSONResponse(status_code=500, content={"success": False, "error": str(e)}) @app.post("/analyze") async def analyze_post(file: UploadFile = File(None), model_choice: str = Form(None)): """Analyze an uploaded image (multipart/form-data). Returns caption JSON.""" try: if file is None: raise HTTPException(status_code=400, detail="file is required") content = await file.read() try: pil_img = Image.open(BytesIO(content)).convert('RGB') except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to read uploaded image: {e}") if vision_language_model is None: raise HTTPException(status_code=503, detail="Florence-2-large model not loaded") model = vision_language_model processor = vision_language_processor caption = process_image_description(model, processor, pil_img) return JSONResponse(content={"success": True, "caption": caption}) except HTTPException: raise except Exception as e: return JSONResponse(status_code=500, content={"success": False, "error": str(e)}) # Get the port from environment variable (for Hugging Face Spaces) port = int(os.environ.get("PORT", 7860)) # Launch FastAPI with uvicorn when run directly if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=port)