숫자의 각 자리수를 더하여 반환하는 문제입니다.

#include <iostream>

using namespace std;
int solution(int n)
{
    int answer = 0;

    while(n > 0){
        answer += (n % 10);
        n /= 10;
    }

    cout << "Hello Cpp" << endl;

    return answer;
}

다른 사람 풀이로는 to_string 사용하여 각 위치의 값을 더하는 방식을 사용하였는데 이 과정에서s[i] - '0'으로 아스키 코드 값을 제거해서 숫자 계산하였습니다.

#include <iostream>
#include <string>

using namespace std;
int solution(int n)
{
    int answer = 0;

    string s = to_string(n);

    for(int i = 0; i < s.size(); i++) answer += (s[i] - '0');

    // [실행] 버튼을 누르면 출력 값을 볼 수 있습니다.
    cout << "Hello Cpp" << endl;

    return answer;
}

올바른 괄호

괄호의 특성을 확인하여 열린 괄호와 닫힌 괄호가 정상적으로 이루어졌는지 파악하는 코드입니다.

#include<string>
#include <iostream>

using namespace std;

bool solution(string s)
{
    bool answer = true;
    int left_open = 0;
    int right_close = 0;
    for(int i = 0; i < s.size(); i++){
        if(s.at(i) == '(') left_open++;
        else if(s.at(i) == ')') right_close++;
        if (right_close > left_open) {answer = false; break;}
    }
    if(right_close != left_open) answer = false;

    cout << "Hello Cpp" << endl;

    return answer;
}

각 괄호의 수에 대한 변수를 계산하여 True, False를 확인했지만 하나의 변수에 대한 증감식을 사용하여 계산하는것이 더욱 깔끔하다고 생각됩니다.(단 ')'가 먼저 들어오는 경우에 대한 예외조건이 필요합니다.)

짝 지어 제거하기

연속된 문자가 나오는 경우에 제거하여 문자열이 남아있는 경우 0 문자열이 모두 사라진 경우 1을 반환하는 코드입니다.

#include <iostream>
#include <string>
using namespace std;

int solution(string s)
{
    int answer = -1;

    for(int j = 0; j < s.size(); ){
        if(s.at(j) == s[j+1]) {
            s.erase(j, 2); 
            j = j-2;
            if(j = s.size()) j = 0;
        }
        else j++;
    }
    if(s.size() == 0) answer = 1;
    else answer = 0;

    cout << s << endl;

    return answer;
}

초기 풀이 코드로 string의 위치를 찾아가 현재 위치와 다음 위치가 같다면 삭제를 진행하는 방식을 사용하였습니다 j+1위치의 경우 string객체의 범위를 벗어나게 되어 at() 함수 대신 배열 접근 방법을 사용하였습니다.

#include <iostream>
#include <stack>
#include <string>

using namespace std;

int solution(string s)
{
    int answer = -1;

    stack<char> cw;
    char tmp;
    for(int j = 0; j < s.size(); j++){
        if(!cw.empty()) tmp = cw.top();
        cw.push(s.at(j));
        if(tmp == cw.top()){
            cw.pop();
            cw.pop();
            tmp = ' ';
        }
    }
    if (cw.empty()) answer = 1;
    else answer = 0;

    return answer;
}

다른 사람들의 풀이를 보고 stack을 사용한 방식입니다. 인덱스로 접근하는 방식이 아니라 인덱스 관련 문제가 존재하지 않습니다.

피보나치 수열

피보나치 수열 문제입니다. 마지막 값에 대해서 1234567로 나머지를 구한 값을 반환합니다.

#include <string>
#include <vector>

using namespace std;
int fibonacci(int n);

int fib[100002];

int solution(int n) {
    int answer = 0;

    answer = fibonacci(n);

    return answer;
}

int fibonacci(int n){
    if(n<=1) {
        fib[n] = n;
        return n;
    }
    else if(fib[n] != 0) return fib[n];
    else {
        fib[n] = (fibonacci(n-1) + fibonacci(n-2)) % 1234567;
        return fib[n];
    }
}

n은 2 이상 100,000 이하인 자연수로 해당 방식에서 마지막 answer에 값을 할당해주는 과정에서 1234567의 값에 대한 나머지를 구하게 되는 경우 int의 범위를 벗어나게 되어 피보나치 함수 내부에서 나눠줌으로서 해당 문제를 해결하였습니다.

숫자의 표현

연속된 자연수의 합으로 입력된 값이 나오는 경우의 수를 출력하는 문제입니다.

#include <string>
#include <vector>

using namespace std;

int solution(int n) {
    int answer = 0;

    int start_num = 1;

    while(true){
        int sum = 0;
        int first_num = start_num;
        if(start_num == n || start_num > (n / 2 + 1)){
            answer++;
            break;
        }
        while(sum < n){
            sum += first_num++;
            if(sum == n) answer++;
        } start_num++;

    }
    return answer;
}

숫자가 시작되는 값에 대해 반복문을 수행하여 경우의 수를 반환하였습니다.

JadenCase 문자열 만들기

#include <string>
#include <vector>

using namespace std;

string solution(string s) {
    string answer = "";

    for(int i = 0; i < s.length(); i++){
        if(s[i - 1] == ' ' && s.at(i) >= 'a' && s.at(i) <= 'z') {
            answer += s.at(i) - ('a'- 'A');
        }
        else if(s[i - 1] != ' ' && s.at(i) >= 'A' && s.at(i) <= 'Z') {
            answer += s.at(i) + ('a'- 'A');
        }
        else if(s.at(0) >= '0' || s.at(0) <= '9') {answer += s.at(i);}
    }
    if(answer.at(0) >= 'a' && answer.at(0) <= 'z') answer.at(0) -= ('a'- 'A');

    return answer;
}

조건문 사용하여 각 조건에 대한 문제 풀이를 진행하였습니다. 아스키 코드로 각 문자에 대한 변환 실행했습니다. 이후 유저들 풀이에서 toupper와 tolower함수 사용하여 풀이한 것을 확인하였습니다.

연속 부분 수열 합의 개수

#include <string>
#include <vector>
#include <set>

using namespace std;

void get_sum(set<int> &sum, const vector<int>& vec, int len);

int solution(vector<int> elements) {
    int answer = 0;
    set<int> sum;

    for(int len = 1; len <= elements.size(); len++){
        get_sum(sum, elements, len);
    }

    answer = sum.size();
    return answer;
}

void get_sum(set<int> &sum, const vector<int>& vec, int len){
    int size = vec.size();

     for(int i = 0; i < size; i++){
        int tmp = 0;

         for(int j = 0; j < len; j++){
            int idx = (i + j) % size;
            tmp += vec[idx];
        }
        sum.insert(tmp);
    }
}

set : sum을 외부 함수로 넘겨 이중 for 문으로 각 벡터에 대해 연산하고 set에 추가하는 방식을 사용하였습니다. 하지만 for 내부에서 이중 for문인 get_sum을 불러와 시간에 대한 복잡도가 높습니다.

#include <string>
#include <vector>
#include <set>

using namespace std;

int solution(vector<int> elements) {
    set<int> S;

    int n = elements.size();

    for (int i = 0 ; i < n ; ++i) {
        int sum = 0;
        for (int j = i ; j < i + n ; ++j) {
            sum += elements[j % n];
            S.insert(sum);
        }
    }

    return S.size();
}

다른 풀이로는 이중 for문으로 계산한 방식으로 이중 for문을 하나의 for문으로 작성하여 시간 복잡도를 줄였습니다.

점프와 순간이동

K칸 이동은 에너지 1 사용, 현재 이동한 위치 x 2의 경우에는 에너지를 소비하지 않을 경우 최소한의 에너지를 사용하여 목적지 n까지 이동하는 문제입니다.

#include <iostream>
using namespace std;

int solution(int n)
{
    int ans = 0;

    while(n != 1){
        if(n % 2 == 1) {n -= 1; ans++;}
        else n /=2;
    } ans++;

    cout << "Hello Cpp" << endl;

    return ans;
}

시작에서 K칸 이동과 이동하는 위치를 계산하는 방법도 있지만 뒤에서 시작하면 더 간편히 써지지 않을까 하여 뒤에서부터 시작하였습니다.

#include <iostream>
using namespace std;

int solution(int n)
{
    int ans = 1;
    int len = 1;
    while(len != n){
        if(len * 2 <= n) len *= 2;
        else {len++; ans++;}
    } 

    cout << "Hello Cpp" << endl;

    return ans;
}

다음은 시작 부분에서 진행한 코드입니다. 다음 코드에서의 문제점은 ans 증가 이후 2배를 하는 것이 더 효율적인지 2배를 한 이후 ans를 증가시키는 것이 더 효율적인지에 대한 보장이 없다는 것입니다. 따라서 해당 코드의 경우에는 BFS 알고리즘을 사용해야 합니다.

#include <iostream>
#include <vector>
#include <queue>

using namespace std;

int solution(int n)
{
    // 현재 위치, 현재까지 쓴 배터리
    queue<pair<int, int>> q;

    q.push({0, 0});

    vector<int> visited(n + 1, 2100000000);
    visited[0] = 0;

    while(!q.empty()){
        int curr = q.front().first;
        int cost = q.front().second;
        q.pop();

        // 적은 배터리로 이곳에 온 적이 있다면, 지금 경로는 폐기
        if(visited[curr] < cost) continue;

        // 순간이동 (*2)
        if(curr * 2 <= n && visited[curr * 2] > cost) {
            visited[curr * 2] = cost;
            q.push({curr * 2, cost});
        }

        // 점프 (+1)
        if(curr + 1 <= n && visited[curr + 1] > cost + 1) {
            visited[curr + 1] = cost + 1;
            q.push({curr + 1, cost + 1});
        }
    }

    return visited[n];
}

DP도 생각해보았으나 DP는 이전 진행된 값이 고정되어야 하기에 쓰지 못하였습니다.

N개의 최소공배수

하나의 벡터에 있는 모든 원소들에 대한 최소 공배수 값을 구하는 문제입니다.

#include <string>
#include <vector>
#include <algorithm>

using namespace std;

int gcd(int a, int b);
int lcm(int a, int b);

int solution(vector<int> arr) {
    int answer = 0;

    for (int i = 0; i < arr.size() - 1; i++){
        arr.at(i + 1) = lcm(arr.at(i), arr.at(i + 1));
    } answer = arr.at(arr.size() -1);

    return answer;
}

int gcd(int a, int b)
{
    int c;
    while (b != 0)
    {
        c = a % b;
        a = b;
        b = c;
    }
    return a;
}
int lcm(int a, int b)
{
    return a * b / gcd(a, b);
}

최대공배수를 구하기 위한 최소공배수 함수 gcd를 생성하였습니다. 이전의 값의 최대 공배수를 새로운 비교값 원소로 설정하였습니다. 최소공배수를 구하는 과정에서 원소들간 작은 수가 높은 수의 약수인 경우를 제거하고 최대공배수를 구하면 효율성이 올라갈 수 있을까 생각해보았지만 제거하는 과정에서의 for문에 들어가는 비용이 더 높아 도중 제외했습니다.

영어 끝말잇기

주어진 vector객체를 기반으로 나열이 되었을 끝말잊기를 틀린 사람이 있는지 파악하고 몇번째 어느 사람이 틀렸는지 이야기하는 문제입니다.

#include <string>
#include <vector>
#include <iostream>
#include <algorithm>

using namespace std;

vector<int> solution(int n, vector<string> words) {
    vector<int> answer;

    for(int i = 1; i < words.size(); i++) {
        if(words[i-1].back() != words[i].front() || 
           find(words.begin(), words.begin() + i, words[i]) != words.begin() + i) {
            answer.push_back((i % n) + 1);
            answer.push_back((i / n) + 1);
            return answer;
        }
    }

    answer.push_back(0);
    answer.push_back(0);

    return answer;
}

if문은 마지막 글자와 첫번째 글자의 맞고 틀림을 검사하는것과 동시에 find 함수로 찾은 이터레이터값이 마지막 위치와 같은지를 확인하여 틀리다면 이전 앞에서 먼저 사용한 것으로 판단하여 해당 위치에서의 반복 횟수와 사람의 인덱스를 answer에 추가하는 방식입니다.

논문을 기반으로 작성하였고 서베이 논문을 포함하여 Multi-Task model 자체에 대한 설명을 하는 글.

이전까지의 딥러닝 모델들은 하나의 입력에 대해 하나의 출력값을 가진다.

Image Classification : 객체의 클래스

Object Detection : 객체의 클래스와 BBox값.

Image Segmentation : 객체의 클래스와 객체의 마스크.

객체의 클래스가 Object Detection과 Image Segmentation에 모두 포함되는 이유는 Head의 구조에 있다.

YOLOv8의 Head는 Decoupled Head를 사용하는데 Classification Branch와 Regression Branch로 구성되어 Classification Branch 에서는 클래스 확률을 예측하고 Regression Branch 에서는 BBox 좌표와 Objectness Score를 예측한다.

각 Head에 대해서 손실 함수의 종류를 선택하고 loss값의 가중치를 조절하는 등의 작업을 통해 균형을 이룬다.

-> YOLOv10의 경우에서는 Classification Branch가 Regression Branch보다 모델의 성능에 덜 영향을 미치는 것을 확인하여 해당 Head에 들어가는 파라미터의 값을 줄이는 작업도 진행한다.

위 단락까지만 보게 된다면 Object detection과 Image Segmentation작업도 하나의 Multi-Task라고 볼 수 있다.(Head라 통칭한 구조 내부에서 Classification과 Detection or Segmentaion작업을 수행하여 출력값이 2개 이상이므로.) 하지만 Detection 작업을 수행하려면 객체가 무었인지부터 파악해야 하는데 이 과정이 Classification 작업이므로 Object Detect과 Image Segmentation작업을 수행할 때 Classification Branch가 존재하는 것에 대하여 Multi-Task Model이라고 엄격히 표시하지는 않는 듯 하다.

(Detection과 Segmentation은 BBox와 Mask 계산에 사용되는 값들이 달라 Multi-task라고 확실히 불린다.)

다시 되돌아와 Multi-Task의 본질은 하나의 Input값에 대하여 2개 이상의 output을 출력하는 작업들을 의미한다.

YOLO series의 경우에는 Input이 이미지이고 output이 classfication, detection, segmentation이 된다.

이미지에 대한 Multi-Task를 목적으로 하는 작업들은 대부분 detection과 segmentation 작업을 합치는 것으로 자율주행에서는 차량과 사람 등에 장애물에 대한 BBox를 구하고 Line Lane과 Drivable Area에 대한 Mask들을 추출한다.

결국 중요한 요점은 모델의 구조를 설계할 때 어디까지 공유를 하고 하이퍼 파라미터들을 어떻게 설정하는지가 된다.

A-YOLOM의 경우 Backbone을 공유하고 Neck과 Head를 모두 분리시키는 방법을 사용한다. 해당 논문에서는 Backbone을 공유하여 특성 추출단계에서의 연산량을 줄이고 AC 블록을 통해 Neck에서의 연산량을 추가적으로 줄임과 동시에 불필요한 특성들을 가져오지 않도록 하였다.

DRMNet의 경우에는 Backbone과 Neck을 공유하고 Head를 분리시켜 사용한다.

Neck을 공유하지 않는 듯 하지만 Detail Branch(Conv) 이후 Segmentic Branch에서 생성된 Neck의 특성맵을 받은 이후에 Head로 분리되는 것을 확인할 수 있다.

DLT-Net의 경우에는 Traffic Object Decoder라는 추가적인 객체를 검출한다 당 논문은 Drivable Area 내부에 Lane Line이 존재하고 Drivable Area로 취급되지 않는 영역에 대해서 Traffic Object가 있는 것에 집중하여 구조를 설계하였다<마지막 특성맵이 차원 수가 256이었으나 128로 줄어든 것은 point-wise conv작업을 진행한 것으로 보인다.(github에서 확인해보려 하였으나 존재하지 않아 확인은 실패.) >. Backbone과 Neck을 공유하고 Context Tensor영역을 추가하여 Drivable Area 내부에 있는 특성맵을 공유하는 형식으로 진행된다. Context Tensor를 통과할 때 Traffic Object는 Concatenation작업을 통해 최종적으로 8045256의 특성맵을 가지게 되고 Lane Line Decoder의 경우에는 Element-wise Addition작업을 수행하여 Drivable Area Decoder와 같은 8045128의 특성맵을 가지게 된다.

YOLOP는 Neck구조까지 공유를 하지만 Detect Head는 모든 특성맵을 사용하는 반면 Segment Head들은 Neck의 가장 큰 사이즈의 특성맵만을 받아 사용하는 구조를 가진다.(사실 YOLOP만의 특성은 아니고 많이 사용한다.)

위 사례들처럼 Multi-Task작업을 구현하는 방식은 여러가지이다.

• Backbone만을 공유하는 사례.
• Neck까지 공유를 하지만 Segment Head에서는 Neck의 일부분만 추출하여 사용하는 사례.
• Neck까지 모두 공유를 하고 Drivable Area의 특성을 다른 Head에 공유하는 사례.

이 외에도 보지 못한 논문들과 사례들이 많겠지만 기본적으로 Backbone은 공유를 하고 Neck은 선택적으로 공유를 하거나 아에 분리시키는 것을 목표로 한다.

특이한 점은 A-YOLOM처럼 Neck의 단계에서 분리를 시킬 때 Segment Neck과 Detect Neck 2가지로 분리시키지 않고 Segment Neck를 2개를 사용하여 총 3개의 Neck을 사용했다는 것이다.

Loss를 기반으로 생각할 때 Multi-Task의 Loss값은 아래와 같이 계산된다.

(각 모델마다 계산하는 방식이 차이는 있지만 전체적인 프레임은 아래와 같다.)

( Loss_1 : Detection loss, Loss_2 : Drivable area loss, Loss_n : Lane line loss <- 부여된 값 예시 )

[ Total Loss = W_1 * Loss_1 + W_2 * Loss_2 + ... + W_n * Loss_n ]

계산된 Loss값은 각 Head에 동일하게 부여되어 최적화 과정을 거치게 되는데 Neck을 공유하게 된다면 Head에서 계산되어 나온 Loss값을 합친다.

( w1 : Detection weight, w2 : Drivable weight, w3 : Lane weight )

[ Total Neck Loss = w1 * Loss_detection_head + w2 * Loss_drivable_head + w3 * Loss_lane_head ]

해당 작업에서 각 작업에 주어지는 Weight값들은 일반적으로 직접 부여하게 된다. (weight값은 영향력, 기여도 등으로 이해해도 좋다.)

이후 Backbone은 Neck이 하나로 이루어져 있다면 하나의 Loss값을 받게 되고 Backbone만을 공유한다면 Neck에서 진행된 과정을 Backbone에서 거치게 된다. 하지만 위에서 언급한 것처럼 Neck을 2개를 사용하고 다시 한번 Neck에서 2개의 Head를 구성하게 되는 경우에는 Loss값을 합치는 과정을 2번 격게 된다. 이 과정에서 각 작업에 대한 Weight값(Neck에서 2번 Backbone에서 2번)을 설정하는 과정도 쉽지 않으며 Head의 수가 달라 절대적인 Gradient값에 대해서도 Head가 많은 경우의 Neck이 높은 값을 가지게 되어 하나의 Head를 가진 Neck의 작업이 낮은 성능을 가지게 되는 경우가 있어 비대칭의 학습이 이루어질 수 있다.

다음과 같은 문제점을 소지하고 있고 해당 작업을 가중치 자동화 작업(Uncertainty-based Weighting)과 중간의 Neck전용 Loss( Auxiliary Loss)값을 추가해주는 방식 등을 통해 완화시킬 수 있지만 다른 구조를 사용하는 편이 더 간편하고 효율성이 높아 Backbone도 공유하고 Neck을 공유함과 동시에 분리시키는 구조는 거의 사용되지 않는다.

종합적으로 평가하였을 때 Multi-Task에서 중요하게 판단하는 부분을 나열하면 다음과 같다.

• Backbone까지 공유할 것인지 Neck까지 공유할 것인가.
• Neck까지 공유하는 경우에 Segmentation 작업에 대해 어느 수준의 정보를 제공할 것인가. ㄴ Detection작업의 경우 작은 객체에 대한 인식이 필요하여 모든 사이즈 레벨의 특성맵을 모두 사용한다. ㄴ 위 과정에서 FPN Network뿐만 아니라 PAN Network도 사용하는 경우가 다수이다.
  • Loss값을 계산할 때 어떠한 방식으로 손실 값을 계산할 것인가. ㄴ Object detection의 경우 CIoU, DIoU, dfl, softmax cross, L1, L2 등 ㄴ Image Segmentation의 경우 Cross-Entropy, Focal, IoU 등
  • 작업을 모두 거쳤을 때 FLOPs값의 크기
  • 작업을 모두 거쳤을 때 정확도와 클래스 인식률.

논문에서는 300Epoch까지 진행하였으나 테스트로는 100Epoch까지만 진행.

train 환경 구성

GPU : Geforce 4070 super

anaconda3 가상환경

python : 3.12.3

github의 requirement.txt 설치

pytorch 설치 : pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118

학습 종료 CLI 학습 결과

논문에서의 결과 (mIoU와 IoU에 대해 동일한 값으로 판단하지 않도록 유의)

A-YOLOM(n) 모델에 대하여 인접하거나 살짝 낮은 결과값을 보임 loss값 추세와 결과물을 확인하였을 때 300Epoch 진행시 논문과 동일한 성능을 가질 것으로 예상된다.

predict 환경 구성

anaconda3 가상환경 python : 3.12.3 github의 requirement.txt 설치 pytorch : 2.2.0 + cu118 torchaudio : 2.2.0 + cu118 torchvision : 0.17.0 + cu118

ultralytics 라이브러리를 pip install로 다운로드 받을 시 sys.path.insert로 경로를 설정해주어도 Anaconda환경에서 먼저 라이브러리를 찾게 되므로 pip install ultralytics 커멘드는 사용하지 말아야 함.

테스트 입력 이미지 사이즈 1280*720 고정

이미지를 Input으로 진행할 때 코드

import sys
import torch

sys.path.insert(0, "C:/YOLOv8-multi-task/ultralytics")

from ultralytics import YOLO

number = 3 #input how many tasks in your work
model = YOLO('C:/YOLOv8-multi-task/runs/multi/yolopm14/weights/best.pt')  # Validate the model
model.predict(source='./img_path', imgsz=(384,672), device=0,name='output_path', save=True, conf=0.25, iou=0.45, show_labels=False, speed=True)

폴더를 경로에 입력해주면 폴더 내부에 있는 모든 이미지에 대해서 예측 작업을 진행해준다.

동영상을 Input으로 진행할 때 코드

import cv2
import time
import sys

sys.path.insert(0, "C:/YOLOv8-multi-task/ultralytics")

from ultralytics import YOLO

number = 3 #input how many tasks in your work
model = YOLO('C:/YOLOv8-multi-task/runs/multi/yolopm14/weights/best.pt')  # Validate the model

# Open the video file
video_path = "C:/yolov10/Seoul_30fps.mp4"
cap = cv2.VideoCapture(video_path)

# Get video properties
fps = cap.get(cv2.CAP_PROP_FPS)  # Frame rate
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))  # Frame width
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))  # Frame height

# Define the codec and create VideoWriter object
fourcc = cv2.VideoWriter_fourcc(*'mp4v')  # Codec for mp4
output_path = "Seoul_predict.mp4"
out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))

black= (0, 0, 0) 
font =  cv2.FONT_HERSHEY_PLAIN

frame_number = 0

# Loop through the video frames
while cap.isOpened():
    # Read a frame from the video
    start_time = time.time()
    success, frame = cap.read()

    if success:
        #Mat객체를 넣어야 하는데 source에는 jpg형태의 객체만 받아들임.
        cv2.imwrite("./runs/save_frame/frame_" + str(frame_number) + ".jpg", frame)
        model.predict(source="./runs/save_frame/frame_" + str(frame_number) + ".jpg", 
            imgsz=(384,672), device=0, name='Seoul', save=True, conf=0.25, iou=0.45, show_labels=False, speed=True)

        print("C:/YOLOv8-multi-task/runs/multi/Seoul/frame_" + str(frame_number) + ".jpg")
        annotated_frame = cv2.imread("C:/YOLOv8-multi-task/runs/multi/Seoul/frame_" + str(frame_number) + ".jpg")

        cv2.imshow("YOLO Inference", annotated_frame)

        frame_number+=1

        end_time = time.time() - start_time
        end_time = f"{end_time * 1000: .2f}"

        annotated_frame = cv2.putText(annotated_frame, "inference and process time :"
                                       + str(end_time) + "ms", (20, 40), font, 2, black, 1, cv2.LINE_AA)

        # Write the annotated frame to the output video
        out.write(annotated_frame)

        # Display the annotated frame
        #cv2.imshow("YOLO Inference", annotated_frame)

        # Break the loop if 'q' is pressed
        if cv2.waitKey(10) & 0xFF == ord("q"):
            break
    else:
        # Break the loop if the end of the video is reached
        break

cap.release()
out.release()
cv2.destroyAllWindows()

불필요한 작업을 진행(저장 -> 호출 -> 저장)하느라 frame당 시간이 오래 걸리는 것을 확인할 수 있다.

frame당 순수한 예측 작업은 평균적으로 50ms 초중반대가 나온다.

Multi-Task란 직역하면 다중 작업을 의미하며 목적은 3개의 model을 사용하여 진행할 작업을 하나의 model 내에서 작동하도록 하는 것이다.

기본적으로 YOLO model은 하나의 작업(Classification, Detect, Segmentaion, Pose 등)을 목표로 만들어진 모델로 두가지 이상의 작업을 수행하기 위해서는 각각의 작업을 수행하는 모델을 학습시킨 이후 얻은 데이터를 조합하여 최종적인 알고리즘을 구현한다.

이번 논문에서는 하나의 YOLO model에서 여러개의 작업을 수행하여 연산량과 모델의 전반적인 크기를 줄이는 방식을 목표로 한다.

Abstract

• 자율주행 관련 연구에서 High precision, lightweights, real-time responsiveness는 필수 요건으로 Detection, Segmentation작업을 동시에 처리하는 A-YOLOM 모델을 설계하였다.

• Neck과 Backbone 사이의 특징에 대해 능동적(논문에서는 적응적 연결로 소개-adaptively concatenates)으로 특징을 연결하여 학습 가능한 매개변수(가중치)를 사용하고 모든 Segmentation 작업에 대해 동일한 손실 함수를 사용한다. ㄴ 동일한 손실 함수를 사용한다는 것은 2개의 Segmentation model로 구성되어 손실함수 선택에 대한 고민을 제거하는 것과 동시에 일관된 학습으로 전체적인 모델들에 대한 안전성을 보장함.

• Segmentation Head를 Convolutional layer로만 구성되게 하여 파라미터 수와 추론 시간을 줄였다.

• BDD100K dataset을 사용하여 기존 모델과 비교하여 긍정적인 결과를 이끌어 내었다.

Introduction

ADS(Autonomous driving systems)은 딥러닝의 발전과 함께 집중을 받았고 lane line segmentation, Drivable area segmentation, Object detection 3개의 작업은 ADS에서 핵심 요소로 평가된다.

카메라로 위 3개의 작업을 수행하는 것은 여러가지 자원과 비용면에서 이점이 있으며 자율 주행이라는 특성상 위 작업은 30 FPS를 초과하는 값을 유지하는 것을 기본으로 한다.

경량 모델과 높은 정밀도를 목표로 하는 작업은 Fast R-CNN(two-stage 방식)과 YOLO(one-stage 방식)에서 진행되었고 YOLO에서는 Object Detection에 중점을 두어 발전해 Segmentation Head가 존재하나 사용되는 손실함수나 평가 방식(loss값 계산)들은 Object Detection작업에 최적화된 값들을 사용해왔다. YOLOv8은 하나의 모델에 대해 하나의 작업만을 구현할 수 있으며 여러 모델을 이용하여 알고리즘을 구현하는 것은 학습 시간과 추론 시간 등의 문제점이 존재한다.

Segmentation 작업 특징 lane line segmentation작업과 Drivable area segmentation에 있어 Drivable area segementation은 이미지의 넓은 영역을 차지하고 주변 상황과 같이 판단하여 High level feature을 필요로 하지만 lane line segmentation은 특징이 길고 단순하게 구성되어 있어 low level feature를 필요로 한다. 이 두 작업을 동시에 하기에는 서로의 정확도에 있어 악영향을 미친다.

Object detection과 Imege segmentation에 대해서도 Detect작업은 Grid Cell 방식을 사용하고 NMS를 통해 결과를 내보내고 Segmentation 작업은 pixel 단위에서 작동하도록 Decoder를 사용하는 방식을 사용한다.

중점은 두 작업 모두 이미지에서 특징을 추출하는 Backbone단계의 구조는 공유할 수 있는 형태를 가진다는 것이다.

해당 논문에서는 하나의 Backbone과 3개의 Neck, Head를 가지는 Architecture를 사용한다. 하나의 Backbone에서 분할되어 나오는 여러개의 Neck사이에 능동적(P1, P2, P3, P4, P5 각 특징을 원하는 대로 Neck에 연결)으로 모듈을 적용하는 방식으로 하여 서로 다른 레벨의 특징을 연결할지 여부를 결정할 수 있다.

Methodology

A-YOLOM 모델은 Encoder-Decoder Architecture를 가진 one-stage 네트워크로 Encoder는 Backbone과 Neck로 구성되었으며 Decoder는 Head로 구성된다.

하나의 Backbone과 세 가지 작업을 위한 3개의 Neck, Head를 단일 모델로 통합한다.

(N 값은 Neck의 수로 해당 논문에서는 2개를 의미한다.)

Encoder (Backbone_Neck)

Backbone

기존 YOLOv5에서 사용되던 Backbone인 SCP-Darknet53을 개선.

YOLOv8의 기초 Backbone과 동일하다.

Neck

3개의 Neck을 활용 { lane lines, drivable areas, object detection }

Neck의 구조는 동일하게 사용하나(그림의 N이 2) 각 목적이 다르게 사용되여 가중치의 값들은 상당히 다르게 진행.

Object Detection 작업의 경우 차량을 감지하여야 하므로 low-level feature보다는 high-level feature의 특징을 조합하여야 하므로 Backbone에서 P3~P5까지의 mid-level과 high-level들을 조합하여 사용하였고 이 과정에서 FPN을 포함하는 PAN구조를 채택하였다. PAN 구조를 사용함으로서 작은 객체와 큰 객체에 대한 정확도를 향상시킨다.

Image Segmentation 작업의 경우 low-level feature과 high-level feature정보를 모두 포함하도록 P1~P5까지의 특징들을 Backbone에서 불러온다. 하지만 lane line에서 high-level feature이 크게 의미가 없는 특징맵의 경우 Adaptive Concatenation Module(AC)을 통해 해당 Feature은 연결하지 않는 등의 작업으로 연산량을 줄이고 정확도를 향상시킨다.

Adaptive Concatenation Module의 알고리즘

Output과정은 초기값 weight(5.0)을 기반으로 특성맵을 연결할지 말지에 대한 여부를 결정한다.

weight값은 sigmoid함수의 파라미터로 사용되어 해당 값이 0.5를 초과하게되면 두 특성맵을 결합하고 1*1 Convolution layer를 통과시켜 채널 수(차원)를 줄이는 과정을 수행한다. 0.5 이하의 값이 나오면 Neck 특성맵만을 C2f 에 통과시킨다.

5번 라인을 간단히 살펴보면 Conv는 Ultralytics에서 제공하는 함수를 의미하며 내부 파라미터는 차례대로 [입력 채널 크기, 출력 채널 크기, kernel size, stride]이다.

학습이 진행되며 weight값이 커져 sigmoid함수를 통과시킬 때 0.5가 넘게 되면 해당 특성맵을 연결하는 과정을 통해 lane line에서는 연결되지 않은 특성맵이 drivable area에서는 연결되는 등의 작업이 이루어질 수 있다.

해당 과정에서 Segmentaion의 Neck를 분리시킨 이유가 나온다. (하나이면 해당 과정에 의미가 약해진다.)

위 과정은 각 작업의 Neck과정에서 불필요한 연산을 줄이며 정확도를 높이게 된다.

• Detection과 Segmentation의 Neck 구조에서 가장 다른 점은 FPN에서 끝나는 Segmentation Neck과는 다르게 Detection Neck에서는 이를 포함하는 PAN network를 사용한다는 것인데 이유는 차량 객체의 경우 멀리 있을 때와 가까이 있을 때의 객체 크기에 상당한 차이를 보이는 반면 lane lines, drivable areas의 경우 멀리 있을 때와 가까이 있을 때의 차이가 크지 않아 민감하게 반응할 이유가 없기 때문에 연산량을 보존하고자 FPN에서 끝낸다. (개인적 의견 첨가)-

Decoder (head)

Backbone과 Neck에서의 작업을 처리하여 예측을 수행 2개의 Segment Head와 1개의 Detect Head를 사용

Head

cv3(cv2(upsample(cv1(x)))을 거치면 해상도가 입력 특성맵의 최종적으로 2배 증가하게 된다.

• 입력 특성맵으로 사용하는 P1의 해상도 크기 기준 2배.

Loss Function

해당 모델에서 손실함수는 다음과 같이 정의된다.

순서대로 Detect, Seg draivable area, Seg lane line 에 대한 손실이다.

객체 탐지(L_det)에서는 DFL loss(Distribution Focal Loss), BCE loss(Binary Cross Entropy), CIoU loss를 조합하여 계산하는데 각각 class, Bbox, Bbox를 계산하는데 사용된다. 위 방식은 YOLOv8의 일반적인 객체 탐지 방법과 동일한 방식이다.

Seg loss에 대해서는 아래와 같이 정의된다.

L_FL은 Focal loss를 의미하고 L_TL은 Tversky loss를 의미한다. ( Tversky loss는 Segmentation에서만 쓰이는 손실 함수로 명확히 설명되어있는 사이트를 찾지 못함.)

각 손실함수에 대한 역할에 대해서 간단히 집고 넘어가자면 Focal loss는 데이터 불균형과 데이터의 난이도에 따른 학습 정도(어느정도로 가중치를 민감하게 수정할지)를 조정하며 Tversky loss는 객체에 대한 경계면을 세세하게 구분하기 위한 손실함수이다.

Experiment

사용한 데이터셋 : BDD100K (자동차, 버스, 트럭 등의 객체들을 Vehicle로 통합)

평가 방식 : mAP50(객체 탐지), mIoU(Drivable area), IoU + Acc(lane line)

사용 장비 : Train 과정 RTX 4090 3개 - Val과정 GTX 1080 Ti GPU 1개로 명시

데이터 증식 기법 : Ultralytics 기본 데이터 증식 기법 사용.

실험 방식 : SGD optimizer, learning rate : 0.01, momentum : 0.937, weight decay : 0.0005, NMS : 0.6

• 3 Epoch까지는 워밍업으로 learning rate값등이 올라간 상태로 시작함. 이미지 사이즈 640*640

평가에 관련하여 multi-task관련 모델(평가를 비교할)들이 부족하여 객체 검출 모델과 영상 분할 모델들을 같이 사용하여 테스트하였다.

총 33만개의 이미지가 포함되고 20만개의 이미지에 Object detection, Segmentationl, Captioning(그림을 단어로 설명하는 작업)에 대한 주석이 포함되어있다.

주석에는 Bounding box, Segmentation mask, Caption in image에 대한 정보가 포함되어있다.

COCO dataset은 "Train2017", "Val2017", "Test2017" 3가지의 디렉리로 구성되어 있는데 Train에는 11.8만의 이미지가 포함되고 Val dataset은 5천개의 이미지가 있으며 Test에는 2만개의 이미지로 구성되어 있다. Test dataset에 대한 주석은 제공되지 않고 Test dataset에 대한 구체적인 성능 평가를 받기 위해서는 COCO eval‎uation server에 제공해야 한다.

Ultralytics에서는 COCO dataset에 대한 구성으로 YAML(야뮬)파일을 사용하며 데이터세트의 경로, 클래스, 기타 정보들을 포함한다.

https://github.com/ultralytics/ultralytics/blob/main/ultralytics/cfg/datasets/coco.yaml

# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license

# COCO 2017 dataset https://cocodataset.org by Microsoft
# Documentation: https://docs.ultralytics.com/datasets/detect/coco/
# Example usage: yolo train data=coco.yaml
# parent
# ├── ultralytics
# └── datasets
#     └── coco  ← downloads here (20.1 GB)

# Train/val/test sets as 1) dir: path/to/imgs, 2) file: path/to/imgs.txt, or 3) list: [path/to/imgs1, path/to/imgs2, ..]
path: ../datasets/coco # dataset root dir
train: train2017.txt # train images (relative to 'path') 118287 images
val: val2017.txt # val images (relative to 'path') 5000 images
test: test-dev2017.txt # 20288 of 40670 images, submit to https://competitions.codalab.org/competitions/20794

# Classes
names:
  0: person
  1: bicycle
  2: car
  3: motorcycle
  4: airplane
  5: bus
  6: train
  7: truck
  8: boat
  9: traffic light
  10: fire hydrant
  11: stop sign
  12: parking meter
  13: bench
  14: bird
  15: cat
  16: dog
  17: horse
  18: sheep
  19: cow
  20: elephant
  21: bear
  22: zebra
  23: giraffe
  24: backpack
  25: umbrella
  26: handbag
  27: tie
  28: suitcase
  29: frisbee
  30: skis
  31: snowboard
  32: sports ball
  33: kite
  34: baseball bat
  35: baseball glove
  36: skateboard
  37: surfboard
  38: tennis racket
  39: bottle
  40: wine glass
  41: cup
  42: fork
  43: knife
  44: spoon
  45: bowl
  46: banana
  47: apple
  48: sandwich
  49: orange
  50: broccoli
  51: carrot
  52: hot dog
  53: pizza
  54: donut
  55: cake
  56: chair
  57: couch
  58: potted plant
  59: bed
  60: dining table
  61: toilet
  62: tv
  63: laptop
  64: mouse
  65: remote
  66: keyboard
  67: cell phone
  68: microwave
  69: oven
  70: toaster
  71: sink
  72: refrigerator
  73: book
  74: clock
  75: vase
  76: scissors
  77: teddy bear
  78: hair drier
  79: toothbrush

# Download script/URL (optional)
download: |
  from ultralytics.utils.downloads import download
  from pathlib import Path

  # Download labels
  segments = True  # segment or box labels
  dir = Path(yaml['path'])  # dataset root dir
  url = 'https://github.com/ultralytics/assets/releases/download/v0.0.0/'
  urls = [url + ('coco2017labels-segments.zip' if segments else 'coco2017labels.zip')]  # labels
  download(urls, dir=dir.parent)
  # Download data
  urls = ['http://images.cocodataset.org/zips/train2017.zip',  # 19G, 118k images
          'http://images.cocodataset.org/zips/val2017.zip',  # 1G, 5k images
          'http://images.cocodataset.org/zips/test2017.zip']  # 7G, 41k images (optional)
  download(urls, dir=dir / 'images', threads=3)

기본적으로 YOLO CLI를 사용하여 훈련을 진행하는 경우에는 따로 다운로드 받을 필요 없이 YAML파일에 저장된 내용대로 경로를 찾아 dataset이 없다면 다운로드 받고 있다면 그대로 사용한다.

YOLO dataset을 다운받게 되면

다음과 같이 디렉터리가 구성되는데 coco 디렉터리 내부에 annotations와 images가 생성된다. annotations는 주석이 달려있는 json파일이 존재하는 파일로 해당 json파일을 YOLO형식에 맞게 수정하면 labels 디렉터리가 생기게 된다.(json2yolo) (YOLO CLI 사용시 해당 과정 불필요) coco dataset format을 살펴보게 되면 annotations 디렉터리에는 아래 그림과 같이

< captions _Tr_Val, instances_Tr_Val, poss _Tr_Val > 로 구성되어 있다.

해당 과정은 github를 살펴보아도 나오지만 jq를 사용하는 방식을 추천한다. https://nepersica.tistory.com/22

다음 과정을 거치면 다음과 같이 줄바꿈이 진행되어 새롭게 파일이 저장된다.

후반부에는 이미지에 대한 annotaions을 제공한다. Bbox와 segmentation points등

{
      "segmentation": [
        [
          260.4,
          231.26,
          215.06,
          274.01,
          194.33,
          307.69,
          195.63,
          329.72,
          168.42,
          355.63,
          120.49,
          382.83,
          112.71,
          415.22,
          159.35,
          457.98,
          172.31,
          483.89,
          229.31,
          504.62,
          275.95,
          500.73,
          288.91,
          495.55,
          344.62,
          605.67,
          395.14,
          634.17,
          480,
          632.87,
          480,
          284.37,
          404.21,
          223.48,
          336.84,
          202.75,
          269.47,
          154.82,
          218.95,
          179.43,
          203.4,
          194.98,
          190.45,
          211.82,
          233.2,
          205.34
        ]
      ],
      "area": 108316.66515000002,
      "iscrowd": 0,
      "image_id": 520301,
      "bbox": [
        112.71,
        154.82,
        367.29,
        479.35
      ],
      "category_id": 18,
      "id": 3186
    },

마지막 영역에는 categorise 정보가 들어간 내용으로

supercategory는 상위 개념의 클래스를 의미하고 내부 name에서 하위 개념의 클래스를 구별해준다.

categories에서 id는 클래스의 번호를 의미한다. (coco.yaml)의 순서와 동일함.

논문에서 소개한 성능표

yolov10l.pt , yolov10n.pt, yolov10s.pt 지표

100epoch 훈련 결과 AP(0.5:0.95)val값은 0.431로 출력

논문과 one-to-one Head에 대해서 0.032(3.2%)의 차이가 존재하는 것을 확인. 해당 값의 차이를 확인하기 위해서 논문을 다시 한번 확인해보기로 하고 체크하였을 때 차이점의 여부에 대해 확인할 수 있었다.

다른 조건들은 모두 동일했으나 Epoch수에 대해 차이점이 있었다.

기본 실험에서는 Ultralytics의 Github를 소개하며 해당 사이트와 동일한 조건으로 테스트하였다고 소개하였으나 세부 영역에서 논문에서는 총 500 Epochs를 돌렸고(3090GPU-8 EA로 학습을 진행) 400 Epochs에 대한 차이값이 0.032의 AP(0.5:0.95)차이점을 나타낸 것으로 확인된다.

500Epoch를 돌리려면 예상 시간으로 150시간정도가 걸릴 것으로 예상되어 테스트는 잠시 미뤄두기로 하였으나 값이 감소되는 추세를 보았을 때 Epoch를 늘렸을 때 0.46까지 도달할 것으로 예상된다. 붉은색 라인이 AP(0.5:0.95)영역에 대한 값으로 0.4294에 대한 AP값을 출력.

해당 커맨드는 coco dataset에 대해서 best.pt모델을 사용하여 val을 진행한 과정.

학습 CLI

학습 시에는 약 8.1정도의 Param.(M)의 값이 나오나 논문에서는 7.2의 Param.(M)를 가지고 있다고 표시한다.

해당 값들의 차이는 one-to-many의 파라미터의 차이로 YOLOv10에서는 추론시 학습이 완료된 one-to-one Head의 파라미터에 의존하여 추론을 하기 때문에 one-to-many Head의 파라미터를 사용하지 않는다.

from ultralytics import YOLOv10

model = YOLOv10('best.pt')
model.model.model[-1].export = True
model.model.model[-1].format = 'onnx'
del model.model.model[-1].cv2
del model.model.model[-1].cv3
model.fuse()

따라서 학습이 완료된 모델에 대해 파라미터를 출력하게 되면 논문에 제시되어있는 파라미터가 출력되는 것을 확인할 수 있다.

https://github.com/THU-MIG/yolov10/issues/178

Regression Head : 회귀 헤드라고 불리며 Bounding Box를 계산하는데 사용된다. YOLO에서는 Onjectness Score(객체가 존재할 확률)를 Regression Head에 포함하여 쓰이는 경우가 대부분이며 YOLOv8에서는 Anchor를 사용하여 계산하고 Anchor를 사용하지 않는 YOLOv10의 경우에는 객체의 중심 좌표와 크기를 one-to-one Head에서 출력된 정보와 비교하며 Bounding Box를 계산한다.

Classification Head : 분류 헤드라고 불리며 해당 객체에 대한 Class를 예측하고 학습하는 Head이다. Detection 작업에서는 예측된 Bounding Box 내에 어떤 Class가 포함되어있는지를 하습하며 클래스에 대한 신뢰도 점수를 기반으로 예측한다.

두 헤드는 각각의 예측과 오류값을 기반으로 동시에 학습된다.

회귀 분류 이미지

관련 영상

PAN은 Path Aggregation Network의 약자로 본래는 Instant Segmentation모델에 대해 제작되었지만 성능이 좋아 여러 AI모델에서 기용하고 있다.

YOLOv3와 yolov8의 모델 구조는 다음과 같다.

해당 두 Architecture에서 자잘한 레이어들에 대한 차이점은 물론 존재하지만 직접적으로 보이는 가장 큰 차이점은 Neck영역으로 구성되는 파트이다.

YOLOv3의 경우 FPN(Feature Pyramid Networks)를 사용하고 있고 YOLOv8의 경우 FPN을 개량한 PAN(Path Aggregation Network)을 사용하고 있다.

FPN은 Top-Down을 기반으로 상위 계층에서 하위 계층으로 Conv레이어를 거치며 레이어가 합쳐지고 PAN은 기존의 FPN구조에 Bottom-Up을 기반으로 하는 구조가 합쳐저서 구성되어 있다.

설명 이전에 Backbone에서 Bottom-Up과정이 수행되지만 각 특정 레이어(Yolov8의 예시에서 4, 6, 9 layer)에서 특징맵을 가져오기 때문에 해당 특징들에 대한 조합이 이루어지지 않는다. 해당 과정을 Neck에서 수행해준다.

두 구조에 대한 차이점을 직관적으로 이야기하면 Head가 받는 특징맵 정보의 질의 차이이다.

FPN을 사용하는 YOLOv3의 경우 Backbone에서 출력된 레이어에 대해 한번의 Conv과정을 통해 low-level feature(edge, color)을 조합한 high-level feature을 학습하게 되된다.

PAN을 사용하는 YOLOv8의 경우 SPPF에서 출력된(YOLOv3를 예시로 들면 하늘색 영역의 Neck) 특징맵이 한번 FPN구조를 따라간 다음 다시 Bottom-Up구조를 거치며 low-level feature를 받게되어 high-level의 특징맵을 가진 2020512*w사이즈를 받는 Head에 대해서도 강한 low-level feature의 정보를 가지게 된다.

예시를 들게 된다면 FPN은 문제집 하나를 푸는 것과 같고 PAN은 문제집 하나를 풀고 여러 유형을 결합한 문제집를 다시 한번 푸는 것과 같다.

PANet( Path Aggregation Network for Instance Segmentation )

신경망의 정보를 Head로 전달하는 과정을 개선하기 위한 논문.

PAN을 설명하기 위한 구조로 (a)영역에 대해 FPN Backbone으로 칭하고 (b)영역에 대해 Bottom-up path augmentaion작업을 수행한다.

Bottom-up path augmentation

FPN에서 바로 Head로 연결되면 low-level featrue의 정보를 Big object에서 직접적으로 활용하기 힘들기 때문에 각 특징의 특성을 직접 연결하는 방식으로 여러 해상도에 대해서 low-level feature과 high-level feature을 사용할 수 있게 한다.

해당 논문에서는 FPN에서 생성한 레이어들의 해상도(P5, P4, P3, P2)에 맞춰 같은 크기의 해상도를 가지는 특징맵(N5, N4, N3, N2)들을 구성한다. P2와 N2는 동일한 특징맵이다. N_n 특징맵들은 3*3_Conv작업을 거치며 해상도가 낮아진 다음 P(n+1) 특징맵들과 Concat레이어를 통해 합쳐지며 새로운 특징맵 N(n+1)을 생성한다. 해당 과정을 반복하며 최종적으로 N5특징 맵을 가지게 된다.

YOLOv10: Real-Time End-to-End Object Detection

직역해 보자면 YOLOv10 : 실시간 객체 검출기(End-to-End 의 형태를 가진)이다.

간단하게 End-to-End의 용어를 집고 넘어가자면 객체의 특징 추출부터 추론까지의 과정을 하나의 Architecture 내에서 진행하는 모델의 구조를 의미한다.

위 이미지는 이전의 전통적으로 진행되던 딥러닝 모델에서는 데이터가 들어오면 그에 대한 특징 추출과 특징에 대한 추정 최종 결과까지 여러 단계의 파이프라인을 지나치게 되며 OUTPUT값을 보내지만 End-to-End 구조의 경우에는 이전에 나누어지던 파이프라인들을 하나의 딥러닝 네트워크 내에서 진행한다. 통상적으로 end2end로 적기도 한다.

Abstract

이전 YOLO 모델부터 이미지 처리를 위해서 컴퓨팅 자원 소모와 처리 시간에 대한 감소를 목적으로 계속해서 발전해왔지만 YOLOv8의 단계까지 왔음에도 해결하지 못한 몇몇 문제점이 존재하는데 해당 논문에서는 후처리 및 모델 아키텍처에서의 YOLO 성능과 효율성의 단계를 증진시키는것을 목적으로 작성되었다.

논문에서 제시한 YOLOv10에서 이전의 YOLO 시리즈들과 대비되어 해결한 내용들은 아래와 같다.

1. 객체를 탐지하고 후처리 과정에서 진행해야 하는 NMS(Non-Maximum-suppression)

2. 특징(feature)을 추출하는 단계에서 모델의 크기가 증가할수록 증가하게 되는 연산 오버헤드(Computational Overhead)

3. 정확도와 효율성에 대한 개선

결과적으로는 YOLOv10은 이전 버전 혹은 다른 모델과 비교하였을 때 하나의 INPUT값에 대해 적은 수의 파라미터와 낮은 수의 FLOPs과 Latency를 가진다.

FLOPs(Floating point Operations) : 부동소수점 연산을 의미하며 여기에서 연산은 사칙연산, log, exp등의 연산이 포함된다.

-FLOPs는 동일 성능 대비 낮을수록 좋다고 생각할 수 있다.(해당 모델에서 소비하는 컴퓨팅 자원이 적다는 것을 의미함)

Latency : 직역은 대기시간으로 딥러닝 추론의 시작부터 완료까지 걸리는 시간을 의미한다, Latency(f)가 붙은 것들은 이미지 후처리에 대한 시간을 제외한 나머지 시간을 의미한다.

Introduction

소개에서는 이제동안 YOLO시리즈가 발전해온 과정과 NMS가 왜 문제가 되는지에 대해서 소개해준다.

NMS는 Non-Maximum-suppression의 약자로 YOLOv8을 기준으로 이미지에서 예측되는 Bounding Box를 모두 특정 임계값에 대해 일차적으로 처리를 하였을 때 남게되는 중복되는 Bounding Box들이 있다.

NMS 미적용	NMS적용

위 이미지를 보았을 때 각 Bounding Box들이 하나의 객체에 대해서 여러개의 후보군들이 있는데 이 때 NMS를 통하여 우측의 사진처럼 하나의 객체에 하나의 Bounding Box만이 남게 된다. 하지만 해당 과정은 병렬적으로 계산되는 것이 아닌 각 Bounding Box에 대해서 1:1매칭으로 비교되는 계산법으 높은 시간 복잡도를 가지고 있어 최종 결과물의 Latency시간에 큰 영향을 미치게 된다.

NMS의 단점으로는 다른 객체에도 영향을 미치게 되는 것으로 클래스별로 계산되는 과정에서 같은 클래스의 Bounding Box가 2개 이상 존재하는 경우에 서로 다른 객체이더라도 하나만 살리는 등의 문제점이 존재한다.

추가적으로 NMS는 end-to-end 구조를 가진 YOLO시리즈에 대해서 부정적인 영향을 미친다고 평가되어왔다.(후처리가 직접적으로 들어가기 때문)

(NMS 계산법에 대한 설명글) (NMS 계산에 대해 개선을 시도하려 하였던 논문)

NMS의 대체제로 DETR(End-to-End Object DEtection-with-TRansfor)이나 RT-DETR(Real-Time-DEtection-with-Transformers)등을 도입해 보았지만 정확도와 추론 오버헤드(다른 객체를 없애는 등)의 문제점이 남아있다는 단점이 존재한다.

Backbone과 Neck 영역에서 계산의 효율성과 정확도에 대한 기능은 충분히 이끌어 내어서 자신들은 Head영역에서 진행하였고 방식으로는 Dual Label Assignments와 Consistent Match. Metric을 채택하였다.

이후 연산 오버헤드와 정확도, 효율성의 개선 방식으로 다음과 같은 방식을 사용하였다.

Efficiency

● Lightweight classification head.

● Spatial-channel decoupled downsampling.

● Rank-guided block design.

Accuracy

● Large-kernel convolution.

● Partial self-attention (PSA).

위 방식들을 통해 YOLOv10 -N / S / M / B / L / X 의 Scale을 가진 모델들을 제작하였고 다음과 같은 성능을 보였다.

APval에서 좌측 값은 one-to-one Head의 값을 나타내며 우측 ' † ' 가 들어간 값은 one-to-many Head의 값을 나타낸다.

Methodology

이제까지의 YOLO모델들은 one-to-many Head를 적용시켜 일반적으로 TAL(Task-ALigned)를 기반으로 생성되었으나 NMS 사후처리에 의존해야 하기 때문에 연산량이 증가되는 문제점이 있고 NMS를 사용하지 않기 위해 제작된 one-to-one Head를 사용하는 모델에 대해서도 추론 오버헤드(존재하는 객체를 인지하지 않는)를 발생시키거나 최적화 되지 않는 문제점이 남아있다.

따라서 해당 논문은 one-to-many Head와 one-to-one Head 두가지를 모두 사용하는 Consist Dual Assignment를 사용하여 NMS를 제거하여 연산량을 해결하고 여러 객체에 대한 정확도를 높이는 작업을 수행함.

◆ Consist Dual Assignment는 Dual Label Assignments영역과 Consistent Match. Metric영역으로 분리되어 사용된다.

Dual Label Assignments(이중 레이블 할당)

one-to-one Head는 하나의 객체에 대해 하나의 예측값을 출력하여 NMS의 사후 처리가 필요하지 않은 Head이나 Supervision에 대한 정보가 충분치 않아 Accuracy와 Convergence speed를 최적화시키지 못한다.

one-to-many Head는 각 객체에 대해 여러가지의 예측값을 출력하는 대신 단일로 사용될 때 NMS와 같은 후처리가 필요하다.

YOLOv10에서는 두 Head를 모두 사용하여 모델의 학습시에는 one-to-one Head와 one-to-many Head에서 나오는 Supervision값들을 사용하여 두 개의 Head와 Backbone 및 Neck영역의 가중치들을 빠르고 정확하게 최적화하며 추론할 때에는 one-to-many Head를 사용하지 않고 one-to-one Head를 사용하여 상위 1개의 Bounding Box를 채택하여 end-to-end 형식의 YOLO 모델을 사용한다. one-to-one매칭에서 훈련 시간이 적은 Hungarian matching과 동일한 성능을 달성한다고 나와있는데 Hungarian matching은 2개의 집합에 대해 최적의 매칭을 잡아 높은 이익의 결과물을 가져오는 알고리즘을 의미한다.

Consistent Matching Metric(일관된 매칭 메트릭)

s : 예측의 앵커 포인트가 인스턴스 내 존재하는 공간의 정보(IoU값으로 계산)

b_hat, b : 예측값과 인스턴스의 bbox

α, β : 의미 예측 작업과 회귀 작업에 사용되는 하이퍼파라미터.

m_o2m=m(α_o2m, β_o2m) : one-to-many에 대한 Metric

m_o2o=m(α_o2o, β_o2o) : one-to-one에 대한 Metric

Consistent Matching Metric에서는 두 Head에서 나온 Supervision들을 토대로 학습을 수행하는 영역이다.

두 Head에서 제공되는 Supervision의 격차를 분석하여 학습을 진행하며 최초 시행시 동일한 값으로 초기화된 두 개의 Head에 대해 검사를 시작하여 동일한 예측값을 생성되는 것을 전제로 동일한 p값과 IoU값을 생성하는 경우에는 두 Head의 결과가 동일한 것으로 판단되고 학습 시 발생하는 차이는 회귀 단계가 아닌 분류에 대한 부분에서 발생한다.

동일한 값을 출력한 예측 값은 공유하고 예측 값이 다른 경우에는 무시되면서 해당 격차는 1-Wasserstein distance로 도출된다.

1-Wasserstein distance의 거리 값이 가장 작은 때의 distance는 Wasserstein distance로 지칭된다. Wasserstein distance에 대한 설명

A는 Supervision의 격차를 의미하며이를 최소화하기 위해 Metric값을 𝛼𝑜2𝑜=𝑟⋅𝛼𝑜2𝑚 및 𝛽𝑜2𝑜=𝑟′⋅𝛽𝑜2𝑚으로 설정되고 이는 하단의 식을 의미한다.

위 과정을 통해 학습 시 one-to-many Head의 최상의 표본은 one-to-one Head에서도 최상의 표본이 되며 두 헤드간의 결과를 일관성을 유지하며 모델을 최적화 시킬 수 있다.

Holistic Efficiency-Accuracy Driven Model Design

효율적으로 모델을 운용하기 위해서 YOLO의 구성 요소 중 계산 비용이 낮은 stem을 제외한 downsampling layers, basic building blocks, head에 대해 설계를 수행함.

(분류 헤드는 객체의 클래스의 확률을 정하며 회귀는 Bounding Box의 좌표와 각 예측에 대한 신뢰도를 계산)

Spatial-channel decoupled downsampling (공간-채널 분리 다운샘플링)

해당 작업은 O(2HWC2 + (9/2)HWC) 계산 비용과 O(2C2+18C)의 매개변수 수로 감소되는 결과를 도출하였다. Pointwise는 3차원 Depth의 레이어들이 있다면 113의 fillter를 통과시켜 1차원 Depth의 레이어 하나가 있도록 하는 기법으로 채널 자원을 조정하며 Depthwise convolution은 3 차원 Depth의 레이어가 있을 경우 각 차원별로 331 filter로 conv를 진행시켜 각 Depth에 대해 합성곱을 진행한다.

해당 방식은 다운 샘플링을 진행하는 동안 정보의 변화를 최소화하면서도 Latency 값의 감소를 유도한다.

Rank-guided block design(순위 기반 블록 설계)

YOLO모델들은 특징을 추출하는 단계에서 모델이 크고 깊은 수준의 특징을 추출할 때 같은 내용의 연산을 중복하여 연산하는 경우가 많다. 이를 해결하기 위해 CIB(Compact Inverted Block)를 사용한다 CIB는 바로 위에서 설명한 depthwise convolution와 pointwise convolution을 채택한 블록 구조이다.

CIB 블록은 ELAN(efficient layer aggregation network)구조에 포함되어 YOLOv10의 기본 구조로 포함된다.

ELAN을 간단하게 설명하자면 특성맵을 받아 여러 채널 그룹으로 분할시키고 각 그룹을 독립적으로 처리하며 서로 다른 해상도의 특징들을 조합하며 학습이 가능한 블록 구조이다.

Accuracy driven model design

모델의 정확도를 높이기 위한 모델 설계로는 Large-kernel convolution과 Partial self-attention(PSA)가 있다.

각 특성 맵은 깊이에 따라 특성간의 거리에 있어 같은 값들을 가지게 되는데 모든 단계에서 33의 kernel size를 유지하게 되면 작은 객체를 감지할 때 얕은 수준의 특성의 학습에 대해 부정적인 영향을 미치고 고해상도 단계에서는 같은 특성의 학습을 계속해서 반복하는 등의 컴퓨팅 소스를 낭비하게 된다. 해당 문제를 해결하기 위해서 깊은 단계의 특성맵들을 계산할 때에는 CIB 내부에서 Dilation rate값을 증가시키면서 kernel size를 77로 증가시키는 작업을 통해 추론-오버헤드(계산량의 증가) 없이 수용영역(kernel size가 증가함에 따른 전역적인 특징의 감지)을 늘린다.

(사진에서는 Kernel size가 3 * 3으로 유지되고 있지만 각각 55와 77이 맞는 표현이다. 그림에서는 추론-오버헤드 없이 진행됨을 강조하기 위해 kernel size를 3*3으로 유지한 것으로 보임.)

(해당 합성곱은 대형 모델에 대해서는 사용하지 않을 수 있다. - 이미 많은 파라미터와 가중치를 가지고 있기 때문이다.)

Partial Self-Attention(PSA)(부분적 자기 주의) - 사진의 C에 해당

Attention에 대한 설명(이미지로 예시를 들다면 각 특성들에 대한 연관성이고 문장으로 예시를 들면 각 단어 각 문장 사이의 연관성의 강도를 의미한다.)

Self-attention은 널리 사용되지만 높은 계산 복잡도와 메모리 사용량을 동반하기에. 1x1 convolution 후에 채널을 두 부분으로 나누어 한 부분만 MHSA(Multi-Head Self-Attention)와 FFN(Feed-forward Network) 블록을 통과하게 하고 Self-Attention 작업은 낮은 해상도를 가지는 파트에서만 적용되어 낮은 계산 비용으로 Self-Attention작업을 수행한다. 그림의 [* N_PSA]는 해당 점선으로 그려진 영역의 반복 횟수를 의미한다.

이후 처음 분리했던 블록과 1*1 convolution에 의해 합쳐진다.

논문에서는 가장 낮은 해상도인 Stage 4 이후에만 배치된다고 하나 Stage 4가 어느 지점을 이야기하는지 정확히 명칭하는 바는 아직 확인하지 못하였다.

해당 작업에서 빠른 추론을 위해 LayerNorm을 BatchNorm으로 변경하는 등의 작업도 수행되었다

Conclusion

NMS 후처리를 대체하기 위해 Consist Dual Assignment을 제안

전체적인 모델의 정확도와 효율성을 개선하기 위해 합성곱을 진행하는 방식과 커널 사이즈 및 Self-Attention을 부분적으로 적용하는 등의 작업을 통해 연산량은 유지(Kernel Size를 증가시키며 Dilation rate값도 증가) 혹은 감소(중복하여 진행하는 연산은 제거)시키면서 정확도를 증가시킴.

레이어 설명

Neck와 Head 분리

Neck영역 추출 이미지

이후 Upsample과정에서는 nearest방식과 scale_factor값에 2를 주어 이미지의 크기를 상승시킨다.

사용되는 torch Upsample 코드 예시

각각에서 추출된 레이어들은 Upsample 레이어와 Cov 레이어를 거치며 각 이미지 사이즈의 특성을 합친 특성맵을 출력하고 이를 Head의 입력으로 사용해 각 사이즈(특성맵의 width, height)의 정보를 가진 Detect 클래스가 생성된다.

Upsample레이어에 관해 10번 레이어와 13번 레이어에 대해 출력 특성 맵의 차원이 다른 것을 확인할 수 있었는데 Upsample레이어 내부에 출력 채널 수를 결정하는 레이어가 존재하나 model.yaml에서는 해당 요소에 대하여 인가하는 파라미터를 찾을 수 없어 현재 파악 중 (c_ 파라미터가 출력 채널의 수로 256이다 이로 인해 13번의 채널 수가 줄어드는 것은 파악할 수 있었으나 10번 레이어를 통과할 때 채널 수가 줄어들이 않는 이유는 확인하지 못함)

각각에서 추출된 레이어들은 Upsample 레이어와 Cov 레이어를 거치며 각 이미지 사이즈의 특성을 합친 특성맵을 출력하고 이를 Head의 입력으로 사용해 각 사이즈(특성맵의 width, height)의 정보를 가진 Detect 클래스가 생성된다.

블록도에서는 서로 다른 Detect에 연결되는 것처럼 보이지만 실제 코드를 살펴보면 각각의 레이어를 하나의 Head가 모두 받는것을 확인할 수 있다.

참고 사이트 https://docs.ultralytics.com/ko/yolov5/tutorials/architecture_description/#44-build-targets https://www.reddit.com/r/Ultralytics/comments/1eolwl8/the_correct_way_to_train_from_a_previously/?rdt=53319

multitask에 대한 YOLO model 구조도 yaml파일

Backbone : 모델을 사용할 때 사전에 학습되어있는 딥러닝 모델에서 특징을 추출할 때 사용되는 기반 신경망 구조.

Backbone영역에 들어가는 Layer 목록

< Conv, C2f, SPPF>

Conv(Convolution Layer)

합성곱 레이어로 이미지의 특징(feature)을 추출하는 기초 레이어.

[ k (kernel) : 커널 크기, s (stride) : 스트라이드, p (pading) = 패딩 ]

kernel에 들어가는 값들에 대해서는 초회차에 무작위로 생성된 이후 다음 학습에 대해 최적화된다.

Conv Layer는 크게 Conv2d, BatchNorm2d, SiLU로 구성되어 있다.

Conv2d

합성곱을 진행하는 파트로 사진의 원본 이미지 혹은 합성곱이 진행된 특성 맵을 입력으로

받아 합성곱을 진행하여 출력으로 내보낸다.

BatchNorm2d

Conv2d로 진행된 특성 맵을 입력으로 받고 각 채널에 대하여 평균과 분산을 계산한 후 정규화를 진행한다.

이후 정규화 된 값에 특정 Scale과 이동 값을 각각 곱하고 더한 이후 특성 맵을 출력으로 내보낸다.

(미니배치 단위로 정규화를 진행하여 Batch + Norm으로 이름이 지어졌다. 2D는 차원(이미지에 대한.))

SiLu

활성화 함수 중 하나로 특성 맵을 입력으로 받아 활성화 값을 곱해준 이후 출력으로 특성맵을 내보낸다.

SiLU vs ReLU 설명하는 사이트

Conv

class Conv(nn.Module):
    """Standard convolution with args(ch_in, ch_out, kernel, stride, padding, groups, dilation, activation)."""

    default_act = nn.SiLU()  # default activation
    #c1 : 입력 채널, c2 : 출력 채널, k : kernel_size, s : stride, p : padding, g : group, d : dilation(필터 간격), act = bias
    #default_act는 활성화 함수
    def __init__(self, c1, c2, k=1, s=1, p=None, g=1, d=1, act=True):
        """Initialize Conv layer with given arguments including activation."""
        super().__init__()
        self.conv = nn.Conv2d(c1, c2, k, s, autopad(k, p, d), groups=g, dilation=d, bias=False)
        #BatchNorm2d 노드가 bn으로 저장
        self.bn = nn.BatchNorm2d(c2)
        #act == True이면 default_act 사용, 아니라면 act로 명시된 활성화함수를 사용. act에 ReLU를 사용하면 ReLu활성화 함수 사용.
        self.act = self.default_act if act is True else act if isinstance(act, nn.Module) else nn.Identity()

    def forward(self, x):
        """Apply convolution, batch normalization and activation to input tensor."""
        return self.act(self.bn(self.conv(x)))

    def forward_fuse(self, x):
        """Perform transposed convolution of 2D data."""
        return self.act(self.conv(x))

Conv2

class Conv2(Conv):
    """Simplified RepConv module with Conv fusing."""
    #c1 : 입력 채널, c2 : 출력 채널, k : kernel_size, s : stride, p : padding, g : group, d : dilation(필터 간격), act = bias
    #Conv를 상속하여 대부분의 파라미터 값을 이어 받는다.
    def __init__(self, c1, c2, k=3, s=1, p=None, g=1, d=1, act=True):
        """Initialize Conv layer with given arguments including activation."""
        super().__init__(c1, c2, k, s, p, g=g, d=d, act=act)
        self.cv2 = nn.Conv2d(c1, c2, 1, s, autopad(1, p, d), groups=g, dilation=d, bias=False)  # add 1x1 conv

    def forward(self, x):
        """Apply convolution, batch normalization and activation to input tensor.
            두 개의 합성곱(conv와 cv2)의 결과를 더한 이후 활성화 함수를 적용."""
        return self.act(self.bn(self.conv(x) + self.cv2(x)))

    def forward_fuse(self, x):
        """Apply fused convolution, batch normalization and activation to input tensor.
            병합된 단일 합성곱의 결과를 활성화 함수에 전달 ->forward와 차이점 : """
        return self.act(self.bn(self.conv(x)))

    def fuse_convs(self):
        """Fuse parallel convolutions.
            병렬 합성곱(conv, cv)을 병합하여 최적화하고 병합 이후 forward_fuse만 수행 -> 연산량 감소 효과"""
        w = torch.zeros_like(self.conv.weight.data)
        i = [x // 2 for x in w.shape[2:]]
        w[:, :, i[0] : i[0] + 1, i[1] : i[1] + 1] = self.cv2.weight.data.clone()
        self.conv.weight.data += w
        self.__delattr__("cv2")
        self.forward = self.forward_fuse

해당 코드를 보면 Conv2와 Conv 클래스 내부에 활성화 함수를 적용시키는 것을 확인할 수 있는데 위 클래스를 각 파트별로 분해시킨 것이 Conv과 Conv2 블록이다. 사용자가 편하게 볼 수 있도록 분할하여 그린 것으로 추정.

C2f(Coordinates-To-Features)

특성맵 처리와 성능 향상 레이어. multi-scale object detection에 대한 성능을 상향. Conv, Split, Bottleneck, Concat 블록으로 이루어져 있다

Conv(Convolution Layer) 이전의 k, s, p에 이어 c가 새로 생겼는데 이는 c_out으로 출력 채널의 수를 의미. 코드의 클래스에서는 c2라는 파라미터 이름으로 들어가 있다. (특성 맵들이 서로 다른 병)

Split

특성 맵 전체를 받아 5 : 5 비율로 특성 맵을 분리시켜주는 블록. 전체 특성 맵(차원)이 128이면 64 : 64로 특성맵을 분리시켜 출력한다. Split이후 라에서 h * w * 0.5c_out으로 표현되는 것을 확인 할 수 있다. C2f의 일부 레이어에서 n = 6d, n = 3d로 나와있는 것이 있는데 이는 후의 bottleneck의 횟수를 정하는 변수이다.

Bottlenect

특성 맵 처리를 담당하는 블록. 입력 특징 맵을 압축하고 확장하는 방식을 통해 연산 결과를 병합한다. 내부 shortcut이라는 파라미터를 확인할 수 있는데 이는 Bottleneck 블록을 수행할 때 사용하는 파라미터로 shortcut이 True이면 입력 값을 따로 보관한 Conv 블록을 거친 특성맵과 Concat블록에서 합쳐지고(이어짐) shortcut이 False면 원본을 보관하지 않고 바로 Conv 블록을 거쳐 특성 맵을 출력한다. 처음 Conv 블록에서 채널 수를 축소하여 출력하는 이유는 모델의 계산에 있어 간결하게 하기 위함이며 불필요하다고 판단되는 특성들을 줄이고 중요하다고 판단되는 특징맵의 수를 늘리는 작업을 수행한다.

class Bottleneck(nn.Module):
    """Standard bottleneck."""

    def __init__(self, c1, c2, shortcut=True, g=1, k=(3, 3), e=0.5):
        """Initializes a bottleneck module with given input/output channels, shortcut option, group, kernels, and
        expansion.
        """
        super().__init__()
        c_ = int(c2 * e)  # hidden channels
        self.cv1 = Conv(c1, c_, k[0], 1)
        self.cv2 = Conv(c_, c2, k[1], 1, g=g)
        self.add = shortcut and c1 == c2

    def forward(self, x):
        """'forward()' applies the YOLO FPN to input data."""
        return x + self.cv2(self.cv1(x)) if self.add else self.cv2(self.cv1(x))

Backbone의 4번 6번 9번 레이어를 보게 된다면 같은 출력값을 가지고 있으나 Concat으로 연결되며 Stride가 추가적으로 적힌 블록도를 확인할 수 있다.

해당 블록도에서 Stride는 일반적인 의미로 사용되는(필터간 간격) Stride가 아닌 다운샘플링이 진행된 총 비율로 4번 레이어에서는 총 8배의 다운샘플링이 진행되었다는 의미이다.

원본 이미지 640640에 대해서 1/2의 다운샘플링 과정이 총 8배가 되었다는 의미로 Conv레이어를 총 3번 거치며 22*2에 대해 640 / 8 = 80이 출력되었다는 것을 확인시켜주는 보조의 의미를 가진다.

이후에 각각 Conv 레이어를 거치며 Stride의 값이 222*2로 6번 레이어에서는 총 16배의 다운샘플링이 진행되었고 9번 레이어에서는 32배의 다운샘플링이 진행되었다는 것을 확인할 수 있다.

6번 레이어 -> 640 / 16 = 40, 8번 레이어 -> 640 / 32 = 20

from ultralytics import YOLO
import multiprocessing
from ultralytics import settings
import matplotlib.pyplot as plt


model = YOLO("yolov10s-seg.yaml")

print("4번 레이어")
print(model.model.model[4])

print("6번 레이어")
print(model.model.model[6])

print("9번 레이어")
print(model.model.model[9])

import torch

model = YOLO("yolov10s-seg.yaml")

dummy_input = torch.randn(1, 3, 640, 640)

x = dummy_input
for idx, layer in enumerate(model.model.model):
    x = layer(x)
    if idx in [4, 6, 9]:  # C2f layers
        print(f"Layer {idx}: Output shape {x.shape}")

Concat

여러개의 레이어를 하나로 합치는 과정으로 같은 채널 크기의 특성맵들을 입력으로 받아 특성맵끼리 이어 하나의 특성 맵을 출력으로 내보낸다.

SPPF(Spatial Pyramid Pooling - Fast)

Conv, MaxPool2d, Concat 블록으로 이루어진 블록.

여러 크기의 특성들에(작은 특성, 중간 특성, 큰 특성) 대해 정보를 결합하여 일반화 된 특성 맵을 제공.

MaxPool2d : Pooling작업을 수행할 때 사용하는 기법

YOLO에서는 AveragePooling이 아닌 MaxPooling기법을 사용함.

Pooling작업을 수행하면 사이즈가 달라져서 Concat 블록에서 합칠 수 없다고 생각할 수 있지만 코드 내부에서 진행할 때 padding작업을 진행해줌으로서 크기에 대한 문제를 해결한다.

class SPPF(nn.Module):
    """Spatial Pyramid Pooling - Fast (SPPF) layer for YOLOv5 by Glenn Jocher."""

    def __init__(self, c1, c2, k=5):
        """
        Initializes the SPPF layer with given input/output channels and kernel size.
        This module is equivalent to SPP(k=(5, 9, 13)).
        레이어 초기화 등등. k는 MaxPooling을 진행할 때 사용하는 커널의 크기.
        """
        super().__init__()
        c_ = c1 // 2  # hidden channels
        self.cv1 = Conv(c1, c_, 1, 1)
        self.cv2 = Conv(c_ * 4, c2, 1, 1)
        self.m = nn.MaxPool2d(kernel_size=k, stride=1, padding=k // 2)

    def forward(self, x):
        """Forward pass through Ghost Convolution block. 순전파"""
        y = [self.cv1(x)]
        y.extend(self.m(y[-1]) for _ in range(3))
        return self.cv2(torch.cat(y, 1))
        class SPPF(nn.Module):
    """Spatial Pyramid Pooling - Fast (SPPF) layer for YOLOv5 by Glenn Jocher."""

    def __init__(self, c1, c2, k=5):
        """
        Initializes the SPPF layer with given input/output channels and kernel size.
        This module is equivalent to SPP(k=(5, 9, 13)).
        레이어 초기화 등등. k는 MaxPooling을 진행할 때 사용하는 커널의 크기.
        """
        super().__init__()
        c_ = c1 // 2  # hidden channels
        self.cv1 = Conv(c1, c_, 1, 1)
        self.cv2 = Conv(c_ * 4, c2, 1, 1)
        self.m = nn.MaxPool2d(kernel_size=k, stride=1, padding=k // 2)

    def forward(self, x):
        """Forward pass through Ghost Convolution block. 순전파"""
        y = [self.cv1(x)]
        y.extend(self.m(y[-1]) for _ in range(3))
        return self.cv2(torch.cat(y, 1))

관련 글 https://github.com/ultralytics/ultralytics/issues/9042 https://github.com/ultralytics/ultralytics/issues/15596 https://blog.roboflow.com/what-is-yolov8/#anchor-free-detection https://github.com/ultralytics/ultralytics/issues/3678 https://github.com/ultralytics/ultralytics/issues/13441

<1-stage detector, 2-stage detector>

1-stage 방식의 대표적인 모델은 YOLO 시리즈와 Retina-Net, SSD 등이 있다.

1-stage 방식이란 Regional Proposal과 Classification이 CNN을 통해 동시에 이루어지는 방식으로 Convolution Layer을 통해 Feature Maps가 생성되면 Output으로 Multi-Class Classification과 Bounding Box Regression을 출력한다.

위 방식의 구조는 Anchor Boxes(앵커 박스)를 찾게되는데 Anchor box란 중심 좌표를 기준으로 여러 크기와 비율을 가지고 생성된 영역이다.

Anchor box 관련 글

돌아와 YOLO 시리즈에서 loss값을 계산하여 학습시키는 모듈은 ultralytics/nn/tasks.py 으로 해당 모듈은 기본적으로 다른 Detection, Segmentation, Poss 등의 작업에 대한 기초 클래스 역할을 하는 BaseModel이 있다.

class BaseModel 은 모델의 기초로서 YOLOv8의 공통적인 레이어 구성과 파라미터 초기화를 초기화하여 이후에 이를 상속받는 모델들에 있어 기초 토대를 만들어준다.

기초 토대를 구성해 준다는 것은 손실 값 계산, 모델의 가중치 로드, 순방향 패스의 수행에 있어 입력값과 출력값 그 사이에 필요한 파라미터(특성 맵을 저장할지에 대한 여부)등 이 있다.

class DetectionModel(BaseModel)
        # Define model
        ch = self.yaml["ch"] = self.yaml.get("ch", ch)  # input channels
        if nc and nc != self.yaml["nc"]:
            LOGGER.info(f"Overriding model.yaml nc={self.yaml['nc']} with nc={nc}")
            self.yaml["nc"] = nc  # override YAML value
        self.model, self.save = parse_model(deepcopy(self.yaml), ch=ch, verbose=verbose)  # model, savelist
        self.names = {i: f"{i}" for i in range(self.yaml["nc"])}  # default names dict
        self.inplace = self.yaml.get("inplace", True)
        self.end2end = getattr(self.model[-1], "end2end", False)

def loss(self, batch, preds=None):
        """
        Compute the loss for the given batch of data.
        Args:
            batch (dict): Dictionary containing image and label data.
            preds (torch.Tensor, optional): Precomputed model predictions. Defaults to None.
        Returns:
            (tuple): A tuple containing the total loss and main three losses in a tensor.
        """
        if not hasattr(self, "criterion"):
            self.criterion = self.init_criterion()

        img = batch["img"]
        # NOTE: preprocess gt_bbox and gt_labels to list.
        bs = len(img)
        batch_idx = batch["batch_idx"]
        gt_groups = [(batch_idx == i).sum().item() for i in range(bs)]
        targets = {
            "cls": batch["cls"].to(img.device, dtype=torch.long).view(-1),
            "loc": batch["loc"].to(img.device, dtype=torch.long).view(-1),
            "action": batch["action"].to(img.device, dtype=torch.long).view(-1),
            "bboxes": batch["bboxes"].to(device=img.device),
            "batch_idx": batch_idx.to(img.device, dtype=torch.long).view(-1),
            "gt_groups": gt_groups,
        }

        preds = self.predict(img, batch=targets) if preds is None else preds
        dec_bboxes, dec_scores, enc_bboxes, enc_scores, dn_meta = preds if self.training else preds[1]
        if dn_meta is None:
            dn_bboxes, dn_scores = None, None
        else:
            dn_bboxes, dec_bboxes = torch.split(dec_bboxes, dn_meta["dn_num_split"], dim=2)
            dn_scores, dec_scores = torch.split(dec_scores, dn_meta["dn_num_split"], dim=2)

        dec_bboxes = torch.cat([enc_bboxes.unsqueeze(0), dec_bboxes])  # (7, bs, 300, 4)
        dec_scores = torch.cat([enc_scores.unsqueeze(0), dec_scores])

        loss = self.criterion(
            (dec_bboxes, dec_scores), targets, dn_bboxes=dn_bboxes, dn_scores=dn_scores, dn_meta=dn_meta
        )
        # NOTE: There are like 12 losses in RTDETR, backward with all losses but only show the main three losses.
        return sum(loss.values()), torch.as_tensor(
            [loss[k].detach() for k in ["loss_giou", "loss_class", "loss_bbox"]], device=img.device
        )

참고할 사이트

ultralytics/models/yolo/detect/train.py, val.py, predict.py

해당 모듈에서는 모델이 학습할 때 쓰이는 loss값이 어떻게 설정되고 계산하는지 포함.

손실 항목은 self.loss_names에 저장.

self.loss_names = "box_loss", "cls_loss", "dfl_loss"

https://github.com/kCW-tb/complex_detection/blob/main/ultralytics/models/yolo/detect/train.py

YOLO dataset document

DATASET.YAML

YOLO model에서 데이터셋 호출을 위해서 설정하는 파일

위치 : ultralytics/ultralytics/cfg/dataset/custum.yaml 데이터셋이 포함되어있는 전체 데이터셋 폴더 위치, path와 내부 학습, 검증, 테스트 데이터셋의 세부 경로를 포함.

ultralytics의 coco8.yaml.
# Train/val/test sets
# 1) dir: path/to/imgs, 2) file: path/to/imgs.txt, or 3) list: [path/to/imgs1, path/to/imgs2, ..]
path: ../datasets/coco8 # dataset root dir
train: images/train # train images
val: images/val # val images
test: # test images

경로를 지정하는 방법으로는 3가지가 존재한다.

(경로를 지정할 떄 lables가 기준이 아닌 images가 기준이 된다.)

1) 경로 : 데이터셋이 들어가 있는 디렉토리의 경로를 지정. (절대경로와 상대경로 모두 가능하다.)

2) 파일 : 이미지의 경로를 나열한 파일을 지정하는 것으로 train의 경우 ../dataset/train/img1.jpg, ../dataset/train/img2.jpg처럼 하나의 txt파일 내부에 각 이미지의 경로를 나열하는 방식으로 저장 후 txt의 경로를 지정한다.

3) 리스트 : 이미지 하나하나의 경로를 직접 적어주는 방식 ["/dataset/train/img1.jpg", "dataset/train/img2.jpg"]를 train:이후 적어주어 이미지 각각을 선택한다.

GlobalWheat2020.yaml 파일이 동일 dataset경로에 있는데 해당 데이터셋은 아래와 같이 하나의 dataset폴더에 모든 데이터셋을 넣고 내부 폴더로 분할시켜주었다.

path: ../datasets/GlobalWheat2020 # dataset root dir
train: # train images
  - images/arvalis_1
  - images/arvalis_2
  - images/arvalis_3
  - images/ethz_1
  - images/rres_1
  - images/inrae_1
  - images/usask_1
val: # val images
  - images/ethz_1
test: # test images
  - images/utokyo_1
  - images/utokyo_2
  - images/nau_1
  - images/uq_1

해당 방식은 리스트를 이용한 방식으로 yaml형식의 파일은 ' : ' 이 변수 명 이후에 적혀있다면 이후 ' - '를 통해 리스트를 생성할 수 있다.

train을 python형식으로 고친다면 train["images/arvalis_1", "images/arvalis_2", "images/arvalis_3"] 과 동일하게 볼 수 있다.

데이터셋 클래스 설정

데이터셋의 경로 설정 이후 해당 데이터셋에 대한 클래스를 지정해줘야 한다.

'names'의 변수 명을 가진 리스트 형태로 작성되며 일반적으로는 해당 클래스 선언 이후 yaml 파일이 끝이난다.

eypoint는 Classes의 일부로 객체 탐지에서 "호랑이"로 추측되는 객체가 존재한다면 해당 객체에 대해 신체 부위 포인트를 지정하기 위해서 Keypoint가 추가적으로 사용되는 데이터이다.

데이터셋 다운로드

대부분의 유저들은 처음 공부하거나 기존 코드를 사용하려 할 때 데이터셋을 미리 구현해두고 코드를 다운받지 않고 코드를 다운로드 받고 이에 해당하는 데이터셋을 찾기 때문에 대중적이거나 공개된 데이터셋의 경우 yaml파일을 읽어들이면서 존재하지 않는다면 파일을 다운로드 받을 수 있게 한다.

내부에 형식이 잘 정리되어있는 경우에는 아래와 같이 링크 하나만 존재하기도 하며

{ data8.yaml }

내부에 형식이 추가적으로 필요한 데이터셋의 경우 추가적인 작업을 진행해주기도 한다

# Download script/URL (optional)
download: https://github.com/ultralytics/assets/releases/download/v0.0.0/dota8.zip```

```# Download script/URL (optional) ------------------------------------------------------------------
download: |
  from ultralytics.utils.downloads import download
  from pathlib import Path

  # Download
  dir = Path(yaml['path'])  # dataset root dir
  urls = ['https://zenodo.org/record/4298502/files/global-wheat-codalab-official.zip',
          'https://github.com/ultralytics/assets/releases/download/v0.0.0/GlobalWheat2020_labels.zip']
  download(urls, dir=dir)

  # Make Directories
  for p in 'annotations', 'images', 'labels':
      (dir / p).mkdir(parents=True, exist_ok=True)

  # Move
  for p in 'arvalis_1', 'arvalis_2', 'arvalis_3', 'ethz_1', 'rres_1', 'inrae_1', 'usask_1', \
           'utokyo_1', 'utokyo_2', 'nau_1', 'uq_1':
      (dir / 'global-wheat-codalab-official' / p).rename(dir / 'images' / p)  # move to /images
      f = (dir / 'global-wheat-codalab-official' / p).with_suffix('.json')  # json file
      if f.exists():
          f.rename((dir / 'annotations' / p).with_suffix('.json'))  # move to /annotations

DATALOADER

DATALOADER-dataset.py

YOLO는 데이터셋 호출시 공통적으로 class YOLODataset을 사용한다.

해당 관련 코드는 ultralytics/ultralytics/cfg/data/dataset.py에 있다.

생성자에 대한 초기화는 다음과 같이 이루어진다.

def __init__(self, *args, data=None, task="detect", **kwargs):
        self.use_segments = task == "segment"
        self.use_keypoints = task == "pose"
        self.use_obb = task == "obb"
        self.data = data
        assert not (self.use_segments and self.use_keypoints), "Can not use both segments and keypoints."
        super().__init__(*args, **kwargs)

head의 속성에 따라 'segment', 'keypoint', 'obb'에 대한 속성을 파악하고 data는 자체적으로 받아들인다.

head의 속성을 입력하지 않는다면 'detect'으로 자동 설정된다.

다음 cache_lables 함수는 데이터셋 중 labeles 정보를 읽을 때 쓰이는 함수로 txt파일을 인덱스 형식으로 처리하여 정보를 읽는 방식을 설정한다.

일반	multitask(A-YOLOM)

우측의 그림은 detection과 segmentation작업을 모두 수행하는 A-YOLOM코드에 대한 정보이다.

DATALOADER-augment.py

(프로젝트 코드로 진행 - loc와 action만 제거하면 원본 코드와 거의 동일하다, 다른 부분은 체크하고 넘길 예정.)

데이터 증식 관련된 함수들과 클래스가 모여있는 장소로 mosaic와 mixup 데이터 증식 기법부터 포함하여 픽셀을 건드리는 수준의 데이터 증식 기법과 공간적 측면을 건드리는 데이터 증식 기법을 가지고 있다.

데이터 증식에 대해서 직접적으로 건드리는 클래스는 class Albumentations이고 이 클래스가 가지고 있는 데이터 증식 기법들은 아래 하단의 사이트에 정리되어있다. Albumentations Documentation

Albumentations는 공간적 속성을 변경하는 spatial_transforms과 pixel에 대해 값 똑은 색상 등을 변환하는 데이터 증식 기법들에 대해 Compose 클래스에 정리한다.

spatial_transforms에 나열되어있는 집합의 원소들은 사용 가능한 데이터 증식 기법들을 나타낸것이다.

def __init__(self, p=1.0):
        """Initialize the transform object for YOLO bbox formatted params."""
        self.p = p
        self.transform = None
        prefix = colorstr("albumentations: ")

        try:
            import albumentations as A

            check_version(A.__version__, "1.0.3", hard=True)  # version requirement

            # List of possible spatial transforms
            spatial_transforms = {
                "Affine",
                "BBoxSafeRandomCrop",
                "CenterCrop",
                "CoarseDropout",
                "Crop",
                "CropAndPad",
                "CropNonEmptyMaskIfExists",
                "D4",
                "ElasticTransform",
                "Flip",
                "GridDistortion",
                "GridDropout",
                "HorizontalFlip",
                "Lambda",
                "LongestMaxSize",
                "MaskDropout",
                "MixUp",
                "Morphological",
                "NoOp",
                "OpticalDistortion",
                "PadIfNeeded",
                "Perspective",
                "PiecewiseAffine",
                "PixelDropout",
                "RandomCrop",
                "RandomCropFromBorders",
                "RandomGridShuffle",
                "RandomResizedCrop",
                "RandomRotate90",
                "RandomScale",
                "RandomSizedBBoxSafeCrop",
                "RandomSizedCrop",
                "Resize",
                "Rotate",
                "SafeRotate",
                "ShiftScaleRotate",
                "SmallestMaxSize",
                "Transpose",
                "VerticalFlip",
                "XYMasking",
            }  # from https://albumentations.ai/docs/getting_started/transforms_and_targets/#spatial-level-transforms

            # Transforms
            T = [
                A.Blur(p=0.01),
                A.MedianBlur(p=0.01),
                A.ToGray(p=0.01),
                A.CLAHE(p=0.01),
                A.RandomBrightnessContrast(p=0.0),
                A.RandomGamma(p=0.0),
                A.ImageCompression(quality_lower=75, p=0.0),
            ]

            # Compose transforms
            self.contains_spatial = any(transform.__class__.__name__ in spatial_transforms for transform in T)
            self.transform = (
                A.Compose(T, bbox_params=A.BboxParams(format="yolo", label_fields=["class_labels"]))
                if self.contains_spatial
                else A.Compose(T)
            )
            LOGGER.info(prefix + ", ".join(f"{x}".replace("always_apply=False, ", "") for x in T if x.p))
        except ImportError:  # package not installed, skip
            pass
        except Exception as e:
            LOGGER.info(f"{prefix}{e}")

사용자는 T로 설정되어 있는 리스트에 원하는 증식 기법을 설정해줄 수 있다.

# Transforms
T = [
    A.Blur(p=0.01),
    A.MedianBlur(p=0.01),
    A.ToGray(p=0.01),
    A.CLAHE(p=0.01),
    A.RandomBrightnessContrast(p=0.0),
    A.RandomGamma(p=0.0),
    A.ImageCompression(quality_lower=75, p=0.0),
]

이후 해당 클래스에서 Compose내역을 확인하게 되는데 Compose 즉 기본 데이터 증식(변환)을 제외하고 사용자가 설정하는 T 리스트를 읽어 적용되는 증식 기법을 확인하게 된다.

# Compose transforms
self.contains_spatial = any(transform.__class__.__name__ in spatial_transforms for transform in T)
self.transform = (
    A.Compose(T, bbox_params=A.BboxParams(format="yolo", label_fields=["class_labels"]))
    if self.contains_spatial
    else A.Compose(T)
)

self.contains_spatial은 any내부의 for in문을 통해 T의 리스트중 spatial_transfoms와 동일한 이름을 가진 데이터 증식 기법이 있는지 확인하게 되고 존재한다면 contains_spatial값을 True로 반환하고 존재하지 않다면 False를 반환한다.

이후 self.transform에서 공간적 변환이 포함된 경우에는 BboxParams를 통해 labels 데이터셋에 있는 segmentation 픽셀의 값이나 Rounding Box의 위치 정보를 이미지와 동일하게 변환해준다. (공간적 증식 이후 픽셀값관련 데이터 증식 진행.)

이후 호출문으로 Compose내역에 따라 데이터 증식을 적용시킨 이후 labels 데이터를 반환한다.

def __call__(self, labels):
        """Generates object detections and returns a dictionary with detection results."""
        if self.transform is None or random.random() > self.p:
            return labels

        if self.contains_spatial:
            cls = labels["cls"]
            loc = labels["loc"]
            action = labels["action"]
            if len(cls):
                im = labels["img"]
                labels["instances"].convert_bbox("xywh")
                labels["instances"].normalize(*im.shape[:2][::-1])
                bboxes = labels["instances"].bboxes
                concatenated_labels = np.concatenate([cls[:, None], loc[:, None], action[:, None]], axis=1)

                # TODO: add supports of segments and keypoints
                new = self.transform(image=im, bboxes=bboxes, class_labels=concatenated_labels)  # transformed
                if len(new["class_labels"]) > 0:  # skip update if no bbox in new im
                    labels["img"] = new["image"]
                    transformed_labels = np.array(new["class_labels"])
                    labels["cls"] = transformed_labels[:, 0]
                    labels["loc"] = transformed_labels[:, 1]
                    labels["action"] = transformed_labels[:, 2:]

                    bboxes = np.array(new["bboxes"], dtype=np.float32)
                labels["instances"].update(bboxes=bboxes)
        else:
            labels["img"] = self.transform(image=labels["img"])["image"]  # transformed

        return labels

인덱스 영역에 대해서 이야기를 하자면 넘파이 형식으로 되어있는 2차원 행렬이 있는 상태를 기반으로 했을 때 다음과 같은 형태를 가지고 있다. transformed_lables가 2차원 행렬 형태이며 {0, 1, 2 : }는 각각에 대한 정보를 가지고 있는 형태.

객체검출 외 분할작업과 추가적으로 위치와 후미등 상태 여부도 파악하는 action의 경우 transformed_labels에 넘파이 형태로 넣은 다음 인덱스 형태로 적용시켰으나 원본 코드에서는 Rounding Box정보와 클래스의 정보만 있기에 바로 적용시킨 것을 확인할 수 있다.

if self.transform is None or random.random() > self.p:
    return labels

if self.contains_spatial:
    cls = labels["cls"]
    if len(cls):
        im = labels["img"]
        labels["instances"].convert_bbox("xywh")
        labels["instances"].normalize(*im.shape[:2][::-1])
        bboxes = labels["instances"].bboxes
        # TODO: add supports of segments and keypoints
        new = self.transform(image=im, bboxes=bboxes, class_labels=cls)  # transformed
        if len(new["class_labels"]) > 0:  # skip update if no bbox in new im
            labels["img"] = new["image"]
            labels["cls"] = np.array(new["class_labels"])
            bboxes = np.array(new["bboxes"], dtype=np.float32)
        labels["instances"].update(bboxes=bboxes)
else:
    labels["img"] = self.transform(image=labels["img"])["image"]  # transformed

return labels

위처럼 구현된 Albumentation 클래스는 같은 모듈의 v8_transforms 클래스를 통해 Mosaic, CopyPaste, RandomPerspective, 등과 함께 Compose 로 구성되어 반환되고 v8_transforms 클래스는 dataset.py의 YOLODataset클래스 내부의 build_transforms 함수에서 선언되어 구현된다.

def build_transforms(self, hyp=None):
    """Builds and appends transforms to the list."""
    if self.augment:
        hyp.mosaic = hyp.mosaic if self.augment and not self.rect else 0.0
        hyp.mixup = hyp.mixup if self.augment and not self.rect else 0.0
        transforms = v8_transforms(self, self.imgsz, hyp)
    else:
        transforms = Compose([LetterBox(new_shape=(self.imgsz, self.imgsz), scaleup=False)])
    transforms.append(
        Format(
            bbox_format="xywh",
            normalize=True,
            return_mask=self.use_segments,
            return_keypoint=self.use_keypoints,
            return_obb=self.use_obb,
            batch_idx=True,
            mask_ratio=hyp.mask_ratio,
            mask_overlap=hyp.overlap_mask,
            bgr=hyp.bgr if self.augment else 0.0,  # only affect training.
        )
    )
    return transforms

yaml

yaml은 YAML Ain't Markup Language로 마크업 단어가 아니라는 정직한 이름을 가지고 있다.

yaml은 기존의 xml이나 json처럼 역할은 같지만 보다 사용자 친화적임에 중점을 두고 있다.

기존의 xml이나 json은 중괄호 대괄호 등에 각각의 역할이 주어지는 반면 yaml은 들여쓰기 하나로 해결해주어 구조 파악 등에 도움을 주고 주석 추가 여부도 다양한 사람이 입문하기 좋은 역할을 하여 YOLO에서 쓰이는 것으로 추정된다.

json같은 경우 labels의 역할도 이미지의 경로와 함께 저장하여 특정 YOLO모델에서 쓰이나 yaml을 사용하는 모델의 대부분인 것은 편의성과 기존 라이브러리 호출에 있어 용이하기 때문에 yaml을 주로 사용한다.

데이터셋의 입력 사이즈의 경우 ultralytics/cfg/default.py에서 설정할수 있다.(명령행인자로 직접 줄 수도 있다.)

# Ultralytics YOLO 🚀, AGPL-3.0 license
# Default training settings and hyperparameters for medium-augmentation COCO training

task: segment # (str) YOLO task, i.e. detect, segment, classify, pose
mode: train # (str) YOLO mode, i.e. train, val, predict, export, track, benchmark

# Train settings -------------------------------------------------------------------------------------------------------
model: # (str, optional) path to model file, i.e. yolov8n.pt, yolov8n.yaml
data: # (str, optional) path to data file, i.e. coco8.yaml
epochs: 100 # (int) number of epochs to train for
time: # (float, optional) number of hours to train for, overrides epochs if supplied
patience: 100 # (int) epochs to wait for no observable improvement for early stopping of training
batch: 32 # (int) number of images per batch (-1 for AutoBatch)
imgsz: 640 # (int | list) input images size as int for train and val modes, or list[w,h] for predict and export modes

해당 모듈을 들여다보면 imgsz라는 변수가 존재하는데 해당 자료형에는 하나의 int형 변수와 list를 넣을 수 있다.

int형 변수를 넣게 된다면 해당 설정한 값이 가장 큰 치수를 기준으로 기존의 이미지의 크기를 변경하여 데이터셋을 구성하고 list 자료형을 넣게 된다면 해당 리스트에 맞게 규격을 맞춰준다.

예시를 들어 원본 1280720에 대해 imgsz를 640으로 넣는다면 데이터셋은 640360으로 이미지를 전처리하여 구성한다.

imgsz를 리스트 형태로 [640*480]으로 구성한다면 원본 이미지가 해당 규격(w,h)에 맞게 설정된다.

만일 이미지의 크기를 정사각형 형태로 640x640으로 구성하고 싶다면 리스트를 [640x640]으로 입력하여도 가능하지만 또 다른 변수 rect를 False에서 True로 변환시켜주는 방법도 있다 해당 변수가 True라면 640으로 입력하여도 정사각형 형태로 변환한다.

rect: False # (bool) rectangular training if mode='train' or rectangular validation if mode='val

windows환경에서 학습을 돌리며 multiprocessing관련 문제가 생겨 num_workers의 값을 0으로 했던 것이 문제였으며 multi-processing.freeze_support()함수를 추가해줌으로 해결

from ultralytics import YOLO
import multiprocessing

if __name__=="__main__":
    multiprocessing.freeze_support()

    model = YOLO("yolov10s-seg.yaml")
    # 배치 크기 8 
    # multiprocession.freeze_support()을 쓰지 않는다면 num_worker 0 고정.
    results = model.train(data="Compete_segment.yaml",pretrained='yolov8s-seg.pt',epochs=100, device=[0], workers=4, batch=8)

num_workers은 학습과정에서 데이터를 불러오는 역할을 수행할 때 사용하는 파라미터로 GPU의 연산 속도를 활용할 수 있도록 CPU의 데이터 전송 속도를 높이는 것이다.

num_workers의 수가 낮다면 GPU에 전송되는 데이터가 늦어 정상적으로 운용을 할 수 없으며 이 경우를 CPU I/O bottleneck현상이라고 하며 num_workers의 수를 높이게 되면 해결된다.

1 Epoch를 진행할 때 위 사진처럼 CPU가 데이터를 GPU에 전송해주는 과정이 존재.

num_workers의 값을 낮게 설정하면 붉은색 선과 같이 시간이 오래 걸려 사이간 GPU를 사용하지 않게 되고 num_workers의 값을 적절히 설정하면 보라색 선과 같이 공백의 시간이 적어 GPU를 온전히 사용할 수 있게 된다. CPU와 GPU에서 num_workers에 대한 설명 글

nvidia-smi

학습을 수행할 때에는 붉은색 박스 내의 정보처럼 1 Epoch를 제외하고선 학습에 5~6분가량을 사용하고 평가지표에 1분정도를 소요함을 알 수 있다.

환경 구성.

중요 환경에 대한 버전.

Package             Version
------------------- --------------------
numpy               1.24.1
opencv-python       4.12.0.88
thop                0.1.1.post2209072238
torch               2.3.1+cu118
torchaudio          2.3.1+cu118
torchvision         0.18.1+cu118

진행 작업

1. 폴더별 가중치만 적용하여 학습.
2. 클래스별 가중치만 적용하여 학습.

class SegmentationTransform:
    def __init__(self, crop_size=[1024, 1024], scale_range=[0.5, 1.5]):
        self.crop_size = crop_size
        self.scale_range = scale_range
        self.mean = [0.485, 0.456, 0.406]
        self.std = [0.229, 0.224, 0.225]
        self.bilinear = transforms.InterpolationMode.BILINEAR
        self.nearest = transforms.InterpolationMode.NEAREST

        # Color Jitter
        self.color_jitter = transforms.ColorJitter(brightness=0.3, contrast=0.3, saturation=0.3, hue=0.1)
        # Gaussian Blur
        self.gaussian_blur = transforms.GaussianBlur(kernel_size=(5, 9), sigma=(0.1, 5))

    def __call__(self, image, label):
        scale_factor = random.uniform(self.scale_range[0], self.scale_range[1])
        width, height = image.size
        new_width, new_height = int(width * scale_factor), int(height * scale_factor)
        image = TF.resize(image, (new_height, new_width), interpolation=self.bilinear)
        label = TF.resize(label, (new_height, new_width), interpolation=self.nearest)

        pad_h = max(self.crop_size[0] - new_height, 0)
        pad_w = max(self.crop_size[1] - new_width, 0)
        if pad_h > 0 or pad_w > 0:
            padding = (0, 0, pad_w, pad_h)
            image = TF.pad(image, padding, fill=0)
            label = TF.pad(label, padding, fill=255)

        # 크롭
        i, j, h, w = transforms.RandomCrop.get_params(image, output_size=self.crop_size)
        image = TF.crop(image, i, j, h, w)
        label = TF.crop(label, i, j, h, w)

        # 좌우 반전
        if random.random() > 0.3:
            image = TF.hflip(image)
            label = TF.hflip(label)

        # 회전
        if random.random() > 0.5:
            angle = random.uniform(-5, 5)
            image = TF.rotate(image, angle, interpolation=self.bilinear, fill=0)
            label = TF.rotate(label, angle, interpolation=self.nearest, fill=255)

        # 색상 변환
        if random.random() > 0.4:
            image = self.color_jitter(image)
        if random.random() > 0.3:
            image = self.gaussian_blur(image)

        image = TF.to_tensor(image)
        image = TF.normalize(image, mean=self.mean, std=self.std)
        label = torch.from_numpy(np.array(label, dtype=np.uint8)).long()

train.py

import os
import argparse
import torch
from torch.utils.data import DataLoader, WeightedRandomSampler
from tqdm import tqdm
from collections import OrderedDict
import json
from pathlib import Path

from DDRNet import DDRNet
from functions import *

def arg_as_dict(s):
    try:
        return json.loads(s)
    except Exception as e:
        raise argparse.ArgumentTypeError(f"Argument must be a JSON-formatted dictionary string. Error: {e}")

def train_and_validate(args):
    device = torch.device(f"cuda:{args.gpu_id}" if torch.cuda.is_available() else "cpu")
    print(f"Initialized training on device: {device}")

    # 데이터셋 폴더별 가중치를 위한 설정
    train_sub_folders = ['cam0', 'cam1', 'cam2', 'cam3', 'cam4', 'cam5', 'set1', 'set2', 'set3']
    val_sub_folders = ['cam0', 'cam1', 'cam2', 'cam3', 'cam4', 'cam5', 'set1', 'set2', 'set3']

    train_dataset = SegmentationDataset(args.dataset_dir, args.crop_size, 'train', args.scale_range, sub_folders=train_sub_folders)

    # --- 폴더별 가중치에 따른 샘플링 확률 계산 ---
    if args.folder_weights:
        print("Applying folder-wise weights for sampling...")
        folder_indices = [sample[1] for sample in train_dataset.samples]
        folder_names_per_sample = [train_sub_folders[i] for i in folder_indices]
        sample_weights = [args.folder_weights.get(name, 1.0) for name in folder_names_per_sample]
        print(f"Sample weights will be based on folder weights: {args.folder_weights}")
        sampler = WeightedRandomSampler(weights=sample_weights, num_samples=len(sample_weights), replacement=True)
        shuffle = False
    else:
        sampler = None
        shuffle = True

    val_dataset = SegmentationDataset(args.dataset_dir, args.crop_size, 'val', args.scale_range, sub_folders=val_sub_folders)

    train_dataloader = DataLoader(train_dataset, batch_size=args.batch_size, shuffle=shuffle, sampler=sampler, num_workers=args.num_workers, pin_memory=True, drop_last=True)
    val_dataloader = DataLoader(val_dataset, batch_size=args.batch_size, shuffle=False, num_workers=args.num_workers, pin_memory=True)

    model = DDRNet(num_classes=args.num_classes).to(device)

    class_weights = None
    if args.class_weights:
        if len(args.class_weights) != args.num_classes:
            raise ValueError(f"Number of class_weights ({len(args.class_weights)}) must match num_classes ({args.num_classes})")
        print(f"Applying class weights: {args.class_weights}")
        class_weights = torch.tensor(args.class_weights, dtype=torch.float).to(device)

    if args.use_ohem:
        print("Using OhemCrossEntropy Loss.")
        criterion = OhemCrossEntropy(ignore_label=255, weight=class_weights)
    else:
        print("Using standard CrossEntropy Loss.")
        criterion = CrossEntropy(ignore_label=255, weight=class_weights)

    optimizer = torch.optim.SGD(model.parameters(), lr=args.lr, momentum=args.momentum, weight_decay=args.weight_decay)
    scheduler = WarmupPolyEpochLR(optimizer, total_epochs=args.epochs, warmup_epochs=args.warmup_epochs)

    start_epoch = 0
    min_val_loss = float('inf')
    if args.loadpath:
        print(f"Loading checkpoint from: {args.loadpath}")
        checkpoint = torch.load(args.loadpath, map_location=device)
        try:
            # DDP 학습 가중치('module.' 접두사) 호환을 위한 처리
            new_state_dict = OrderedDict()
            for k, v in checkpoint['model_state_dict'].items():
                name = k[7:] if k.startswith('module.') else k
                new_state_dict[name] = v
            model.load_state_dict(new_state_dict, strict=False)

            optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
            scheduler.load_state_dict(checkpoint['scheduler_state_dict'])
            start_epoch = checkpoint['epoch'] + 1
            min_val_loss = checkpoint.get('loss', float('inf'))
            print(f"Resuming training from epoch {start_epoch}, with min_val_loss: {min_val_loss:.4f}")
        except KeyError:
            print("Old checkpoint format. Loading model state_dict only.")
            load_state_dict(model, checkpoint)

    os.makedirs(args.result_dir, exist_ok=True)
    log_path = os.path.join(args.result_dir, "log.txt")
    with open(log_path, 'a' if start_epoch > 0 else 'w') as f:
        if start_epoch == 0:
            f.write("Epoch\t\tTrain-loss\t\tVal-loss\t\tlearningRate\n")

    for epoch in range(start_epoch, args.epochs):
        model.train()
        total_train_loss = 0.0
        loop = tqdm(train_dataloader, desc=f"Train [{epoch+1}/{args.epochs}]", ncols=100)

        for i, (imgs, labels) in enumerate(loop):
            optimizer.zero_grad(set_to_none=True)
            imgs, labels = imgs.to(device), labels.to(device)
            outputs = model(imgs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            total_train_loss += loss.item()
            loop.set_postfix(loss=loss.item(), avg_loss=total_train_loss/(i+1), lr=scheduler.get_last_lr()[0])

        avg_train_loss = total_train_loss / len(train_dataloader)
        scheduler.step()

        avg_val_loss_str = "N/A"
        if (epoch + 1) % 5 == 0 or (epoch + 1) == args.epochs:
            model.eval()
            total_val_loss = 0.0
            with torch.no_grad():
                loop_val = tqdm(val_dataloader, desc=f"Val [{epoch+1}/{args.epochs}]", ncols=100)
                for imgs, labels in loop_val:
                    imgs, labels = imgs.to(device), labels.to(device)
                    outputs = model(imgs)
                    loss = criterion(outputs, labels)
                    total_val_loss += loss.item()

            avg_val_loss = total_val_loss / len(val_dataloader)
            avg_val_loss_str = f"{avg_val_loss:.4f}"
            print(f"\nEpoch {epoch+1}: Train Loss = {avg_train_loss:.4f}, Validation Loss = {avg_val_loss:.4f}")

            if avg_val_loss < min_val_loss:
                min_val_loss = avg_val_loss
                best_path = os.path.join(args.result_dir, "model_best.pth")
                torch.save({'model_state_dict': model.state_dict()}, best_path)
                print(f"Best model saved at epoch {epoch+1} with val loss {min_val_loss:.4f}")

        lr = scheduler.get_last_lr()[0]
        with open(log_path, "a") as f:
            f.write(f"\n{epoch + 1}\t\t{avg_train_loss:.4f}\t\t{avg_val_loss_str}\t\t{lr:.8f}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="DDRNet Weighted Training Script")

    parser.add_argument("--dataset_dir", type=str, default="./data")
    parser.add_argument("--result_dir", type=str, default="output")
    parser.add_argument("--loadpath", type=str, default=None)
    parser.add_argument("--epochs", type=int, default=400)
    parser.add_argument("--num_classes", type=int, default=19)
    parser.add_argument("--gpu_id", type=int, default=0)

    parser.add_argument("--lr", type=float, default=1e-2)
    parser.add_argument("--batch_size", type=int, default=8)
    parser.add_argument("--momentum", type=float, default=0.9)
    parser.add_argument("--weight_decay", type=float, default=5e-4)
    parser.add_argument("--warmup_epochs", type=int, default=5)

    parser.add_argument("--crop_size", default=[512, 1024], type=arg_as_list)
    parser.add_argument("--scale_range", default=[0.75, 1.5], type=arg_as_list)
    parser.add_argument("--num_workers", type=int, default=os.cpu_count())

    # 폴더 가중치 조절. (1.0이 기본 가중치)
    parser.add_argument("--folder_weights", type=arg_as_dict, default={"cam0":1.0, "cam1":1.0, "cam2":1.0, "cam3":0.8, "cam4":1.0, "cam5":1.0, "set1":1.5, "set2":1.8, "set3":1.5},
                        help='{"cam0": 1.0, "set1": 2.0}')
    # 클래스 가중치 조절 (1.0이 기본 가중치로 픽셀 수에 따라서 조절.)
    parser.add_argument("--class_weights", type=arg_as_list, default=[2.0166, 3.481, 4.0911, 3.9912, 3.9619, 2.0864, 1.8396, 4.3168, 3.79, 6.4674, 5.7661, 5.642, 8.4116, 5.9525, 2.2137, 5.2137, 6.1661, 4.195, 1.0],
                        help='List of weights for each class. "[1.0, 1.5, 0.8]"')

    parser.add_argument("--use_ohem", action='store_true', help="Use OHEM Cross Entropy loss")

    args = parser.parse_args()

    result_dir = Path(args.result_dir)
    result_dir.mkdir(parents=True, exist_ok=True)

    train_and_validate(args)

학습 실행 최종 명령어.

python train_weight.py --loadpath ./DDRNet_cityscape.pth --epoch 300 --batch_size 16

학습 결과

클래스 & 폴더별 | 클래스 | 폴더별 |---|---|---| | | | |

test dataset predict 결과

inference time

모델 구조에 대해서 현재 환경에 약 9.3ms소요

결과 종합

모델명: DDRNet

데이터셋 : 제공된 데이터셋 7 : 2 : 1로 분할시켜 학습 및 테스트 진행

import os
import shutil
from pathlib import Path

def split_dataset(base_dir):
    main_folders = ['colormap', 'image', 'labelmap']

    source_base_path = Path(base_dir) / 'image' / 'train'

    if not source_base_path.exists():
        return

    # train 폴더 내의 하위 폴더들(cam0, cam1, set1 등) 목록 가져오기
    try:
        sub_folders = [d.name for d in source_base_path.iterdir() if d.is_dir()]
    except OSError as e:
        print(f"문제가 발생했습니다")
        return

    print("데이터셋 분할을 시작")

    for sub_folder in sub_folders:
        print(f"\n📁 [{sub_folder}] 폴더 처리 중...")

        source_sub_folder_path = source_base_path / sub_folder

        try:
            files = sorted([f.name for f in source_sub_folder_path.iterdir() if f.is_file()])
        except FileNotFoundError:
            print(f"  '{source_sub_folder_path}' 폴더를 찾을 수 없습니다..")
            continue

        if not files:
            print(f"  '{sub_folder}' 폴더에 파일이 없습니다.")
            continue

        for main_folder in main_folders:
            for split_type in ['train', 'val', 'test']:
                dest_path = Path(base_dir) / main_folder / split_type / sub_folder
                dest_path.mkdir(parents=True, exist_ok=True)

        moved_counts = {'train': 0, 'val': 0, 'test': 0}
        for i in range(0, len(files), 10):
            chunk = files[i:i+10]

            # 10개 미만이면 train으로 이동
            if len(chunk) < 10:
                split_map = {'train': chunk}
            # 10개이면 7:2:1로 분할
            else:
                split_map = {
                    'train': chunk[0:7],
                    'val': chunk[7:9],
                    'test': chunk[9:10]
                }
            for split_type, files_to_move in split_map.items():
                if not files_to_move:
                    continue

                for file_name in files_to_move:
                    moved_counts[split_type] += 1
                    for main_folder in main_folders:
                        source_file = Path(base_dir) / main_folder / 'train' / sub_folder / file_name
                        dest_file = Path(base_dir) / main_folder / split_type / sub_folder / file_name

                        if source_file.exists():
                            shutil.move(str(source_file), str(dest_file))

        print(f"  - ✅ Train: {moved_counts['train']}개 파일 이동 완료")
        print(f"  - ✅ Validation: {moved_counts['val']}개 파일 이동 완료")
        print(f"  - ✅ Test: {moved_counts['test']}개 파일 이동 완료")


if __name__ == '__main__':
    base_directory = 'C:/etri/data'  

    split_dataset(base_directory)

테스트 코드

prediction

import os
import argparse
from glob import glob
from PIL import Image
import numpy as np
from tqdm import tqdm
import torch
import torch.nn.functional as F
from torchvision import transforms
from DDRNet import DDRNet
from torch.utils.data import Dataset, DataLoader
import matplotlib.cm as cm
from collections import OrderedDict


class TestSegmentationDataset(Dataset):
    def __init__(self, root_dir, subset='test'):
        self.image_dir = os.path.join(root_dir, "image", subset)
        self.image_paths = sorted(glob(os.path.join(self.image_dir, "*", "*.*"), recursive=True))
        self.to_tensor = transforms.ToTensor()

    def __len__(self):
        return len(self.image_paths)

    def __getitem__(self, idx):
        img_path = self.image_paths[idx]
        img = Image.open(img_path).convert("RGB")
        tensor = self.to_tensor(img)
        return tensor, img_path

# 단일 GPU
def load_model(weight_path, num_classes, device):
    model = DDRNet(num_classes=num_classes)

    checkpoint = torch.load(weight_path, map_location=device)
    if 'model_state_dict' in checkpoint:
        state_dict = checkpoint['model_state_dict']
    else:
        state_dict = checkpoint

    new_state_dict = OrderedDict()
    for k, v in state_dict.items():
        name = k[7:] if k.startswith('module.') else k # 'module.' 접두사를 제거
        new_state_dict[name] = v

    model.load_state_dict(new_state_dict)

    model = model.to(device)
    model.eval()

    return model

# 예측 결과를 이미지 파일로 저장
def save_prediction(pred, save_path, colormap_root, num_classes):
    pred_np = pred.squeeze().cpu().numpy().astype(np.uint8)

    os.makedirs(os.path.dirname(save_path), exist_ok=True)
    Image.fromarray(pred_np).save(save_path)

    normed = pred_np.astype(np.float32) / (num_classes - 1) 
    cmap = cm.get_cmap('turbo')
    colored = cmap(normed)
    rgb = (colored[:, :, :3] * 255).astype(np.uint8)
    rgb_img = Image.fromarray(rgb)

    try:
        rel_path = os.path.relpath(save_path, start=os.path.dirname(save_path))
        cmap_path = os.path.join(colormap_root, os.path.dirname(os.path.relpath(save_path, start=args.result_dir)), rel_path)
    except ValueError: # 다른 드라이브에 있을 경우 대비
        rel_path = Path(save_path).name
        cmap_path = os.path.join(colormap_root, rel_path)

    os.makedirs(os.path.dirname(cmap_path), exist_ok=True)
    rgb_img.save(cmap_path)

# 전체 테스트
def test(args):
    # 단일 GPU 사용
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    print(f"Using device: {device}")

    dataset = TestSegmentationDataset(args.dataset_dir, subset=args.subset)
    if not dataset.image_paths:
        print(f"Error: No images found in '{dataset.image_dir}'. Please check the path and subset.")
        return

    dataloader = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=2)

    model = load_model(args.weight_path, args.num_classes, device)
    colormap_root = os.path.join(args.result_dir, "colormap")

    with torch.inference_mode():
        for img_tensor, img_path_tuple in tqdm(dataloader, desc="Predicting..."):
            img_path = img_path_tuple[0]
            img_tensor = img_tensor.to(device)

            output = model(img_tensor)
            if isinstance(output, tuple):
                output = output[0]

            pred = torch.argmax(output, dim=1)

            rel_path = os.path.relpath(img_path, start=os.path.join(args.dataset_dir, "image"))
            save_path = os.path.join(args.result_dir, rel_path)

            save_prediction(pred, save_path, colormap_root, args.num_classes)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--dataset_dir", type=str, default="./data", help="Path to dataset root directory")
    parser.add_argument("--weight_path", type=str, default="./output/model_best.pth", help="Path to model weight (.pth)")
    parser.add_argument("--result_dir", type=str, default="./result", help="Directory to save results")
    parser.add_argument("--num_classes", type=int, default=19, help="Number of segmentation classes")
    parser.add_argument("--subset", type=str, default="test", help="Which subset to run prediction on (e.g., 'test', 'val', 'train')")

    args = parser.parse_args()

    os.makedirs(args.result_dir, exist_ok=True)

    test(args)

mIoU 계산

import os
import argparse
import numpy as np
from PIL import Image
from glob import glob
from tqdm import tqdm
from sklearn.metrics import confusion_matrix
from pathlib import Path

def load_image(path):
    return np.array(Image.open(path)).astype(np.uint8)

def compute_miou(confusion, num_classes):
    """
    1. mIoU (All): NaN을 제외한 모든 클래스(IoU=0 포함)의 평균
    2. mIoU (>0): IoU가 0보다 큰 클래스들만의 평균
    """
    ious = []
    for cls in range(num_classes):
        TP = confusion[cls, cls]
        FP = confusion[:, cls].sum() - TP
        FN = confusion[cls, :].sum() - TP

        denom = TP + FP + FN
        if denom == 0:
            iou = float('nan')
        else:
            iou = TP / denom
        ious.append(iou)

    # mIoU (All Classes) 계산
    # NaN 값을 무시하고 평균을 계산.
    miou_all = np.nanmean(ious)

    # mIoU (IoU > 0 Classes Only) 계산
    # IoU가 NaN이 아니고 0보다 큰 값들만 계산.
    positive_ious = [iou for iou in ious if not np.isnan(iou) and iou > 0]

    # 0이 제거된 iou 값들의 평균을 계산
    if not positive_ious:
        miou_positive = 0.0
    else:
        miou_positive = np.mean(positive_ious) 

    return miou_all, miou_positive, ious

def evaluate(result_dir, label_dir, num_classes):
    pred_paths = sorted(glob(os.path.join(result_dir, "**", "*_leftImg8bit.png"), recursive=True))
    print(f'Found {len(pred_paths)} segmentation result images in {result_dir}')

    if not pred_paths:
        print("Error: No prediction files found. Please check the 'result_dir' path and file names.")
        return

    all_confusion = np.zeros((num_classes, num_classes), dtype=np.int64)

    for pred_path in tqdm(pred_paths, desc="Evaluating"):
        sub_folder = Path(pred_path).parent.name
        file_id = os.path.basename(pred_path).replace("_leftImg8bit.png", "")

        label_path = os.path.join(label_dir, sub_folder, f"{file_id}_gtFine_CategoryId.png")

        if not os.path.exists(label_path):
            print(f"Label not found at {label_path}, skipping.")
            continue

        pred = load_image(pred_path).flatten()
        label = load_image(label_path).flatten()

        mask = label != 255
        pred = pred[mask]
        label = label[mask]

        pred = np.clip(pred, 0, num_classes - 1)
        label = np.clip(label, 0, num_classes - 1)

        conf = confusion_matrix(label, pred, labels=list(range(num_classes)))
        all_confusion += conf

    miou_all, miou_positive, ious = compute_miou(all_confusion, num_classes)

    print("\n--- Evaluation Results ---")
    print(f"📊 mIoU (All Classes, IoU=0 포함): {miou_all:.4f}")
    print(f"📊 mIoU (Positive Classes, IoU>0 제외): {miou_positive:.4f}")
    print("--------------------------")

    for i, iou in enumerate(ious):
        print(f"Class {i}: IoU = {iou:.4f}" if not np.isnan(iou) else f"Class {i}: IoU = NaN (ignored in mean)")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Calculate mIoU for semantic segmentation results.")

    parser.add_argument("--result_dir", type=str, default="C:/ETRI/result/test", 
                        help="Predicted *_leftImg8bit.png files가 있는 상위 디렉토리")
    parser.add_argument("--label_dir", type=str, default="C:/ETRI/data/labelmap/test", 
                        help="정답 레이블 *_gtFine_CategoryId.png files가 있는 상위 디렉토리")
    parser.add_argument("--num_classes", type=int, default=19, help="세그먼테이션 클래스 수")

    args = parser.parse_args()

    evaluate(args.result_dir, args.label_dir, args.num_classes)

클래스 & 폴더별 | 클래스 | 폴더별 |---|---|---| | | | |

개선방법 1 : ** 데이터 증식 추가** 개선방법 2 : 손실함수 변경

DDRNet에 대한 개선 방향

• 데이터 증식 방법 조절(우천에 최적화된 방향으로)
• 손실함수 변경(weighted CE)
• mIoU계산에 대한 조절(test방향과 많이 다른 것을 확인 - 50 mIoU가 실제 test에서는 33점대로 떨어져 cam0~cam5에 대해서 학습이 실제 test 데이터셋과는 많이 차이가 나는것을 확인.)

환경 구성.

중요 환경에 대한 버전.

Package             Version
------------------- --------------------
numpy               1.24.1
opencv-python       4.12.0.88
thop                0.1.1.post2209072238
torch               2.3.1+cu118
torchaudio          2.3.1+cu118
torchvision         0.18.1+cu118

진행 작업

1. 폴더별 가중치 조절 코드 추가.
2. 클래스별 가중치 조절 코드 추가.
3. 기초 데이터 증식 코드 추가.

class SegmentationTransform:
    def __init__(self, crop_size=[1024, 1024], scale_range=[0.5, 1.5]):
        self.crop_size = crop_size
        self.scale_range = scale_range
        self.mean = [0.485, 0.456, 0.406]
        self.std = [0.229, 0.224, 0.225]
        self.bilinear = transforms.InterpolationMode.BILINEAR
        self.nearest = transforms.InterpolationMode.NEAREST

        # Color Jitter
        self.color_jitter = transforms.ColorJitter(brightness=0.3, contrast=0.3, saturation=0.3, hue=0.1)
        # Gaussian Blur
        self.gaussian_blur = transforms.GaussianBlur(kernel_size=(5, 9), sigma=(0.1, 5))

    def __call__(self, image, label):
        scale_factor = random.uniform(self.scale_range[0], self.scale_range[1])
        width, height = image.size
        new_width, new_height = int(width * scale_factor), int(height * scale_factor)
        image = TF.resize(image, (new_height, new_width), interpolation=self.bilinear)
        label = TF.resize(label, (new_height, new_width), interpolation=self.nearest)

        pad_h = max(self.crop_size[0] - new_height, 0)
        pad_w = max(self.crop_size[1] - new_width, 0)
        if pad_h > 0 or pad_w > 0:
            padding = (0, 0, pad_w, pad_h)
            image = TF.pad(image, padding, fill=0)
            label = TF.pad(label, padding, fill=255)

        # 크롭
        i, j, h, w = transforms.RandomCrop.get_params(image, output_size=self.crop_size)
        image = TF.crop(image, i, j, h, w)
        label = TF.crop(label, i, j, h, w)

        # 좌우 반전
        if random.random() > 0.3:
            image = TF.hflip(image)
            label = TF.hflip(label)

        # 회전
        if random.random() > 0.5:
            angle = random.uniform(-5, 5)
            image = TF.rotate(image, angle, interpolation=self.bilinear, fill=0)
            label = TF.rotate(label, angle, interpolation=self.nearest, fill=255)

        # 색상 변환
        if random.random() > 0.4:
            image = self.color_jitter(image)
        if random.random() > 0.3:
            image = self.gaussian_blur(image)

        image = TF.to_tensor(image)
        image = TF.normalize(image, mean=self.mean, std=self.std)
        label = torch.from_numpy(np.array(label, dtype=np.uint8)).long()

train.py

import os
import argparse
import torch
from torch.utils.data import DataLoader, WeightedRandomSampler
from tqdm import tqdm
from collections import OrderedDict
import json
from pathlib import Path

from DDRNet import DDRNet
from functions import *

def arg_as_dict(s):
    try:
        return json.loads(s)
    except Exception as e:
        raise argparse.ArgumentTypeError(f"Argument must be a JSON-formatted dictionary string. Error: {e}")

def train_and_validate(args):
    device = torch.device(f"cuda:{args.gpu_id}" if torch.cuda.is_available() else "cpu")
    print(f"Initialized training on device: {device}")

    # 데이터셋 폴더별 가중치를 위한 설정
    train_sub_folders = ['cam0', 'cam1', 'cam2', 'cam3', 'cam4', 'cam5', 'set1', 'set2', 'set3']
    val_sub_folders = ['cam0', 'cam1', 'cam2', 'cam3', 'cam4', 'cam5', 'set1', 'set2', 'set3']

    train_dataset = SegmentationDataset(args.dataset_dir, args.crop_size, 'train', args.scale_range, sub_folders=train_sub_folders)

    # --- 폴더별 가중치에 따른 샘플링 확률 계산 ---
    if args.folder_weights:
        print("Applying folder-wise weights for sampling...")
        folder_indices = [sample[1] for sample in train_dataset.samples]
        folder_names_per_sample = [train_sub_folders[i] for i in folder_indices]
        sample_weights = [args.folder_weights.get(name, 1.0) for name in folder_names_per_sample]
        print(f"Sample weights will be based on folder weights: {args.folder_weights}")
        sampler = WeightedRandomSampler(weights=sample_weights, num_samples=len(sample_weights), replacement=True)
        shuffle = False
    else:
        sampler = None
        shuffle = True

    val_dataset = SegmentationDataset(args.dataset_dir, args.crop_size, 'val', args.scale_range, sub_folders=val_sub_folders)

    train_dataloader = DataLoader(train_dataset, batch_size=args.batch_size, shuffle=shuffle, sampler=sampler, num_workers=args.num_workers, pin_memory=True, drop_last=True)
    val_dataloader = DataLoader(val_dataset, batch_size=args.batch_size, shuffle=False, num_workers=args.num_workers, pin_memory=True)

    model = DDRNet(num_classes=args.num_classes).to(device)

    class_weights = None
    if args.class_weights:
        if len(args.class_weights) != args.num_classes:
            raise ValueError(f"Number of class_weights ({len(args.class_weights)}) must match num_classes ({args.num_classes})")
        print(f"Applying class weights: {args.class_weights}")
        class_weights = torch.tensor(args.class_weights, dtype=torch.float).to(device)

    if args.use_ohem:
        print("Using OhemCrossEntropy Loss.")
        criterion = OhemCrossEntropy(ignore_label=255, weight=class_weights)
    else:
        print("Using standard CrossEntropy Loss.")
        criterion = CrossEntropy(ignore_label=255, weight=class_weights)

    optimizer = torch.optim.SGD(model.parameters(), lr=args.lr, momentum=args.momentum, weight_decay=args.weight_decay)
    scheduler = WarmupPolyEpochLR(optimizer, total_epochs=args.epochs, warmup_epochs=args.warmup_epochs)

    start_epoch = 0
    min_val_loss = float('inf')
    if args.loadpath:
        print(f"Loading checkpoint from: {args.loadpath}")
        checkpoint = torch.load(args.loadpath, map_location=device)
        try:
            # DDP 학습 가중치('module.' 접두사) 호환을 위한 처리
            new_state_dict = OrderedDict()
            for k, v in checkpoint['model_state_dict'].items():
                name = k[7:] if k.startswith('module.') else k
                new_state_dict[name] = v
            model.load_state_dict(new_state_dict, strict=False)

            optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
            scheduler.load_state_dict(checkpoint['scheduler_state_dict'])
            start_epoch = checkpoint['epoch'] + 1
            min_val_loss = checkpoint.get('loss', float('inf'))
            print(f"Resuming training from epoch {start_epoch}, with min_val_loss: {min_val_loss:.4f}")
        except KeyError:
            print("Old checkpoint format. Loading model state_dict only.")
            load_state_dict(model, checkpoint)

    os.makedirs(args.result_dir, exist_ok=True)
    log_path = os.path.join(args.result_dir, "log.txt")
    with open(log_path, 'a' if start_epoch > 0 else 'w') as f:
        if start_epoch == 0:
            f.write("Epoch\t\tTrain-loss\t\tVal-loss\t\tlearningRate\n")

    for epoch in range(start_epoch, args.epochs):
        model.train()
        total_train_loss = 0.0
        loop = tqdm(train_dataloader, desc=f"Train [{epoch+1}/{args.epochs}]", ncols=100)

        for i, (imgs, labels) in enumerate(loop):
            optimizer.zero_grad(set_to_none=True)
            imgs, labels = imgs.to(device), labels.to(device)
            outputs = model(imgs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            total_train_loss += loss.item()
            loop.set_postfix(loss=loss.item(), avg_loss=total_train_loss/(i+1), lr=scheduler.get_last_lr()[0])

        avg_train_loss = total_train_loss / len(train_dataloader)
        scheduler.step()

        avg_val_loss_str = "N/A"
        if (epoch + 1) % 5 == 0 or (epoch + 1) == args.epochs:
            model.eval()
            total_val_loss = 0.0
            with torch.no_grad():
                loop_val = tqdm(val_dataloader, desc=f"Val [{epoch+1}/{args.epochs}]", ncols=100)
                for imgs, labels in loop_val:
                    imgs, labels = imgs.to(device), labels.to(device)
                    outputs = model(imgs)
                    loss = criterion(outputs, labels)
                    total_val_loss += loss.item()

            avg_val_loss = total_val_loss / len(val_dataloader)
            avg_val_loss_str = f"{avg_val_loss:.4f}"
            print(f"\nEpoch {epoch+1}: Train Loss = {avg_train_loss:.4f}, Validation Loss = {avg_val_loss:.4f}")

            if avg_val_loss < min_val_loss:
                min_val_loss = avg_val_loss
                best_path = os.path.join(args.result_dir, "model_best.pth")
                torch.save({'model_state_dict': model.state_dict()}, best_path)
                print(f"Best model saved at epoch {epoch+1} with val loss {min_val_loss:.4f}")

        lr = scheduler.get_last_lr()[0]
        with open(log_path, "a") as f:
            f.write(f"\n{epoch + 1}\t\t{avg_train_loss:.4f}\t\t{avg_val_loss_str}\t\t{lr:.8f}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="DDRNet Weighted Training Script")

    parser.add_argument("--dataset_dir", type=str, default="./data")
    parser.add_argument("--result_dir", type=str, default="output")
    parser.add_argument("--loadpath", type=str, default=None)
    parser.add_argument("--epochs", type=int, default=400)
    parser.add_argument("--num_classes", type=int, default=19)
    parser.add_argument("--gpu_id", type=int, default=0)

    parser.add_argument("--lr", type=float, default=1e-2)
    parser.add_argument("--batch_size", type=int, default=8)
    parser.add_argument("--momentum", type=float, default=0.9)
    parser.add_argument("--weight_decay", type=float, default=5e-4)
    parser.add_argument("--warmup_epochs", type=int, default=5)

    parser.add_argument("--crop_size", default=[512, 1024], type=arg_as_list)
    parser.add_argument("--scale_range", default=[0.75, 1.5], type=arg_as_list)
    parser.add_argument("--num_workers", type=int, default=os.cpu_count())

    # 폴더 가중치 조절. (1.0이 기본 가중치)
    parser.add_argument("--folder_weights", type=arg_as_dict, default={"cam0":1.0, "cam1":1.0, "cam2":1.0, "cam3":0.8, "cam4":1.0, "cam5":1.0, "set1":1.5, "set2":1.8, "set3":1.5},
                        help='{"cam0": 1.0, "set1": 2.0}')
    # 클래스 가중치 조절 (1.0이 기본 가중치로 픽셀 수에 따라서 조절.)
    parser.add_argument("--class_weights", type=arg_as_list, default=[2.0166, 3.481, 4.0911, 3.9912, 3.9619, 2.0864, 1.8396, 4.3168, 3.79, 6.4674, 5.7661, 5.642, 8.4116, 5.9525, 2.2137, 5.2137, 6.1661, 4.195, 1.0],
                        help='List of weights for each class. "[1.0, 1.5, 0.8]"')

    parser.add_argument("--use_ohem", action='store_true', help="Use OHEM Cross Entropy loss")

    args = parser.parse_args()

    result_dir = Path(args.result_dir)
    result_dir.mkdir(parents=True, exist_ok=True)

    train_and_validate(args)


# 학습 시작 명령어 기본.
python train.py \
    --dataset_dir "./data" \
    --result_dir "./output_001" \
    --loadpath "./DDRNet23s_cityscape.pth" \
    --epochs 300 \
    --batch_size 16 \
    --lr 1e-2 \
    --folder_weights '{"cam0":1.0, "cam1":1.0, "cam2":1.0, "cam3":1.0, "cam4":1.0, "cam5":1.0, "set1":1.5, "set2":1.5, "set3":1.5}' \
    --class_weights '[0.5, 1.0, 1.0, 1.0, 1.0, 1.5, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0]' \
    --use_ohem

학습 실행 최종 명령어.

python train_weight.py --loadpath ./output/model_best.pth --epoch 300 --batch_size 16

학습 loss값이 일정해지는 수준까지 학습 진행을 예측하고 300Epoch까지 진행. 추가 학습을 진행하였으나 loss값이 정체되어 정지.

학습 결과

학습이 중간에 끊겨서 다시 시작하는 과정에서 LR값이 초기화

학습이 끊기며 LR값이 초기화 되는 과정에서 잠시 loss값에 변동이 존재.

test dataset predict 결과

mIoU값 계산.

클래스가 없는 경우를 포함하였을 때	클래스가 없는 경우를 제외하였을 때

픽셀 수가 낮은 데이터들은 확실히 데이터가 증가함. (0.1 단위로 증가한 값들도 존재) 픽셀 수가 다수인 0, 2, 3, 18번 같은 경우에는 정확도가 감소하는 현상 발견. 클래스와 폴더의 가중치를 동시에 준 결과로 클래스와 폴더의 가중치 값들이 각각 어느정도 IoU값에 영향을 주는지 확인이 필요.

inference time

모델 구조에 대해서 현재 환경에 약 9.3ms소요

결과 종합

모델명: DDRNet

데이터셋 : 제공된 데이터셋 7 : 2 : 1로 분할시켜 학습 및 테스트 진행

import os
import shutil
from pathlib import Path

def split_dataset(base_dir):
    main_folders = ['colormap', 'image', 'labelmap']

    source_base_path = Path(base_dir) / 'image' / 'train'

    if not source_base_path.exists():
        return

    # train 폴더 내의 하위 폴더들(cam0, cam1, set1 등) 목록 가져오기
    try:
        sub_folders = [d.name for d in source_base_path.iterdir() if d.is_dir()]
    except OSError as e:
        print(f"문제가 발생했습니다")
        return

    print("데이터셋 분할을 시작")

    for sub_folder in sub_folders:
        print(f"\n📁 [{sub_folder}] 폴더 처리 중...")

        source_sub_folder_path = source_base_path / sub_folder

        try:
            files = sorted([f.name for f in source_sub_folder_path.iterdir() if f.is_file()])
        except FileNotFoundError:
            print(f"  '{source_sub_folder_path}' 폴더를 찾을 수 없습니다..")
            continue

        if not files:
            print(f"  '{sub_folder}' 폴더에 파일이 없습니다.")
            continue

        for main_folder in main_folders:
            for split_type in ['train', 'val', 'test']:
                dest_path = Path(base_dir) / main_folder / split_type / sub_folder
                dest_path.mkdir(parents=True, exist_ok=True)

        moved_counts = {'train': 0, 'val': 0, 'test': 0}
        for i in range(0, len(files), 10):
            chunk = files[i:i+10]

            # 10개 미만이면 train으로 이동
            if len(chunk) < 10:
                split_map = {'train': chunk}
            # 10개이면 7:2:1로 분할
            else:
                split_map = {
                    'train': chunk[0:7],
                    'val': chunk[7:9],
                    'test': chunk[9:10]
                }
            for split_type, files_to_move in split_map.items():
                if not files_to_move:
                    continue

                for file_name in files_to_move:
                    moved_counts[split_type] += 1
                    for main_folder in main_folders:
                        source_file = Path(base_dir) / main_folder / 'train' / sub_folder / file_name
                        dest_file = Path(base_dir) / main_folder / split_type / sub_folder / file_name

                        if source_file.exists():
                            shutil.move(str(source_file), str(dest_file))

        print(f"  - ✅ Train: {moved_counts['train']}개 파일 이동 완료")
        print(f"  - ✅ Validation: {moved_counts['val']}개 파일 이동 완료")
        print(f"  - ✅ Test: {moved_counts['test']}개 파일 이동 완료")


if __name__ == '__main__':
    base_directory = 'C:/etri/data'  

    split_dataset(base_directory)

테스트 코드

prediction

import os
import argparse
from glob import glob
from PIL import Image
import numpy as np
from tqdm import tqdm
import torch
import torch.nn.functional as F
from torchvision import transforms
from DDRNet import DDRNet
from torch.utils.data import Dataset, DataLoader
import matplotlib.cm as cm
from collections import OrderedDict


class TestSegmentationDataset(Dataset):
    def __init__(self, root_dir, subset='test'):
        self.image_dir = os.path.join(root_dir, "image", subset)
        self.image_paths = sorted(glob(os.path.join(self.image_dir, "*", "*.*"), recursive=True))
        self.to_tensor = transforms.ToTensor()

    def __len__(self):
        return len(self.image_paths)

    def __getitem__(self, idx):
        img_path = self.image_paths[idx]
        img = Image.open(img_path).convert("RGB")
        tensor = self.to_tensor(img)
        return tensor, img_path

# 단일 GPU
def load_model(weight_path, num_classes, device):
    model = DDRNet(num_classes=num_classes)

    checkpoint = torch.load(weight_path, map_location=device)
    if 'model_state_dict' in checkpoint:
        state_dict = checkpoint['model_state_dict']
    else:
        state_dict = checkpoint

    new_state_dict = OrderedDict()
    for k, v in state_dict.items():
        name = k[7:] if k.startswith('module.') else k # 'module.' 접두사를 제거
        new_state_dict[name] = v

    model.load_state_dict(new_state_dict)

    model = model.to(device)
    model.eval()

    return model

# 예측 결과를 이미지 파일로 저장
def save_prediction(pred, save_path, colormap_root, num_classes):
    pred_np = pred.squeeze().cpu().numpy().astype(np.uint8)

    os.makedirs(os.path.dirname(save_path), exist_ok=True)
    Image.fromarray(pred_np).save(save_path)

    normed = pred_np.astype(np.float32) / (num_classes - 1) 
    cmap = cm.get_cmap('turbo')
    colored = cmap(normed)
    rgb = (colored[:, :, :3] * 255).astype(np.uint8)
    rgb_img = Image.fromarray(rgb)

    try:
        rel_path = os.path.relpath(save_path, start=os.path.dirname(save_path))
        cmap_path = os.path.join(colormap_root, os.path.dirname(os.path.relpath(save_path, start=args.result_dir)), rel_path)
    except ValueError: # 다른 드라이브에 있을 경우 대비
        rel_path = Path(save_path).name
        cmap_path = os.path.join(colormap_root, rel_path)

    os.makedirs(os.path.dirname(cmap_path), exist_ok=True)
    rgb_img.save(cmap_path)

# 전체 테스트
def test(args):
    # 단일 GPU 사용
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    print(f"Using device: {device}")

    dataset = TestSegmentationDataset(args.dataset_dir, subset=args.subset)
    if not dataset.image_paths:
        print(f"Error: No images found in '{dataset.image_dir}'. Please check the path and subset.")
        return

    dataloader = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=2)

    model = load_model(args.weight_path, args.num_classes, device)
    colormap_root = os.path.join(args.result_dir, "colormap")

    with torch.inference_mode():
        for img_tensor, img_path_tuple in tqdm(dataloader, desc="Predicting..."):
            img_path = img_path_tuple[0]
            img_tensor = img_tensor.to(device)

            output = model(img_tensor)
            if isinstance(output, tuple):
                output = output[0]

            pred = torch.argmax(output, dim=1)

            rel_path = os.path.relpath(img_path, start=os.path.join(args.dataset_dir, "image"))
            save_path = os.path.join(args.result_dir, rel_path)

            save_prediction(pred, save_path, colormap_root, args.num_classes)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--dataset_dir", type=str, default="./data", help="Path to dataset root directory")
    parser.add_argument("--weight_path", type=str, default="./output/model_best.pth", help="Path to model weight (.pth)")
    parser.add_argument("--result_dir", type=str, default="./result", help="Directory to save results")
    parser.add_argument("--num_classes", type=int, default=19, help="Number of segmentation classes")
    parser.add_argument("--subset", type=str, default="test", help="Which subset to run prediction on (e.g., 'test', 'val', 'train')")

    args = parser.parse_args()

    os.makedirs(args.result_dir, exist_ok=True)

    test(args)

mIoU 계산

import os
import argparse
import numpy as np
from PIL import Image
from glob import glob
from tqdm import tqdm
from sklearn.metrics import confusion_matrix
from pathlib import Path

def load_image(path):
    return np.array(Image.open(path)).astype(np.uint8)

def compute_miou(confusion, num_classes):
    """
    1. mIoU (All): NaN을 제외한 모든 클래스(IoU=0 포함)의 평균
    2. mIoU (>0): IoU가 0보다 큰 클래스들만의 평균
    """
    ious = []
    for cls in range(num_classes):
        TP = confusion[cls, cls]
        FP = confusion[:, cls].sum() - TP
        FN = confusion[cls, :].sum() - TP

        denom = TP + FP + FN
        if denom == 0:
            iou = float('nan')
        else:
            iou = TP / denom
        ious.append(iou)

    # mIoU (All Classes) 계산
    # NaN 값을 무시하고 평균을 계산.
    miou_all = np.nanmean(ious)

    # mIoU (IoU > 0 Classes Only) 계산
    # IoU가 NaN이 아니고 0보다 큰 값들만 계산.
    positive_ious = [iou for iou in ious if not np.isnan(iou) and iou > 0]

    # 0이 제거된 iou 값들의 평균을 계산
    if not positive_ious:
        miou_positive = 0.0
    else:
        miou_positive = np.mean(positive_ious) 

    return miou_all, miou_positive, ious

def evaluate(result_dir, label_dir, num_classes):
    pred_paths = sorted(glob(os.path.join(result_dir, "**", "*_leftImg8bit.png"), recursive=True))
    print(f'Found {len(pred_paths)} segmentation result images in {result_dir}')

    if not pred_paths:
        print("Error: No prediction files found. Please check the 'result_dir' path and file names.")
        return

    all_confusion = np.zeros((num_classes, num_classes), dtype=np.int64)

    for pred_path in tqdm(pred_paths, desc="Evaluating"):
        sub_folder = Path(pred_path).parent.name
        file_id = os.path.basename(pred_path).replace("_leftImg8bit.png", "")

        label_path = os.path.join(label_dir, sub_folder, f"{file_id}_gtFine_CategoryId.png")

        if not os.path.exists(label_path):
            print(f"Label not found at {label_path}, skipping.")
            continue

        pred = load_image(pred_path).flatten()
        label = load_image(label_path).flatten()

        mask = label != 255
        pred = pred[mask]
        label = label[mask]

        pred = np.clip(pred, 0, num_classes - 1)
        label = np.clip(label, 0, num_classes - 1)

        conf = confusion_matrix(label, pred, labels=list(range(num_classes)))
        all_confusion += conf

    miou_all, miou_positive, ious = compute_miou(all_confusion, num_classes)

    print("\n--- Evaluation Results ---")
    print(f"📊 mIoU (All Classes, IoU=0 포함): {miou_all:.4f}")
    print(f"📊 mIoU (Positive Classes, IoU>0 제외): {miou_positive:.4f}")
    print("--------------------------")

    for i, iou in enumerate(ious):
        print(f"Class {i}: IoU = {iou:.4f}" if not np.isnan(iou) else f"Class {i}: IoU = NaN (ignored in mean)")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Calculate mIoU for semantic segmentation results.")

    parser.add_argument("--result_dir", type=str, default="C:/ETRI/result/test", 
                        help="Predicted *_leftImg8bit.png files가 있는 상위 디렉토리")
    parser.add_argument("--label_dir", type=str, default="C:/ETRI/data/labelmap/test", 
                        help="정답 레이블 *_gtFine_CategoryId.png files가 있는 상위 디렉토리")
    parser.add_argument("--num_classes", type=int, default=19, help="세그먼테이션 클래스 수")

    args = parser.parse_args()

    evaluate(args.result_dir, args.label_dir, args.num_classes)

개선방법 1 : 폴더별 가중치 조절 개선방법 2 : 클래스별 가중치 조절 개선방법 3 : 데이터 증식 추가

DDRNet에 대한 개선 방향

• 데이터 증식 방법 추가
• 폴더에 가중치를 주는 방향과 클래스별로 가중치를 주는 방향에 대해 효율적인 방향 탐구 필요.
• Train dataset(set1) 중 일부 데이터가 차량 본넷의 클래스가 잘못 라벨링이 되어있는것을 확인, 해당 데이터를 6번(차량) 클래스에서 18번(기타) 클래스로 변경
• 모델 구조에서 반복된 연산으로 인해서 연산량은 늘어나나 효율은 낮은 구조에 대해 개선.

환경 구성.

중요 환경에 대한 버전.

Package             Version
------------------- --------------------
numpy               1.24.1
opencv-python       4.12.0.88
thop                0.1.1.post2209072238
torch               2.3.1+cu118
torchaudio          2.3.1+cu118
torchvision         0.18.1+cu118

ㅇㅇ #### 진행 작업

1. BaseLine code Train 코드 개선 및 수행. 학습된 Epoch까지의 데이터에 대해서 추론 및 mIoU 계산.
2. Backbone을 freeze하여 Backbone의 가중치를 그대로 가진 상태로 추론 작업을 수행 가능하도록 개선.
3. 다중 GPU작업에 맞춰진 환경을 단일 GPU 환경으로 개선.
4. DataLoader나 학습 파라미터 등의 인자를 parser로 받아 조절 가능하게 개선.
5. 학습이 길어지는 경우(중간에 끊어야 하는 경우)를 대비해서 CheckPoint model을 받아 학습을 이어 받을 수 있도록 개선
6. 데이터셋의 일부(20%)를 Validation작업에 수행하기 위해서 데이터를 이동.
7. train에 대해서 eval의 과정이 없는 코드에 eval DataLoader를 사용하여 train 중간에 eval과정을 거치도록 개선.

train.py

import os
import argparse
import torch
from torch.utils.data import DataLoader
from tqdm import tqdm
from collections import OrderedDict
from DDRNet import DDRNet
from functions import *
from pathlib import Path

def train_and_validate(args):
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    print(f"Initialized single GPU training on device: {device}")

    train_dataset = SegmentationDataset(args.dataset_dir, args.crop_size, 'train', args.scale_range)
    val_dataset = SegmentationDataset(args.dataset_dir, args.crop_size, 'val', args.scale_range)

    print(f"DataLoader settings: num_workers={args.num_workers}, pin_memory={args.pin_memory}, shuffle={args.shuffle}, drop_last={args.drop_last}")
    train_dataloader = DataLoader(train_dataset, batch_size=args.batch_size, shuffle=args.shuffle, num_workers=args.num_workers, pin_memory=args.pin_memory, drop_last=args.drop_last)
    val_dataloader = DataLoader(val_dataset, batch_size=args.batch_size, shuffle=False, num_workers=args.num_workers, pin_memory=args.pin_memory)

    model = DDRNet(num_classes=args.num_classes).to(device)

    criterion = CrossEntropy(ignore_label=255)

    if args.freeze_backbone:
        print("❄️ Freezing backbone layers...")
        backbone_layer_names = ['conv1', 'layer1', 'layer2', 'layer3', 'layer4', 'spp'] 
        for name, param in model.named_parameters():
            if any(name.startswith(layer_name) for layer_name in backbone_layer_names):
                param.requires_grad = False

    params_to_update = [p for p in model.parameters() if p.requires_grad]
    print(f"Total parameters: {len(list(model.parameters()))}, Trainable parameters: {len(params_to_update)}")

    optimizer = torch.optim.SGD(params_to_update, lr=args.lr, momentum=args.momentum, weight_decay=args.weight_decay)
    scheduler = WarmupPolyEpochLR(optimizer, total_epochs=args.epochs, warmup_epochs=args.warmup_epochs, warmup_ratio=5e-4)

    start_epoch = 0
    min_val_loss = float('inf')
    if args.loadpath is not None:
        print(f"Loading checkpoint from: {args.loadpath}")
        checkpoint = torch.load(args.loadpath, map_location=device)
        try:
            new_state_dict = OrderedDict()
            for k, v in checkpoint['model_state_dict'].items():
                name = k[7:] if k.startswith('module.') else k
                new_state_dict[name] = v
            model.load_state_dict(new_state_dict, strict=False)

            optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
            scheduler.load_state_dict(checkpoint['scheduler_state_dict'])
            start_epoch = checkpoint['epoch'] + 1
            min_val_loss = checkpoint.get('loss', float('inf'))
            print(f"Resuming training from epoch {start_epoch}, with min_val_loss: {min_val_loss:.4f}")
        except KeyError:
            print("Old checkpoint format. Loading model state_dict only.")
            new_state_dict = OrderedDict()
            for k, v in checkpoint.items():
                if k.startswith('module.'): name = k[7:]
                elif k.startswith('model.'): name = k[6:]
                else: name = k
                new_state_dict[name] = v
            model.load_state_dict(new_state_dict, strict=False)

    os.makedirs(args.result_dir, exist_ok=True)
    log_path = os.path.join(args.result_dir, "log.txt")
    mode = 'a' if start_epoch > 0 else 'w'
    with open(log_path, mode) as f:
        if start_epoch == 0: f.write("Epoch\t\tTrain-loss\t\tVal-loss\t\tlearningRate\n")

    for epoch in range(start_epoch, args.epochs):
        model.train()
        total_train_loss = 0.0
        loop = tqdm(train_dataloader, desc=f"Train [{epoch+1}/{args.epochs}]", ncols=100)

        for i, (imgs, labels) in enumerate(loop):
            optimizer.zero_grad(set_to_none=True)
            imgs, labels = imgs.to(device), labels.to(device)
            outputs = model(imgs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            total_train_loss += loss.item()
            loop.set_postfix(loss=loss.item(), avg_loss=total_train_loss/(i+1), lr=scheduler.get_last_lr()[0])

        avg_train_loss = total_train_loss / len(train_dataloader)
        scheduler.step()

        avg_val_loss_str = "N/A"
        if (epoch + 1) % 5 == 0 or (epoch + 1) == args.epochs:
            model.eval()
            total_val_loss = 0.0
            with torch.no_grad():
                loop_val = tqdm(val_dataloader, desc=f"Val [{epoch+1}/{args.epochs}]", ncols=100)
                for i, (imgs, labels) in enumerate(loop_val):
                    imgs, labels = imgs.to(device), labels.to(device)
                    outputs = model(imgs)
                    loss = criterion(outputs, labels)
                    total_val_loss += loss.item()

            avg_val_loss = total_val_loss / len(val_dataloader)
            avg_val_loss_str = f"{avg_val_loss:.4f}"

            print(f"\nEpoch {epoch+1}: Train Loss = {avg_train_loss:.4f}, Validation Loss = {avg_val_loss:.4f}")

            if avg_val_loss < min_val_loss:
                min_val_loss = avg_val_loss
                ckp_path = os.path.join(args.result_dir, "model_best.pth")
                state_to_save = {
                    'epoch': epoch, 'model_state_dict': model.state_dict(),
                    'optimizer_state_dict': optimizer.state_dict(), 'scheduler_state_dict': scheduler.state_dict(),
                    'loss': min_val_loss,
                }
                torch.save(state_to_save, ckp_path)
                print(f"Best model saved at epoch {epoch+1} with val loss {min_val_loss:.4f}")

            ckp_path = os.path.join(args.result_dir, f"model_epoch{epoch+1}.pth")
            state_to_save = {
                'epoch': epoch, 'model_state_dict': model.state_dict(),
                'optimizer_state_dict': optimizer.state_dict(), 'scheduler_state_dict': scheduler.state_dict(),
                'loss': avg_val_loss,
            }
            torch.save(state_to_save, ckp_path)

        lr = scheduler.get_last_lr()[0]
        with open(log_path, "a") as f:
            log_entry = f"\n{epoch + 1}\t\t{avg_train_loss:.4f}\t\t{avg_val_loss_str}\t\t{lr:.8f}"
            f.write(log_entry)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="DDRNet Training Script")

    parser.add_argument("--dataset_dir", type=str, default="./data", help="Path to dataset root")
    parser.add_argument("--loadpath", type=str, default=None, help="Path to checkpoint for resuming training")
    parser.add_argument("--result_dir", type=str, default="output", help="Directory to save results")
    parser.add_argument("--epochs", type=int, default=400, help="Total number of training epochs")
    parser.add_argument("--num_classes", type=int, default=19, help="Number of segmentation classes")

    parser.add_argument("--lr", type=float, default=1e-2, help="Initial learning rate")
    parser.add_argument("--batch_size", type=int, default=8, help="Training batch size")
    parser.add_argument("--momentum", type=float, default=0.9, help="Momentum for SGD optimizer")
    parser.add_argument("--weight_decay", type=float, default=5e-4, help="Weight decay for SGD optimizer")
    parser.add_argument("--warmup_epochs", type=int, default=5, help="Number of warmup epochs for scheduler")

    parser.add_argument("--crop_size", default=[512, 1024], type=arg_as_list, help="Crop size (H W)")
    parser.add_argument("--scale_range", default=[0.75, 1.5], type=arg_as_list, help="Resize input scale range")

    parser.add_argument("--num_workers", type=int, default=os.cpu_count(), help="Number of workers for DataLoader")
    parser.add_argument("--no_pin_memory", action="store_false", dest="pin_memory", help="Disable pin_memory for DataLoader")
    parser.add_argument("--no_shuffle", action="store_false", dest="shuffle", help="Disable shuffling for training data")
    parser.add_argument("--no_drop_last", action="store_false", dest="drop_last", help="Disable drop_last for training data")
    parser.set_defaults(pin_memory=True, shuffle=True, drop_last=True)

    parser.add_argument("--freeze_backbone", action='store_true', help="Freeze backbone layers for fine-tuning")

    args = parser.parse_args()

    result_dir = Path(args.result_dir)
    result_dir.mkdir(parents=True, exist_ok=True)

    train_and_validate(args)

DDRNet23s_imagenet.pth파일의 가중치를 받아서 진행. Backbone을 freeze하고 학습을 진행하는 경우 전체 파라미터의 1/3정도만 학습이 되고 학습 진행에 있어서 Train의 Loss값이 너무 느리게 학습되는 현상이 발견되어 Backbone freeze작업을 수행하지 않고 전체적으로 모두 수행하기로 함. Backbone을 imagenet과 cityscape로 모두 학습을 수행했으나 유의미한 차이를 발견하지 못함.

학습 실행 최종 명령어.

python backbone_freeze_train.py --loadpath ./DDRNet_cityscape.pth --batch_size 16

학습 Epoch은 200Epoch으로 진행하였을 때 계속해서 Loss값이 낮아지는 경향이 있어 CheckPoint로 이어서 학습하기로 하고 크게 500으로 설정.

학습 결과

낮게나마 낮아지던 loss값이 300Epoch가까이 진행되었을 때 무의미하다고 판단하여 학습을 종료

test dataset predict 결과 (200Epoch에 대한 추론)

mIoU값 계산.

클래스가 없는 경우를 포함하였을 때	클래스가 없는 경우를 제외하였을 때
	0(주행가능영역)
	1(인도)
	2(도로노면표시)
	3(차선)
	4(연석)
	5(벽,울타리)
	6(승용차)
	7(트럭)
	8(버스)
	9(바이크, 자전거)
	10(기타 차량)
	11(보행자)
	12(라이더)
	13(교통용 콘 및 봉)
	14(기타 수직 물체)
	15(건물)
	16(교통 표지)
	17(교통 신호)
	18 (기타)

• 차량 객체 중 트럭의 가중치 낮음.
• 바이크, 기타 차량, 라이더 등에 대해 test로 넣은 이미지에 없는 지 0으로 mIoU결과값 추론
• 교통용 콘, 봉, 교통 표지, 교통 신호 등 작은 객체에 대한 정확도 낮음.
• loss값에 비교하여 mIoU값이 불안정. 18번 클래스 기타에 대해 loss값이 맞춰진 것으로 추측

inference time

모델 구조에 대해서 현재 환경에 약 9.3ms

DDRNet에 대한 개선 방향

• 폴더와 클래스별로 가중치를 다르게 개선
• 폴더별로 클래스 분포를 확인
• 증강 기법 추가(우천, 강한 강원 등에 대비) - 색상에 제한되는 객체(신호등의 경우 Red, Blue, Green)를 신경쓸 필요가 없으니 색상 변환 등도 추가할 예정.
• inference time을 개선할 방법 모색

모델 비교

mIoU값에 대해서는 Deeplabv3가 가장 높게 나오는 중이나 모델의 크기와 추론 시간에 대한 정확한 정보 필요. YOLOv11_m은 YOLOv11_s 학습 결과가 나온 이후에 해당 값에 대해서 비교. 추론 시간이 모델의 사이즈가 낮아짐에 따라서 추론 시간에 장점이 있을 것으로 보임. DDRNet은 추론 시간과 mIoU 값 등 여러 개선이 필요.

주의 요소 : 추론 시간 계산 시 장비에 따라 차이가 있음을 주의 (RTX 4070 super, RTX 5070(컴퓨터실))

Deeplabv3는 학습 진행 중이므로 설정한 Epoch이 진행된 모델 기준으로 mIoU값과 loss값 다시 정리 예정.

https://github.com/chenjun2hao/DDRNet.pytorch 해당 깃허브의 코드로 custom dataset 학습 시도

ETRI dataset에서 다운받은 format은 images -> jpg와 labels -> txt형태로 구성.

해당 깃허브에서는 labels의 형태를 mask를 받아 학습하는 형태로 구성. (data loader에 따라서 mask를 그대로 사용하기도 하고 txt형태의 데이터를 받기도 한다)

label을 불러오는데에는 lst파일을 이용하여 경로를 받는다.

1. txt를 png형태로 데이터셋을 변형

2. 해당 경로에 따른 lst파일 생성

3. 파일 변환이 정상적으로 되었는지 테스트.

1. txt2png 코드

   import os
   import cv2
   import numpy as np
   

#IMAGE_DIR = r"C:\DDRNet\data\ETRI\train\images" #IMAGE_DIR = r"C:\DDRNet\data\ETRI\val\images" IMAGE_DIR = r"C:\DDRNet\data\ETRI\test\images"

#TXT_LABEL_DIR = r"C:\DDRNet\data\ETRI\train\labels" #TXT_LABEL_DIR = r"C:\DDRNet\data\ETRI\val\labels" TXT_LABEL_DIR = r"C:\DDRNet\data\ETRI\test\labels"

#OUTPUT_PNG_DIR = r"C:\DDRNet\data\ETRI\train\masks_png" #OUTPUT_PNG_DIR = r"C:\DDRNet\data\ETRI\val\masks_png" OUTPUT_PNG_DIR = r"C:\DDRNet\data\ETRI\test\masks_png"

CLASS_TO_ID = {str(i): i for i in range(42)}

def convert_txt_to_png(): os.makedirs(OUTPUT_PNG_DIR, exist_ok=True)

txt_files = [f for f in os.listdir(TXT_LABEL_DIR) if f.endswith('.txt')]
print(f"{len(txt_files)}개의 .txt 파일을 변환.")

processed_count = 0
for txt_filename in txt_files:
    base_filename = os.path.splitext(txt_filename)[0]

    img_path = None
    for ext in ['.jpg', '.jpeg', '.png']:
        potential_path = os.path.join(IMAGE_DIR, base_filename + ext)
        if os.path.exists(potential_path):
            img_path = potential_path
            break

    if not img_path:
        print(f"원본 이미지가 존재하지 않음.")
        continue

    img = cv2.imread(img_path)
    if img is None:
        print(f"경로 오류")
        continue
    height, width, _ = img.shape

    mask = np.zeros((height, width), dtype=np.uint8)

    txt_path = os.path.join(TXT_LABEL_DIR, txt_filename)
    with open(txt_path, 'r') as f:
        for line in f.readlines():
            parts = line.strip().split()
            if len(parts) < 3:
                continue

            class_id_str = parts[0]

            if class_id_str not in CLASS_TO_ID:
                continue
            mask_value = CLASS_TO_ID[class_id_str]

            if len(parts[1:]) % 2 != 0:
                continue

            try:
                normalized_coords = np.array(parts[1:], dtype=np.float32).reshape((-1, 2))

                pixel_coords = (normalized_coords * np.array([width, height])).astype(np.int32)

            except ValueError:
                continue

            cv2.fillPoly(mask, [pixel_coords], color=mask_value)

    output_path = os.path.join(OUTPUT_PNG_DIR, base_filename + '.png')
    cv2.imwrite(output_path, mask)
    processed_count += 1
    if processed_count % 100 == 0:
        print(f"{processed_count}/{len(txt_files)} 파일 처리 완료...")

print(f"변환 완료")

if name == 'main': convert_txt_to_png()

2. lst 파일 생성
```python
import os

def create_lst_files(base_path):
    splits = ['train', 'val', 'test']

    for split in splits:
        # 예: C:\DDRNet\data\ETRI\train\images
        image_dir = os.path.join(base_path, split, 'images')

        if not os.path.isdir(image_dir):
            print(f"폴더의 위치 찾을 수 없음.")
            continue

        lst_content = []

        image_files = os.listdir(image_dir)

        jpg_files = sorted([f for f in image_files if f.lower().endswith('.jpg')])

        for image_file in jpg_files:
            base_name = os.path.splitext(image_file)[0]

            image_path_relative = f"{split}/images/{base_name}.jpg"
            label_path_relative = f"{split}/labels/{base_name}.txt"

            line = f"{image_path_relative} {label_path_relative}"
            lst_content.append(line)

        if lst_content:
            lst_file_path = os.path.join(base_path, f"{split}.lst")
            with open(lst_file_path, 'w') as f:
                f.write('\n'.join(lst_content))
            print(f"✅ '{lst_file_path}' (총 {len(lst_content)} 줄)")

dataset_base_path = r'C:\DDRNet\data\ETRI'
create_lst_files(dataset_base_path)

3. 파일 변환이 정상적으로 되었는지 테스트.

   import os
   import cv2
   import numpy as np
   from PIL import Image
   

PNG_MASK_DIR = r"C:\DDRNet\data\ETRI\train\masks_png" OUTPUT_VIS_DIR = r"C:\DDRNet\data\ETRI\train\masks_visualized"

dataset이 총 42종이므로 클래스를 41까지 색상을 분리해서 확인

COLOR_PALETTE = [ (0, 0, 0), # 0: 배경 (Black) (128, 0, 0), # 1: Dark Red (0, 128, 0), # 2: Dark Green (128, 128, 0), # 3: Dark Yellow (0, 0, 128), # 4: Dark Blue (128, 0, 128), # 5: Dark Magenta (0, 128, 128), # 6: Dark Cyan (128, 128, 128), # 7: Gray (64, 0, 0), # 8: (192, 0, 0), # 9: (64, 128, 0), # 10: (192, 128, 0), # 11: (64, 0, 128), # 12: (192, 0, 128), # 13: (64, 128, 128), # 14: (192, 128, 128), # 15: (0, 64, 0), # 16: (128, 64, 0), # 17: (0, 192, 0), # 18: (128, 192, 0), # 19: (0, 64, 128), # 20: (128, 64, 128), # 21: (0, 192, 128), # 22: (128, 192, 128), # 23: (64, 64, 0), # 24: (192, 64, 0), # 25: (64, 192, 0), # 26: (192, 192, 0), # 27: (64, 64, 128), # 28: (192, 64, 128), # 29: (64, 192, 128), # 30: (192, 192, 128), # 31: (0, 0, 64), # 32: (128, 0, 64), # 33: (0, 128, 64), # 34: (128, 128, 64), # 35: (0, 0, 192), # 36: (128, 0, 192), # 37: (0, 128, 192), # 38: (128, 128, 192), # 39: (64, 0, 64), # 40: (192, 0, 64), # 41: ]

def visualize_masks(): os.makedirs(OUTPUT_VIS_DIR, exist_ok=True)

png_files = [f for f in os.listdir(PNG_MASK_DIR) if f.endswith('.png')]
print(f"총 {len(png_files)}개의 .png 마스크를 시각화합니다.")

processed_count = 0
for png_filename in png_files:
    png_path = os.path.join(PNG_MASK_DIR, png_filename)

    mask = cv2.imread(png_path, cv2.IMREAD_UNCHANGED)

    if mask is None:
        print(f"경고: 마스크 파일 {png_path}를 읽을 수 없습니다. 건너뜁니다.")
        continue

    height, width = mask.shape
    colored_mask = np.zeros((height, width, 3), dtype=np.uint8)

    # 각 픽셀의 클래스 ID에 따라 색상 적용
    for class_id in range(len(COLOR_PALETTE)):
        indices = (mask == class_id)

        colored_mask[indices, 0] = COLOR_PALETTE[class_id][0] # Blue 채널
        colored_mask[indices, 1] = COLOR_PALETTE[class_id][1] # Green 채널
        colored_mask[indices, 2] = COLOR_PALETTE[class_id][2] # Red 채널

    output_path = os.path.join(OUTPUT_VIS_DIR, png_filename)
    cv2.imshow("visualize", colored_mask)
    cv2.waitKey()
    #cv2.imwrite(output_path, colored_mask)

    processed_count += 1
    if processed_count % 100 == 0:
        print(f"{processed_count}/{len(png_files)} 파일 시각화 완료...")

if name == 'main': visualize_masks()

```

작동 영상

데이터셋 조정을 마치고 학습을 진행한 결과.

명령어 python tools/train.py --cfg experiments/cityscapes/ddrnet_39.yaml prompt 내용

loss값이 0으로 고정되고 Acc값도 소수점의 자리에 위치하는 것을 확인.

dataset에 255와 같은 mask 값이 존재하는지 데이터 결함 테스트도 진행했으나 dataset 자체에는 문제가 없다는 것을 확인. numpy 버전에 대해서 문제가 있는것으로 추정되나 현재 GPU와 알맞은 pytorch 버전에 대해서 환경을 맞출 수 없기에 다른 github code를 사용하는 방법으로 회선하여 학습을 시도해보는 것을 목표로 진행하기로 함.

학습 관련 기능이 들어있는 github DDRNet.pytorch, deci.ai 시도 Segmentation-Pytorch 학습 시도 중

Cityscapes dataset이 아닌 custom dataset을 학습하기에 data loader와 pkl등의 코드 생성중

Segmentation 작업을 수행하는 모델들은 속도를 챙기기 위해 Dilated Convolution등을 사용하는데 해당 Conv layer의 단점은 픽셀의 정보를 건너뛰어 수집하기 때문에 특정 패턴의 정보를 받아들이지 못하는 경우가 많다.

DDRNet에서는 Dilated Convolution작업을 수행하지 않는것은 아니지만 정보의 보존을 위해 Dual-resolution network(이중 해상도 네트워크)를 제안한다.

DDRNet은 하나의 trunk에서 시작하여 high-resolution feature maps과 다운샘플링을 거친 rich sementic information maps을 추출한다. 두 branch는 정보 융합을 위해 bilateral connection이 진행된다.

rich semantic information maps의 경우 이후 Segment Head에 들어가기 이전 DAPPM 모듈 내에서 multi-scale context information을 추출하고 융합하여 최종 특성맵을 완성시킨다.

branch : 데이터 처리 경로. trunk : 병렬 branch로 나뉘기 전 공통 부분. bilateral connection : 양방향 연결 DAPPM(Deep Aggregation Pyramid Pooling Module) :

DDRNet Architecture

RB(Residual Basic Blocks) : Convolution Layer로써 특성 추출. RBB(Bottleneck Block) : feature dimension을 확장하여 다양한 종류의 특징 생성. Seg.Head : Segment Head(점선 Head와 실선 Head 구별)

RB 1/32의 상단에 RB 1/8에서 점선으로 Seg. Head가 존재하는데 해당 Head는 Auxiliary Loss(보조 손실값)으로 DNN이 역전파를 진행하는 과정에서 초반 레이어에 영향이 적게 미치는 문제를 해결하기 위하여 중간의 Loss값을 계산하여 이후에 합치는 과정을 거친다. 비교적 메인 Seg. Head보다는 간단한 구조로 제작되어있다.

Deep Supervision

학습이 완료되고 난 이후에는 모델에 $$L_a$$는 포함되지 않는다.

$$L_f = L_n + α*L_a$$

$$L_f$$ : 최종 손실 $$L_n$$ : 일반 손실 $$α * L_a$$ : 보조 손실

논문에서는 α값을 0.4로 두어 가중치 값을 조정하였다.

bilateral fusion

저해상도의 특징맵과 고해상도의 특징맵의 정보를 교환하는 네트워크로 좌측은 저 해상도의 branch에서 가져온 특성맵이고 우측은 고해상도의 branch에서 가져온 특성맵이다.

Low-resolution branch는 3 x 3 convolution layer에 대해 256의 채널 수를 가지고 High-resolution brach는 3 x 3 convolution layer에 대해 128의 채널 수를 가진다. Low-resolution branch에서 High-resolution brach로 넘어가는 경우에는 1 x 1 convolution layer를 거치며 채널 수를 128로 줄이고 이후 UpSample과정을 통해 이미지 사이즈를 늘린다. High-resolution brach에서 Low-resolution branch로 넘어가는 경우에는 3 x 3 Convolution layer를 Stride값을 2로 설정하여 채널 수를 256으로 늘리는 것과 동시에 사이즈를 줄인다.

Low-resolution branch의 채널 수가 많은 이유는 낮은 해상도에 대해 더 깊은 수준의 특징을 가지고 싶어하기 때문이며 High-resolution brach가 높은 해상도에 대해서 특성은 추출하나 너무 연산량과 정확도에 대해 중간 지점을 맞추기 위함이다.

RB 1/8에서 RB 1/32로 넘어가는 경우는 UpSample을 x4와 x8을 하는 과정을 거치며 RB 1/32에서 RB 1/8로 넘어가는 경우는 stride 값을 4와 8로 설정하는 과정을 거친다.

DAPPM Module

• kernel = 5, stride=2 : 1/128 size 특성맵, 이후 1 x 1 conv를 거치고 UpSample하며 채널 수 조정
• kernel = 9, stride=4 : 1/256 size 특성맵, 이후 1 x 1 conv를 거치고 UpSample하며 채널 수 조정
• kernel = 17, stride=8 : 1/512 size 특성맵, 이후 1 x 1 conv를 거치고 UpSample하며 채널 수 조정
• kernel = H * W : (1, 1) size의 특성맵, 이후 1 x 1 conv를 거치고 UpSample하며 채널 수 조정

각 UpSample을 통해 확장된 특성맵은 3 x 3 Conv layer를 거치며 부자연스러운 특성들을 정비한다.

X : 1 / 64의 사이즈를 가진 특성맵 $$y_i$$ : 1 / 64의 사이즈를 가진 특성맵

$$y_i$$

• $$i = 1$$
• $$1 < i < n$$
• $$i = n$$

1번을 제외한 나머지는 UpSample과정을 거치고 난 후 이전 사이즈의 정보를 더한 이후 Conv(3 * 3)을 진행.

위 과정을 거치며 각기 다른 size의 정보를 작은 size부터 넓은 size의 정보로 누적시켜 사이즈간 정보를 정교하게 만든다.

$$C(1 * 1)$$ : 1 x 1 Conv layer $$C(3 * 3)$$ : 3 x 3 Conv layer U : UpSample P : Pooling payer, P(global)은 Kernel = H x W을 의미 i : scale 크기, 그림에서 i의 최대값은 n n : scale의 마지막 j : kernel size k : stride

성능 지표

GTX 2080Ti로 돌린 모델과 비교했을 때 MIoU값과 FPS가 높은 것을 확인할 수 있다.

Carla 주행 경로 선택 과정

Client의 ego 차량은 PythonAPI.carla.agents.navigation폴더의 모듈들에 의해서 경로를 설정하고 조정하는데 그 중 global_route_planner에 경로를 설정하는 알고리즘이 들어있다.

GlobalRoutePlanner클래스를 생성할 때 _build_topology()함수를 통해 Carla Map의 Road Segment의 목록(list)을 받는다. 도로의 시작 waypoint와 끝 waypoint 객체에 대한 3D 좌표를 받고 도로에서 이동 가능한 영역의 waypoint들의 집합을 path에 저장한다. path에 저장되는 값은 이동 가능한 waypoint들의 리스트를 나타내는 것으로 단위는 하나의 곡선 도로나 교차로와 교차로를 잇는 직선을 의미한다.(각 차선에 대해 구별-> 2차선의 경우 2개)

path에 저장된 경로들은 build_graph()함수를 통해 실제 경로 탐색에 쓰일(path_search) netwrokx 그래프를 생성한다.

이후 find_loose_ends()와 lane_change_link()함수를 실행하여 그래프 상에 막다른 길이나 차선 변경이 가능한 영역에 대해 가중치 값이 0인 특수 엣지를 추가하는 등의 작업을 수행한다. (가중치값이란 후에 A* 알고리즘을 통해 최단 경로 계산에 쓰이는 값을 의미)

GlobalRoutePlanner클래스를 생성한 이후 path_search()함수를 반복적으로 실행하여 A* 알고리즘을 기반으로 경로 비용을 계산하여 최단 경로를 구성하는 path내에 포함된 값들의 리스트를 반환한다.

def _path_search(self, origin, destination):
    start, end = self._localize(origin), self._localize(destination)

    route = nx.astar_path(
        self._graph, source=start[0], target=end[0],
        heuristic=self._distance_heuristic, weight='length')
    route.append(end[1])
    return route

A* 알고리즘 (Client)

한점의 출발 지점에서 목표 지점까지 가는 최단 경로를 찾아내는 그래프 탐색 알고리즘으로 Carla에서 사용하는 path planning의 기반이다.

$$f(n) = g(n) + h(n)$$

• $$g(n)$$ : 출발 지점부터 현재 지점(n)까지에 대한 비용.
• $$h(n)$$ : 현재 지점(n)부터 목표 지점까지 도달하기까지 예상되는 비용.
• $$f(n)$$ : 총 예상 비용(출발 지점부터 목표 지점까지)

n은 현재 위치로 현재 위치까지 오기까지의 비용과 앞으로 예상되는 비용들을 계산할 때 중심 위치이다. n은 여러 위치에서 계산되며 그 중 최종 값인 f(n)이 가장 적게 나오는 값을 최단 경로로 설정한다.

| 0 | 0 | 0 | 0 | 
| 0 | 0 | 0 | 0 | 
| 0 | 0 | 0 | 0 | 
| 0 | 0 | 0 | 0 | 

좌측 하단을 출발로 우측 상단으로 이동한다고 가정하고 한칸당 1의 비용을 가질 때 [1, 2]의 위치를 n이라고 가정하게 되면 n까지 가는 방법은 여러가지가 있게 된다.

| 0 | 7 | 8 | 9 | 
| 0 | 6 | 0 | 0 | 
| 0 | 5 | 4 | 0 | 
| 1 | 2 | 3 | 0 | 

다음의 경로로 움직이게 될 경우 n을 가기까지 걸린 비용은 5가 되고 n부터 목표 지점까지는 4의 비용을 가져 총 9의 비용을 가진다. $$f(n) = g(n) + h(n)$$ 가 $$9 = 5 + 4$$ 형태로 이루어진 것이다.

| 6 | 7 | 8 | 9 | 
| 5 | 4 | 0 | 0 | 
| 0 | 3 | 0 | 0 | 
| 1 | 2 | 0 | 0 | 

다음의 경우는 $$f(n) = g(n) + h(n)$$ 가 $$9 = 3 + 6$$ 형태로 이루어진 것이다.

A* 알고리즘은 다음과 같은 그래프에서 최단 비용을 소모하는 경로를 찾아 최종 경로를 찾게 된다.

| 0 | 0 | 0 | 7 | 
| 0 | 4 | 5 | 6 | 
| 0 | 3 | 0 | 0 | 
| 1 | 2 | 0 | 0 | 

$$f(n) = g(n) + h(n)$$ -> $$7 = 3 + 4$$

차선 변경의 경우는 비용이 0이 되어 차선 변경이 들어가더라도 비용값에 대해 영향을 미치게 되지 않게 설정된다.

Server에서 사용하는 알고리즘

Carla Simulator는 Server내에서 Actor의 상태를 파악하고 각 Actor가 현재 위치한 waypoint의 정보에 따라 차선, 신호등, 인근 차량 등을 고려하여 경로를 계산한다. 계산하는 주기는 world.tick()마다 반복된다.

경로 추종 제어의 순서는 다음과 같다.

1. 목표 지점(waypoint)설정

차량의 현재 위치에서 가능한 경로 중 다음 waypoint 하나를 목표 지점으로 선정.

2. 차량 제어 값 계산

목표 waypoint까지 도달하기 위해 가속, 제동, 회전 등의 값을 계산.

3. 명령 수행

계산된 각 값들을 각 Actor에 전달하여 수행.

4. 최종 목적지까지 반복

각 world.tick()마다 1단계부터 3단계를 반복하여 수행.

A* 알고리즘을 사용하지 않는 이유는 Server에 돌아다니는 차량의 경우 최종 목적지가 정해지지 않고 waypoint(1m)단위로 랜덤으로 이동하기 때문이다. 따라서 매 world.tick()마다 이동 가능한 waypoint를 찾아 그 중 하나를 선택하고 이동하는 방식을 사용하기에 특정한 알고리즘을 사용하지 않는다.