import json import os import time from datetime import timedelta def main(): """ 여러 COCO 형식의 JSON 파일을 하나의 COCO 형식 JSON 파일로 합치는 스크립트입니다. 이 스크립트는 ID 충돌을 방지하고 카테고리를 일관되게 유지합니다. 사용법: 1. annotations_dir 변수에 annotation JSON 파일들이 있는 디렉토리 경로를 지정합니다. 2. output_json_path 변수에 합쳐진 COCO 형식 JSON 파일을 저장할 경로를 지정합니다. 3. 스크립트를 실행합니다. 주의사항: - 각 JSON 파일 내의 'images'와 'annotations' 리스트가 존재해야 합니다. - 카테고리는 모든 JSON 파일에서 동일해야 합니다. 그렇지 않으면 스크립트가 오류를 발생시킵니다. """ start_time = time.time() # 스크립트 시작 시간 기록 # 1. COCO 데이터 구조 초기화 merged_coco = { "info": {}, "licenses": [], "categories": [], "images": [], "annotations": [] } # 카테고리 이름을 ID로 매핑하기 위한 딕셔너리 category_name_to_id = {} categories_initialized = False # 이미지 ID 매핑 next_image_id = 1 # 이미지 ID는 1부터 시작 # 어노테이션 ID 매핑 next_annotation_id = 1 # 어노테이션 ID는 1부터 시작 # 어노테이션 JSON 파일들이 있는 디렉토리 경로 annotations_dir = '/DATA2/ltb/si/it1_worker_helmet/releases/it1-worker-helmet-250326-4class-background/annotations' # 합쳐진 COCO JSON 파일을 저장할 경로 output_json_path = '/DATA2/ltb/mmyolo/it1_250326_1819_4class/it1_250326_1819_4class.json' # 디렉토리가 존재하는지 확인 if not os.path.isdir(annotations_dir): print(f"Error: 지정된 어노테이션 디렉토리 '{annotations_dir}'가 존재하지 않습니다.") return # 디렉토리 내의 JSON 파일 목록 가져오기 json_files = [f for f in os.listdir(annotations_dir) if f.endswith('.json')] total_files = len(json_files) if total_files == 0: print(f"Error: 디렉토리 '{annotations_dir}'에 JSON 파일이 존재하지 않습니다.") return print(f"총 {total_files}개의 JSON 파일을 처리합니다.") for idx, json_file in enumerate(json_files): json_path = os.path.join(annotations_dir, json_file) try: with open(json_path, 'r', encoding='utf-8') as f: data = json.load(f) except json.JSONDecodeError as e: print(f"Warning: '{json_file}' 파일을 JSON으로 파싱하는 중 오류 발생: {e}. 이 파일은 건너뜁니다.") continue except Exception as e: print(f"Warning: '{json_file}' 파일을 여는 중 오류 발생: {e}. 이 파일은 건너뜁니다.") continue # 첫 번째 파일의 'info', 'licenses', 'categories'를 설정 if not categories_initialized: merged_coco['info'] = data.get('info', {}) merged_coco['licenses'] = data.get('licenses', []) # 카테고리 초기화 categories = data.get('categories', []) for cat in categories: cat_id = cat.get('id') cat_name = cat.get('name') if cat_id is None or cat_name is None: print(f"Warning: '{json_file}' 파일의 카테고리 정보가 불완전합니다. 이 파일은 건너뜁니다.") break merged_coco['categories'].append(cat) category_name_to_id[cat_name] = cat_id categories_initialized = True else: # 이후 파일의 카테고리가 첫 파일과 동일한지 확인 categories = data.get('categories', []) for cat in categories: cat_name = cat.get('name') cat_id = cat.get('id') if cat_name not in category_name_to_id: print(f"Warning: '{json_file}' 파일에 새로운 카테고리 '{cat_name}'이 발견되었습니다. 이 파일은 건너뜁니다.") break elif category_name_to_id[cat_name] != cat_id: print(f"Warning: '{json_file}' 파일의 카테고리 ID가 기존과 다릅니다. 이 파일은 건너뜁니다.") break else: pass # 모든 카테고리가 일치 # Continue processing # If break was not called, else executes # So continue # No action needed # If break was called, skip this file # So, to skip, need to use a flag # Use a flag to determine if we should continue processing should_continue = True for cat in categories: cat_name = cat.get('name') cat_id = cat.get('id') if cat_name not in category_name_to_id or category_name_to_id[cat_name] != cat_id: should_continue = False break if not should_continue: print(f"Warning: '{json_file}' 파일의 카테고리가 기존과 일치하지 않습니다. 이 파일은 건너뜁니다.") continue # 이미지 리스트 가져오기 images = data.get('images', []) if not images: print(f"Warning: '{json_file}' 파일에 'images' 리스트가 비어 있습니다. 이 파일은 건너뜁니다.") continue # 어노테이션 리스트 가져오기 annotations = data.get('annotations', []) if not annotations: print(f"Warning: '{json_file}' 파일에 'annotations' 리스트가 비어 있습니다. 이 파일은 건너뜁니다.") # 파일이 유효할 수도 있으니 계속 진행 # continue # 이미지 ID 매핑: 원본 이미지 ID -> 새로운 이미지 ID original_image_id_to_new = {} for img in images: original_image_id = img.get('id') if original_image_id is None: print(f"Warning: '{json_file}' 파일의 이미지에 'id'가 없습니다. 이 이미지는 건너뜁니다.") continue # 새로운 이미지 ID 할당 new_image_id = next_image_id next_image_id += 1 # 업데이트된 이미지 딕셔너리 new_image = { "id": new_image_id, "license": img.get('license', 1), # 기본값 1 "file_name": img.get('file_name'), "width": img.get('width'), "height": img.get('height'), "date_captured": img.get('date_captured', "") } merged_coco['images'].append(new_image) # 매핑 저장 original_image_id_to_new[original_image_id] = new_image_id # 어노테이션 처리 for ann in annotations: original_image_id = ann.get('image_id') if original_image_id is None: print(f"Warning: '{json_file}' 파일의 어노테이션에 'image_id'가 없습니다. 이 어노테이션은 건너뜁니다.") continue if original_image_id not in original_image_id_to_new: print(f"Warning: '{json_file}' 파일의 어노테이션이 참조하는 이미지 ID '{original_image_id}'가 존재하지 않습니다. 이 어노테이션은 건너뜁니다.") continue new_image_id = original_image_id_to_new[original_image_id] category_id = ann.get('category_id') if category_id is None: print(f"Warning: '{json_file}' 파일의 어노테이션에 'category_id'가 없습니다. 이 어노테이션은 건너뜁니다.") continue # 어노테이션의 카테고리 ID가 기존과 일치하는지 확인 # 카테고리가 이미 매핑되어 있으므로, 그대로 사용 if category_id not in [cat['id'] for cat in merged_coco['categories']]: print(f"Warning: '{json_file}' 파일의 어노테이션에 존재하지 않는 'category_id' '{category_id}'가 있습니다. 이 어노테이션은 건너뜁니다.") continue bbox = ann.get('bbox') if not bbox or len(bbox) != 4: print(f"Warning: '{json_file}' 파일의 어노테이션에 'bbox'가 없거나 형식이 잘못되었습니다. 이 어노테이션은 건너뜁니다.") continue area = ann.get('area') if area is None: # bbox를 이용해 계산 x, y, w, h = bbox area = w * h segmentation = ann.get('segmentation', []) iscrowd = ann.get('iscrowd', 0) new_annotation = { "id": next_annotation_id, "image_id": new_image_id, "category_id": category_id, "bbox": bbox, "area": area, "segmentation": segmentation, "iscrowd": iscrowd } merged_coco['annotations'].append(new_annotation) next_annotation_id += 1 # 진행 상황 출력 (퍼센테이지) progress = ((idx + 1) / total_files) * 100 print(f"진행 상황: {progress:.2f}% ({idx + 1}/{total_files} 파일 처리 완료)", end='\r') print("\n모든 파일 처리가 완료되었습니다.") # 카테고리 중복 여부 확인 및 출력 if merged_coco['categories']: print("카테고리 목록:") for cat in merged_coco['categories']: print(f"- ID: {cat['id']}, Name: {cat['name']}") else: print("카테고리가 존재하지 않습니다.") # 합쳐진 COCO 형식 JSON 파일 저장 try: with open(output_json_path, 'w', encoding='utf-8') as f: json.dump(merged_coco, f, ensure_ascii=False, indent=2) print(f"합쳐진 COCO JSON 파일이 성공적으로 저장되었습니다: {output_json_path}") except Exception as e: print(f"Error: COCO JSON 파일을 저장하는 중 오류 발생: {e}") return # 총 소요 시간 출력 elapsed_time = time.time() - start_time print(f"작업 소요 시간: {str(timedelta(seconds=int(elapsed_time)))}") if __name__ == "__main__": main()