도메인 적응

import torch
import torch.nn as nn
from torch.autograd import Function

# Gradient Reversal Layer - 도메인 적응의 핵심
class GradientReversal(Function):
    @staticmethod
    def forward(ctx, x, alpha):
        ctx.alpha = alpha
        return x.view_as(x)

    @staticmethod
    def backward(ctx, grad_output):
        return grad_output.neg() * ctx.alpha, None

class DANN(nn.Module):
    def __init__(self, num_classes=10, backbone='resnet50'):
        super().__init__()
        # 특징 추출기 (소스와 타겟이 공유)
        self.feature_extractor = torch.hub.load(
            'pytorch/vision', backbone, pretrained=True
        )
        feature_dim = self.feature_extractor.fc.in_features
        self.feature_extractor.fc = nn.Identity()

        # 레이블 분류기
        self.classifier = nn.Sequential(
            nn.Linear(feature_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(256, num_classes)
        )

        # 도메인 분류기 (소스=0, 타겟=1 구분)
        self.domain_classifier = nn.Sequential(
            nn.Linear(feature_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(256, 2)
        )

    def forward(self, x, alpha=1.0):
        features = self.feature_extractor(x)
        class_output = self.classifier(features)
        # Gradient Reversal로 도메인 구분자를 '속이는' 학습
        reversed_features = GradientReversal.apply(features, alpha)
        domain_output = self.domain_classifier(reversed_features)
        return class_output, domain_output

# 학습 루프
def train_dann(model, source_loader, target_loader, epochs=100):
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
    ce_loss = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        # Alpha 스케줄링 (점진적 증가)
        p = epoch / epochs
        alpha = 2. / (1. + np.exp(-10 * p)) - 1

        for (src_x, src_y), (tgt_x, _) in zip(source_loader, target_loader):
            # 소스 도메인 분류 손실
            class_out, domain_out = model(src_x, alpha)
            class_loss = ce_loss(class_out, src_y)
            src_domain_loss = ce_loss(domain_out, torch.zeros(len(src_x)).long())

            # 타겟 도메인 적대적 손실
            _, tgt_domain_out = model(tgt_x, alpha)
            tgt_domain_loss = ce_loss(tgt_domain_out, torch.ones(len(tgt_x)).long())

            total_loss = class_loss + src_domain_loss + tgt_domain_loss
            optimizer.zero_grad()
            total_loss.backward()
            optimizer.step()

import torch
from transformers import CLIPProcessor, CLIPModel

# CLIP을 활용한 프롬프트 기반 도메인 적응
class CLIPDomainAdapter:
    def __init__(self, model_name="openai/clip-vit-large-patch14"):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)

    def create_domain_prompts(self, class_names, source_domain, target_domain):
        """도메인 특화 프롬프트 생성"""
        prompts = []
        for cls in class_names:
            # 타겟 도메인에 맞는 프롬프트
            prompts.append(f"a {target_domain} photo of a {cls}")
            prompts.append(f"a {cls} in {target_domain} conditions")
        return prompts

    def adapt_and_classify(self, image, class_names, target_domain="night"):
        """도메인 적응된 제로샷 분류"""
        prompts = [f"a {target_domain} photo of a {cls}" for cls in class_names]
        inputs = self.processor(
            text=prompts,
            images=image,
            return_tensors="pt",
            padding=True
        ).to(self.device)

        with torch.no_grad():
            outputs = self.model(**inputs)
            logits = outputs.logits_per_image
            probs = logits.softmax(dim=1)

        return class_names[probs.argmax()], probs.max().item()

# 사용 예시: 주간 학습 -> 야간 적용
adapter = CLIPDomainAdapter()
classes = ["car", "pedestrian", "bicycle", "truck"]
predicted_class, confidence = adapter.adapt_and_classify(
    night_image, classes, target_domain="nighttime"
)
print(f"예측: {predicted_class} (신뢰도: {confidence:.2%})")