Commit f127e23c authored by 叶璨铭's avatar 叶璨铭
Browse files

codes update

parent 7679e682
Loading
Loading
Loading
Loading
+4 −0
Original line number Diff line number Diff line
*.zip
*.pdf
tmp
ag_automm_tutorial_imgcls
AutogluonModels
data
### VisualStudioCode template
.vscode/*
!.vscode/settings.json
+2 −0
Original line number Diff line number Diff line
@@ -220,6 +220,8 @@ def init():
    visual_res, valid_rects = process(current_img, var_dict['processing_step']['value'])    
    cv2.imshow(window_name, visual_res)


if __name__ == "__main__":
    init()

    # 3. 启动标注循环
+34 −0
Original line number Diff line number Diff line
#%%
# labelGo 人工修正只作用于 txt,现在通过这个程序把png修正过来。
#%%
from labeler_main import *
pages = len(list(input_folder.glob("*.PNG")))
pages
#%%
from tqdm import tqdm
bar = tqdm(range(pages))
for i in bar:
    current_num = i+1
    current_img = cv_imread(img_path(current_num))
    image_size = (current_img.shape[1], current_img.shape[0])
    yolo_path = yolo_txt_path(current_num)
    
    path = inner_png_path(current_num, "*").relative_to(output_folder).as_posix()
    for f in output_folder.glob(path):
        f.unlink()
    
    with open(yolo_path) as f:
        lines = f.readlines()
        for j, line in enumerate(lines):
            ss = line.strip().split()
            if len(ss)==5:
                yolo_box = tuple(map(float, ss[1:]))
                cv_rect = yolo_xywh2cv_rect(image_size, yolo_box)
                y1, y2, x1, x2 = cv_rect2slice(cv_rect)
                if y1==y2 or x1==x2:
                    print("warning: empty rect")
                    continue
                roi = current_img[y1:y2, x1:x2]
                cv_imwrite(inner_png_path(current_num, j).as_posix(), roi)
                
# %%
+191 −0
Original line number Diff line number Diff line
#%%
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision.datasets import VisionDataset
# 'os' module provides functions for interacting with the operating system 
import os
# 'Numpy' is used for mathematical operations on large, multi-dimensional arrays and matrices
import numpy as np
# 'Pandas' is used for data manipulation and analysis
import pandas as pd
# 'Matplotlib' is a data visualization library for 2D and 3D plots, built on numpy
from matplotlib import pyplot as plt
# 'Seaborn' is based on matplotlib; used for plotting statistical graphics
import seaborn as sns
# to suppress warnings
import warnings

import numpy as np
import pandas as pd
from sklearn.tree import DecisionTreeClassifier
from sklearn.tree import DecisionTreeRegressor
import matplotlib.pyplot as plt
import matplotlib.cm as cm
from sklearn import tree
from sklearn.tree import _tree
from sklearn.base import is_classifier # 用于判断是回归树还是分类树
from dtreeviz.colors import adjust_colors # 用于分类树颜色(色盲友好模式)
import seaborn as sns #用于回归树颜色
from matplotlib.colors import Normalize # 用于标准化RGB数值
import graphviz # 插入graphviz库
import os
plt.style.use("default")
warnings.filterwarnings("ignore") 
plt.rcParams['font.sans-serif']=['SimHei'] #用来正常显示中文标签
plt.rcParams['axes.unicode_minus']=False #用来正常显示负号
# https://blog.csdn.net/wtySama/article/details/105316240
if os.path.exists('~/.cache/matplotlib'):
    os.rmdir('~/.cache/matplotlib')
#%%
import matplotlib
print(matplotlib.matplotlib_fname())
#%%
import pandas as pd
csv_path = '../../ztk/北京-天津-内蒙古.txt'
df = pd.read_csv(csv_path, sep='\t', index_col=0, 
                 )
df.head()
#%%
df.info()
#%%
# 观察标签分布
def see_discrete_dist(df, col_name):
    print(df[col_name].value_counts())
    # plt.figure(figsize=(10, 5))
    plt.figure(figsize=(30, 10))
    plt.title(f"{col_name} distribution")
    plt.bar(df[col_name].value_counts().index, df[col_name].value_counts().values)
see_discrete_dist(df, '器名')
#%%
df['器名'].value_counts().value_counts()

#%%
from pathlib import Path
this_file = Path(__file__).resolve()
this_directory = this_file.parent
project_directory = this_directory.parent.parent.parent
picture_path = "../../../public/data/3.pngs/ChineseUnearthedBronzes-01-Beijing_Tianjin-InnerMongolia"
picture_path = Path(picture_path).resolve()
picture_path
#%%
from tqdm import tqdm
bar = tqdm(df.index)
res = []
for i in bar:
    page, number = df.loc[i, ['页码', '编号']]
    label0, label1, label2 = df.loc[i, ['器名', '朝代', '出土地址']]
    page_txt_path = picture_path / f'{page}页-{page}.txt'
    with open(page_txt_path, 'r', encoding='utf-8') as f:
        lines = f.read().strip().splitlines()
        for j, line in enumerate(lines):
            splits = line.strip().split()
            if len(splits)==5 and splits[0]==str(number):
                res.append(dict(文物编号=number, 
                                图片路径=(picture_path / f'{page}页-{page}-内图{j}.png')
                                .relative_to(project_directory).as_posix(),
                                器名=label0, 朝代=label1, 
                                出土地址=label2))
res_pd = pd.DataFrame(res)
res_pd.head()
#%%
res_directory = project_directory/'public/data/4.datasets/中国出土青铜器/北京天津内蒙古/'
res_directory.mkdir(exist_ok=True, parents=True)
res_pd.to_csv(res_directory/'all.csv', index=False)

#%%
see_discrete_dist(res_pd, '器名')
res_pd['器名'].value_counts().value_counts()
#%%
cnts = res_pd['器名'].value_counts()
single_instance_class = [idx for idx in cnts.index if cnts[idx]==1]
single_instance_class
res_pd[res_pd['器名'].isin(single_instance_class)].to_csv(res_directory/'single_instance_classes.csv', index=False)

#%%
normal_pd = res_pd[~res_pd['器名'].isin(single_instance_class)]
len(normal_pd), len(res_pd)
#%%
normal_pd['器名'].value_counts().value_counts()
#%%
from sklearn.model_selection import train_test_split, StratifiedKFold, StratifiedShuffleSplit
# for train_index, test_index in StratifiedShuffleSplit(n_splits=1, 
#                                                     #   test_size=0.2, 
#                                                       test_size=1/3, 
#                                                       random_state=0).split(normal_pd, normal_pd['器名']):
#     train_pd = normal_pd.iloc[train_index]
#     test_pd = normal_pd.iloc[test_index]
# print(len(train_pd), len(test_pd))
# train_pd.head()
from sklearn.utils import _safe_indexing
from itertools import chain, combinations
from collections import Counter
def my_stratified_split(*arrays, test_size=1/3, random_state=0, stratify=normal_pd['器名']):
    randomer = np.random.RandomState(random_state)
    stratify = np.asarray(stratify)
    all_indices = np.arange(len(arrays[0]))
    train_res, test_res = [], []
    # cnts = stratify.value_counts()
    cnts = Counter(stratify)
    for cls, cnt in cnts.items():
        to_be_test = max(1, int(cnt*test_size))
        cls_indices = all_indices[stratify==cls]
        assert len(cls_indices)==cnt
        test_chosen = randomer.choice(cls_indices, to_be_test, 
                                      replace=False) # 不能重复选一样的
        train_chosen = np.setdiff1d(cls_indices, test_chosen)
        train_res+=train_chosen.tolist()
        test_res+=test_chosen.tolist()
        print(f"{cls}选择了 {len(test_chosen)}/{len(cls_indices)} 个作为测试集")
    # return arrays.loc[train_res], arrays.loc[test_res]
    return list(
            chain.from_iterable(
                (_safe_indexing(a, train_res), _safe_indexing(a, test_res)) for a in arrays
            )
        )

# train_pd, test_pd = train_test_split(normal_pd, test_size=1/2, random_state=0,
#                                      stratify=normal_pd['器名'])
train_pd, test_pd = my_stratified_split(normal_pd, 
                                        test_size=1/3, random_state=0,
                                     stratify=normal_pd['器名'])        

# %%
see_discrete_dist(train_pd, '器名')
see_discrete_dist(test_pd, '器名')

# %%
def cover_check(train_pd, test_pd):
    trains = train_pd['器名'].unique()
    tests =     test_pd['器名'].unique()
    not_in = [i for i in tests if i not in trains]
    print(not_in)
    train_redundacy = [i for i in trains if i not in tests]
    print(train_redundacy)
    if len(train_redundacy)>0:
        print(train_pd[train_pd['器名']==train_redundacy[0]])
    if len(not_in)>0:
        print(test_pd[test_pd['器名']==not_in[0]])
cover_check(train_pd, test_pd)
# %%

train_pd.to_csv(res_directory/'train.csv', index=False)
test_pd.to_csv(res_directory/'test.csv', index=False)

# %%
cnts = res_pd['器名'].value_counts()
major_classes = [idx for idx in cnts.index if cnts[idx]>5]
major_classes
#%%
major_pd = res_pd[res_pd['器名'].isin(major_classes)]
major_classes_train_pd, major_classes_test_pd = train_test_split(
    major_pd, test_size=1/3, random_state=0, stratify=major_pd['器名'])
cover_check(major_classes_train_pd, major_classes_test_pd)

# %%
major_classes_train_pd.to_csv(res_directory/'major_classes_train.csv', 
                              index=False)
major_classes_test_pd.to_csv(res_directory/'major_classes_test.csv',
                            index=False)    
# %%
+94 −0
Original line number Diff line number Diff line
from typing import Callable, Literal, Optional, Any, Tuple
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision.datasets import VisionDataset
from pathlib import Path
import numpy as np
import pandas as pd
from PIL import Image

from sklearn.preprocessing import LabelEncoder

this_file = Path(__file__).resolve()
this_directory = this_file.parent
project_directory = Path("/data/users/yecanming/P_CV_for_Archaeology/Chinese-Bronze-Ware").resolve()
dataset_directory = project_directory/'public/data/4.datasets/中国出土青铜器/北京天津内蒙古'

class CUB_BTI(VisionDataset):
    def __init__(self, 
                root: str=dataset_directory.as_posix(),
                image_root: str = project_directory.as_posix(),
                transforms: Optional[Callable] = None,
                transform: Optional[Callable] = None,
                target_transform: Optional[Callable] = None,
                type:Literal['train', 'test', 'major_classes_train', 'major_classes_test', 'single_instance_classes']='train',
                download=False, 
                target:Literal['器名', '朝代', '出土地址']='器名'):
        # transform和target_transform是对图片和标签的变换
        # https://github.com/pytorch/vision/issues/215
        super().__init__(root, transforms, transform, target_transform)
        if image_root is None: image_root = root
        self.type = type
        # 保证数据存在
        if download: self.download()
        if not self._check_integrity(): raise RuntimeError('Dataset not found or corrupted.')
        # 读取table
        root = Path(root).resolve()
        image_root = Path(image_root).resolve()
        table_path = root/f'{type}.csv'
        self.table = pd.read_csv(table_path)
        # 从table继续读取 data 和 targets
        # self.targets = self.table[target].astype(int).values
        self.targets = LabelEncoder().fit_transform(self.table[target])
        self.data = np.array([np.array(Image.open(image_root/i)) for i in self.table['图片路径'].values])
    
    def num_classes(self):
        return len(np.unique(self.targets))
    def __getitem__(self, index: int) -> Tuple[Any, Any]:
        """
        Args:
            index (int): Index
        Returns:
            tuple: (image, target) where target is index of the target class.
        """
        img, target = self.data[index], self.targets[index]

        # doing this so that it is consistent with all other datasets
        # to return a PIL Image
        img = Image.fromarray(img)

        # 父类没有任何能力, transform 要自己调用
        if self.transform is not None:
            img = self.transform(img)

        if self.target_transform is not None:
            target = self.target_transform(target)

        return img, target
    
    def __len__(self) -> int:
        return len(self.table)
    
    def download(self):
        raise NotImplementedError
    def _check_integrity(self):
        root, type = self.root, self.type
        root = Path(root).resolve()
        if not root.exists(): raise FileNotFoundError(f'root {root} does not exist.')
        table_path = root/f'{type}.csv'
        if not table_path.exists(): raise FileNotFoundError(f'table {table_path} does not exist.')
        return True
        

#%%
if __name__ == '__main__':
    train_ds = CUB_BTI(root=dataset_directory.as_posix(), 
                       image_root=project_directory.as_posix(),
                       type='train')
    test_ds = CUB_BTI(root=dataset_directory.as_posix(),
                      image_root=project_directory.as_posix(),
                       type='test')
    train_ds[0]
# %%
Loading