Spaces:
Sleeping
Sleeping
Update app.py
Browse files
app.py
CHANGED
|
@@ -26,6 +26,9 @@ default_harmful_objects = ["knife", "gun", "weapon", "blood", "syringe", "bomb",
|
|
| 26 |
# Case folder storage (in-memory for demo; persist to disk in production)
|
| 27 |
case_folders = {}
|
| 28 |
|
|
|
|
|
|
|
|
|
|
| 29 |
# 🔍 Core Features
|
| 30 |
# ----------------
|
| 31 |
# Resize image to max 1024x1024 while preserving aspect ratio
|
|
@@ -106,6 +109,12 @@ def manage_case_folder(case_name: str, results: List[dict], action: str = "add")
|
|
| 106 |
|
| 107 |
# 🛠️ User Options & Controls
|
| 108 |
# --------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 109 |
# Format results as a detailed report
|
| 110 |
def format_detailed_report(results, timestamp, keyword: str = ""):
|
| 111 |
filtered_results = keyword_search(results, keyword)
|
|
@@ -182,6 +191,8 @@ async def predict(
|
|
| 182 |
custom_weights: str = "",
|
| 183 |
description_level: str = "detailed"
|
| 184 |
):
|
|
|
|
|
|
|
| 185 |
if len(files) > 3:
|
| 186 |
raise HTTPException(status_code=400, detail="Maximum 3 images allowed.")
|
| 187 |
|
|
@@ -204,6 +215,8 @@ async def predict(
|
|
| 204 |
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
| 205 |
|
| 206 |
for idx, file in enumerate(files, 1):
|
|
|
|
|
|
|
| 207 |
try:
|
| 208 |
# Check file size (limit to 10MB)
|
| 209 |
image_data = await file.read()
|
|
@@ -218,21 +231,29 @@ async def predict(
|
|
| 218 |
|
| 219 |
# Core Features
|
| 220 |
if metadata:
|
|
|
|
|
|
|
| 221 |
result["metadata"] = extract_metadata(image_data)
|
| 222 |
|
| 223 |
if description:
|
|
|
|
|
|
|
| 224 |
prompt_desc = "<image> Provide a " + ("brief description of the image." if description_level == "basic" else "detailed description of the image, including objects, colors, people, and environmental context.")
|
| 225 |
inputs = processor(text=[prompt_desc], images=[image], return_tensors="pt", padding=True)
|
| 226 |
outputs = model.generate(**inputs, max_new_tokens=256 if description_level == "basic" else 512)
|
| 227 |
result["description"] = processor.decode(outputs[0], skip_special_tokens=True).replace(prompt_desc, "").strip() or "No description generated."
|
| 228 |
|
| 229 |
if signs:
|
|
|
|
|
|
|
| 230 |
prompt_ocr = "<image> Extract all visible text in the image, such as road signs, license plates, or billboards, with exact wording."
|
| 231 |
inputs_ocr = processor(text=[prompt_ocr], images=[image], return_tensors="pt", padding=True)
|
| 232 |
ocr_outputs = model.generate(**inputs_ocr, max_new_tokens=256)
|
| 233 |
result["signs"] = processor.decode(ocr_outputs[0], skip_special_tokens=True).replace(prompt_ocr, "").strip() or "None detected"
|
| 234 |
|
| 235 |
if harmful:
|
|
|
|
|
|
|
| 236 |
prompt_detect = f"<image> Identify any harmful objects ({', '.join(harmful_objects)}) in the image. List them explicitly and estimate confidence (0-100%) for each detection based on clarity."
|
| 237 |
inputs_detect = processor(text=[prompt_detect], images=[image], return_tensors="pt", padding=True)
|
| 238 |
detect_outputs = model.generate(**inputs_detect, max_new_tokens=256)
|
|
@@ -245,18 +266,24 @@ async def predict(
|
|
| 245 |
result["harmful_objects"] = harmful_detected if harmful_detected else [{"object": "None", "confidence": 0}]
|
| 246 |
|
| 247 |
if faces:
|
|
|
|
|
|
|
| 248 |
prompt_faces = "<image> Detect faces and estimate attributes such as age range (e.g., child, adult, senior), gender (male, female, unknown), and emotional cues (e.g., neutral, angry, scared)."
|
| 249 |
inputs_faces = processor(text=[prompt_faces], images=[image], return_tensors="pt", padding=True)
|
| 250 |
faces_outputs = model.generate(**inputs_faces, max_new_tokens=256)
|
| 251 |
result["faces"] = processor.decode(faces_outputs[0], skip_special_tokens=True).replace(prompt_faces, "").strip() or "No faces detected"
|
| 252 |
|
| 253 |
if objects:
|
|
|
|
|
|
|
| 254 |
prompt_objects = "<image> Identify and localize key objects (e.g., vehicles, weapons, bags) in the image. Provide object names and approximate bounding box coordinates (x_min, y_min, x_max, y_max) in the image."
|
| 255 |
inputs_objects = processor(text=[prompt_objects], images=[image], return_tensors="pt", padding=True)
|
| 256 |
objects_outputs = model.generate(**inputs_objects, max_new_tokens=256)
|
| 257 |
result["objects"] = [{"object": obj.strip(), "bbox": "(unknown)"} for obj in processor.decode(objects_outputs[0], skip_special_tokens=True).replace(prompt_objects, "").split(",") if obj.strip()] or [{"object": "None", "bbox": "N/A"}]
|
| 258 |
|
| 259 |
if activity:
|
|
|
|
|
|
|
| 260 |
prompt_activity = "<image> Describe any activities or actions occurring in the image, such as walking, running, or driving."
|
| 261 |
inputs_activity = processor(text=[prompt_activity], images=[image], return_tensors="pt", padding=True)
|
| 262 |
activity_outputs = model.generate(**inputs_activity, max_new_tokens=256)
|
|
@@ -264,21 +291,29 @@ async def predict(
|
|
| 264 |
|
| 265 |
# Investigation-Specific Features
|
| 266 |
if clothing:
|
|
|
|
|
|
|
| 267 |
prompt_clothing = "<image> Identify clothing items and their colors worn by people in the image."
|
| 268 |
inputs_clothing = processor(text=[prompt_clothing], images=[image], return_tensors="pt", padding=True)
|
| 269 |
clothing_outputs = model.generate(**inputs_clothing, max_new_tokens=256)
|
| 270 |
result["clothing"] = processor.decode(clothing_outputs[0], skip_special_tokens=True).replace(prompt_clothing, "").strip() or "No clothing detected"
|
| 271 |
|
| 272 |
if scene:
|
|
|
|
|
|
|
| 273 |
prompt_scene = "<image> Classify the scene type (e.g., indoor, outdoor, urban, rural) and estimate the time of day (e.g., day, night, dusk)."
|
| 274 |
inputs_scene = processor(text=[prompt_scene], images=[image], return_tensors="pt", padding=True)
|
| 275 |
scene_outputs = model.generate(**inputs_scene, max_new_tokens=256)
|
| 276 |
result["scene_context"] = processor.decode(scene_outputs[0], skip_special_tokens=True).replace(prompt_scene, "").strip() or "No context determined"
|
| 277 |
|
| 278 |
if threat_score:
|
|
|
|
|
|
|
| 279 |
result["threat_score"] = calculate_threat_score(result.get("harmful_objects", []), weights)
|
| 280 |
|
| 281 |
if similarity:
|
|
|
|
|
|
|
| 282 |
inputs_emb = processor(images=[image], return_tensors="pt", padding=True)
|
| 283 |
with torch.no_grad():
|
| 284 |
emb = model.vision_model(inputs_emb["pixel_values"]).last_hidden_state.mean(dim=1).cpu().numpy()
|
|
@@ -310,12 +345,14 @@ async def predict(
|
|
| 310 |
if similarity and len(image_embeddings) > 1:
|
| 311 |
base_embedding = image_embeddings[0]
|
| 312 |
for i in range(1, len(image_embeddings)):
|
|
|
|
|
|
|
| 313 |
sim = np.dot(base_embedding, image_embeddings[i].T) / (
|
| 314 |
np.linalg.norm(base_embedding) * np.linalg.norm(image_embeddings[i])
|
| 315 |
)
|
| 316 |
results[i]["similarity_to_image_1"] = float(sim[0][0])
|
| 317 |
|
| 318 |
-
return {"results": results, "analysis_timestamp": timestamp}
|
| 319 |
|
| 320 |
# Gradio interface
|
| 321 |
def gradio_predict(
|
|
@@ -345,6 +382,8 @@ def gradio_predict(
|
|
| 345 |
case_folder: str,
|
| 346 |
case_action: str
|
| 347 |
):
|
|
|
|
|
|
|
| 348 |
images = [image_1, image_2, image_3]
|
| 349 |
images = [img for img in images if img is not None]
|
| 350 |
if not images:
|
|
@@ -379,6 +418,11 @@ def gradio_predict(
|
|
| 379 |
errors = []
|
| 380 |
|
| 381 |
for idx, image in enumerate(images, 1):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 382 |
try:
|
| 383 |
# Convert Gradio image input to PIL and resize
|
| 384 |
image = Image.fromarray(image).convert("RGB")
|
|
@@ -391,21 +435,41 @@ def gradio_predict(
|
|
| 391 |
|
| 392 |
# Core Features
|
| 393 |
if metadata:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 394 |
result["metadata"] = extract_metadata(image_data.getvalue())
|
| 395 |
|
| 396 |
if description:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 397 |
prompt_desc = "<image> Provide a " + ("brief description of the image." if description_level == "basic" else "detailed description of the image, including objects, colors, people, and environmental context.")
|
| 398 |
inputs = processor(text=[prompt_desc], images=[image], return_tensors="pt", padding=True)
|
| 399 |
outputs = model.generate(**inputs, max_new_tokens=256 if description_level == "basic" else 512)
|
| 400 |
result["description"] = processor.decode(outputs[0], skip_special_tokens=True).replace(prompt_desc, "").strip() or "No description generated."
|
| 401 |
|
| 402 |
if signs:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 403 |
prompt_ocr = "<image> Extract all visible text in the image, such as road signs, license plates, or billboards, with exact wording."
|
| 404 |
inputs_ocr = processor(text=[prompt_ocr], images=[image], return_tensors="pt", padding=True)
|
| 405 |
ocr_outputs = model.generate(**inputs_ocr, max_new_tokens=256)
|
| 406 |
result["signs"] = processor.decode(ocr_outputs[0], skip_special_tokens=True).replace(prompt_ocr, "").strip() or "None detected"
|
| 407 |
|
| 408 |
if harmful:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 409 |
prompt_detect = f"<image> Identify any harmful objects ({', '.join(harmful_objects)}) in the image. List them explicitly and estimate confidence (0-100%) for each detection based on clarity."
|
| 410 |
inputs_detect = processor(text=[prompt_detect], images=[image], return_tensors="pt", padding=True)
|
| 411 |
detect_outputs = model.generate(**inputs_detect, max_new_tokens=256)
|
|
@@ -418,18 +482,33 @@ def gradio_predict(
|
|
| 418 |
result["harmful_objects"] = harmful_detected if harmful_detected else [{"object": "None", "confidence": 0}]
|
| 419 |
|
| 420 |
if faces:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 421 |
prompt_faces = "<image> Detect faces and estimate attributes such as age range (e.g., child, adult, senior), gender (male, female, unknown), and emotional cues (e.g., neutral, angry, scared)."
|
| 422 |
inputs_faces = processor(text=[prompt_faces], images=[image], return_tensors="pt", padding=True)
|
| 423 |
faces_outputs = model.generate(**inputs_faces, max_new_tokens=256)
|
| 424 |
result["faces"] = processor.decode(faces_outputs[0], skip_special_tokens=True).replace(prompt_faces, "").strip() or "No faces detected"
|
| 425 |
|
| 426 |
if objects:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 427 |
prompt_objects = "<image> Identify and localize key objects (e.g., vehicles, weapons, bags) in the image. Provide object names and approximate bounding box coordinates (x_min, y_min, x_max, y_max) in the image."
|
| 428 |
inputs_objects = processor(text=[prompt_objects], images=[image], return_tensors="pt", padding=True)
|
| 429 |
objects_outputs = model.generate(**inputs_objects, max_new_tokens=256)
|
| 430 |
result["objects"] = [{"object": obj.strip(), "bbox": "(unknown)"} for obj in processor.decode(objects_outputs[0], skip_special_tokens=True).replace(prompt_objects, "").split(",") if obj.strip()] or [{"object": "None", "bbox": "N/A"}]
|
| 431 |
|
| 432 |
if activity:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 433 |
prompt_activity = "<image> Describe any activities or actions occurring in the image, such as walking, running, or driving."
|
| 434 |
inputs_activity = processor(text=[prompt_activity], images=[image], return_tensors="pt", padding=True)
|
| 435 |
activity_outputs = model.generate(**inputs_activity, max_new_tokens=256)
|
|
@@ -437,18 +516,33 @@ def gradio_predict(
|
|
| 437 |
|
| 438 |
# Investigation-Specific Features
|
| 439 |
if clothing:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 440 |
prompt_clothing = "<image> Identify clothing items and their colors worn by people in the image."
|
| 441 |
inputs_clothing = processor(text=[prompt_clothing], images=[image], return_tensors="pt", padding=True)
|
| 442 |
clothing_outputs = model.generate(**inputs_clothing, max_new_tokens=256)
|
| 443 |
result["clothing"] = processor.decode(clothing_outputs[0], skip_special_tokens=True).replace(prompt_clothing, "").strip() or "No clothing detected"
|
| 444 |
|
| 445 |
if scene:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 446 |
prompt_scene = "<image> Classify the scene type (e.g., indoor, outdoor, urban, rural) and estimate the time of day (e.g., day, night, dusk)."
|
| 447 |
inputs_scene = processor(text=[prompt_scene], images=[image], return_tensors="pt", padding=True)
|
| 448 |
scene_outputs = model.generate(**inputs_scene, max_new_tokens=256)
|
| 449 |
result["scene_context"] = processor.decode(scene_outputs[0], skip_special_tokens=True).replace(prompt_scene, "").strip() or "No context determined"
|
| 450 |
|
| 451 |
if threat_score:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 452 |
result["threat_score"] = calculate_threat_score(result.get("harmful_objects", []), weights)
|
| 453 |
|
| 454 |
# User Options & Controls
|
|
@@ -460,6 +554,11 @@ def gradio_predict(
|
|
| 460 |
result["comments"] = comments.strip()
|
| 461 |
|
| 462 |
if similarity:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 463 |
inputs_emb = processor(images=[image], return_tensors="pt", padding=True)
|
| 464 |
with torch.no_grad():
|
| 465 |
emb = model.vision_model(inputs_emb["pixel_values"]).last_hidden_state.mean(dim=1).cpu().numpy()
|
|
@@ -467,11 +566,7 @@ def gradio_predict(
|
|
| 467 |
|
| 468 |
results.append(result)
|
| 469 |
except Exception as e:
|
| 470 |
-
|
| 471 |
-
"image_id": f"Image_{idx}",
|
| 472 |
-
"error": str(e)
|
| 473 |
-
}
|
| 474 |
-
errors.append(error_result)
|
| 475 |
result = {
|
| 476 |
"image_id": f"Image_{idx}",
|
| 477 |
"timestamp": timestamp,
|
|
@@ -499,6 +594,11 @@ def gradio_predict(
|
|
| 499 |
if similarity and len(image_embeddings) > 1:
|
| 500 |
base_embedding = image_embeddings[0]
|
| 501 |
for i in range(1, len(image_embeddings)):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 502 |
sim = np.dot(base_embedding, image_embeddings[i].T) / (
|
| 503 |
np.linalg.norm(base_embedding) * np.linalg.norm(image_embeddings[i])
|
| 504 |
)
|
|
@@ -517,10 +617,18 @@ def gradio_predict(
|
|
| 517 |
|
| 518 |
# Handle case folder
|
| 519 |
case_output = ""
|
| 520 |
-
if case_folder.strip():
|
| 521 |
case_output = manage_case_folder(case_folder, results, case_action)
|
| 522 |
|
| 523 |
-
# Format output
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 524 |
output = f"**Analysis Timestamp**: {timestamp}\n\n"
|
| 525 |
headers = ["Image ID"]
|
| 526 |
if metadata:
|
|
@@ -610,47 +718,72 @@ def gradio_predict(
|
|
| 610 |
output += "\n**Detailed Report**:\n" + format_detailed_report(results, timestamp, keyword_search)
|
| 611 |
|
| 612 |
# Append case folder output
|
| 613 |
-
if case_output:
|
| 614 |
output += "\n**Case Folder**:\n" + case_output
|
| 615 |
|
| 616 |
return output
|
| 617 |
|
| 618 |
# Gradio interface
|
| 619 |
-
|
| 620 |
-
|
| 621 |
-
|
| 622 |
-
|
| 623 |
-
|
| 624 |
-
gr.Image(label="Upload Image
|
| 625 |
-
gr.
|
| 626 |
-
gr.
|
| 627 |
-
|
| 628 |
-
|
| 629 |
-
gr.
|
| 630 |
-
|
| 631 |
-
|
| 632 |
-
|
| 633 |
-
|
| 634 |
-
|
| 635 |
-
|
| 636 |
-
|
| 637 |
-
|
| 638 |
-
|
| 639 |
-
|
| 640 |
-
|
| 641 |
-
|
| 642 |
-
|
| 643 |
-
gr.
|
| 644 |
-
|
| 645 |
-
|
| 646 |
-
|
| 647 |
-
|
| 648 |
-
|
| 649 |
-
|
| 650 |
-
|
| 651 |
-
|
| 652 |
-
|
| 653 |
-
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 654 |
|
| 655 |
if __name__ == "__main__":
|
| 656 |
# Launch Gradio interface for Hugging Face Spaces
|
|
|
|
| 26 |
# Case folder storage (in-memory for demo; persist to disk in production)
|
| 27 |
case_folders = {}
|
| 28 |
|
| 29 |
+
# Cancellation flag
|
| 30 |
+
cancel_flag = False
|
| 31 |
+
|
| 32 |
# 🔍 Core Features
|
| 33 |
# ----------------
|
| 34 |
# Resize image to max 1024x1024 while preserving aspect ratio
|
|
|
|
| 109 |
|
| 110 |
# 🛠️ User Options & Controls
|
| 111 |
# --------------------------
|
| 112 |
+
# Cancel analysis function
|
| 113 |
+
def cancel_analysis():
|
| 114 |
+
global cancel_flag
|
| 115 |
+
cancel_flag = True
|
| 116 |
+
return "Analysis cancellation requested. Please wait for the current operation to stop."
|
| 117 |
+
|
| 118 |
# Format results as a detailed report
|
| 119 |
def format_detailed_report(results, timestamp, keyword: str = ""):
|
| 120 |
filtered_results = keyword_search(results, keyword)
|
|
|
|
| 191 |
custom_weights: str = "",
|
| 192 |
description_level: str = "detailed"
|
| 193 |
):
|
| 194 |
+
global cancel_flag
|
| 195 |
+
cancel_flag = False # Reset cancellation flag
|
| 196 |
if len(files) > 3:
|
| 197 |
raise HTTPException(status_code=400, detail="Maximum 3 images allowed.")
|
| 198 |
|
|
|
|
| 215 |
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
| 216 |
|
| 217 |
for idx, file in enumerate(files, 1):
|
| 218 |
+
if cancel_flag:
|
| 219 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 220 |
try:
|
| 221 |
# Check file size (limit to 10MB)
|
| 222 |
image_data = await file.read()
|
|
|
|
| 231 |
|
| 232 |
# Core Features
|
| 233 |
if metadata:
|
| 234 |
+
if cancel_flag:
|
| 235 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 236 |
result["metadata"] = extract_metadata(image_data)
|
| 237 |
|
| 238 |
if description:
|
| 239 |
+
if cancel_flag:
|
| 240 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 241 |
prompt_desc = "<image> Provide a " + ("brief description of the image." if description_level == "basic" else "detailed description of the image, including objects, colors, people, and environmental context.")
|
| 242 |
inputs = processor(text=[prompt_desc], images=[image], return_tensors="pt", padding=True)
|
| 243 |
outputs = model.generate(**inputs, max_new_tokens=256 if description_level == "basic" else 512)
|
| 244 |
result["description"] = processor.decode(outputs[0], skip_special_tokens=True).replace(prompt_desc, "").strip() or "No description generated."
|
| 245 |
|
| 246 |
if signs:
|
| 247 |
+
if cancel_flag:
|
| 248 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 249 |
prompt_ocr = "<image> Extract all visible text in the image, such as road signs, license plates, or billboards, with exact wording."
|
| 250 |
inputs_ocr = processor(text=[prompt_ocr], images=[image], return_tensors="pt", padding=True)
|
| 251 |
ocr_outputs = model.generate(**inputs_ocr, max_new_tokens=256)
|
| 252 |
result["signs"] = processor.decode(ocr_outputs[0], skip_special_tokens=True).replace(prompt_ocr, "").strip() or "None detected"
|
| 253 |
|
| 254 |
if harmful:
|
| 255 |
+
if cancel_flag:
|
| 256 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 257 |
prompt_detect = f"<image> Identify any harmful objects ({', '.join(harmful_objects)}) in the image. List them explicitly and estimate confidence (0-100%) for each detection based on clarity."
|
| 258 |
inputs_detect = processor(text=[prompt_detect], images=[image], return_tensors="pt", padding=True)
|
| 259 |
detect_outputs = model.generate(**inputs_detect, max_new_tokens=256)
|
|
|
|
| 266 |
result["harmful_objects"] = harmful_detected if harmful_detected else [{"object": "None", "confidence": 0}]
|
| 267 |
|
| 268 |
if faces:
|
| 269 |
+
if cancel_flag:
|
| 270 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 271 |
prompt_faces = "<image> Detect faces and estimate attributes such as age range (e.g., child, adult, senior), gender (male, female, unknown), and emotional cues (e.g., neutral, angry, scared)."
|
| 272 |
inputs_faces = processor(text=[prompt_faces], images=[image], return_tensors="pt", padding=True)
|
| 273 |
faces_outputs = model.generate(**inputs_faces, max_new_tokens=256)
|
| 274 |
result["faces"] = processor.decode(faces_outputs[0], skip_special_tokens=True).replace(prompt_faces, "").strip() or "No faces detected"
|
| 275 |
|
| 276 |
if objects:
|
| 277 |
+
if cancel_flag:
|
| 278 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 279 |
prompt_objects = "<image> Identify and localize key objects (e.g., vehicles, weapons, bags) in the image. Provide object names and approximate bounding box coordinates (x_min, y_min, x_max, y_max) in the image."
|
| 280 |
inputs_objects = processor(text=[prompt_objects], images=[image], return_tensors="pt", padding=True)
|
| 281 |
objects_outputs = model.generate(**inputs_objects, max_new_tokens=256)
|
| 282 |
result["objects"] = [{"object": obj.strip(), "bbox": "(unknown)"} for obj in processor.decode(objects_outputs[0], skip_special_tokens=True).replace(prompt_objects, "").split(",") if obj.strip()] or [{"object": "None", "bbox": "N/A"}]
|
| 283 |
|
| 284 |
if activity:
|
| 285 |
+
if cancel_flag:
|
| 286 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 287 |
prompt_activity = "<image> Describe any activities or actions occurring in the image, such as walking, running, or driving."
|
| 288 |
inputs_activity = processor(text=[prompt_activity], images=[image], return_tensors="pt", padding=True)
|
| 289 |
activity_outputs = model.generate(**inputs_activity, max_new_tokens=256)
|
|
|
|
| 291 |
|
| 292 |
# Investigation-Specific Features
|
| 293 |
if clothing:
|
| 294 |
+
if cancel_flag:
|
| 295 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 296 |
prompt_clothing = "<image> Identify clothing items and their colors worn by people in the image."
|
| 297 |
inputs_clothing = processor(text=[prompt_clothing], images=[image], return_tensors="pt", padding=True)
|
| 298 |
clothing_outputs = model.generate(**inputs_clothing, max_new_tokens=256)
|
| 299 |
result["clothing"] = processor.decode(clothing_outputs[0], skip_special_tokens=True).replace(prompt_clothing, "").strip() or "No clothing detected"
|
| 300 |
|
| 301 |
if scene:
|
| 302 |
+
if cancel_flag:
|
| 303 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 304 |
prompt_scene = "<image> Classify the scene type (e.g., indoor, outdoor, urban, rural) and estimate the time of day (e.g., day, night, dusk)."
|
| 305 |
inputs_scene = processor(text=[prompt_scene], images=[image], return_tensors="pt", padding=True)
|
| 306 |
scene_outputs = model.generate(**inputs_scene, max_new_tokens=256)
|
| 307 |
result["scene_context"] = processor.decode(scene_outputs[0], skip_special_tokens=True).replace(prompt_scene, "").strip() or "No context determined"
|
| 308 |
|
| 309 |
if threat_score:
|
| 310 |
+
if cancel_flag:
|
| 311 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 312 |
result["threat_score"] = calculate_threat_score(result.get("harmful_objects", []), weights)
|
| 313 |
|
| 314 |
if similarity:
|
| 315 |
+
if cancel_flag:
|
| 316 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 317 |
inputs_emb = processor(images=[image], return_tensors="pt", padding=True)
|
| 318 |
with torch.no_grad():
|
| 319 |
emb = model.vision_model(inputs_emb["pixel_values"]).last_hidden_state.mean(dim=1).cpu().numpy()
|
|
|
|
| 345 |
if similarity and len(image_embeddings) > 1:
|
| 346 |
base_embedding = image_embeddings[0]
|
| 347 |
for i in range(1, len(image_embeddings)):
|
| 348 |
+
if cancel_flag:
|
| 349 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Cancelled"}
|
| 350 |
sim = np.dot(base_embedding, image_embeddings[i].T) / (
|
| 351 |
np.linalg.norm(base_embedding) * np.linalg.norm(image_embeddings[i])
|
| 352 |
)
|
| 353 |
results[i]["similarity_to_image_1"] = float(sim[0][0])
|
| 354 |
|
| 355 |
+
return {"results": results, "analysis_timestamp": timestamp, "status": "Completed"}
|
| 356 |
|
| 357 |
# Gradio interface
|
| 358 |
def gradio_predict(
|
|
|
|
| 382 |
case_folder: str,
|
| 383 |
case_action: str
|
| 384 |
):
|
| 385 |
+
global cancel_flag
|
| 386 |
+
cancel_flag = False # Reset cancellation flag
|
| 387 |
images = [image_1, image_2, image_3]
|
| 388 |
images = [img for img in images if img is not None]
|
| 389 |
if not images:
|
|
|
|
| 418 |
errors = []
|
| 419 |
|
| 420 |
for idx, image in enumerate(images, 1):
|
| 421 |
+
if cancel_flag:
|
| 422 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 423 |
+
if results:
|
| 424 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 425 |
+
return output
|
| 426 |
try:
|
| 427 |
# Convert Gradio image input to PIL and resize
|
| 428 |
image = Image.fromarray(image).convert("RGB")
|
|
|
|
| 435 |
|
| 436 |
# Core Features
|
| 437 |
if metadata:
|
| 438 |
+
if cancel_flag:
|
| 439 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 440 |
+
if results:
|
| 441 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 442 |
+
return output
|
| 443 |
result["metadata"] = extract_metadata(image_data.getvalue())
|
| 444 |
|
| 445 |
if description:
|
| 446 |
+
if cancel_flag:
|
| 447 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 448 |
+
if results:
|
| 449 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 450 |
+
return output
|
| 451 |
prompt_desc = "<image> Provide a " + ("brief description of the image." if description_level == "basic" else "detailed description of the image, including objects, colors, people, and environmental context.")
|
| 452 |
inputs = processor(text=[prompt_desc], images=[image], return_tensors="pt", padding=True)
|
| 453 |
outputs = model.generate(**inputs, max_new_tokens=256 if description_level == "basic" else 512)
|
| 454 |
result["description"] = processor.decode(outputs[0], skip_special_tokens=True).replace(prompt_desc, "").strip() or "No description generated."
|
| 455 |
|
| 456 |
if signs:
|
| 457 |
+
if cancel_flag:
|
| 458 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 459 |
+
if results:
|
| 460 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 461 |
+
return output
|
| 462 |
prompt_ocr = "<image> Extract all visible text in the image, such as road signs, license plates, or billboards, with exact wording."
|
| 463 |
inputs_ocr = processor(text=[prompt_ocr], images=[image], return_tensors="pt", padding=True)
|
| 464 |
ocr_outputs = model.generate(**inputs_ocr, max_new_tokens=256)
|
| 465 |
result["signs"] = processor.decode(ocr_outputs[0], skip_special_tokens=True).replace(prompt_ocr, "").strip() or "None detected"
|
| 466 |
|
| 467 |
if harmful:
|
| 468 |
+
if cancel_flag:
|
| 469 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 470 |
+
if results:
|
| 471 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 472 |
+
return output
|
| 473 |
prompt_detect = f"<image> Identify any harmful objects ({', '.join(harmful_objects)}) in the image. List them explicitly and estimate confidence (0-100%) for each detection based on clarity."
|
| 474 |
inputs_detect = processor(text=[prompt_detect], images=[image], return_tensors="pt", padding=True)
|
| 475 |
detect_outputs = model.generate(**inputs_detect, max_new_tokens=256)
|
|
|
|
| 482 |
result["harmful_objects"] = harmful_detected if harmful_detected else [{"object": "None", "confidence": 0}]
|
| 483 |
|
| 484 |
if faces:
|
| 485 |
+
if cancel_flag:
|
| 486 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 487 |
+
if results:
|
| 488 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 489 |
+
return output
|
| 490 |
prompt_faces = "<image> Detect faces and estimate attributes such as age range (e.g., child, adult, senior), gender (male, female, unknown), and emotional cues (e.g., neutral, angry, scared)."
|
| 491 |
inputs_faces = processor(text=[prompt_faces], images=[image], return_tensors="pt", padding=True)
|
| 492 |
faces_outputs = model.generate(**inputs_faces, max_new_tokens=256)
|
| 493 |
result["faces"] = processor.decode(faces_outputs[0], skip_special_tokens=True).replace(prompt_faces, "").strip() or "No faces detected"
|
| 494 |
|
| 495 |
if objects:
|
| 496 |
+
if cancel_flag:
|
| 497 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 498 |
+
if results:
|
| 499 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 500 |
+
return output
|
| 501 |
prompt_objects = "<image> Identify and localize key objects (e.g., vehicles, weapons, bags) in the image. Provide object names and approximate bounding box coordinates (x_min, y_min, x_max, y_max) in the image."
|
| 502 |
inputs_objects = processor(text=[prompt_objects], images=[image], return_tensors="pt", padding=True)
|
| 503 |
objects_outputs = model.generate(**inputs_objects, max_new_tokens=256)
|
| 504 |
result["objects"] = [{"object": obj.strip(), "bbox": "(unknown)"} for obj in processor.decode(objects_outputs[0], skip_special_tokens=True).replace(prompt_objects, "").split(",") if obj.strip()] or [{"object": "None", "bbox": "N/A"}]
|
| 505 |
|
| 506 |
if activity:
|
| 507 |
+
if cancel_flag:
|
| 508 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 509 |
+
if results:
|
| 510 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 511 |
+
return output
|
| 512 |
prompt_activity = "<image> Describe any activities or actions occurring in the image, such as walking, running, or driving."
|
| 513 |
inputs_activity = processor(text=[prompt_activity], images=[image], return_tensors="pt", padding=True)
|
| 514 |
activity_outputs = model.generate(**inputs_activity, max_new_tokens=256)
|
|
|
|
| 516 |
|
| 517 |
# Investigation-Specific Features
|
| 518 |
if clothing:
|
| 519 |
+
if cancel_flag:
|
| 520 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 521 |
+
if results:
|
| 522 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 523 |
+
return output
|
| 524 |
prompt_clothing = "<image> Identify clothing items and their colors worn by people in the image."
|
| 525 |
inputs_clothing = processor(text=[prompt_clothing], images=[image], return_tensors="pt", padding=True)
|
| 526 |
clothing_outputs = model.generate(**inputs_clothing, max_new_tokens=256)
|
| 527 |
result["clothing"] = processor.decode(clothing_outputs[0], skip_special_tokens=True).replace(prompt_clothing, "").strip() or "No clothing detected"
|
| 528 |
|
| 529 |
if scene:
|
| 530 |
+
if cancel_flag:
|
| 531 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 532 |
+
if results:
|
| 533 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 534 |
+
return output
|
| 535 |
prompt_scene = "<image> Classify the scene type (e.g., indoor, outdoor, urban, rural) and estimate the time of day (e.g., day, night, dusk)."
|
| 536 |
inputs_scene = processor(text=[prompt_scene], images=[image], return_tensors="pt", padding=True)
|
| 537 |
scene_outputs = model.generate(**inputs_scene, max_new_tokens=256)
|
| 538 |
result["scene_context"] = processor.decode(scene_outputs[0], skip_special_tokens=True).replace(prompt_scene, "").strip() or "No context determined"
|
| 539 |
|
| 540 |
if threat_score:
|
| 541 |
+
if cancel_flag:
|
| 542 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 543 |
+
if results:
|
| 544 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 545 |
+
return output
|
| 546 |
result["threat_score"] = calculate_threat_score(result.get("harmful_objects", []), weights)
|
| 547 |
|
| 548 |
# User Options & Controls
|
|
|
|
| 554 |
result["comments"] = comments.strip()
|
| 555 |
|
| 556 |
if similarity:
|
| 557 |
+
if cancel_flag:
|
| 558 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 559 |
+
if results:
|
| 560 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 561 |
+
return output
|
| 562 |
inputs_emb = processor(images=[image], return_tensors="pt", padding=True)
|
| 563 |
with torch.no_grad():
|
| 564 |
emb = model.vision_model(inputs_emb["pixel_values"]).last_hidden_state.mean(dim=1).cpu().numpy()
|
|
|
|
| 566 |
|
| 567 |
results.append(result)
|
| 568 |
except Exception as e:
|
| 569 |
+
errors.append({"image_id": f"Image_{idx}", "error": str(e)})
|
|
|
|
|
|
|
|
|
|
|
|
|
| 570 |
result = {
|
| 571 |
"image_id": f"Image_{idx}",
|
| 572 |
"timestamp": timestamp,
|
|
|
|
| 594 |
if similarity and len(image_embeddings) > 1:
|
| 595 |
base_embedding = image_embeddings[0]
|
| 596 |
for i in range(1, len(image_embeddings)):
|
| 597 |
+
if cancel_flag:
|
| 598 |
+
output = f"**Analysis Cancelled at**: {timestamp}\n"
|
| 599 |
+
if results:
|
| 600 |
+
output += "\n**Partial Results**:\n" + format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 601 |
+
return output
|
| 602 |
sim = np.dot(base_embedding, image_embeddings[i].T) / (
|
| 603 |
np.linalg.norm(base_embedding) * np.linalg.norm(image_embeddings[i])
|
| 604 |
)
|
|
|
|
| 617 |
|
| 618 |
# Handle case folder
|
| 619 |
case_output = ""
|
| 620 |
+
if case_folder.strip() and not cancel_flag:
|
| 621 |
case_output = manage_case_folder(case_folder, results, case_action)
|
| 622 |
|
| 623 |
+
# Format output
|
| 624 |
+
output = format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors)
|
| 625 |
+
|
| 626 |
+
# Reset cancellation flag
|
| 627 |
+
cancel_flag = False
|
| 628 |
+
return output
|
| 629 |
+
|
| 630 |
+
# Helper function to format results
|
| 631 |
+
def format_results(results, timestamp, keyword_search, json_export, detailed_report, case_folder, case_action, filter_attributes, annotation, flag_images, comments, errors):
|
| 632 |
output = f"**Analysis Timestamp**: {timestamp}\n\n"
|
| 633 |
headers = ["Image ID"]
|
| 634 |
if metadata:
|
|
|
|
| 718 |
output += "\n**Detailed Report**:\n" + format_detailed_report(results, timestamp, keyword_search)
|
| 719 |
|
| 720 |
# Append case folder output
|
| 721 |
+
if case_folder.strip() and case_output:
|
| 722 |
output += "\n**Case Folder**:\n" + case_output
|
| 723 |
|
| 724 |
return output
|
| 725 |
|
| 726 |
# Gradio interface
|
| 727 |
+
with gr.Blocks() as iface:
|
| 728 |
+
gr.Markdown("# VisionSage: Image Analysis for Crime Investigation")
|
| 729 |
+
gr.Markdown("Upload up to 3 images (up to 10MB each, any resolution) and select outputs for investigative analysis. Features include metadata extraction, descriptions, text recognition, object detection, facial analysis, activity recognition, clothing detection, threat scoring, and case management. Export results as JSON or a detailed report.")
|
| 730 |
+
|
| 731 |
+
with gr.Row():
|
| 732 |
+
image_1 = gr.Image(label="Upload Image 1 (up to 10MB)")
|
| 733 |
+
image_2 = gr.Image(label="Upload Image 2 (up to 10MB)")
|
| 734 |
+
image_3 = gr.Image(label="Upload Image 3 (up to 10MB)")
|
| 735 |
+
|
| 736 |
+
with gr.Row():
|
| 737 |
+
with gr.Column():
|
| 738 |
+
description = gr.Checkbox(label="Description", value=True)
|
| 739 |
+
signs = gr.Checkbox(label="Signs", value=True)
|
| 740 |
+
harmful = gr.Checkbox(label="Harmful Objects", value=True)
|
| 741 |
+
similarity = gr.Checkbox(label="Similarity", value=True)
|
| 742 |
+
faces = gr.Checkbox(label="Facial Attributes", value=False)
|
| 743 |
+
objects = gr.Checkbox(label="Localized Objects", value=False)
|
| 744 |
+
scene = gr.Checkbox(label="Scene Context", value=False)
|
| 745 |
+
metadata = gr.Checkbox(label="Metadata", value=False)
|
| 746 |
+
activity = gr.Checkbox(label="Activity Recognition", value=False)
|
| 747 |
+
clothing = gr.Checkbox(label="Clothing/Colors", value=False)
|
| 748 |
+
threat_score = gr.Checkbox(label="Threat Score", value=False)
|
| 749 |
+
combined = gr.Checkbox(label="Combined (All Outputs)", value=False)
|
| 750 |
+
|
| 751 |
+
with gr.Column():
|
| 752 |
+
json_export = gr.Checkbox(label="JSON Export", value=False)
|
| 753 |
+
detailed_report = gr.Checkbox(label="Detailed Report", value=False)
|
| 754 |
+
custom_harmful = gr.Textbox(label="Custom Harmful Objects (comma-separated, e.g., handgun, crowbar)", placeholder="Enter objects to detect")
|
| 755 |
+
custom_weights = gr.Textbox(label="Custom Weights (e.g., knife:2.0,gun:3.0)", placeholder="Enter object:weight pairs")
|
| 756 |
+
description_level = gr.Radio(label="Description Level", choices=["basic", "detailed"], value="detailed")
|
| 757 |
+
keyword_search = gr.Textbox(label="Keyword Search", placeholder="Enter keywords to filter results")
|
| 758 |
+
filter_attributes = gr.Textbox(label="Filter by Attributes (comma-separated)", placeholder="e.g., red, adult, urban")
|
| 759 |
+
annotation = gr.Textbox(label="Manual Annotation", placeholder="Add labels or notes to images")
|
| 760 |
+
flag_images = gr.Checkbox(label="Flag Important Images", value=False)
|
| 761 |
+
comments = gr.Textbox(label="Investigator Comments", placeholder="Add comments or notes")
|
| 762 |
+
case_folder = gr.Textbox(label="Case Folder Name", placeholder="Enter case folder name")
|
| 763 |
+
case_action = gr.Radio(label="Case Folder Action", choices=["add", "view", "clear"], value="add")
|
| 764 |
+
|
| 765 |
+
with gr.Row():
|
| 766 |
+
submit_button = gr.Button("Submit")
|
| 767 |
+
cancel_button = gr.Button("Cancel Analysis")
|
| 768 |
+
|
| 769 |
+
output = gr.Textbox(label="Investigation Results", placeholder="Results will appear here...")
|
| 770 |
+
|
| 771 |
+
submit_button.click(
|
| 772 |
+
fn=gradio_predict,
|
| 773 |
+
inputs=[
|
| 774 |
+
image_1, image_2, image_3,
|
| 775 |
+
description, signs, harmful, similarity, faces, objects, scene, metadata, activity, clothing, threat_score, combined,
|
| 776 |
+
json_export, detailed_report, custom_harmful, custom_weights, description_level, keyword_search, filter_attributes,
|
| 777 |
+
annotation, flag_images, comments, case_folder, case_action
|
| 778 |
+
],
|
| 779 |
+
outputs=output
|
| 780 |
+
)
|
| 781 |
+
|
| 782 |
+
cancel_button.click(
|
| 783 |
+
fn=cancel_analysis,
|
| 784 |
+
inputs=[],
|
| 785 |
+
outputs=output
|
| 786 |
+
)
|
| 787 |
|
| 788 |
if __name__ == "__main__":
|
| 789 |
# Launch Gradio interface for Hugging Face Spaces
|