이안류 CCTV 데이터셋

1. 이안류 CCTV 데이터셋

​AI Hub의 '이안류 CCTV 데이터'는 우리나라 주요 해수욕장(해운대, 송정, 대천, 중문, 낙산)에서 이안류 발생 여부와 위치를 모니터링하기 위해 구축된 인공지능 학습용 데이터셋입니다. 해수욕장 주변에 설치된 CCTV 영상을 이미지로 변환하여, 이안류 발생 여부와 위치를 가시화하는 모델 개발에 활용할 수 있습니다. 이 데이터셋은 이안류 탐지 및 예측 시스템 개발에 필수적인 자료를 제공하며, 해수욕객의 안전을 위한 응용 서비스 구성에 활용될 수 있습니다. ​이안류는 해안에서 먼 바다로 빠르게 이동하는 폭이 좁은 바닷물의 흐름으로, 기상 상태가 양호한 경우에도 나타나며, 얕은 곳에 있던 해수욕객을 순식간에 수심이 깊은 먼 바다로 이동시켜 인명사고를 유발할 수 있습니다. 따라서 이러한 데이터셋은 해수욕장 안전 관리 및 이안류 예측 모델 개발에 중요한 역할을 합니다.

 

아래 AIHub에서 이안류 CCTV 데이터셋을 다운로드 받습니다.

https://www.aihub.or.kr/aihubdata/data/view.do?currMenu=115&topMenu=100&aihubDataSe=data&dataSetSn=71297

AI-Hub

 

!pip install ultralytics opencv-python

 

import os
import random
import shutil
import cv2
import glob
import json
import yaml
import ultralytics
import matplotlib.pyplot as plt
from torchvision import transforms
from tqdm import tqdm
from ultralytics import YOLO
from ultralytics.utils.plotting import Annotator

 

ultralytics.checks()

 

data_root = '/content/drive/MyDrive/Colab Notebooks'
file_root = f"{data_root}/{pjt_name}/data"

pjt_name = 'ripcurrent'

train_root = f"{data_root}/{pjt_name}/train"
valid_root = f"{data_root}/{pjt_name}/valid"
test_root = f"{data_root}/{pjt_name}/test"

for folder in [train_root, valid_root, test_root]:
    if not os.path.exists(folder):
        os.makedirs(folder)
    for s in ['images', 'labels']:
        s_folder = f"{folder}/{s}"
        if not os.path.exists(s_folder):
            os.makedirs(s_folder)

 

file_list = glob.glob(f"{file_root}/labels/*.json")

 

# Bounding box 형태 변형하기
def json_to_yolo_bbox(bbox, w, h):
    # xcenter = xmin, xmax /2
    # ycenter = ymin, ymax /2
    x_center = ((bbox[0][0]+ bbox[1][0])/2)/w
    y_center = ((bbox[0][1]+ bbox[3][1])/2)/h
    width = (bbox[1][0]- bbox[0][0])/w
    height = (bbox[3][1]- bbox[0][1])/h

    return [x_center, y_center, width, height]

 

# Json -> Txt
file = file_list[3]
print(file)
result = set()
with open(file, 'r') as f:
    json_data = json.load(f)
    width, height = list(map(int, json_data["image_info"]["resolution"].split(',')))
    cls = 0

    if json_data["annotations"].get("drawing"):
        for b in json_data["annotations"]["drawing"]:
            yolo_bbox = json_to_yolo_bbox(b, width, height)
            bbox_string = " ".join([str(x) for x in yolo_bbox])
            result.add(f"{cls} {bbox_string}")

    result = list(result)
    if result:
        with open(file.replace('json', 'txt'), 'w', encoding = 'utf-8') as t:
            t.write("\n".join(result))
            print(file)

 

for file in tqdm(file_list):
    result = set()
    with open(file, 'r') as f:
        json_data = json.load(f)
        width, height = list(map(int, json_data["image_info"]["resolution"].split(',')))
        cls = 0
        num_b = json_data["annotations"]["bounding_count"]
        if num_b > 0:
            for b in json_data["annotations"]["drawing"]:
                yolo_bbox = json_to_yolo_bbox(b, width, height)
                bbox_string = " ".join([str(x) for x in yolo_bbox])
                result.add(f"{cls} {bbox_string}")
            result = list(result)
            if result:
                with open(file.replace('json', 'txt'), "w", encoding="utf-8") as f:
                    f.write("\n".join(result))

 

random.seed(2025)
file_list = glob.glob(f"{file_root}/labels/*.txt")

random.shuffle(file_list)
test_ratio = 0.1
num_file = len(file_list)

test_list = file_list[:int(num_file*test_ratio)]
valid_list = file_list[int(num_file*test_ratio):int(num_file*test_ratio)*2]
train_list = file_list[int(num_file*test_ratio)*2:]

 

# 라벨데이터 복사하기 & 대응하는 이미지 데이터 복사하기
for i in tqdm(test_list):
    txt_name = i.split('/')[-1]
    shutil.copyfile(i, f"{test_root}/labels/{txt_name}")
    img_path = i.replace('labels', 'images').replace('TL', 'TS').replace('JSON', '이미지').replace('txt', 'jpg')
    img_name = img_path.split('/')[-1]
    shutil.copyfile(img_path, f"{test_root}/images/{img_name}")

for i in tqdm(valid_list):
    txt_name = i.split('/')[-1]
    shutil.copyfile(i, f"{valid_root}/labels/{txt_name}")
    img_path = i.replace('labels', 'images').replace('TL','TS').replace('JSON', '이미지').replace('txt', 'jpg')
    jpg_name = img_path.split('/')[-1]
    shutil.copyfile(img_path, f"{valid_root}/images/{jpg_name}")

for i in tqdm(train_list):
    txt_name = i.split('/')[-1]
    shutil.copyfile(i, f"{train_root}/labels/{txt_name}")
    img_path = i.replace('labels', 'images').replace('TL','TS').replace('JSON', '이미지').replace('txt', 'jpg')
    jpg_name = img_path.split('/')[-1]
    shutil.copyfile(img_path, f"{train_root}/images/{jpg_name}")

 

pjt_root = '/content/drive/MyDrive/Colab Notebooks/ripcurrent'

 

# Config 파일 만들기
import yaml
data = dict()

data['train'] = train_root
data['val'] = valid_root
data['test'] = test_root
data['nc'] = 1
data['names'] = ['yes']

with open(f'{pjt_root}/rip.yaml', 'w') as f:
    yaml.dump(data, f)

 

%cd /content/drive/MyDrive/Colab Notebooks/ripcurrent

 

# 모델 불러오기 & 학습 Setting하기
from ultralytics import YOLO
model = YOLO('yolov8s.pt')
results= model.train(data = 'rip.yaml', epochs = 10, batch = 8, imgsz = 224, device = 0, workers = 1, amp= False, patience = 30, name = 'rip_s')

 

result_folder = f'{pjt_root}/runs/detect/rip_s'

 

model = YOLO(f'{result_folder}/weights/best.pt')

metrics = model.val(split = 'test')

print(metrics.box.map)
print(metrics.box.map50)

 

test_root = f"{data_root}/{pjt_name}/test"

test_file_list = glob.glob(f"{test_root}/images/*")
random.shuffle(test_file_list)

 

# 모델 불러오기
model = YOLO(f'{result_folder}/weights/best.pt')

test_data_transform = transforms.Compose([transforms.ToTensor()])

color_dict = [(0, 0, 255)]

 

# Inference 진행하기
test_img = cv2.imread(test_file_list[0])
img_src = cv2.cvtColor(test_img, cv2.COLOR_BGR2RGB)
results = model(img_src)

 

print(results[0].boxes)

 

test_img = cv2.imread(test_file_list[0])
img_src = cv2.cvtColor(test_img, cv2.COLOR_BGR2RGB)
results = model(img_src)[0]

annotator = Annotator(img_src)
boxes = results.boxes
for box in boxes:
    b = box.xyxy[0]
    cls = box.cls
    annotator.box_label(b, model.names[int(cls)], color_dict[int(cls)])
img_src = annotator.result()
plt.imshow(img_src)
plt.show()

 

plt.figure(figsize = (12,6))
for idx in range(6):
    test_img = cv2.imread(test_file_list[idx])
    img_src = cv2.cvtColor(test_img, cv2.COLOR_BGR2RGB)
    results = model(test_img)

    for result in results:
        annotator = Annotator(img_src)
        boxes = result.boxes
        for box in boxes:
            b = box.xyxy[0]  # get box coordinates in (top, left, bottom, right) format
            cls = box.cls
            annotator.box_label(b, model.names[int(cls)], color_dict[int(cls)])
    img_src = annotator.result()

    plt.subplot(2, 3, (idx+1))
    plt.imshow(img_src)
plt.show()