연금술사

딥러닝 모델에서 이미지 분할 작업의 성능을 향상시키기 위해 다양한 손실 함수가 사용됩니다. 이번 글에서는 BCE부터 MCCLoss까지 다양한 손실 함수들을 정리해보겠습니다.

순서와 간단한 요약 아래와 같습니다.

1. BCE (Binary Cross Entropy): 기본적인 손실 함수 중 하나로, 주로 이진 분류 문제에서 사용되는 함수.
2. DICE: 두 집합간의 유성을 측정하는 지표입니다. $ \text{DICE}= 2|AUB| / (|A|+|B|)$
3. Jaccard loss: 두 유사성을 측정하느 또 다른 방법. $\text{Jaccard}=|A \cap B| / |A \cup B|$
4. Tversky Loss: Jaccard 손실 함수의 가중치 버전.  $\text{Tversky}=\frac{ |A \cap B| }{ |A \cap B| + \alpha |A-B| + \beta |B-A| } $
5. Focal loss: 클래스 불균형을 해결하기 위한 손실함수 $\text{Focal L}= -\frac{1}{N} \sum_{i=1}^{N}(1-p_{i})^{\lambda}log(p_{i})$
6. SoftBCEWithLogitsLoss: BCE에 label smoothing이 적용
7. MCCLoss: Confusion matrix의 각 항목을 지표화 시켜놓은 손실함

BCE (Binary Cross Entropy)

기본적인 손실 함수 중 하나로, 주로 이진 분류 문제에서 사용됩니다. BCE는 다음과 같은 수식으로 정의됩니다:

$\text{BCE} = - \frac{1}{N} \sum_{i=1}^{N} \left[ y_i \log(p_i) + (1 - y_i) \log(1 - p_i) \right]$

여기서 $y_i$​는 실제 라벨, $p_{i}$​는 예측 확률입니다. BCE는 픽셀 단위로 계산되기 때문에 이미지 분할 문제에도 적용할 수 있습니다.

DICE

DICE 손실 함수는 분할 작업에서 자주 사용되는 또 다른 손실 함수입니다. DICE 계수는 두 집합 간의 유사성을 측정하는 지표로, 손실 함수로 사용될 때는 다음과 같이 정의됩니다:

$\text{DICE} = \frac{2|A \cap B|}{|A| + |B|}$

손실 함수로 사용할 때는 1에서 DICE 계수를 뺀 값을 사용합니다. DICE 손실 함수는 미분 가능하여 역전파 과정에서 사용할 수 있습니다. IoU랑 거의 흡사하나, IoU은 Intersection인 $ A \cap B| $에 그치지만, 분자 더 집중할 수 있도록 2배를 하였습니다.

Jaccard Loss

Jaccard 계수는 두 집합 간의 유사성을 측정하는 또 다른 방법입니다. Jaccard 손실 함수는 다음과 같이 정의됩니다:

$​\text{Jaccard} = \frac{|A \cap B|}{|A \cup B|}$

Jaccard 손실 함수 역시 픽셀 단위로 계산되며, 미분 가능하여 역전파 과정에서 사용할 수 있습니다

Tversky Loss

Tversky 손실 함수는 Jaccard 손실 함수의 가중치 버전입니다. Tversky 지수는 다음과 같이 정의됩니다:

$\text{Tversky}(A,B; \alpha, \beta) = \frac{|A \cap B|}{|A \cap B| + \alpha|A - B| + \beta|B - A|} $
​여기서 $\alpha$ 와 $\beta$ 는 가중치로, 두 값에 따라 민감도를 조절할 수 있습니다.

Focal Loss

Focal Loss는 클래스 불균형 문제를 해결하기 위해 제안된 손실 함수입니다. Focal Loss는 다음과 같이 정의됩니다:

$ \text{Focal Loss} = - \frac{1}{N} \sum_{i=1}^{N} (1 - p_i)^{\gamma} \log(p_i)$

여기서 $\gamma$ 는 조정 가능한 매개변수로, 일반적으로 2로 설정됩니다. Focal Loss는 특히 어려운 클레스를 학습하는 데 유리합니다.

SoftBCEWeightLogitLoss

SoftBCEWithLogitsLoss는 label smoothing 기법을 적용한 BCE 손실 함수입니다. 이는 모델이 더 일반화된 예측을 할 수 있도록 도와줍니다. 수식은 다음과 같습니다:

$\text{SoftBCEWithLogitsLoss} = - \frac{1}{N} \sum_{i=1}^{N} \left[ y_i' \log(\sigma(p_i)) + (1 - y_i') \log(1  \sigma(p_i)) \right]$

여기서 $y_i' = y_i (1 - \epsilon) + 0.5 \epsilon$이며, $ \epsilon $은 smoothing 계수, $\sigma$ 은 시그모이드 함수입니다.

BCE와의 차이는 라벨($y'$)을 그대로 쓰는게 아니라 [0,1]사이의 값을 사용한다는 것입니다. 이 SoftBCE을 사용하려면 통상의 라벨이 0또는 1이기 때문에, Label을 수정해서 label smoothing을 적용해야합니다.

$\text{BCE} = - \frac{1}{N} \sum_{i=1}^{N} \left[ y_i \log(p_i) + (1 - y_i) \log(1 - p_i) \right]$

MCC

MCC(Matthews Correlation Coefficient) 손실 함수는 분류 문제에서의 상관 관계를 측정하는 지표로, 다음과 같이 정의됩니다:

$\text{MCC} = \frac{TP \cdot TN - FP \cdot FN}{\sqrt{(TP + FP)(TP + FN)(TN + FP)(TN + FN)}}$​

여기서 TP는 True Positive, TN은 True Negative, FP는 False Positive, FN은 False Negative를 의미합니다. MCCLoss는 분할 작업에서의 성능을 효과적으로 평가할 수 있습니다.

import torch
import torch.nn.functional as F

def mcc_loss(y_true, y_pred):
    y_pred_pos = torch.round(torch.clamp(y_pred, 0, 1))
    y_pred_neg = 1 - y_pred_pos
    
    y_pos = torch.round(torch.clamp(y_true, 0, 1))
    y_neg = 1 - y_pos
    
    TP = torch.sum(y_pos * y_pred_pos)
    TN = torch.sum(y_neg * y_pred_neg)
    FP = torch.sum(y_neg * y_pred_pos)
    FN = torch.sum(y_pos * y_pred_neg)
    
    numerator = TP * TN - FP * FN
    denominator = torch.sqrt((TP + FP) * (TP + FN) * (TN + FP) * (TN + FN))
    
    return 1 - (numerator / (denominator + 1e-7))  # 1e-7은 ZeroDivision 방지

# 예시 사용법
y_true = torch.tensor([1, 0, 1, 1], dtype=torch.float32)
y_pred = torch.tensor([0.9, 0.1, 0.8, 0.7], dtype=torch.float32)
loss = mcc_loss(y_true, y_pred)
print(loss)

기본개념: 코루틴, 메인루틴, 서브루틴

코루틴을 이해하기위해서는 메인루틴과 서브루틴을 이해하고 있으면, 더 쉽게 이해가 됩니다[이전포스팅]. 이 3가지 루틴을 정리하면 아래와 같습니다.

• 메인루틴(Main routine): 메인루틴은 보통 프로그램의 시작점이며, 프로그램의 주 흐름을 담당합니다. 메인루틴은 일련의 작업을 수행하고 다른 서브루틴이나 코루틴을 호출할 수 있습니다. 프로그램이 시작되는 메인코드라고 생각하면 됩니다.
• 서브루틴(subroutine): 메인루틴에서 호출되는 함수, 또는 서브루틴에서 호출되는 함수들을 의미합니다. 즉, 다른 루틴에서 호출되는 경우를 의미합니다.
• 코루틴(Coroutine): 메인루틴에서 호출되지만, 코루틴은 실행되는 도중에, 일시중단되어 다시 메인루틴으로 전환되었다가 다시 코루틴으러 전환될 수 있는 "제어흐름"이 가능한 루틴을 의미합니다.

코루틴과 제너레이터

코루틴과 제너레이터 중, 코루틴이 더 일반적인 개념이며 코루틴 중에 반복가능한 데이터를생성하는 한 종류가 제너레이터라고 할 수 있습니다. 이 제너레이터와 코루틴을 굳이 비교하자면 아래와 같이 비교해볼 수 있습니다.

코루틴의 간단한 예시입니다.

def simple_coroutine():
    print("Coroutine started")
    x = yield  # 처음 호출될 때까지 대기
    print(f"Received: {x}")
    y = yield x * 2  # yield 우측에 있는 값은 반환하는 값입니다.
    print(f"Received: {y}")

# 코루틴 생성
coro = simple_coroutine()

# 코루틴 시작
print(next(coro))  # Coroutine started, 첫 번째 yield까지 실행

# 데이터 전송
print(coro.send(10))  # Received: 10

# 데이터 전송
coro.send(20)  # Received: 20

위의 예시처럼 코루틴을 시작하려면 반드시 "next()"을 호출하여 실행시켜야 합니다. 그렇지 않으면 아래와 같이 에러가 발생합니다. 즉, coroutine 시작지점까지는 함수가 실행되어 대기중이어야합니다.

Cell In[24], line 1
----> 1 coro.send(20)

TypeError: can't send non-None value to a just-started generator

코루틴과 마찬가지로, generator도 yield의 반환이 없으면 StopIteration 예외가 발생합니다.

In [1]: def simple_generator():
   ...:     yield 1
   ...:     yield 2
   ...:     yield 3
   ...: 

In [2]: gen = simple_generator()

In [4]: next(gen)
Out[4]: 1

In [5]: next(gen)
Out[5]: 2

In [6]: next(gen)
Out[6]: 3

In [7]: next(gen)
StopIteration                             Traceback (most recent call last)
Cell In[7], line 1
----> 1 next(gen)

StopIteration:

코루틴의 4가지 상태: GEN_CREATE, GEN_RUNNING, GEN_SUSPENDED, GEN_CLOSE

1. GEN_CREATED (생성 상태): 코루틴이 생성되었지만 아직 시작되지 않은 상태입니다.

• 특징: __next__() 또는 send(None)이 호출되기 전의 상태입니다.

def coroutine_example():
    yield

coro = coroutine_example()
print(coro)  # 코루틴 객체가 생성된 상태, 아직 실행되지 않음

2. GEN_RUNNING (실행 중 상태): 코루틴이 현재 실행 중인 상태입니다.

• 특징: 코루틴이 실행되고 있으며, 내부적으로 yield 문을 만날 때까지 실행됩니다. 실행 중에 다시 재개될 수 없습니다.

def coroutine_example():
    print("Running")
    yield

coro = coroutine_example()
next(coro)  # "Running" 출력, 실행 중 상태

3. GEN_SUSPENDED (일시 중지 상태): 코루틴이 yield 문에서 일시 중지된 상태입니다. 주의해야할 것은 RUNNING 상태는 yield을 만나기 전까지 실행되는 것인데, 여기서는 yield가 이미 실행된 상태를 의미합니다.

그리고, inpsect.getgeneratorstate 함수로 코루틴의 상태를 조회해보면 GEN_SUSPENDED을 확인할 수 있습니다.

>>> def coroutine_example():
>>>     yield "Suspended"
>>>  

>>> coro = coroutine_example()
>>> print(next(coro))

>>> import inspect
>>> inspect.getgeneratorstate(coro)
>>> 'GEN_SUSPENDED'

4. GEN_CLOSED(종료상태): 코루틴이 종료된 상태입니다. 종료되면 다시 시작할 수 없습니다.

• 특징: 모든 코드가 실행되었거나 close() 메서드가 호출된 상태입니다.

def coroutine_example():
    yield "Running"
    return "Done"

coro = coroutine_example()
print(next(coro))  # "Running" 출력
try:
    next(coro)  # StopIteration 예외 발생
except StopIteration as e:
    print(e.value)  # "Done" 출력, 종료 상태

각 상태를 연속선상으로 확인해보면 아래처럼 확인해볼 수 있습니다.

import inspect

def coroutine_example():
    print("Coroutine started")
    x = yield "Suspended at first yield"  # GEN_SUSPENDED 상태로 진입
    print(f"Received: {x}")
    y = yield "Suspended at second yield"  # 다시 GEN_SUSPENDED 상태로 진입
    print(f"Received: {y}")
    return "Done"  # GEN_CLOSED 상태로 진입

# 코루틴 객체 생성 (GEN_CREATED 상태)
coro = coroutine_example()
print(inspect.getgeneratorstate(coro))  # GEN_CREATED

# 코루틴 시작 (GEN_RUNNING 상태)
print(next(coro))  # "Coroutine started" 출력, "Suspended at first yield" 반환
print(inspect.getgeneratorstate(coro))  # GEN_SUSPENDED

# 데이터 전송 및 재개 (GEN_RUNNING 상태로 변환 후 GEN_SUSPENDED 상태로 다시 변환)
print(coro.send(10))  # "Received: 10" 출력, "Suspended at second yield" 반환
print(inspect.getgeneratorstate(coro))  # GEN_SUSPENDED

# 데이터 전송 및 종료 (GEN_RUNNING 상태로 변환 후 GEN_CLOSED 상태로 변환)
try:
    coro.send(20)  # "Received: 20" 출력, StopIteration 예외 발생
except StopIteration as e:
    print(e.value)  # "Done" 출력
print(inspect.getgeneratorstate(coro))  # GEN_CLOSED

while True 구문을 이용한 코루틴

while True를 사용한 코루틴으로 지수 이동 평균 (Exponential Moving Average, EMA)을 계산하는 것들을 해볼 수 있습니다.

아래는 그 예제입니다.

def ema_coroutine(initial_ema, alpha):
    ema = initial_ema
    print(f"Starting EMA with initial value: {ema}")
    while True:
        new_value = yield ema
        ema = (alpha * new_value) + ((1 - alpha) * ema)
        print(f"Updated EMA to: {ema}")

# 코루틴 생성 및 초기화
alpha = 0.1
initial_ema = 50  # 초기 EMA 값
ema_gen = ema_coroutine(initial_ema, alpha)
current_ema = next(ema_gen)  # 코루틴 시작, 최초의 EMA 값을 받음

# 새로운 값으로 EMA 업데이트
print(ema_gen.send(55))  # 새 데이터 포인트 55를 전달하고 업데이트된 EMA 출력
print(ema_gen.send(60))  # 새 데이터 포인트 60을 전달하고 업데이트된 EMA 출력

# 코루틴 종료
ema_gen.close()

yield 에서 값을 반환하는 순서를 주의할 필요가 있습니다.

1. send(50): 코루틴에 값을 보내면 new_value에 할당합니다. 이후 print()문까지 도달하여 출력합니다. 
2. yield ema: 그리고나서 업데이트된 ema을 반환하여 50.5가 출력됩니다. 

코루틴을 초기화해주면 데코레이터: coroutine

DICE 손실 함수(Dice Loss)는  Sørensen–Dice coefficient 라고도 불리며 주로 의료 영상 분석과 같은 분야에서 세그멘테이션 문제에 많이 사용됩니다. 이 손실 함수는 이진 분류 작업에서 두 샘플 집합의 유사도를 측정하기 위해 사용되며, 특히 불균형한 데이터셋에서 좋은 성능을 보입니다.

DICE 계수는 두 샘플 집합의 유사도를 측정하는데 사용되며, 다음과 같이 정의됩니다:

$DICE = \frac{2\times|X \cap Y|}{|X|+|Y|}$​

1. $|X \cap Y|$: 두 집합의 교집합의 크기입니다.
2. |X|, |Y|: 각각 집합의 크기입니다.

DICE 손실함수

DICE 손실 함수는 1에서 DICE 계수를 뺀 값으로 정의됩니다. 이는 계수가 1에 가까울수록 손실이 작아지며, 예측과 실제 값 사이의 유사도가 높음을 의미합니다. 손실 함수는 다음과 같이 표현됩니다:

DICE Loss=1−DICE

이 손실 함수는 특히 클래스 간 불균형이 클 때 유용하며, 소수 클래스의 중요한 특징을 놓치지 않도록 도와줍니다.

import torch

class DiceLoss(torch.nn.Module):
    def __init__(self, smooth=1.):
        super(DiceLoss, self).__init__()
        self.smooth = smooth

    def forward(self, inputs, targets):
        inputs = torch.sigmoid(inputs)
        intersection = (inputs * targets).sum()
        dice = (2. * intersection + self.smooth) / (inputs.sum() + targets.sum() + self.smooth)
        return 1 - dice

DICE loss function의 미분 가능성(differentiable)

객체인식(Object detection)에서의 IoU(Intesection over Union)은 직접 미분이 안되어, 미분 가능한 형태로 변경하여 계산합니다[UnitBox]. Segmentation에서의 DICE score은 미분이 가능합니다. 특히, segementation은 픽셀단위로 계산하기 때문에 예측픽셀과 정답픽셀간의 차이를 직접 계산할 수 있습니다.

DICE loss은 아래와 같이 정의됩니다.

DICE loss = 1- DICE coefficient  (1)

여기서 TP, FP, FN을 아래와 같이 정의할 수 있습니다. 

• $p_{i}$: predicted probability
• $g_{i}$: label
• TP= $\sum_{i} p_{i}g_{i} $ : 예측과 실제가 둘 다 1인 경우
• FP= $\sum_{i} p_{i}(1-g_{i}) $: 실제가 0일 때, 예측이 1인 경우. $(1- g_{i})$로 indication을 넣음
• FN= $\sum_{i} (1-p_{i})(g_{i})$: 실제가 1일 떄, 예측이 0인 경우의 합

(1)식에서 DICE coefficient만 구하면 loss은 구할 수 있습니다. 따라서, 위를 정의들을 이용하여, DICE coefficient을 다시 정의해보겠습니다.

$\text{DICE coefficient} = \frac{ 2 \sum_{i} p_{i}g_{i} }{\sum_{i} p_{i} + \sum_{i}g_{i} }$

이를 DICE coefficient식에 대입하고 p에 대해서 미분합니다.

$\frac{\partial L}{\partial p_{i}} = \frac{\partial}{\partial p_{i}} (1 - \frac{ 2 \sum_{i} p_{i}g_{i} }{\sum_{i} p_{i} + \sum_{i}g_{i} })$ (2)

$=-\frac{2g_{i} (\sum_{i} p_{i} + \sum_{i}g_{i}) - 2(\sum_{i}p_{i}g_{i})}{(\sum_{i} p_{i} + \sum_{i}g_{i})^{2}}$ (3)

(3)은 나눗셈의 미분으로 계산합니다. 분모는 분모의 제곱으로 들어가고, 분자는 분자미분*분모 + 분자*분모미분으로 계산됩니다. $\frac{f(x)}{g(x)}=\frac{f'(x)g(x) + f(x)g'(x)}{g(x)^{2}} $

(3)식을 보면 DICE loss은 예측값 $p_{i}$에 대해서의 미분값이며, gradient descent을 이용하여 최적화가 가능합니다. 이 식은 예측값 $p_{i}$을 얼마만큼 조정하냐에 따라 손실함수가 바뀐다는 것이냐는거고, 다시 딥러닝 parameter에 대해서 미분하는게 필요하니, 체인룰을 이용하여 ($\frac{\partial L }{\partial \theta } = \frac{\partial L }{\partial p_{i}} \frac{\partial p_{i} }{\partial \theta }$)해볼 수 있습니다.

요약

• 딥러닝에서의 stride: 컨볼루션 연산에서 커널의 이동 범위를 나타냅니다.
• numpy 배열에서의 stride: 배열의 각 차원에서 다음 요소로 이동하기 위한 메모리 상의 바이트 수를 나타냅니다.

보통 딥러닝하시는 분들이면 딥러닝의 CNN layer에서 stride을 들어보셨을 텐데, numpy 에서의 stride가 어떤 개념인지 정리해보겠습니다.

numpy 배열에서의 Stride: 배열의 각 차원에서 다음 요소로 이동하기 위한 메모리 상의 바이트 수

numpy 배열에서의 stride는 메모리에서 다음 요소로 이동하기 위해 건너뛰어야 하는 바이트 수를 나타냅니다. numpy 배열은 연속된 메모리 블록을 사용하여 데이터를 저장하며, stride는 배열의 각 차원을 따라 이동할 때의 바이트 수를 지정합니다. 배열이 슬라이싱되거나 뒤집힐 때 stride가 음수가 될 수 있습니다. 이는 배열이 역순으로 참조되고 있음을 의미합니다.

예를 들어, numpy.array을 3, 3 shape의 unit8 타입의 행렬을 만들었다고 가정해보겠습니다. 여기서 [0,0]의 위치에서 그 다음인 [0, 1]을 읽으려면 uint8이기 때문에 8bit(1byte)을 이동해서 읽어야합니다. 한편, 그 다음행인 [1, 0]을 읽어야하면, 한 행이 3개의 원소를 담고 있기에, 3*8bit(3byte)을 이동해서 읽어야합니다. Axis 1이 컬럼이며, Axis 0이 행이기에, Axis[0](행)을 이동하려면 3bytes (stride=3)으로 표시되어야하며, Axis[1](열)을 이동하려면 1byte (stride=1)을 이동해야합니다.

아래와 같이 3, 3 정수형의 uint8을 생성해보겠습니다. 이 배열의 `.strides`속성은 축이 2개이기에, 2개의 값이 튜플로 (3, 1)로 반환됩니다. 그리고, 위에서 언급했던 것처럼 stride가 (3, 1)입니다.

>>> import numpy as np

>>> arr = np.array([
    [1, 2, 3],
    [4, 5, 6],
    [7, 8, 9]
], dtype=np.uint8)

>>> print("배열:")
>>> print(arr)
>>> print("Stride:", arr.strides)

배열:
[[1 2 3]
 [4 5 6]
 [7 8 9]]
Stride: (3, 1)

Stride은 음수도 될 수 있습니다. 이 행렬을 뒤집는 경우, 다음 원소를 읽기위해서는 이전의 메모리를 참조해야합니다. 예를 들어, 행렬을 한번 뒤짚은 경우라면, `array[0]`이후에 `array[1]`을 읽으려면 +로 바이트를 이동시켜야하는게 아니라 (-)으로 바이트를 이동시켜야합니다.

>>> import numpy as np

>>> arr = np.array([1, 2, 3, 4, 5])
>>> reversed_arr = arr[::-1]

>>> print("원래 배열:", arr)
>>> print("뒤집힌 배열:", reversed_arr)
>>> print("원래 배열의 stride:", arr.strides)
>>> print("뒤집힌 배열의 stride:", reversed_arr.strides)
원래 배열: [1 2 3 4 5]
뒤집힌 배열: [5 4 3 2 1]
원래 배열의 stride: (8,)
뒤집힌 배열의 stride: (-8,)

딥러닝에서의 Stride: 컨볼루션 연산에서 커널의 이동 범위

딥러닝에서의 stride는 주로 컨볼루션 연산에서 사용됩니다. 컨볼루션 연산에서 stride는 커널이 입력 이미지 위를 얼마나 많이 움직이는지를 의미합니다. 예를 들어, stride가 1이면 커널이 한 픽셀씩 이동하며 연산을 수행하고, stride가 2이면 두 픽셀씩 건너뛰면서 연산을 수행합니다.

연관 에러: ValueError: At least one stride in the given numpy array is negative, and tensors with negative strides are not currently supported. (You can probably work around this by making a copy of your array with array.copy().)

이미지를 전처리하는 과정에서 flip 등의 연산이 들어가게되면, 위의 예시처럼 stride가 음수가 되는 경우, 위의 에러가 발생합니다. "적어도 하나 이상의 stride가 음수가 됨을 의미합니다". 따라서 이 경우에는 flip한 array을 다시 메모리에 재할당하여(=copy)하여 처리하면 해결이됩니다.

# 변경전

transform = A.Compose(
    [
        A.RandomCrop(height=224, width=224, p=1),
        A.Resize(224, 224),
        A.HorizontalFlip(p=0.5),
        A.VerticalFlip(p=0.5),
        A.RandomRotate90(p=0.5), 
        A.Normalize(mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225)),
        ToTensorV2(),
    ]    
)


train_dataset = dataset(train_images, train_masks, transform)
train_dataset[0] # 에러 발생

# 변경

if else구문이 아닌, 구문에서의 else의 사용

1. for - else 구문:

for loop가 정상적으로 완료되면 else 구문이 실행됩니다. break으로 중단되는 경우 else 블록은 실행되지 않습니다.

numbers = [1, 2, 3, 4, 5]

for number in numbers:
    if number == 3:
        print("Found 3!")
        break
else:
    print("3 is not in the list.")

2. try-except-else:

try-except에서 예외가 발생하지 않을 때, else 블록이 실행됩니다.

이거 왜 쓰냐는 질문이 종종있는데요. 아래의 예시를 들어보겠습니다. 아래의 try-except구문에서 FileNoteFoundError을 open()함수에서 발생할 수 있다는 것을 경험적으로 알 수 있습니다. 

try:
    file = open('example.txt', 'r')
    content = file.read()
    print(content)
    file.close()
except FileNotFoundError:
    print("File not found.")

이는 open()에서 자주 예외가 발생할 수 있는데, 만일 open이 아니라 my_fun1(), my_fun2()라면 어느구문에서 해당예외가 발생한다고 예측할 수 있을까요? 즉, 실제로 예외가 발생하지 않는 경우에도 try 블록에 많은 코드가 포함되어있어서 디버깅이 어렵습니다. try-except구문에서는 예외가 발생할 수 있는 함수의 호출을 최소한의 블록으로 담아야 더 명시적입니다. 

try:
    my_func1()
    my_func2()
except FileNotFoundError:
    print("file not found)

따라서, 이를 처리하기위해 아래와 같이 try-except-else 구문으로 처리해볼 수 있습니다. 

try:
    file = open('example.txt', 'r')  #open()에 대해서만 예외처리
except FileNotFoundError:
    print("File not found.")
else:
    content = file.read()
    print(content)
    file.close()

이는 EAFP(Easier to ask for forgiveness than permission, 허락보다 용서가 쉽다) 코딩스타일로 `if os.path.exist()`등으로 처리가 가능하나, 파이썬에서는 EAFP스타일을 사용을 많이합니다.

with 구문

with 구문은 크게 2가지로 구성됩니다.

1. `__enter__`: with 구문의 실행 (진입)
2. `__exit__`: with 구문의 종료. + with 구문에 끝에 finally 절의 역할이 수행됩니다.

with 구문은 컨텍스트 관리자 객체를 사용하여, 코드블록의 진입과 종료 시에 특별한 동작을 합니다. 컨텍스트 관리자 객체는 __enter__와 __exit__ 메서드를 구현하여, 코드 블록 진입 시에 __enter__ 메서드를, 종료 시에 __exit__ 메서드를 호출하게 합니다. 즉, 컨텍스트 관리자를 이용하면, 자원 해제, 예외 처리 등을 신뢰성 있게 관리할 수 있습니다.

아래의 예시를 하나 들어보겠습니다. 아래의 예시에서는 파일 객체를 반환하고,

1. with구문으로 실행될때 (__enter__)가 실행되고,
2. with 구문이 종료되면 (__exit__)이 종료되는 것 과 같습니다. 

with open('example.txt', 'r') as file:
    content = file.read()
    print(content)

실제로 `__enter__`, `__exit__`이 있는지 살펴보겠습니다.

In [1]: # open 함수로 반환된 파일 객체의 __enter__와 __exit__ 메서드 확인
   ...: with open('example.txt', 'r') as file:
   ...:     print(hasattr(file, '__enter__'))  # True 출력
   ...:     print(hasattr(file, '__exit__'))   # True 출력
   ...:     content = file.read()
   ...:     print(content)
   ...:
True
True

또 다른 예시로는 사용자 정의 컨테스트를 만들 수 있습니다. 이 때도, 마찬가지로 `__enter__`, `__exit__`의 메직메서드의 구현이 되어있어야합니다.

class Resource:
    def __enter__(self):
        print('Resource allocated')
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        print('Resource released')

# 컨텍스트 관리자 사용
with Resource() as resource:
    print('Using the resource')

Resource allocated  # with 구문이 실행될 때
Using the resource  # print구문
Resource released   # with 구문의 종료

sqlite3나 mysqlDB등을 이용할때의 db connection을 명시적으로 종료하지않고 with구문으로 컨텍스트매니저로 사용할 수 있습니다.

import sqlite3

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')
    results = cursor.fetchall()
    print(results)

contextlib: python 표준라이브러리

1. @contextmanager: 이 데코레이터는 컨텍스트 메니저를 생성할때 코드를 줄여줍니다.

@contextmanager 데코레이터는 컨텍스트 관리자를 생성할 때 사용됩니다. 이 데코레이터를 사용하면 __enter__와 __exit__ 메서드를 명시적으로 구현하지 않고도 컨텍스트 관리자를 만들 수 있습니다. 대신, 함수 내부에서 yield를 사용하여 진입 및 종료 시 실행할 코드를 정의합니다.

아래와 같은 예시를 들어보겠습니다. 

from contextlib import contextmanager
import time

@contextmanager
def timer():
    start = time.time()
    try:
        yield
    finally:
        end = time.time()
        print(f'Elapsed time: {end - start} seconds')

# 사용 예시
with timer():   # start = time.time()이 서브루틴에서 할당됨. yield로 메인루틴으로 전환
    time.sleep(2)  # yield 이후 메인루틴의 실행, with 구문종료시 서브루틴의 finally 구문으로 전환

또 다른 예시로,DB의  FastAPI의 contextmanager을 디팬던시로 사용하는 경우입니다. 

FastAPI에서 contextmanager역할로 `__enter__`, `__exit__`메서드를 구현했지만, contextlib 내의 contextmangager 데코레이터를 이용하면 그럴필요는 없습니다.

아래와 같이 구현이 가능합니다.

from contextlib import contextmanager

@contextmanager
def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

# FastAPI dependency 사용 예시
@app.get("/items/")
async def read_items(db: DBSession = Depends(get_db)):
    # 데이터베이스 작업
    items = db.execute("SELECT * FROM items").fetchall()
    return items

객체표현: __repr__, __str__ 차이

파이썬에서 객체의 설명을 보려면, `repr()`, `str()`함수가 필요합니다. 이 두 함수는  `__repr__`와 `__str__`이 정의되어야 확인할 수 있습니다.  이 두 함수의 차이점은 아래와 같습니다.

공통점

• 두 함수 모두 반환되는 형태가 `str` 입니다.

차이점

• `repr`, `__repr__`은 주로 개발자을 위해 사용되며, 내부 디버깅을 위해 사용되는 표현방식입니다. 주로 `obj`을 직접 호출할 때 사용되고, 문자열 그 자체로 객체가 생성되도록 작성을 합니다.
• `str`, `__str__`은 주로 사용자를 위해 사용되며, 보기쉬운 형태로 전환하거나 타입 캐스팅을 하는 용도로 사용됩니다. 또한, `__str__`이 정의되어있지 않다면, `__rerp__`이 호출되기도 합니다.

from dataclasses import dataclass

@dataclass
class MyClass:
    attribute1: int
    attribute2: str

    def __repr__(self):
        return f"MyClass(attribute1={self.attribute1}, attribute2={self.attribute2!r})"

# 객체 생성
obj = MyClass(attribute1=10, attribute2="example")

# repr() 함수 호출
print(repr(obj))  # MyClass(attribute1=10, attribute2='example')

# str() 함수 호출 - __str__이 정의되지 않았으므로 __repr__이 대신 호출됩니다.
print(str(obj))  # MyClass(attribute1=10, attribute2='example')

from dataclasses import dataclass

@dataclass
class MyClass:
    attribute1: int
    attribute2: str

    def __str__(self):
        return f"MyClass with attribute1: {self.attribute1} and attribute2: {self.attribute2}"

# 객체 생성
obj = MyClass(attribute1=10, attribute2="example")

# repr() 함수 호출 - __repr__이 정의되지 않았으므로 기본 구현이 호출됩니다.
print(repr(obj))  # MyClass(attribute1=10, attribute2='example')

# str() 함수 호출
print(str(obj))  # MyClass with attribute1: 10 and attribute2: example

위의 코드에서 추가로 `MyClass(attribute1=10, attribute2='example')`을 그대로 호출해보면 객체가 생성이 가능합니다.

a = MyClass(attribute1=10, attribute2='example')
a
MyClass(attribute1=10, attribute2='example')

그 외에도, byte 시퀀스로 호출하려면 `byte()`함수를 호출하여 사용이 가능합니다. 이 때도, `__bytes__`매직메서드가 구현되어있어야합니다.

from dataclasses import dataclass

@dataclass
class MyClass:
    attribute1: int
    attribute2: str

    def __repr__(self):
        return f"MyClass(attribute1={self.attribute1}, attribute2={self.attribute2!r})"

    def __str__(self):
        return f"MyClass with attribute1: {self.attribute1} and attribute2: {self.attribute2}"

    def __bytes__(self):
        # 객체를 바이트 시퀀스로 변환
        return f"MyClass(attribute1={self.attribute1}, attribute2={self.attribute2})".encode('utf-8')

# 객체 생성
obj = MyClass(attribute1=10, attribute2="example")
print(bytes(obj))  # b'MyClass(attribute1=10, attribute2=example)'
b'MyClass(attribute1=10, attribute2=example)'

객체 생성의 방법: @classmethod을 이용한 팩토리패턴

아래의 코드는 @classmethod에서 자기 자신의 클레스의 인스턴스를 생성 패턴입니다. 생성자를 꼭 만들필요없고, 대안적으로 아래와 같은 디자인 패턴이 가능합니다.

@dataclass
class MulticlassMetrics:
    accuracy: float
    auroc: float
    prauc: float

    @classmethod
    def calculate(
        cls, labels: np.ndarray, confidences: np.ndarray
    ) -> MulticlassMetrics:
        pred_label = np.argmax(confidences, axis=1)
        accuracy = (pred_label == labels).mean()
        auroc = calculate_auroc_ovr(confidences, labels)
        prauc = calculate_prauc_ovr(confidences, labels)

        return cls(accuracy, auroc, prauc)

    def to_dict(self, prefix=str()) -> dict:
        return {
            prefix + metric_name: round(value, 5)
            for metric_name, value in asdict(self).items()
        }

클레스 해시하기

파이썬 Set과 같은 기본자료구조에 넣으려면, 객체는 모두 hashable해야합니다. 즉, 해시값에 대해서 유일해야합니다. 이 내부적으로 hash()함수를 이용해서 객체를 해시하는데, 이를 이용하려면 `__hash__`을 정의해야합니다. 

이 디자인의 예시는 아래와 같습니다. HPO라는 온톨로지(개념)의 데이터클레스입니다. HPO의 id가 유일하므로 이 hash가 다르면 id가 내용이 각각 다름이 보장됩니다. 따라서, `__hash__`을 id을 기준으로 hash합니다.

@dataclass
class HPO:
    """HPO (node)"""

    id: str
    name: str = str()
    definition: str = str()
    synonyms: Set[str] = field(default_factory=set)
    xref: Set[str] = field(default_factory=set)
    vector: np.ndarray = np.empty(shape=(1,))
    depth: int = 0
    ic: float = -1

    subclasses: Set[HPO] = field(default_factory=set, repr=False)
    ancestors: Set[HPO] = field(default_factory=set, repr=False)

    def __hash__(self) -> int:
        return hash(self.id)

    def _get_subclasses_recursively(self, subclass: HPO, container: set) -> Set[HPO]:
        for sub_subclass in subclass.subclasses:
            container.add(sub_subclass.id)
            self._get_subclasses_recursively(sub_subclass, container)

        return container

클레스 내 속성 보호하기: read only attributes : @property

파이썬에서는 비공개 속성(private attribute)을 생성하는 방법은 없지만, 수정을 막는 방법은 있습니다. @property을 이용하여 readonly로 작성할 수 있습니다.

아래의 예제 코드를 보면, all_subclasses을 `self._all_subclasses`로 저장해두고 @property로 읽어오려고합니다. 그리고, `_all_subclasses`을 `all_subclasses `을 통해서 볼 수는 있지만 직접 수정을 막기위해서, 감쌌습니다.

@dataclass
class HPO:
    """HPO (node)"""

    id: str
    name: str = str()
    definition: str = str()
    synonyms: Set[str] = field(default_factory=set)
    xref: Set[str] = field(default_factory=set)
    vector: np.ndarray = np.empty(shape=(1,))
    depth: int = 0
    ic: float = -1

    subclasses: Set[HPO] = field(default_factory=set, repr=False)
    ancestors: Set[HPO] = field(default_factory=set, repr=False)

    def __hash__(self) -> int:
        return hash(self.id)

    def _get_subclasses_recursively(self, subclass: HPO, container: set) -> Set[HPO]:
        for sub_subclass in subclass.subclasses:
            container.add(sub_subclass.id)
            self._get_subclasses_recursively(sub_subclass, container)

        return container

    @property
    def all_subclasses(self) -> List[HPO]:
        """현재 클래스를 포함하여 모든 하위 클래스를 찾아 반환
        Return
            List[HPO]: 모든 하위 클래스의 ID 목록
        Example:
            >>> self.all_subclasses
            [
                HPO(...),
                HPO(...),
                ...
                HPO(...),
            ]
        """
        if hasattr(self, "_all_subclasses"):
            return self._all_subclasses

        container = set()
        self._get_subclasses_recursively(self, container)

        self._all_subclasses = [HPO(hpo_id) for hpo_id in container]
        return self._all_subclasses

__slot__ 으로 클레스 속성 메모리를 효율화 하기

파이썬에서는 객체의 속성이 생기면 `__dict__`라는 딕셔너리형 속성에 저장합니다. 따라서, 매우 빠르게 속성을 접근할 수 있습니다. 하지만, 반대로 속성이 매우 많은 객체라면, 메모리부담이 매우 커지게됩니다. 

아래와 같이 `__slots__`이라는 정의하면, 인스턴스에 `__dict__`라는 딕셔너리 속성이 저장되지않고, 고정된 튜플로만 관리합니다. 따라서, 동적으로 속성을 추가할 수는 없지만, dict객체를 따로 들고있지 않아도되니 속도가 빠르거나 메모리가 효율적으로 동작할 수 있습니다.

class Person:
    # __slots__를 사용하여 객체의 속성을 제한합니다.
    __slots__ = ('name', 'age')

    def __init__(self, name, age):
        self.name = name
        self.age = age

# 객체 생성
person1 = Person('John', 30)

# 속성에 접근
print(person1.name)  # 출력: John
print(person1.age)   # 출력: 30

# 새로운 속성 추가 시 에러 발생
# person1.address = '123 Street'  # 'Person' object has no attribute 'address'

# __dict__를 사용하여 객체의 속성을 확인할 때 에러 발생
# print(person1.__dict__)  # AttributeError: 'Person' object has no attribute '__dict__'