| 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): | |
| 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] | |
| ans = {"image_id": img_id, "id": box_id, "area": area, "category_id": cat_id, "bbox": subbox, "segment": flattened_segmnt} | |
| obj ={"url": image_url, "answer":[ans]} | |
| solutions.append(obj) | |
| box_id += 1 | |
| img_id += 1 | |
| 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", []) | |
| # 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) | |
| # 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() | |