カスタム PyTorch スクリプトをインポートする#

import numpy as np
import pandas as pd
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader

from logging import getLogger

logger = getLogger(__name__)


device = 'cuda' if torch.cuda.is_available() else 'cpu'

batch_size = 64
epochs = 5

# define structured dataset class
class TabularDataset(Dataset):
  def __init__(self, data, input_cols=None, output_col=None):
    self.n = data.shape[0]
    if output_col:
      self.y = data[output_col].astype(np.float32).values.reshape(-1, 1)
    else:
      self.y =  np.zeros((self.n, 1))
    if input_cols:
        self.x = data[input_cols].astype(np.float32).values
    else:
        input_cols = [col for col in data.columns if col != output_col]
        self.x = data[input_cols].astype(np.float32).values

  def __len__(self):
    return self.n

  def __getitem__(self, idx):
    return [self.y[idx], self.x[idx]]

# define model class
class Model(nn.Module):
    def __init__(self, in_features=13, h1=32, h2=8, out_features=1):
        # how many layers?
        super().__init__()
        self.fc1 = nn.Linear(in_features, h1)
        self.fc2 = nn.Linear(h1, h2)
        self.out = nn.Linear(h2, out_features)
    
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.out(x)
        return x


def pytorch_model_train(event):
    # forepaas get data
    x_df, y_df = get_train_dataset()
    
    data = pd.concat([x_df,y_df], axis=1)
    output_feature = "is_overcrowded"
    # get data
    train_data = TabularDataset(data=data,input_cols=None,output_col=output_feature)
    train_loader = DataLoader(dataset = train_data, batch_size=batch_size, shuffle=True )
    logger.info(len(train_data))

    model = Model().to(device)
    model.train()

    # some additional params
    criterion = nn.BCEWithLogitsLoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=0.1)

    for epoch in range(epochs):
        epoch_loss = 0
        epoch_accuracy = 0 
        for y, x in train_loader:
            x = x.to(device)
            y  = y.to(device)
            # Forward Pass
            preds = model(x)
            loss = criterion(preds, y)
            # Backward Pass and Optimization
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            acc = ((preds.argmax(dim=1) == y).float().mean())
            epoch_accuracy += acc/len(train_loader)
            epoch_loss += loss/len(train_loader)
        logger.info('Epoch : {}, train accuracy : {}, train loss : {}'.format(epoch+1, epoch_accuracy,epoch_loss))

    return model

train_transforms =  transforms.Compose([
        transforms.Resize(150),
        transforms.RandomResizedCrop(150),
        transforms.ToTensor(),
    ])

class dataset(torch.utils.data.Dataset):
    def __init__(self,image_list,label_list,transform=None):
        self.file_list_df = image_list
        self.label_list_df = label_list
        self.transform = transform
        
    def __len__(self):
        self.filelength = len(self.file_list_df)
        return self.filelength
    
    def __getitem__(self,idx):
        file_list = self.file_list_df["input"].values.tolist()
        img_path = file_list[idx]
        img = Image.open(img_path)
        resized_img = img.resize((vertical_px, horizontal_px))
        img = resized_img.convert(mode='L')
        image_2_npArray_2_tensor = transforms.ToTensor()(img)
        label_list = self.label_list_df["output"].values.tolist()
        with open(label_list[idx],'r') as file:
            label = int(file.read())
        return image_2_npArray_2_tensor,torch.tensor(label)

...


def pytorch_train(event):
    # forepaas get data
    x_df, y_df = get_train_dataset()
    train_feature,val_feature,train_label,val_label= train_test_split(x_df, y_df, test_size=0.2)

    # get data
    train_data = dataset(train_feature,train_label)
    val_data = dataset(val_feature,val_label, transform=test_transforms)
    ...