blog

import torch

import intel_extension_for_pytorch as ipex  # isort: skip

import re
import sys


def custom_warning(message, category, filename, lineno, file=None, line=None):
    filename = re.sub("^.*?lib", "{virtualenv}/lib", filename)
    sys.stderr.write(
        "{filename}:{lineno}\n{category}:{message}\n".format(
            message=message,
            category=category,
            filename=filename,
            lineno=lineno,
        )
    )

import os

import cv2
import numpy as np
from PIL import Image
from torchvision.io import read_image
from torchvision.transforms.v2 import functional


def load_image(annotation_files, label_names, image_dir):

    data = []
    labels = []
    num_chars = []
    datasets = []

    for annotation_file in annotation_files:
        with open(annotation_file, "r") as f:
            while line := f.readline():
                file_name, label_name = line.strip().split()
                datasets.append([file_name, label_name])

    for file_name, label_name in datasets:
        if file_name.endswith("-0000.png"):
            # 1番目のカラムは読み飛ばす
            continue

        # 画像に含まれる文字
        chars = list(label_name)

        # 画像に含まれる文字数
        num_chars.append(len(chars))

        # ラベルの分割
        for char in chars:
            if char not in label_names:
                continue

            # ラベルをOne-Hotエンコーディングする
            label = torch.zeros(len(label_names))
            label_index = label_names.index(char)
            label[label_index] = 1
            labels.append(label)

        # 画像の読み込み
        image_path = os.path.join(image_dir, file_name)
        image = read_image(image_path)

        tmp = functional.to_pil_image(image)
        tmp = np.array(tmp)

        # 画像の分割
        # 少し線を太くする
        dilated = cv2.dilate(tmp, None, iterations=1)
        # 輪郭を抽出する
        contours, _ = cv2.findContours(
            dilated, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
        )

        # バウンディングボックスを求める
        bboxes = []
        for contour in contours:
            x, y, w, h = cv2.boundingRect(contour)
            bboxes.append([x, y, w, h])

        if len(chars) != len(bboxes):
            raise ValueError(
                "number of bounding box should be {} but found {}. image: {}".format(
                    len(chars), len(bboxes), file_name
                )
            )

        if len(chars) > 1:
            # "10"の画像を読み込んで"0","1"と輪郭が取れるので
            # 輪郭をx軸でソートする
            bboxes = sorted(bboxes, key=lambda bbox: bbox[0])

        for x, y, w, h in bboxes:
            data.append(
                functional.pil_to_tensor(
                    Image.fromarray(tmp[y : y + h, x : x + w])
                )
            )

    return data, labels, num_chars

teams = ["a", "b", "c"]

label_names = [str(i) for i in range(10)]
label_names.extend(["+", "-", "/", "n", "N"])

image_dir = "data/pism/image"
annotation_files = [
    os.path.join(image_dir, f"table-1-{team}-label.txt") for team in teams
]

annotation_files

['data/pism/image/table-1-a-label.txt',
 'data/pism/image/table-1-b-label.txt',
 'data/pism/image/table-1-c-label.txt']

data, labels, num_chars = load_image(annotation_files, label_names, image_dir)

len(data), len(labels), len(num_chars), sum(num_chars)

(1042, 1042, 980, 1042)

import matplotlib.pyplot as plt

fig, axs = plt.subplots(28, 40, figsize=(15, 10))
axs = axs.flatten()

i = j = 0

for k, num_char in enumerate(num_chars):
    # 2桁の数字が表示される列の場合
    if k % 35 in [0, 31, 32, 33, 34]:
        if num_char == 1:
            # 1桁ならダミーの画像を追加して表示する
            axs[j].imshow(Image.fromarray(np.zeros((1, 1))), cmap=plt.cm.gray)
            axs[j].set_axis_off()
            j += 1
        else:
            # 2桁なら画像を2つとも表示する
            axs[j].imshow(functional.to_pil_image(data[i]), cmap=plt.cm.gray)
            axs[j].set_axis_off()
            i += 1
            j += 1
    axs[j].imshow(functional.to_pil_image(data[i]), cmap=plt.cm.gray)
    axs[j].set_axis_off()
    i += 1
    j += 1


plt.tight_layout()
plt.show()

from nobituk import PillowPad, Trim
from torchvision.transforms import v2

image_width = 90
image_height = 120

transforms = v2.Compose(
    [
        Trim(),
        PillowPad(size=(image_width, image_height)),
        v2.Grayscale(),
        v2.ToDtype(torch.float32, scale=True),
    ]
)

from torch.utils.data import Dataset


class TransformDataset(Dataset):
    def __init__(self, data, target, transform=None, target_transform=None):
        self.data = data
        self.target = target
        self.transform = transform
        self.transform = transform
        self.target_transform = target_transform

    def __len__(self):
        return len(data)

    def __getitem__(self, i):
        x = self.data[i]
        if self.transform:
            x = self.transform(self.data[i])
        t = self.target[i]
        if self.target_transform:
            t = self.target_transform(self.target[i])
        return x, t

dataset = TransformDataset(data, labels, transforms)

from torch.utils.data import random_split

train_size = int(0.6 * len(dataset))
test_size = len(dataset) - train_size

train_dataset, test_dataset = random_split(dataset, [train_size, test_size])

from torch.utils.data import DataLoader

batch_size = 30
num_workers = 2

dataloader = DataLoader(
    dataset,
    batch_size=batch_size,
    shuffle=False,
    num_workers=num_workers,
)

train_loader = DataLoader(
    train_dataset,
    batch_size=batch_size,
    shuffle=True,
    num_workers=num_workers,
)

test_loader = DataLoader(
    train_dataset,
    batch_size=batch_size,
    shuffle=True,
    num_workers=num_workers,
)

from torch import nn


class CNN(nn.Module):

    def __init__(self):

        super().__init__()

        conv1 = nn.Sequential(
            nn.Conv2d(
                in_channels=1,
                out_channels=32,
                kernel_size=(3, 3),
            ),
            nn.ReLU(),
            nn.MaxPool2d(
                kernel_size=(2, 2),
            ),
        )

        conv2 = nn.Sequential(
            nn.Conv2d(
                in_channels=32,
                out_channels=64,
                kernel_size=(3, 3),
            ),
            nn.ReLU(),
            nn.MaxPool2d(
                kernel_size=(2, 2),
            ),
        )

        convolve = nn.Sequential(
            conv1,
            conv2,
        )

        connect = nn.Sequential(
            nn.LazyLinear(2048),
            nn.ReLU(),
            nn.Linear(2048, 256),
            nn.ReLU(),
            nn.Linear(256, 15),
        )

        self.layers = nn.Sequential(
            convolve,
            nn.Flatten(),
            connect,
        )

    def forward(self, x):
        return self.layers(x)

import warnings

device = "xpu"

with warnings.catch_warnings(
    action="default",
):
    warnings.showwarning = custom_warning

    model = CNN().to(device)

{virtualenv}/lib/python3.11/site-packages/torch/nn/modules/lazy.py:180
<class 'UserWarning'>:Lazy modules are a new feature under heavy development so changes to the API or functionality can happen at any moment.

print(model)

CNN(
  (layers): Sequential(
    (0): Sequential(
      (0): Sequential(
        (0): Conv2d(1, 32, kernel_size=(3, 3), stride=(1, 1))
        (1): ReLU()
        (2): MaxPool2d(kernel_size=(2, 2), stride=(2, 2), padding=0, dilation=1, ceil_mode=False)
      )
      (1): Sequential(
        (0): Conv2d(32, 64, kernel_size=(3, 3), stride=(1, 1))
        (1): ReLU()
        (2): MaxPool2d(kernel_size=(2, 2), stride=(2, 2), padding=0, dilation=1, ceil_mode=False)
      )
    )
    (1): Flatten(start_dim=1, end_dim=-1)
    (2): Sequential(
      (0): LazyLinear(in_features=0, out_features=2048, bias=True)
      (1): ReLU()
      (2): Linear(in_features=2048, out_features=256, bias=True)
      (3): ReLU()
      (4): Linear(in_features=256, out_features=15, bias=True)
    )
  )
)

criterion = nn.CrossEntropyLoss().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

from tqdm import tqdm


def train(epoch, dataloader, model, criterion, optimizer):

    model.train()
    model, optimizer = ipex.optimize(model, optimizer=optimizer)

    train_loss_history = []
    train_accuracy_history = []

    with tqdm(dataloader, unit="batch") as tbat:
        tbat.set_description(f"Epoch {epoch} Train")
        for images, labels in tbat:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)

            loss = criterion(outputs, labels)
            train_loss_history.append(loss.item())
            loss.backward()

            correct = (
                (outputs.argmax(dim=1) == labels.argmax(dim=1)).sum().item()
            )
            accuracy = correct / len(images)
            train_accuracy_history.append(accuracy)

            optimizer.step()
            optimizer.zero_grad()

            tbat.set_postfix(loss=loss.item(), accuracy=100 * accuracy)

    return train_loss_history, train_accuracy_history

def test(epoch, dataloader, model, criterion):

    size = len(dataloader.dataset)
    num_batches = len(dataloader)

    model.eval()
    model = ipex.optimize(model)

    test_loss, test_correct = 0, 0
    test_loss_history = []
    test_accuracy_history = []

    with torch.no_grad():
        for images, labels in dataloader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)

            loss = criterion(outputs, labels).item()
            test_loss_history.append(loss)
            test_loss += loss

            correct = (
                (outputs.argmax(dim=1) == labels.argmax(dim=1)).sum().item()
            )
            test_accuracy_history.append(correct / len(images))
            test_correct += correct

    test_correct /= size
    test_loss /= num_batches
    print(
        f"Epoch {epoch} Test Error: Accuracy: {(100*test_correct)}%, Avg loss: {test_loss}"
    )

    return test_loss_history, test_accuracy_history

epochs = 5
total_train_loss_history = []
total_train_accuracy_history = []
total_test_loss_history = []
total_test_accuracy_history = []

for epoch in range(epochs):

    # train
    epoch_loss_history, epoch_accuracy_history = train(
        epoch,
        train_loader,
        model,
        criterion,
        optimizer,
    )

    total_train_loss_history.extend(epoch_loss_history)
    total_train_accuracy_history.extend(epoch_accuracy_history)

    # test
    epoch_loss_history, epoch_accuracy_history = test(
        epoch,
        test_loader,
        model,
        criterion,
    )

    total_test_loss_history.extend(epoch_loss_history)
    total_test_accuracy_history.extend(epoch_accuracy_history)

Epoch 0 Train: 100%|██████████| 21/21 [00:02<00:00,  9.10batch/s, accuracy=100, loss=0.00975]


Epoch 0 Test Error: Accuracy: 98.56%, Avg loss: 0.05610149020018677


Epoch 1 Train: 100%|██████████| 21/21 [00:01<00:00, 11.60batch/s, accuracy=100, loss=0.0109]


Epoch 1 Test Error: Accuracy: 99.83999999999999%, Avg loss: 0.006129331724271954


Epoch 2 Train: 100%|██████████| 21/21 [00:01<00:00, 11.60batch/s, accuracy=100, loss=0.00412]


Epoch 2 Test Error: Accuracy: 99.83999999999999%, Avg loss: 0.009472846632152573


Epoch 3 Train: 100%|██████████| 21/21 [00:01<00:00, 11.61batch/s, accuracy=100, loss=0.00361]


Epoch 3 Test Error: Accuracy: 99.83999999999999%, Avg loss: 0.0019177654740772034


Epoch 4 Train: 100%|██████████| 21/21 [00:01<00:00, 11.59batch/s, accuracy=100, loss=1.92e-6]


Epoch 4 Test Error: Accuracy: 100.0%, Avg loss: 0.00019342863193319966

fig, axs = plt.subplots(1, 2, figsize=(10, 4))

axs[0].set_title("loss")
axs[0].plot(total_train_loss_history, label="train")
axs[0].plot(total_test_loss_history, label="test")
axs[0].set_xlabel("#batch")
axs[0].legend()

axs[1].set_title("accuracy")
axs[1].plot(total_train_accuracy_history, label="train")
axs[1].plot(total_test_accuracy_history, label="test")
axs[1].set_xlabel("#batch")
axs[1].legend()

plt.tight_layout()
plt.show()

torch.save(model.state_dict(), "data/pism/53-2-231-table-1.cnn.pth")

model.load_state_dict(torch.load("data/pism/53-2-231-table-1.cnn.pth"))

<All keys matched successfully>

def predict(dataloader, model):
    model.eval()
    model = ipex.optimize(model)
    result = []
    with torch.no_grad():
        for images, labels in dataloader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            result.extend(outputs.tolist())
    return result

result = predict(dataloader, model)

result[0]

[-14.32275390625,
 20.943679809570312,
 -1.1709486246109009,
 -6.548696517944336,
 -7.440320014953613,
 1.206632375717163,
 -12.301752090454102,
 6.903934001922607,
 -6.3719305992126465,
 -2.277395725250244,
 -1.5444855690002441,
 -10.331376075744629,
 -6.116647720336914,
 -14.116082191467285,
 -20.127317428588867]

decoded_result = []
for r in result:
    decoded_result.append(label_names[r.index(max(r))])

shift_table_data = []

i = 0

staff_shift_table_data = []
for j, num_char in enumerate(num_chars):

    if j > 0 and j % 35 == 0:
        shift_table_data.append(staff_shift_table_data)
        staff_shift_table_data = []

    if j % 35 in [0, 31, 32, 33, 34]:
        if num_char == 1:
            staff_shift_table_data.append(int(decoded_result[i]))
        else:
            num = decoded_result[i]
            i += 1
            num += decoded_result[i]
            staff_shift_table_data.append(int(num))
    else:
        staff_shift_table_data.append(decoded_result[i])

    i += 1

shift_table_data.append(staff_shift_table_data)

import pandas as pd

pd.options.display.max_columns = 40
pd.options.display.max_rows = 100

shift_table = pd.DataFrame(
    shift_table_data,
    columns=list(range(1, len(shift_table_data[0]) + 1)),
)

shift_table

answer = pd.read_csv(
    "data/pism/53-2-231-table-1.csv",
    header=None,
    skiprows=1,
    names=list(range(1, 36)),
)
answer

answer.compare(shift_table)