patent - gyeroung1209

'''
---gyeroung1209/patent Auto Generate Code---
Author		: gyeroung1209
Project Name: patent
Project Link: https://blockai.kr/gyeroung1209/patent (BlockAI)
Create Date	: 2023-09-04

---Requirements---
# 사용자의 환경(OS, CUDA 등)에 따라 라이브러리 버전을 맞춰주세요
pip install torch==2.0 torchvision==0.15.2 torchtext==0.15.2 torchaudio==2.0.2
pip install pytorch-lightning==2.0.4
pip install tqdm
pip install pandas
pip install scikit-learn
pip install transformers

---Folder Structure---
--📂 data
   |--📄 train.csv
   |--📄 val.csv
   |--📄 test.csv
--📄 patent.py
--📄 patent.ipynb
--📄 requirements.txt
'''

import os
import argparse
import copy
import random
from glob import glob

from tqdm import tqdm
import numpy as np
import pandas as pd
from sklearn import preprocessing
from sklearn.model_selection import train_test_split

import torch
import pytorch_lightning as pl

import transformers

path_sep = os.sep


# https://pytorch.org/tutorials/beginner/basics/data_tutorial.html#creating-a-custom-dataset-for-your-files
class Dataset(torch.utils.data.Dataset):
    def __init__(self, inputs, targets=[]):
        self.inputs = inputs
        self.targets = targets

    # 학습 및 추론 과정에서 데이터를 1개씩 꺼내오는 곳
    def __getitem__(self, idx):
        # 정답이 있다면 if문을, 없다면 else문을 수행합니다
        if len(self.targets) == 0:
            return (torch.tensor(self.inputs[idx]['input_ids']), torch.tensor(self.inputs[idx]['attention_mask']), torch.tensor(self.inputs[idx]['token_type_ids']))
        else:
            return (torch.tensor(self.inputs[idx]['input_ids']), torch.tensor(self.inputs[idx]['attention_mask']), torch.tensor(self.inputs[idx]['token_type_ids'])), torch.tensor(self.targets[idx])

    # 입력하는 개수만큼 데이터를 사용합니다
    # 'return 100'이면 1에폭에 100개의 데이터만 사용합니다
    def __len__(self):
        return len(self.inputs)


# https://pytorch-lightning.readthedocs.io/en/stable/extensions/datamodules.html
class Dataloader(pl.LightningDataModule):
# 데이터의 종류에 따라 코드 수정이 필요할 수 있습니다
    def __init__(self, data_folder, batch_size, train_ratio, shuffle):
        super().__init__()
        self.data_folder = data_folder
        self.batch_size = batch_size
        self.train_ratio = train_ratio
        self.shuffle = shuffle

        self.train_dataset = None
        self.val_dataset = None
        self.test_dataset = None
        self.predict_dataset = None

        self.tokenizer = transformers.AutoTokenizer.from_pretrained('beomi/kollama-33b')

    def tokenizing(self, dataframe, text_columns):
        data = []
        for idx, item in tqdm(dataframe.iterrows(), desc='tokenizing', total=len(dataframe)):
            input_text = item[text_columns[0]]
            for column in text_columns[1:]:
                input_text += self.tokenizer.sep_token + item[column]
            outputs = self.tokenizer(input_text, add_special_tokens=True, padding='max_length', truncation=True)
            data.append(outputs)

        return data

    def set_preprocessing(self, data):
        self.target_columns = ["SSno"]
        self.encoder_columns = ["SSno"]
        self.delete_columns = ['claims', 'documentId']
        self.text_columns = ["invention_title", "abstract"]

        # 미사용 컬럼들을 삭제합니다
        data = data.drop(columns=self.delete_columns)

        # Scikit-learning 전처리 함수를 생성하고 현재 데이터셋에 맞게 설정합니다
        self.encoders = {column: preprocessing.LabelEncoder().fit(data[column]) for column in self.encoder_columns}

    def preprocessing(self, data):
        # 미사용 컬럼들을 삭제합니다
        data = data.drop(columns=self.delete_columns)

        # https://scikit-learn.org/stable/modules/generated/sklearn.preprocessing.LabelEncoder.html
        # 선택 컬럼들을 라벨 인코더로 변환합니다
        # 추후 encoders[컬럼명].inverse_transform(복원할 값) 함수로 원본 값을 복원할 수 있습니다
        for column in self.encoder_columns:
            if column in data.columns:
                data[column] = self.encoders[column].transform(data[column])

        # 타겟 데이터가 없으면 빈 배열을 리턴합니다.
        try:
            targets = data[self.target_columns].values.tolist()
            inputs = self.tokenizing(data.drop(self.target_columns, axis=1), self.text_columns)
        except:
            targets = []
            inputs = self.tokenizing(data, self.text_columns)

        return inputs, targets

    def setup(self, stage='fit'):
        if stage == 'fit':
            total_data = pd.read_csv(os.path.join(self.data_folder, 'train.csv'), sep=',', index_col=None, encoding='utf-8')
            self.set_preprocessing(total_data)

            # 학습 데이터와 검증 데이터셋을 비율에 맞춰 분리합니다
            train_data = total_data.sample(frac=self.train_ratio)
            val_data = total_data.drop(train_data.index)
            # DataFrame의 index를 새롭게 초기화해줍니다.
            train_data.reset_index(drop=True, inplace=True)
            val_data.reset_index(drop=True, inplace=True)

            # 학습데이터 준비
            train_inputs, train_targets = self.preprocessing(train_data)

            # 검증데이터 준비
            val_inputs, val_targets = self.preprocessing(val_data)

            # train 데이터만 shuffle을 적용해줍니다, 필요하다면 val, test 데이터에도 shuffle을 적용할 수 있습니다
            self.train_dataset = Dataset(train_inputs, train_targets)
            self.val_dataset = Dataset(val_inputs, val_targets)
        else:
            # 평가데이터 준비
            test_data = pd.read_csv(os.path.join(self.data_folder, 'test.csv'), sep=',', index_col=None, encoding='utf-8')
            test_inputs, test_targets = self.preprocessing(test_data)
            self.test_dataset = Dataset(test_inputs, test_targets)
            self.predict_dataset = Dataset(test_inputs, [])

    def train_dataloader(self):
        return torch.utils.data.DataLoader(self.train_dataset, batch_size=self.batch_size, shuffle=args.shuffle)

    def val_dataloader(self):
        return torch.utils.data.DataLoader(self.val_dataset, batch_size=self.batch_size)

    def test_dataloader(self):
        return torch.utils.data.DataLoader(self.test_dataset, batch_size=self.batch_size)

    def predict_dataloader(self):
        return torch.utils.data.DataLoader(self.predict_dataset, batch_size=self.batch_size)


# https://pytorch-lightning.readthedocs.io/en/stable/common/lightning_module.html
class Model(pl.LightningModule):
    def __init__(self):
        super().__init__()
        self.save_hyperparameters()

        self.automodel_1 = transformers.AutoModel.from_pretrained(pretrained_model_name_or_path='beomi/kollama-33b')
        self.linear_1 = torch.nn.Linear(in_features=256, out_features=256)
        self.linear_2 = torch.nn.Linear(in_features=256, out_features=564)

        self.dropout_1 = torch.nn.Dropout(p=0.1)
        self.gelu_1 = torch.nn.GELU()
        self.crossentropyloss_1 = torch.nn.CrossEntropyLoss()

    def forward(self, x_patentcsv):
        x_0 = self.automodel_1(*x_patentcsv)[0][:, 0, :] # [CLS] 토큰의 벡터만 가져옵니다.
        x_0 = self.dropout_1(x_0)
        x_0 = self.linear_1(x_0)
        x_0 = self.gelu_1(x_0)
        x_0 = self.dropout_1(x_0)
        x_0 = self.linear_2(x_0)

        return x_0

    def training_step(self, batch, batch_idx):
        x_patentcsv, y = batch
        x_0 = self(x_patentcsv)
        loss = self.crossentropyloss_1(x_0, y.long().squeeze())
        self.log("train_loss", loss)

        return loss

    def validation_step(self, batch, batch_idx):
        x_patentcsv, y = batch
        x_0 = self(x_patentcsv)
        loss = self.crossentropyloss_1(x_0, y.long().squeeze())
        self.log("val_loss", loss)

        return loss

    def test_step(self, batch, batch_idx):
        x_patentcsv, y = batch
        x_0 = self(x_patentcsv)
        loss = self.crossentropyloss_1(x_0, y.long().squeeze())
        self.log("test_loss", loss)

        return loss

    def predict_step(self, batch, batch_idx):
        x_patentcsv = batch
        x_0 = self(x_patentcsv)

        return x_0

    def configure_optimizers(self):
        # 이곳에서 lr 값을 변경할 수 있습니다
        optimizer = torch.optim.AdamW(self.parameters(), lr=0.0005)
        return optimizer


if __name__ == '__main__':
    # https://docs.python.org/ko/3/library/argparse.html
    # 하이퍼 파라미터 등 각종 설정값을 입력받습니다
    # 터미널 실행 예시 : python3 run.py --batch_size=64 ...
    # 실행 시 '--batch_size=64' 같은 인자를 입력하지 않으면 default 값이 기본으로 실행됩니다
    parser = argparse.ArgumentParser()
    parser.add_argument('--data_folder', default='./data')
    parser.add_argument('--batch_size', default=50)
    parser.add_argument('--max_epoch', default=1)
    parser.add_argument('--shuffle', default=True)
    parser.add_argument('--train_ratio', default=0.8)
    args = parser.parse_args()

    dataloader = Dataloader(args.data_folder, args.batch_size, args.train_ratio, args.shuffle)
    model = Model()
    # https://pytorch-lightning.readthedocs.io/en/stable/common/trainer.html
    # 학습 및 추론을 위한 Trainer 설정
    trainer = pl.Trainer(accelerator='gpu', devices=1, max_epochs=args.max_epoch)
    trainer.fit(model=model, datamodule=dataloader)
    # trainer.test(model=model, datamodule=dataloader)
    # predictions = trainer.predict(model=model, datamodule=dataloader)