게시글 본문내용
|
|
다음검색
> RoI Pooling 수행 > Region of Interest Pooling
> 제안된 영역에 대해 classification(분류)와 regression(회귀) 수행
> train.py의 에폭 도는 부분 수정 > 클래스 추가
print("Start training") start_time = time.time() lp = LivePlot() # 객체 생성 for epoch in range(args.start_epoch, args.epochs): if args.distributed: train_sampler.set_epoch(epoch) train_loss, train_acc = train_one_epoch(model, optimizer, data_loader, device, epoch, args.print_freq, scaler) lr_scheduler.step() if args.output_dir: checkpoint = { "model": model_without_ddp.state_dict(), "optimizer": optimizer.state_dict(), "lr_scheduler": lr_scheduler.state_dict(), "args": args, "epoch": epoch, } if args.amp: checkpoint["scaler"] = scaler.state_dict() utils.save_on_master(checkpoint, os.path.join(args.output_dir, f"model_{epoch}.pth")) utils.save_on_master(checkpoint, os.path.join(args.output_dir, "checkpoint.pth")) # added by 2sungryul, save the full model with its structure and weight torch.save(model_without_ddp, os.path.join(args.output_dir, "model.pth")) # evaluate after every epoch val_loss, val_acc = evaluate(model, data_loader_test, device=device) lp.update_all(epoch, train_loss, val_loss, train_acc, val_acc) lp.show() total_time = time.time() - start_time total_time_str = str(datetime.timedelta(seconds=int(total_time))) print(f"Training time {total_time_str}") | cs |
> engine.py 함수 수정
def train_one_epoch(model, optimizer, data_loader, device, epoch, print_freq, scaler=None): model.train() # 모델을 학습 모드로 설정 metric_logger = utils.MetricLogger(delimiter=" ") header = f"Epoch: [{epoch}]" running_loss = 0.0 total_samples = 0 coco_evaluator = CocoEvaluator(get_coco_api_from_dataset(data_loader.dataset), iou_types=["bbox"]) for images, targets in metric_logger.log_every(data_loader, print_freq, header): images = list(image.to(device) for image in images) targets = [{k: (v.to(device) if isinstance(v, torch.Tensor) else v) for k, v in t.items()} for t in targets] # 모델에 이미지와 타겟을 전달하여 손실 계산 (학습 모드) with torch.autocast(device_type='cpu',enabled=scaler is not None): loss_dict = model(images, targets) losses = sum(loss for loss in loss_dict.values()) optimizer.zero_grad() if scaler is not None: scaler.scale(losses).backward() scaler.step(optimizer) scaler.update() else: losses.backward() optimizer.step() batch_size = len(images) running_loss += losses.item() * batch_size total_samples += batch_size model.eval() outputs = model(images) model.train() res = {} for i, output in enumerate(outputs): image_id = targets[i]["image_id"] # image_id가 문자열이면 정수로 변환 if isinstance(image_id, torch.Tensor): image_id = image_id.item() elif isinstance(image_id, str): try: image_id = int(image_id) # 문자열이면 정수로 변환 except ValueError: raise ValueError(f"Invalid image_id: {image_id}. Expected int or convertible string.") res[image_id] = { "boxes": output["boxes"].detach().cpu(), "scores": output["scores"].detach().cpu(), "labels": output["labels"].detach().cpu() } coco_evaluator.update(res) metric_logger.update(loss=losses.item()) # 에포크별 손실 계산 epoch_loss = running_loss / total_samples # COCO evaluator 처리 coco_evaluator.synchronize_between_processes() coco_evaluator.accumulate() coco_evaluator.summarize() # mAP 계산 coco_stats = coco_evaluator.coco_eval['bbox'].stats train_mAP = coco_stats[0] # mAP (IoU=0.50:0.95) print(f"epcoh_loss: {epoch_loss}") print(f"train_mAP: {train_mAP}") return epoch_loss, train_mAP # 훈련 손실 및 mAP 반환 def _get_iou_types(model): model_without_ddp = model if isinstance(model, torch.nn.parallel.DistributedDataParallel): model_without_ddp = model.module iou_types = ["bbox"] if isinstance(model_without_ddp, torchvision.models.detection.MaskRCNN): iou_types.append("segm") if isinstance(model_without_ddp, torchvision.models.detection.KeypointRCNN): iou_types.append("keypoints") return iou_types @torch.inference_mode() def evaluate(model, data_loader_test, device): model.eval() coco = get_coco_api_from_dataset(data_loader_test.dataset) coco_evaluator = CocoEvaluator(coco, iou_types=["bbox"]) running_loss = 0.0 total_samples = 0 with torch.no_grad(): for images, targets in data_loader_test: images = list(image.to(device) for image in images) targets = [{k: (v.to(device) if isinstance(v, torch.Tensor) else v) for k, v in t.items()} for t in targets] model.train() loss_dict = model(images, targets) losses = sum(loss for loss in loss_dict.values()) model.eval() batch_size = len(images) running_loss += losses.item() * batch_size total_samples += batch_size # 모델의 예측을 coco evaluator로 업데이트 outputs = model(images) #res = {target["image_id"].item(): output for target, output in zip(targets, outputs)} res = {} for i, output in enumerate(outputs): image_id = targets[i]["image_id"] # image_id가 문자열이면 정수로 변환 if isinstance(image_id, torch.Tensor): image_id = image_id.item() elif isinstance(image_id, str): try: image_id = int(image_id) # 문자열이면 정수로 변환 except ValueError: raise ValueError(f"Invalid image_id: {image_id}. Expected int or convertible string.") res[image_id] = { "boxes": output["boxes"].detach().cpu(), "scores": output["scores"].detach().cpu(), "labels": output["labels"].detach().cpu() } coco_evaluator.update(res) epoch_loss = running_loss / total_samples coco_evaluator.synchronize_between_processes() coco_evaluator.accumulate() coco_evaluator.summarize() # mAP 값 추출 coco_stats = coco_evaluator.coco_eval['bbox'].stats val_mAP = coco_stats[0] # mAP (IoU=0.50:0.95) print(f"epcoh_loss: {epoch_loss}") print(f"val_mAP: {val_mAP}") return epoch_loss, val_mAP # 검증 손실 및 mAP 반환 | cs |
from collections import defaultdict import torch import transforms as reference_transforms def get_modules(use_v2): # We need a protected import to avoid the V2 warning in case just V1 is used if use_v2: import torchvision.transforms.v2 import torchvision.tv_tensors return torchvision.transforms.v2, torchvision.tv_tensors else: return reference_transforms, None class DetectionPresetTrain: # Note: this transform assumes that the input to forward() are always PIL # images, regardless of the backend parameter. def __init__( self, *, data_augmentation, hflip_prob=0.5, mean=(123.0, 117.0, 104.0), backend="pil", use_v2=False, ): T, tv_tensors = get_modules(use_v2) transforms = [] backend = backend.lower() if backend == "tv_tensor": transforms.append(T.ToImage()) elif backend == "tensor": transforms.append(T.PILToTensor()) elif backend != "pil": raise ValueError(f"backend can be 'tv_tensor', 'tensor' or 'pil', but got {backend}") if data_augmentation == "hflip": transforms += [T.RandomHorizontalFlip(p=hflip_prob)] elif data_augmentation == "lsj": transforms += [ T.ScaleJitter(target_size=(1024, 1024), antialias=True), # TODO: FixedSizeCrop below doesn't work on tensors! reference_transforms.FixedSizeCrop(size=(1024, 1024), fill=mean), T.RandomHorizontalFlip(p=hflip_prob), ] elif data_augmentation == "multiscale": transforms += [ T.RandomShortestSize(min_size=(480, 512, 544, 576, 608, 640, 672, 704, 736, 768, 800), max_size=1333), T.RandomHorizontalFlip(p=hflip_prob), ] elif data_augmentation == "ssd": fill = defaultdict(lambda: mean, {tv_tensors.Mask: 0}) if use_v2 else list(mean) transforms += [ T.RandomPhotometricDistort(), T.RandomZoomOut(fill=fill), T.RandomIoUCrop(), T.RandomHorizontalFlip(p=hflip_prob), ] elif data_augmentation == "ssdlite": transforms += [ T.RandomIoUCrop(), T.RandomHorizontalFlip(p=hflip_prob), ] else: raise ValueError(f'Unknown data augmentation policy "{data_augmentation}"') if backend == "pil": # Note: we could just convert to pure tensors even in v2. transforms += [T.ToImage() if use_v2 else T.PILToTensor()] transforms += [T.ToDtype(torch.float, scale=True)] if use_v2: transforms += [ T.ConvertBoundingBoxFormat(tv_tensors.BoundingBoxFormat.XYXY), T.SanitizeBoundingBoxes(), T.ToPureTensor(), ] self.transforms = T.Compose(transforms) def __call__(self, img, target): return self.transforms(img, target) class DetectionPresetEval: def __init__(self, backend="pil", use_v2=False): T, _ = get_modules(use_v2) transforms = [] backend = backend.lower() if backend == "pil": # Note: we could just convert to pure tensors even in v2? transforms += [T.ToImage() if use_v2 else T.PILToTensor()] elif backend == "tensor": transforms += [T.PILToTensor()] elif backend == "tv_tensor": transforms += [T.ToImage()] else: raise ValueError(f"backend can be 'tv_tensor', 'tensor' or 'pil', but got {backend}") transforms += [T.ToDtype(torch.float, scale=True)] if use_v2: transforms += [T.ToPureTensor()] self.transforms = T.Compose(transforms) def __call__(self, img, target): return self.transforms(img, target) | cs |
사용한 증식 목록
from ultralytics import YOLO import torch import time # 시간 측정을 위한 모듈 추가 def train_yolo(data_yaml, epochs=50, imgsz=640, save_path="/home/jetson/work/Pytorch/dataset2/output_yolo/yolo_custom.pt"): """ YOLO 모델 학습 및 저장 """ model = YOLO("yolov8n.pt") # YOLOv8n 모델 로드 start_time = time.time() # 학습 시작 시간 기록 model.train(data=data_yaml, epochs=epochs, imgsz=imgsz) end_time = time.time() # 학습 종료 시간 기록 total_time = end_time - start_time # 훈련 시간 계산 print(f"Total training time: {total_time:.2f} seconds") # 훈련 시간 출력 model.save(save_path) # 학습된 모델 저장 print(f"Model saved at: {save_path}") return save_path def main(): device = 'cuda' if torch.cuda.is_available() else 'cpu' print(f"Using device: {device}") # YOLO 학습 데이터 설정 data_yaml = "/home/jetson/work/Pytorch/vision/references/detection/data.yaml" # YOLO 커스텀 학습 진행 trained_model = train_yolo(data_yaml) if __name__ == "__main__": main() | cs |
import os import json import shutil # 클래스 정의 (labels.txt 기반) class_names = ['big robot', 'small robot'] # JSON에서 YOLO TXT로 변환하는 함수 def convert_json_to_yolo(json_path, output_txt_path, img_width, img_height): with open(json_path, 'r') as f: data = json.load(f) if img_width == 0 or img_height == 0: print(f"Warning: Image size not found in {json_path}") return with open(output_txt_path, 'w') as f: for shape in data.get('shapes', []): label = shape.get('label') if label not in class_names: continue # 정의되지 않은 클래스는 무시 class_id = class_names.index(label) # bounding box 좌표 (예: [[x1, y1], [x2, y2]]) points = shape.get('points', []) if len(points) != 2: continue x1, y1 = points[0] x2, y2 = points[1] # YOLO 형식으로 변환 (중심 좌표와 너비/높이 정규화) x_center = (x1 + x2) / 2 / img_width y_center = (y1 + y2) / 2 / img_height width = abs(x2 - x1) / img_width height = abs(y2 - y1) / img_height # YOLO TXT 형식으로 기록 f.write(f"{class_id} {x_center:.6f} {y_center:.6f} {width:.6f} {height:.6f}\n") # 디렉토리 처리 및 파일 이동 함수 def process_directory(input_dir, output_img_dir, output_label_dir): if not os.path.exists(output_img_dir): os.makedirs(output_img_dir) if not os.path.exists(output_label_dir): os.makedirs(output_label_dir) for file in os.listdir(input_dir): if file.endswith('.json'): json_path = os.path.join(input_dir, file) with open(json_path, 'r') as f: data = json.load(f) img_width = data.get('imageWidth', 0) img_height = data.get('imageHeight', 0) # 이미지 파일 이름 추출 img_name = os.path.splitext(file)[0] + '.jpg' img_path = os.path.join(input_dir, img_name) # TXT 파일 이름 txt_name = os.path.splitext(file)[0] + '.txt' txt_path = os.path.join(output_label_dir, txt_name) # 이미지 파일이 존재하면 이동 if os.path.exists(img_path): shutil.move(img_path, os.path.join(output_img_dir, img_name)) print(f"Moved image: {img_name} to {output_img_dir}") else: print(f"Warning: Image {img_name} not found in {input_dir}") continue # JSON을 TXT로 변환 convert_json_to_yolo(json_path, txt_path, img_width, img_height) print(f"Converted: {file} to {txt_name}") # 실행 base_dir = '/home/jetson/work/Pytorch/dataset2' output_base_dir = os.path.join(base_dir, 'dataset') image_base_dir = os.path.join(output_base_dir, 'images') label_base_dir = os.path.join(output_base_dir, 'labels') for folder in ['train', 'val']: input_dir = os.path.join(base_dir, folder) output_img_dir = os.path.join(image_base_dir, folder) output_label_dir = os.path.join(label_base_dir, folder) print(f"Processing {folder}...") process_directory(input_dir, output_img_dir, output_label_dir) print("Dataset restructuring and conversion completed!") | cs |
# 추가 작성
백본 가중치 고정 후 훈련
freeze로 가중치 고정을 할 수 있음
앞서 사용한 YOLOv8의 구조
백본 층이 0-9층인 걸로 보아 총 10개가 모델의 앞 구조에 있음을 확인
from ultralytics import YOLO import torch import time def train_yolo(data_yaml, epochs=50, imgsz=640, freeze_layers=10, save_path="/home/jetson/work/Pytorch/dataset2/output_yolo/yolo_custom.pt"): """ YOLO 모델 학습 및 저장 (백본 가중치 고정) """ model = YOLO("yolov8n.pt") # YOLOv8n 모델 로드 print(model.model) # 모델 구조 출력 start_time = time.time() # 학습 시작 시간 기록 model.train(data=data_yaml, epochs=epochs, imgsz=imgsz, freeze=freeze_layers, device="cuda") end_time = time.time() # 학습 종료 시간 기록 total_time = end_time - start_time # 훈련 시간 계산 print(f"Total training time: {total_time:.2f} seconds") # 훈련 시간 출력 model.save(save_path) # 학습된 모델 저장 print(f"Model saved at: {save_path}") return save_path def main(): device = 'cuda' if torch.cuda.is_available() else 'cpu' print(f"Using device: {device}") # YOLO 학습 데이터 설정 data_yaml = "/home/jetson/work/Pytorch/vision/references/detection/data.yaml" # YOLO 커스텀 학습 진행 (백본 고정) trained_model = train_yolo(data_yaml, freeze_layers=10) if __name__ == "__main__": main() | cs |
freeze할 층을 10개로 잡아 가중치 고정시켜서 학습
훈련 시작 시 아래 사진과 같이 Freezing layer 문구가 뜨면 됨
결과
첫댓글 Yolo훈련시 백본의 가중치는 고정하고 나머지만 훈련한건요?
로스,정확도 그래프 추가할것, Yolo에서 저장해준는 것도 첨부할것
훈련검증에 사용하지 않은 테스트 영상 10장 이상으로 테스트 결과 추가할 것