<jupyter_start><jupyter_text>Цель ноутбука: изучение метода Few Shots Learning  В RTSD не хватает 14 знаков:| Знак | Описание | Источник || ------------- | ------------- | ---- || 1.6 | Пересечение равнозначных дорог | - || 1.31 | Туннель | - || 2.4 | Уступите дорогу | GTSRB Recognition || 3.21 | Конец запрещения обгона | GTSRB Recognition || 3.22 | Обгон грузовым автомобилям запрещен | GTSRB Recognition || 3.23 | Конец запрещения обгона грузовым автомобилям | GTSRB Recognition || 3.24-90 | Огр 90 | - || 3.24-100 | Огр 100 | GTSRB Recognition || 3.24-110 | Огр 110 | - || 3.24-120 | Огр 120 | GTSRB Recognition || 3.24-130 | Огр 130 | - || 3.25 | Конец огр. максимальной скорости | GTSRB Recognition || 3.31 | Конец всех ограничений | GTSRB Recognition || 6.3.2 | Зона для разворота | - | Инициализация библиотек<jupyter_code>import albumentations as A
import matplotlib.pyplot as plt
import numpy as np
import random
import torch
from torch import nn
import seaborn as sns
import pandas as pd
import os
import pathlib
import shutil
import cv2
import PIL
import cv2
import sys
from datetime import datetime

TEXT_COLOR = 'black'
# Зафиксируем состояние случайных чисел
RANDOM_STATE = 42
np.random.seed(RANDOM_STATE)
torch.manual_seed(RANDOM_STATE)
random.seed(RANDOM_STATE)
%matplotlib inline
plt.rcParams["figure.figsize"] = (17,10)

IN_COLAB = False
USE_COLAB_GPU = False

try:
    import google.colab
    IN_COLAB = True
    USE_COLAB_GPU = True
    from google.colab import drive
except:
    if IN_COLAB:
        print('[!]YOU ARE IN COLAB, BUT DIDNT MOUND A DRIVE. Model wont be synced[!]')

        if not os.path.isfile(CURRENT_FILE_NAME):
            print("FIX ME")
        IN_COLAB = False

    else:
        pass

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
device<jupyter_output><empty_output><jupyter_text>Инициализация основных путей и папки src<jupyter_code>if not IN_COLAB:
    PROJECT_ROOT = pathlib.Path(os.path.join(os.curdir, os.pardir))
else:
    PROJECT_ROOT = pathlib.Path('..')
    
DATA_DIR = PROJECT_ROOT / 'data'
NOTEBOOKS_DIR = PROJECT_ROOT / 'notebooks'

SRC_DIR = PROJECT_ROOT / 'src'
sys.path.append(SRC_DIR)
import torch.nn.functional as F

class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(1, 32, 3, 1)
        self.conv2 = nn.Conv2d(32, 64, 3, 1)
        # self.dropout1 = nn.Dropout2d(0.25)
        # self.dropout2 = nn.Dropout2d(0.5)
        self.fc1 = nn.Linear(9216, 128, bias=False)

    def forward(self, x):
        x = self.conv1(x)
        x = F.relu(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = F.max_pool2d(x, 2)
        # x = self.dropout1(x)
        x = torch.flatten(x, 1)
        x = self.fc1(x)
        return x

from torchvision.models import resnet18

model = Net() #resnet18()
# model.conv1 = nn.Conv2d(1, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False)
# model.fc = nn.Linear(in_features=512, out_features=128, bias=True)
model = model.to(device)
model
from torchvision import datasets, transforms

from pytorch_metric_learning import distances, losses, miners, reducers, testers
from pytorch_metric_learning.utils.accuracy_calculator import AccuracyCalculator
transform = transforms.Compose(
    [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
)

dataset1 = datasets.MNIST(".", train=True, download=True, transform=transform)
dataset2 = datasets.MNIST(".", train=False, transform=transform)

batch_size = 384
train_loader = torch.utils.data.DataLoader(dataset1, batch_size=batch_size, shuffle=True)
test_loader = torch.utils.data.DataLoader(dataset2, batch_size=batch_size)
### convenient function from pytorch-metric-learning ###
def get_all_embeddings(dataset, model):
    tester = testers.BaseTester(dataloader_num_workers=0)
    return tester.get_all_embeddings(dataset, model)


### MNIST code originally from https://github.com/pytorch/examples/blob/master/mnist/main.py ###
def train(model, loss_func, mining_func, device, train_loader, optimizer, epoch):
    print('train was called')
    model.train()
    for batch_idx, (data, labels) in enumerate(train_loader):
        # print(data)
        # print(labels)
        # input()
        data, labels = data.to(device), labels.to(device)
        optimizer.zero_grad()
        embeddings = model(data)
        indices_tuple = mining_func(embeddings, labels)
        # print(indices_tuple)
        loss = loss_func(embeddings, labels, indices_tuple)
        print(loss)
        loss.backward()
        optimizer.step()
        if batch_idx % 20 == 0:
            print(
                "Epoch {} Iteration {}: Loss = {}, Number of mined triplets = {}".format(
                    epoch, batch_idx, loss, mining_func.num_triplets
                )
            )

def init_normal(m):
    if type(m) == nn.Linear:
        nn.init.uniform_(m.weight)
    if isinstance(m, nn.Conv2d):
        torch.nn.init.xavier_uniform(m.weight.data)
        
model.apply(init_normal)

### compute accuracy using AccuracyCalculator from pytorch-metric-learning ###
def test(train_set, test_set, model, accuracy_calculator):
    train_embeddings, train_labels = get_all_embeddings(train_set, model)
    test_embeddings, test_labels = get_all_embeddings(test_set, model)
    train_labels = train_labels.squeeze(1)
    test_labels = test_labels.squeeze(1)
    print("Computing accuracy")
    accuracies = accuracy_calculator.get_accuracy(
        test_embeddings, train_embeddings, test_labels, train_labels, False
    )
    print("Test set accuracy (Precision@1) = {}".format(accuracies["precision_at_1"]))

import torch.optim as optim

optimizer = optim.Adam(model.parameters(), lr=0.005)
num_epochs = 1


### pytorch-metric-learning stuff ###
distance = distances.CosineSimilarity()
reducer = reducers.ThresholdReducer(low=0)
loss_func = losses.TripletMarginLoss(margin=0.2, distance=distance, reducer=reducer)
mining_func = miners.TripletMarginMiner(
    margin=0.2, distance=distance, type_of_triplets="semihard"
)
accuracy_calculator = AccuracyCalculator(include=("precision_at_1", ), k=1)
### pytorch-metric-learning stuff ###


for epoch in range(1, num_epochs + 1):
    train(model, loss_func, mining_func, device, train_loader, optimizer, epoch)
    test(dataset1, dataset2, model, accuracy_calculator)
from tqdm.notebook import trange, tqdm
from sklearn.manifold import TSNE
from sklearn.decomposition import PCA
from umap import UMAP
import plotly.express as px

@torch.no_grad()
def plotSmth(model, loader, device='cpu', dim3=False, fcn='PCA'):
    model.eval()
    
    plt.figure(figsize=(32, 16))    
    # clean the figure
    plt.clf()
    
    n_components = 3 if dim3 else 2
    
    fcn = fcn.upper()
    
    if fcn == 'PCA':
        reducer = PCA(n_components=n_components, random_state=RANDOM_STATE) 
    elif fcn == 'TSNE':
        reducer = TSNE(n_components=n_components, init='random', random_state=RANDOM_STATE)
    elif fcn == 'UMAP':
        reducer = UMAP(n_components=n_components, init='random', random_state=RANDOM_STATE)
    else:
        assert False, "wrong fcn arg"
    
    pbar = tqdm(enumerate(loader),
                    total=len(loader), 
                    position=0,
                    leave=False)

    MODEL_ARRAY_OUT_SIZE = (0, 128)
    model_out_arr = np.empty(MODEL_ARRAY_OUT_SIZE, dtype=np.float32)
    target_arr = np.empty((0, 1), dtype=np.int32)

    for idx, (data, target) in pbar:

        data = data.to(device)
        out = model(data).detach().cpu().numpy()
        # print(out.shape)
        model_out_arr = np.append(model_out_arr, out, axis=0)
        target = target.detach().cpu().numpy()
        # print(target)
        target_arr = np.append(target_arr, target)

        # if idx > 1:
        #     break

    # print(len(target_arr)) 
    # print(target_arr)
    X_embedded = reducer.fit_transform(model_out_arr)
    target_arr = np.char.mod('%d', target_arr)
    # print(target_arr)
    
    if not dim3:
        fig = px.scatter(X_embedded, x=0, y=1, color=target_arr)
    else:
        fig = px.scatter_3d(X_embedded, x=0, y=1, z=2, color=target_arr)    
    fig.show()
    
    
test_loader = torch.utils.data.DataLoader(dataset2, batch_size=256)
# model.apply(init_normal)
plotSmth(model, test_loader, device=device, dim3=True, fcn='umap')
plt.show()
enumerate(train_loader)[0]
model(dataset1[1][0][None, ...].to(device)).shape
res = get_all_embeddings(dataset1, model)
res[0].shape
from torchvision.transforms.functional import pad
pad(dataset1[1], 1)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1226601557958377, 'fraction_numerical': 0.025810952214858737, 'mean_word_length': 3.480208333333333, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 10, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '27847', 'license': 'MIT', 'n_tokens_mistral': 3196, 'n_tokens_neox': 3075, 'n_words': 654, 'path': 'notebooks/SignDetectorAndClassifier/notebooks/FewShotLearning_MNIST.ipynb', 'repo_name': 'lsd-maddrive/adas_system', 'size': '8601'}