import gradio as gr from PIL import Image from ultralytics import YOLO import requests import json import logging logging.basicConfig(level=logging.INFO) model = YOLO("BP_Multiple_Objects_Complicated_v1.pt") def detect_objects(images): results = model(images) all_bboxes = [] all_bboxes2 = [] all_segments = [] for result in results: boxes = result.boxes.xywhn.tolist() boxes2 = result.boxes.xywh.tolist() all_bboxes.append(boxes) all_bboxes2.append(boxes2) if result.masks is not None: masks = result.masks.xyn sub_arrays = [arr.tolist() for arr in masks] else: sub_arrays = [] all_segments.append(sub_arrays) return all_bboxes, all_bboxes2, all_segments def create_solutions(image_urls, all_bboxes, all_bboxes2, all_segments, file_ids): solutions = [] img_id = 1 box_id = 1 cat_id = 1 for image_url, bbox, bbox2, segmnt, file_id in zip(image_urls, all_bboxes, all_bboxes2, all_segments, file_ids): temp=[] for subbox, subbox2, subsegmnt in zip(bbox, bbox2, segmnt): w = subbox2[2] h = subbox2[3] area = w * h flattened_segmnt = [item for sublist in subsegmnt for item in sublist] ans = {"image_id": img_id, "id": box_id, "area": area, "category_id": cat_id, "bbox": subbox, "segment": flattened_segmnt} box_id += 1 temp.append(ans) img_id += 1 obj ={"url": image_url, "answer":temp, "qcUser" : None, "normalfileID" : file_id} solutions.append(obj) return solutions # def send_results_to_api(data, result_url): # headers = {"Content-Type": "application/json"} # response = requests.post(result_url, json=data, headers=headers) # if response.status_code == 200: # return response.json() # else: # return {"error": f"Failed to send results to API: {response.status_code}"} def process_images(params): try: params = json.loads(params) except json.JSONDecodeError as e: logging.error(f"Invalid JSON input: {e.msg} at line {e.lineno} column {e.colno}") return {"error": f"Invalid JSON input: {e.msg} at line {e.lineno} column {e.colno}"} image_urls = params.get("urls", []) if not params.get("normalfileID",[]): file_ids = [None]*len(image_urls) else: file_ids = params.get("normalfileID",[]) # api = params.get("api", "") # job_id = params.get("job_id", "") if not image_urls: logging.error("Missing required parameters: 'urls'") return {"error": "Missing required parameters: 'urls'"} try: images = [Image.open(requests.get(url, stream=True).raw) for url in image_urls] except Exception as e: logging.error(f"Error loading images: {e}") return {"error": f"Error loading images: {str(e)}"} all_bboxes, all_bboxes2, all_segments = detect_objects(images) solutions = create_solutions(image_urls, all_bboxes, all_bboxes2, all_segments, file_ids) # result_url = f"{api}/{job_id}" # send_results_to_api(solutions, result_url) return json.dumps({"solutions": solutions}) inputt = gr.Textbox(label="Parameters (JSON format)") outputs = gr.JSON() application = gr.Interface(fn=process_images, inputs=inputt, outputs=outputs, title="Multiple Object Segmentation with API Integration") application.launch() # import gradio as gr # from PIL import Image # from ultralytics import YOLO # import requests # import json # model = YOLO("BP_Multiple_Objects_Complicated_v1.pt") # def detect_objects(images): # results = model(images) # all_bboxes = [] # all_bboxes2 = [] # all_segments = [] # for result in results: # boxes = result.boxes.xywhn.tolist() # boxes2 = result.boxes.xywh.tolist() # all_bboxes.append(boxes) # all_bboxes2.append(boxes2) # masks = result.masks.xyn # sub_arrays = [arr.tolist() for arr in masks] # all_segments.append(sub_arrays) # return all_bboxes, all_bboxes2, all_segments # def create_solutions(image_urls, all_bboxes, all_bboxes2, all_segments): # solutions = [] # img_id =1 # box_id =1 # cat_id =1 # for image_url, bbox, bbox2, segmnt in zip(image_urls, all_bboxes, all_bboxes2, all_segments): # for subbox, subbox2, subsegmnt in zip(bbox, bbox2, segmnt): # w = subbox2[2] # h = subbox2[3] # area = w*h # flattened_segmnt = [item for sublist in subsegmnt for item in sublist] # obj = {"image_id":img_id, "image_url": image_url, "id":box_id, "area":area, "category_id":cat_id, "bbox": subbox, "segment":flattened_segmnt} # Create an object for each image # box_id +=1 # solutions.append(obj) # img_id +=1 # return solutions # def send_results_to_api(data, result_url): # # Example function to send results to an API # headers = {"Content-Type": "application/json"} # response = requests.post(result_url, json=data, headers=headers) # if response.status_code == 200: # return response.json() # Return any response from the API if needed # else: # return {"error": f"Failed to send results to API: {response.status_code}"} # def process_images(params): # # Parse the JSON string into a dictionary # params = json.loads(params) # image_urls = params.get("image_urls", []) # api = params.get("api", "") # job_id = params.get("job_id", "") # images = [Image.open(requests.get(url, stream=True).raw) for url in image_urls] # images from URLs # all_bboxes, all_bboxes2, all_segments = detect_objects(images) # Perform object detection # solutions = create_solutions(image_urls, all_bboxes, all_bboxes2, all_segments) # Create solutions with image URLs and bounding boxes # result_url = f"{api}/{job_id}" # # send_results_to_api(solutions, result_url) # return json.dumps({"solutions": solutions}, indent=4) # inputt = gr.Textbox(label="Parameters (JSON format)") # outputs = gr.JSON() # application = gr.Interface(fn=process_images, inputs=inputt, outputs=outputs, title="Multiple Object Segmentation with API Integration") # application.launch()