prithivMLmods commited on
Commit
de485bb
·
verified ·
1 Parent(s): adfd115

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +343 -343
app.py CHANGED
@@ -1,344 +1,344 @@
1
- import os
2
- import cv2
3
- import tempfile
4
- import spaces
5
- import gradio as gr
6
- import numpy as np
7
- import torch
8
- import matplotlib
9
- import matplotlib.pyplot as plt
10
- from PIL import Image, ImageDraw
11
- from typing import Iterable
12
-
13
- from transformers import (
14
- Sam3Model, Sam3Processor,
15
- Sam3VideoModel, Sam3VideoProcessor,
16
- Sam3TrackerModel, Sam3TrackerProcessor
17
- )
18
-
19
- device = "cuda" if torch.cuda.is_available() else "cpu"
20
- print(f"🖥️ Using compute device: {device}")
21
-
22
- print("⏳ Loading SAM3 Models permanently into memory...")
23
-
24
- try:
25
- # 1. Load Image Segmentation Model (Text)
26
- print(" ... Loading Image Text Model")
27
- IMG_MODEL = Sam3Model.from_pretrained("facebook/sam3").to(device)
28
- IMG_PROCESSOR = Sam3Processor.from_pretrained("facebook/sam3")
29
-
30
- # 2. Load Image Tracker Model (Click)
31
- print(" ... Loading Image Tracker Model")
32
- TRK_MODEL = Sam3TrackerModel.from_pretrained("facebook/sam3").to(device)
33
- TRK_PROCESSOR = Sam3TrackerProcessor.from_pretrained("facebook/sam3")
34
-
35
- # 3. Load Video Segmentation Model
36
- print(" ... Loading Video Model")
37
- # Using bfloat16 for video to optimize VRAM
38
- VID_MODEL = Sam3VideoModel.from_pretrained("facebook/sam3").to(device, dtype=torch.bfloat16)
39
- VID_PROCESSOR = Sam3VideoProcessor.from_pretrained("facebook/sam3")
40
-
41
- print("✅ All Models loaded successfully!")
42
-
43
- except Exception as e:
44
- print(f"❌ CRITICAL ERROR LOADING MODELS: {e}")
45
- IMG_MODEL = None
46
- IMG_PROCESSOR = None
47
- TRK_MODEL = None
48
- TRK_PROCESSOR = None
49
- VID_MODEL = None
50
- VID_PROCESSOR = None
51
-
52
-
53
- # --- UTILS ---
54
- def apply_mask_overlay(base_image, mask_data, opacity=0.5):
55
- """Draws segmentation masks on top of an image."""
56
- if isinstance(base_image, np.ndarray):
57
- base_image = Image.fromarray(base_image)
58
- base_image = base_image.convert("RGBA")
59
-
60
- if mask_data is None or len(mask_data) == 0:
61
- return base_image.convert("RGB")
62
-
63
- if isinstance(mask_data, torch.Tensor):
64
- mask_data = mask_data.cpu().numpy()
65
- mask_data = mask_data.astype(np.uint8)
66
-
67
- # Handle dimensions
68
- if mask_data.ndim == 4: mask_data = mask_data[0]
69
- if mask_data.ndim == 3 and mask_data.shape[0] == 1: mask_data = mask_data[0]
70
-
71
- num_masks = mask_data.shape[0] if mask_data.ndim == 3 else 1
72
- if mask_data.ndim == 2:
73
- mask_data = [mask_data]
74
- num_masks = 1
75
-
76
- try:
77
- color_map = matplotlib.colormaps["rainbow"].resampled(max(num_masks, 1))
78
- except AttributeError:
79
- import matplotlib.cm as cm
80
- color_map = cm.get_cmap("rainbow").resampled(max(num_masks, 1))
81
-
82
- rgb_colors = [tuple(int(c * 255) for c in color_map(i)[:3]) for i in range(num_masks)]
83
- composite_layer = Image.new("RGBA", base_image.size, (0, 0, 0, 0))
84
-
85
- for i, single_mask in enumerate(mask_data):
86
- mask_bitmap = Image.fromarray((single_mask * 255).astype(np.uint8))
87
- if mask_bitmap.size != base_image.size:
88
- mask_bitmap = mask_bitmap.resize(base_image.size, resample=Image.NEAREST)
89
-
90
- fill_color = rgb_colors[i]
91
- color_fill = Image.new("RGBA", base_image.size, fill_color + (0,))
92
- mask_alpha = mask_bitmap.point(lambda v: int(v * opacity) if v > 0 else 0)
93
- color_fill.putalpha(mask_alpha)
94
- composite_layer = Image.alpha_composite(composite_layer, color_fill)
95
-
96
- return Image.alpha_composite(base_image, composite_layer).convert("RGB")
97
-
98
- def draw_points_on_image(image, points):
99
- """Draws red dots on the image to indicate click locations."""
100
- if isinstance(image, np.ndarray):
101
- image = Image.fromarray(image)
102
-
103
- draw_img = image.copy()
104
- draw = ImageDraw.Draw(draw_img)
105
-
106
- for pt in points:
107
- x, y = pt
108
- r = 8 # Radius of point
109
- draw.ellipse((x-r, y-r, x+r, y+r), fill="red", outline="white", width=4)
110
-
111
- return draw_img
112
-
113
- @spaces.GPU
114
- def run_image_segmentation(source_img, text_query, conf_thresh=0.5):
115
- if IMG_MODEL is None or IMG_PROCESSOR is None:
116
- raise gr.Error("Models failed to load on startup.")
117
-
118
- if source_img is None or not text_query:
119
- raise gr.Error("Please provide an image and a text prompt.")
120
-
121
- try:
122
- pil_image = source_img.convert("RGB")
123
- model_inputs = IMG_PROCESSOR(images=pil_image, text=text_query, return_tensors="pt").to(device)
124
-
125
- with torch.no_grad():
126
- inference_output = IMG_MODEL(**model_inputs)
127
-
128
- processed_results = IMG_PROCESSOR.post_process_instance_segmentation(
129
- inference_output,
130
- threshold=conf_thresh,
131
- mask_threshold=0.5,
132
- target_sizes=model_inputs.get("original_sizes").tolist()
133
- )[0]
134
-
135
- annotation_list = []
136
- raw_masks = processed_results['masks'].cpu().numpy()
137
- raw_scores = processed_results['scores'].cpu().numpy()
138
-
139
- for idx, mask_array in enumerate(raw_masks):
140
- label_str = f"{text_query} ({raw_scores[idx]:.2f})"
141
- annotation_list.append((mask_array, label_str))
142
-
143
- return (pil_image, annotation_list)
144
-
145
- except Exception as e:
146
- raise gr.Error(f"Error during image processing: {e}")
147
-
148
- @spaces.GPU
149
- def run_image_click_gpu(input_image, x, y, points_state, labels_state):
150
- if TRK_MODEL is None or TRK_PROCESSOR is None:
151
- raise gr.Error("Tracker Model failed to load.")
152
-
153
- if input_image is None: return input_image, [], []
154
- if points_state is None: points_state = []; labels_state = []
155
-
156
- # Append new point
157
- points_state.append([x, y])
158
- labels_state.append(1) # 1 indicates a positive click (foreground)
159
-
160
- try:
161
- # Prepare inputs format: [Batch, Point_Group, Point_Idx, Coord]
162
- input_points = [[points_state]]
163
- input_labels = [[labels_state]]
164
-
165
- inputs = TRK_PROCESSOR(images=input_image, input_points=input_points, input_labels=input_labels, return_tensors="pt").to(device)
166
-
167
- with torch.no_grad():
168
- # multimask_output=True usually helps with ambiguity, but let's default to best mask for simplicity here
169
- outputs = TRK_MODEL(**inputs, multimask_output=False)
170
-
171
- # Post process
172
- masks = TRK_PROCESSOR.post_process_masks(outputs.pred_masks.cpu(), inputs["original_sizes"], binarize=True)[0]
173
-
174
- # Overlay mask
175
- # masks shape is [1, 1, H, W] for single object tracking
176
- final_img = apply_mask_overlay(input_image, masks[0])
177
-
178
- # Draw the visual points on top
179
- final_img = draw_points_on_image(final_img, points_state)
180
-
181
- return final_img, points_state, labels_state
182
-
183
- except Exception as e:
184
- print(f"Tracker Error: {e}")
185
- return input_image, points_state, labels_state
186
-
187
- def image_click_handler(image, evt: gr.SelectData, points_state, labels_state):
188
- # Wrapper to handle the Gradio select event
189
- x, y = evt.index
190
- return run_image_click_gpu(image, x, y, points_state, labels_state)
191
-
192
- def calc_timeout_duration(vid_file, *args):
193
- return args[-1] if args else 60
194
-
195
- @spaces.GPU(duration=calc_timeout_duration)
196
- def run_video_segmentation(source_vid, text_query, frame_limit, time_limit):
197
- if VID_MODEL is None or VID_PROCESSOR is None:
198
- raise gr.Error("Video Models failed to load on startup.")
199
-
200
- if not source_vid or not text_query:
201
- raise gr.Error("Missing video or prompt.")
202
-
203
- try:
204
- video_cap = cv2.VideoCapture(source_vid)
205
- vid_fps = video_cap.get(cv2.CAP_PROP_FPS)
206
- vid_w = int(video_cap.get(cv2.CAP_PROP_FRAME_WIDTH))
207
- vid_h = int(video_cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
208
-
209
- video_frames = []
210
- counter = 0
211
- while video_cap.isOpened():
212
- ret, frame = video_cap.read()
213
- if not ret or (frame_limit > 0 and counter >= frame_limit): break
214
- video_frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
215
- counter += 1
216
- video_cap.release()
217
-
218
- session = VID_PROCESSOR.init_video_session(video=video_frames, inference_device=device, dtype=torch.bfloat16)
219
- session = VID_PROCESSOR.add_text_prompt(inference_session=session, text=text_query)
220
-
221
- temp_out_path = tempfile.mktemp(suffix=".mp4")
222
- video_writer = cv2.VideoWriter(temp_out_path, cv2.VideoWriter_fourcc(*'mp4v'), vid_fps, (vid_w, vid_h))
223
-
224
- for model_out in VID_MODEL.propagate_in_video_iterator(inference_session=session, max_frame_num_to_track=len(video_frames)):
225
- post_processed = VID_PROCESSOR.postprocess_outputs(session, model_out)
226
- f_idx = model_out.frame_idx
227
- original_pil = Image.fromarray(video_frames[f_idx])
228
-
229
- if 'masks' in post_processed:
230
- detected_masks = post_processed['masks']
231
- if detected_masks.ndim == 4: detected_masks = detected_masks.squeeze(1)
232
- final_frame = apply_mask_overlay(original_pil, detected_masks)
233
- else:
234
- final_frame = original_pil
235
-
236
- video_writer.write(cv2.cvtColor(np.array(final_frame), cv2.COLOR_RGB2BGR))
237
-
238
- video_writer.release()
239
- return temp_out_path, "Video processing completed successfully.✅"
240
-
241
- except Exception as e:
242
- return None, f"Error during video processing: {str(e)}"
243
-
244
- custom_css="""
245
- #col-container { margin: 0 auto; max-width: 1200px; }
246
- #main-title h1 { font-size: 2.2em !important; }
247
- """
248
-
249
- with gr.Blocks() as demo:
250
- with gr.Column(elem_id="col-container"):
251
- gr.Markdown("# **SAM3: Segment Anything Model 3**", elem_id="main-title")
252
- gr.Markdown("Segment objects in image or video using **SAM3** with Text Prompts or Interactive Clicks.")
253
-
254
- with gr.Tabs():
255
- with gr.Tab("Image Segmentation"):
256
- with gr.Row():
257
- with gr.Column(scale=1):
258
- image_input = gr.Image(label="Upload Image", type="pil", height=350)
259
- txt_prompt_img = gr.Textbox(label="Text Prompt", placeholder="e.g., cat, face, car wheel")
260
- with gr.Accordion("Advanced Settings", open=False):
261
- conf_slider = gr.Slider(0.0, 1.0, value=0.45, step=0.05, label="Confidence Threshold")
262
-
263
- btn_process_img = gr.Button("Segment Image", variant="primary")
264
-
265
- with gr.Column(scale=1.5):
266
- image_result = gr.AnnotatedImage(label="Segmented Result", height=410)
267
-
268
- gr.Examples(
269
- examples=[
270
- ["examples/player.jpg", "player in white", 0.5],
271
- ],
272
- inputs=[image_input, txt_prompt_img, conf_slider],
273
- outputs=[image_result],
274
- fn=run_image_segmentation,
275
- cache_examples=False,
276
- label="Image Examples"
277
- )
278
-
279
- btn_process_img.click(
280
- fn=run_image_segmentation,
281
- inputs=[image_input, txt_prompt_img, conf_slider],
282
- outputs=[image_result]
283
- )
284
-
285
- with gr.Tab("Video Segmentation"):
286
- with gr.Row():
287
- with gr.Column():
288
- video_input = gr.Video(label="Upload Video", format="mp4", height=320)
289
- txt_prompt_vid = gr.Textbox(label="Text Prompt", placeholder="e.g., person running, red car")
290
-
291
- with gr.Row():
292
- frame_limiter = gr.Slider(10, 500, value=60, step=10, label="Max Frames")
293
- time_limiter = gr.Radio([60, 120, 180], value=60, label="Timeout (seconds)")
294
-
295
- btn_process_vid = gr.Button("Segment Video", variant="primary")
296
-
297
- with gr.Column():
298
- video_result = gr.Video(label="Processed Video")
299
- process_status = gr.Textbox(label="System Status", interactive=False)
300
-
301
- gr.Examples(
302
- examples=[
303
- ["examples/sample_video.mp4", "players", 120, 120],
304
- ],
305
- inputs=[video_input, txt_prompt_vid, frame_limiter, time_limiter],
306
- outputs=[video_result, process_status],
307
- fn=run_video_segmentation,
308
- cache_examples=False,
309
- label="Video Examples"
310
- )
311
-
312
- btn_process_vid.click(
313
- run_video_segmentation,
314
- inputs=[video_input, txt_prompt_vid, frame_limiter, time_limiter],
315
- outputs=[video_result, process_status]
316
- )
317
-
318
- with gr.Tab("Image Click Segmentation"):
319
- with gr.Row():
320
- with gr.Column(scale=1):
321
- img_click_input = gr.Image(type="pil", label="Upload Image", interactive=True, height=450)
322
-
323
- with gr.Row():
324
- img_click_clear = gr.Button("Clear Points & Reset", variant="primary")
325
-
326
- st_click_points = gr.State([])
327
- st_click_labels = gr.State([])
328
-
329
- with gr.Column(scale=1):
330
- img_click_output = gr.Image(type="pil", label="Result Preview", height=450, interactive=False)
331
-
332
- img_click_input.select(
333
- image_click_handler,
334
- inputs=[img_click_input, st_click_points, st_click_labels],
335
- outputs=[img_click_output, st_click_points, st_click_labels]
336
- )
337
-
338
- img_click_clear.click(
339
- lambda: (None, [], []),
340
- outputs=[img_click_output, st_click_points, st_click_labels]
341
- )
342
-
343
- if __name__ == "__main__":
344
  demo.launch(css=custom_css, theme=gr.themes.Soft(), ssr_mode=False, mcp_server=True, show_error=True)
 
1
+ import os
2
+ import cv2
3
+ import tempfile
4
+ import spaces
5
+ import gradio as gr
6
+ import numpy as np
7
+ import torch
8
+ import matplotlib
9
+ import matplotlib.pyplot as plt
10
+ from PIL import Image, ImageDraw
11
+ from typing import Iterable
12
+
13
+ from transformers import (
14
+ Sam3Model, Sam3Processor,
15
+ Sam3VideoModel, Sam3VideoProcessor,
16
+ Sam3TrackerModel, Sam3TrackerProcessor
17
+ )
18
+
19
+ device = "cuda" if torch.cuda.is_available() else "cpu"
20
+ print(f"🖥️ Using compute device: {device}")
21
+
22
+ print("⏳ Loading SAM3 Models permanently into memory...")
23
+
24
+ try:
25
+ # 1. Load Image Segmentation Model (Text)
26
+ print(" ... Loading Image Text Model")
27
+ IMG_MODEL = Sam3Model.from_pretrained("strangervisionhf/sam3.1-st-bf16").to(device)
28
+ IMG_PROCESSOR = Sam3Processor.from_pretrained("strangervisionhf/sam3.1-st-bf16")
29
+
30
+ # 2. Load Image Tracker Model (Click)
31
+ print(" ... Loading Image Tracker Model")
32
+ TRK_MODEL = Sam3TrackerModel.from_pretrained("strangervisionhf/sam3.1-st-bf16").to(device)
33
+ TRK_PROCESSOR = Sam3TrackerProcessor.from_pretrained("strangervisionhf/sam3.1-st-bf16")
34
+
35
+ # 3. Load Video Segmentation Model
36
+ print(" ... Loading Video Model")
37
+ # Using bfloat16 for video to optimize VRAM
38
+ VID_MODEL = Sam3VideoModel.from_pretrained("strangervisionhf/sam3.1-st-bf16").to(device, dtype=torch.bfloat16)
39
+ VID_PROCESSOR = Sam3VideoProcessor.from_pretrained("strangervisionhf/sam3.1-st-bf16")
40
+
41
+ print("✅ All Models loaded successfully!")
42
+
43
+ except Exception as e:
44
+ print(f"❌ CRITICAL ERROR LOADING MODELS: {e}")
45
+ IMG_MODEL = None
46
+ IMG_PROCESSOR = None
47
+ TRK_MODEL = None
48
+ TRK_PROCESSOR = None
49
+ VID_MODEL = None
50
+ VID_PROCESSOR = None
51
+
52
+
53
+ # --- UTILS ---
54
+ def apply_mask_overlay(base_image, mask_data, opacity=0.5):
55
+ """Draws segmentation masks on top of an image."""
56
+ if isinstance(base_image, np.ndarray):
57
+ base_image = Image.fromarray(base_image)
58
+ base_image = base_image.convert("RGBA")
59
+
60
+ if mask_data is None or len(mask_data) == 0:
61
+ return base_image.convert("RGB")
62
+
63
+ if isinstance(mask_data, torch.Tensor):
64
+ mask_data = mask_data.cpu().numpy()
65
+ mask_data = mask_data.astype(np.uint8)
66
+
67
+ # Handle dimensions
68
+ if mask_data.ndim == 4: mask_data = mask_data[0]
69
+ if mask_data.ndim == 3 and mask_data.shape[0] == 1: mask_data = mask_data[0]
70
+
71
+ num_masks = mask_data.shape[0] if mask_data.ndim == 3 else 1
72
+ if mask_data.ndim == 2:
73
+ mask_data = [mask_data]
74
+ num_masks = 1
75
+
76
+ try:
77
+ color_map = matplotlib.colormaps["rainbow"].resampled(max(num_masks, 1))
78
+ except AttributeError:
79
+ import matplotlib.cm as cm
80
+ color_map = cm.get_cmap("rainbow").resampled(max(num_masks, 1))
81
+
82
+ rgb_colors = [tuple(int(c * 255) for c in color_map(i)[:3]) for i in range(num_masks)]
83
+ composite_layer = Image.new("RGBA", base_image.size, (0, 0, 0, 0))
84
+
85
+ for i, single_mask in enumerate(mask_data):
86
+ mask_bitmap = Image.fromarray((single_mask * 255).astype(np.uint8))
87
+ if mask_bitmap.size != base_image.size:
88
+ mask_bitmap = mask_bitmap.resize(base_image.size, resample=Image.NEAREST)
89
+
90
+ fill_color = rgb_colors[i]
91
+ color_fill = Image.new("RGBA", base_image.size, fill_color + (0,))
92
+ mask_alpha = mask_bitmap.point(lambda v: int(v * opacity) if v > 0 else 0)
93
+ color_fill.putalpha(mask_alpha)
94
+ composite_layer = Image.alpha_composite(composite_layer, color_fill)
95
+
96
+ return Image.alpha_composite(base_image, composite_layer).convert("RGB")
97
+
98
+ def draw_points_on_image(image, points):
99
+ """Draws red dots on the image to indicate click locations."""
100
+ if isinstance(image, np.ndarray):
101
+ image = Image.fromarray(image)
102
+
103
+ draw_img = image.copy()
104
+ draw = ImageDraw.Draw(draw_img)
105
+
106
+ for pt in points:
107
+ x, y = pt
108
+ r = 8 # Radius of point
109
+ draw.ellipse((x-r, y-r, x+r, y+r), fill="red", outline="white", width=4)
110
+
111
+ return draw_img
112
+
113
+ @spaces.GPU
114
+ def run_image_segmentation(source_img, text_query, conf_thresh=0.5):
115
+ if IMG_MODEL is None or IMG_PROCESSOR is None:
116
+ raise gr.Error("Models failed to load on startup.")
117
+
118
+ if source_img is None or not text_query:
119
+ raise gr.Error("Please provide an image and a text prompt.")
120
+
121
+ try:
122
+ pil_image = source_img.convert("RGB")
123
+ model_inputs = IMG_PROCESSOR(images=pil_image, text=text_query, return_tensors="pt").to(device)
124
+
125
+ with torch.no_grad():
126
+ inference_output = IMG_MODEL(**model_inputs)
127
+
128
+ processed_results = IMG_PROCESSOR.post_process_instance_segmentation(
129
+ inference_output,
130
+ threshold=conf_thresh,
131
+ mask_threshold=0.5,
132
+ target_sizes=model_inputs.get("original_sizes").tolist()
133
+ )[0]
134
+
135
+ annotation_list = []
136
+ raw_masks = processed_results['masks'].cpu().numpy()
137
+ raw_scores = processed_results['scores'].cpu().numpy()
138
+
139
+ for idx, mask_array in enumerate(raw_masks):
140
+ label_str = f"{text_query} ({raw_scores[idx]:.2f})"
141
+ annotation_list.append((mask_array, label_str))
142
+
143
+ return (pil_image, annotation_list)
144
+
145
+ except Exception as e:
146
+ raise gr.Error(f"Error during image processing: {e}")
147
+
148
+ @spaces.GPU
149
+ def run_image_click_gpu(input_image, x, y, points_state, labels_state):
150
+ if TRK_MODEL is None or TRK_PROCESSOR is None:
151
+ raise gr.Error("Tracker Model failed to load.")
152
+
153
+ if input_image is None: return input_image, [], []
154
+ if points_state is None: points_state = []; labels_state = []
155
+
156
+ # Append new point
157
+ points_state.append([x, y])
158
+ labels_state.append(1) # 1 indicates a positive click (foreground)
159
+
160
+ try:
161
+ # Prepare inputs format: [Batch, Point_Group, Point_Idx, Coord]
162
+ input_points = [[points_state]]
163
+ input_labels = [[labels_state]]
164
+
165
+ inputs = TRK_PROCESSOR(images=input_image, input_points=input_points, input_labels=input_labels, return_tensors="pt").to(device)
166
+
167
+ with torch.no_grad():
168
+ # multimask_output=True usually helps with ambiguity, but let's default to best mask for simplicity here
169
+ outputs = TRK_MODEL(**inputs, multimask_output=False)
170
+
171
+ # Post process
172
+ masks = TRK_PROCESSOR.post_process_masks(outputs.pred_masks.cpu(), inputs["original_sizes"], binarize=True)[0]
173
+
174
+ # Overlay mask
175
+ # masks shape is [1, 1, H, W] for single object tracking
176
+ final_img = apply_mask_overlay(input_image, masks[0])
177
+
178
+ # Draw the visual points on top
179
+ final_img = draw_points_on_image(final_img, points_state)
180
+
181
+ return final_img, points_state, labels_state
182
+
183
+ except Exception as e:
184
+ print(f"Tracker Error: {e}")
185
+ return input_image, points_state, labels_state
186
+
187
+ def image_click_handler(image, evt: gr.SelectData, points_state, labels_state):
188
+ # Wrapper to handle the Gradio select event
189
+ x, y = evt.index
190
+ return run_image_click_gpu(image, x, y, points_state, labels_state)
191
+
192
+ def calc_timeout_duration(vid_file, *args):
193
+ return args[-1] if args else 60
194
+
195
+ @spaces.GPU(duration=calc_timeout_duration)
196
+ def run_video_segmentation(source_vid, text_query, frame_limit, time_limit):
197
+ if VID_MODEL is None or VID_PROCESSOR is None:
198
+ raise gr.Error("Video Models failed to load on startup.")
199
+
200
+ if not source_vid or not text_query:
201
+ raise gr.Error("Missing video or prompt.")
202
+
203
+ try:
204
+ video_cap = cv2.VideoCapture(source_vid)
205
+ vid_fps = video_cap.get(cv2.CAP_PROP_FPS)
206
+ vid_w = int(video_cap.get(cv2.CAP_PROP_FRAME_WIDTH))
207
+ vid_h = int(video_cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
208
+
209
+ video_frames = []
210
+ counter = 0
211
+ while video_cap.isOpened():
212
+ ret, frame = video_cap.read()
213
+ if not ret or (frame_limit > 0 and counter >= frame_limit): break
214
+ video_frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
215
+ counter += 1
216
+ video_cap.release()
217
+
218
+ session = VID_PROCESSOR.init_video_session(video=video_frames, inference_device=device, dtype=torch.bfloat16)
219
+ session = VID_PROCESSOR.add_text_prompt(inference_session=session, text=text_query)
220
+
221
+ temp_out_path = tempfile.mktemp(suffix=".mp4")
222
+ video_writer = cv2.VideoWriter(temp_out_path, cv2.VideoWriter_fourcc(*'mp4v'), vid_fps, (vid_w, vid_h))
223
+
224
+ for model_out in VID_MODEL.propagate_in_video_iterator(inference_session=session, max_frame_num_to_track=len(video_frames)):
225
+ post_processed = VID_PROCESSOR.postprocess_outputs(session, model_out)
226
+ f_idx = model_out.frame_idx
227
+ original_pil = Image.fromarray(video_frames[f_idx])
228
+
229
+ if 'masks' in post_processed:
230
+ detected_masks = post_processed['masks']
231
+ if detected_masks.ndim == 4: detected_masks = detected_masks.squeeze(1)
232
+ final_frame = apply_mask_overlay(original_pil, detected_masks)
233
+ else:
234
+ final_frame = original_pil
235
+
236
+ video_writer.write(cv2.cvtColor(np.array(final_frame), cv2.COLOR_RGB2BGR))
237
+
238
+ video_writer.release()
239
+ return temp_out_path, "Video processing completed successfully.✅"
240
+
241
+ except Exception as e:
242
+ return None, f"Error during video processing: {str(e)}"
243
+
244
+ custom_css="""
245
+ #col-container { margin: 0 auto; max-width: 1200px; }
246
+ #main-title h1 { font-size: 2.2em !important; }
247
+ """
248
+
249
+ with gr.Blocks() as demo:
250
+ with gr.Column(elem_id="col-container"):
251
+ gr.Markdown("# **SAM3: Segment Anything Model 3**", elem_id="main-title")
252
+ gr.Markdown("Segment objects in image or video using **SAM3** with Text Prompts or Interactive Clicks.")
253
+
254
+ with gr.Tabs():
255
+ with gr.Tab("Image Segmentation"):
256
+ with gr.Row():
257
+ with gr.Column(scale=1):
258
+ image_input = gr.Image(label="Upload Image", type="pil", height=350)
259
+ txt_prompt_img = gr.Textbox(label="Text Prompt", placeholder="e.g., cat, face, car wheel")
260
+ with gr.Accordion("Advanced Settings", open=False):
261
+ conf_slider = gr.Slider(0.0, 1.0, value=0.45, step=0.05, label="Confidence Threshold")
262
+
263
+ btn_process_img = gr.Button("Segment Image", variant="primary")
264
+
265
+ with gr.Column(scale=1.5):
266
+ image_result = gr.AnnotatedImage(label="Segmented Result", height=410)
267
+
268
+ gr.Examples(
269
+ examples=[
270
+ ["examples/player.jpg", "player in white", 0.5],
271
+ ],
272
+ inputs=[image_input, txt_prompt_img, conf_slider],
273
+ outputs=[image_result],
274
+ fn=run_image_segmentation,
275
+ cache_examples=False,
276
+ label="Image Examples"
277
+ )
278
+
279
+ btn_process_img.click(
280
+ fn=run_image_segmentation,
281
+ inputs=[image_input, txt_prompt_img, conf_slider],
282
+ outputs=[image_result]
283
+ )
284
+
285
+ with gr.Tab("Video Segmentation"):
286
+ with gr.Row():
287
+ with gr.Column():
288
+ video_input = gr.Video(label="Upload Video", format="mp4", height=320)
289
+ txt_prompt_vid = gr.Textbox(label="Text Prompt", placeholder="e.g., person running, red car")
290
+
291
+ with gr.Row():
292
+ frame_limiter = gr.Slider(10, 500, value=60, step=10, label="Max Frames")
293
+ time_limiter = gr.Radio([60, 120, 180], value=60, label="Timeout (seconds)")
294
+
295
+ btn_process_vid = gr.Button("Segment Video", variant="primary")
296
+
297
+ with gr.Column():
298
+ video_result = gr.Video(label="Processed Video")
299
+ process_status = gr.Textbox(label="System Status", interactive=False)
300
+
301
+ gr.Examples(
302
+ examples=[
303
+ ["examples/sample_video.mp4", "players", 120, 120],
304
+ ],
305
+ inputs=[video_input, txt_prompt_vid, frame_limiter, time_limiter],
306
+ outputs=[video_result, process_status],
307
+ fn=run_video_segmentation,
308
+ cache_examples=False,
309
+ label="Video Examples"
310
+ )
311
+
312
+ btn_process_vid.click(
313
+ run_video_segmentation,
314
+ inputs=[video_input, txt_prompt_vid, frame_limiter, time_limiter],
315
+ outputs=[video_result, process_status]
316
+ )
317
+
318
+ with gr.Tab("Image Click Segmentation"):
319
+ with gr.Row():
320
+ with gr.Column(scale=1):
321
+ img_click_input = gr.Image(type="pil", label="Upload Image", interactive=True, height=450)
322
+
323
+ with gr.Row():
324
+ img_click_clear = gr.Button("Clear Points & Reset", variant="primary")
325
+
326
+ st_click_points = gr.State([])
327
+ st_click_labels = gr.State([])
328
+
329
+ with gr.Column(scale=1):
330
+ img_click_output = gr.Image(type="pil", label="Result Preview", height=450, interactive=False)
331
+
332
+ img_click_input.select(
333
+ image_click_handler,
334
+ inputs=[img_click_input, st_click_points, st_click_labels],
335
+ outputs=[img_click_output, st_click_points, st_click_labels]
336
+ )
337
+
338
+ img_click_clear.click(
339
+ lambda: (None, [], []),
340
+ outputs=[img_click_output, st_click_points, st_click_labels]
341
+ )
342
+
343
+ if __name__ == "__main__":
344
  demo.launch(css=custom_css, theme=gr.themes.Soft(), ssr_mode=False, mcp_server=True, show_error=True)