Spaces:
Sleeping
Sleeping
| """SAM3 open-vocabulary panoptic segmentation API for Hugging Face Spaces.""" | |
| from __future__ import annotations | |
| import base64 | |
| import io | |
| import os | |
| import traceback | |
| import gradio as gr | |
| import numpy as np | |
| import spaces | |
| import torch | |
| from PIL import Image | |
| from transformers import Sam3Model, Sam3Processor | |
| HF_TOKEN = os.environ.get("HF_TOKEN") | |
| MODEL_ID = "facebook/sam3" | |
| DEVICE = "cuda" | |
| # ZeroGPU emulates CUDA during startup and swaps in a real GPU for @spaces.GPU. | |
| processor = Sam3Processor.from_pretrained(MODEL_ID, token=HF_TOKEN) | |
| model = Sam3Model.from_pretrained(MODEL_ID, token=HF_TOKEN).to(DEVICE) | |
| model.eval() | |
| def _encode_mask(mask_bool: np.ndarray) -> str: | |
| buffer = io.BytesIO() | |
| Image.fromarray(mask_bool.astype(np.uint8) * 255, mode="L").save( | |
| buffer, format="PNG" | |
| ) | |
| return base64.b64encode(buffer.getvalue()).decode("ascii") | |
| def _gpu_duration(image, concepts, conf, mask_threshold=0.5) -> int: | |
| del image, conf, mask_threshold | |
| concept_count = len( | |
| [value for value in str(concepts or "").split(",") if value.strip()] | |
| ) | |
| return min(120, max(30, concept_count * 20)) | |
| def api_panoptic(image, concepts, conf, mask_threshold=0.5): | |
| if image is None: | |
| return {"error": "no image provided", "detections": []} | |
| image = image.convert("RGB") | |
| width, height = image.size | |
| concept_list = [ | |
| value.strip() for value in str(concepts or "").split(",") if value.strip() | |
| ] | |
| if not concept_list: | |
| return { | |
| "version": "4", | |
| "model": MODEL_ID, | |
| "width": width, | |
| "height": height, | |
| "detections": [], | |
| } | |
| print( | |
| f"SAM3 request: size={width}x{height} concepts={len(concept_list)}", | |
| flush=True, | |
| ) | |
| detections = [] | |
| try: | |
| for concept in concept_list: | |
| inputs = processor( | |
| images=image, text=concept, return_tensors="pt" | |
| ).to(DEVICE) | |
| with torch.inference_mode(): | |
| outputs = model(**inputs) | |
| target_sizes = ( | |
| inputs["original_sizes"].tolist() | |
| if "original_sizes" in inputs | |
| else [[height, width]] | |
| ) | |
| result = processor.post_process_instance_segmentation( | |
| outputs, | |
| threshold=float(conf), | |
| mask_threshold=float(mask_threshold), | |
| target_sizes=target_sizes, | |
| )[0] | |
| masks = result["masks"] | |
| scores = result["scores"] | |
| boxes = result.get("boxes") | |
| for index in range(len(scores)): | |
| mask = masks[index] | |
| mask_array = ( | |
| mask.detach().cpu().numpy() | |
| if hasattr(mask, "detach") | |
| else np.asarray(mask) | |
| ) | |
| mask_bool = ( | |
| mask_array | |
| if mask_array.dtype == bool | |
| else mask_array > 0.5 | |
| ) | |
| box = ( | |
| boxes[index].detach().cpu().numpy().tolist() | |
| if boxes is not None | |
| else [0, 0, 0, 0] | |
| ) | |
| detections.append( | |
| { | |
| "label": concept, | |
| "score": float(scores[index]), | |
| "box": box, | |
| "mask_png_b64": _encode_mask(mask_bool), | |
| } | |
| ) | |
| except Exception: | |
| traceback.print_exc() | |
| raise | |
| print(f"SAM3 response: detections={len(detections)}", flush=True) | |
| return { | |
| "version": "4", | |
| "model": MODEL_ID, | |
| "width": width, | |
| "height": height, | |
| "detections": detections, | |
| } | |
| with gr.Blocks(title="SAM3 Panoptic") as demo: | |
| gr.Markdown("# SAM3 Panoptic API") | |
| with gr.Row(): | |
| input_image = gr.Image(type="pil", label="Image") | |
| output_json = gr.JSON(label="Detections") | |
| concept_text = gr.Textbox( | |
| label="Concepts", | |
| value="person, car, road, building, tree", | |
| ) | |
| confidence = gr.Slider( | |
| 0.0, 1.0, value=0.4, step=0.05, label="Confidence" | |
| ) | |
| mask_threshold = gr.Slider( | |
| 0.05, 0.95, value=0.5, step=0.05, label="Mask threshold" | |
| ) | |
| gr.Button("Segment").click( | |
| api_panoptic, | |
| [input_image, concept_text, confidence, mask_threshold], | |
| output_json, | |
| api_name="api_panoptic", | |
| ) | |
| if __name__ == "__main__": | |
| demo.queue(default_concurrency_limit=1).launch() | |