Python初心者実戦の機械学習簡単な検証コード識別を実現(二):画像二値化ノイズ除去

画像の二値化ノイズ除去

import os
from PIL import Image
import numpy as np
import pandas as pd
from pandas import DataFrame
from threading import Thread
import queue
import time


class Clean(Thread):
    def __init__(self, que, out):
        """
        :param que:
        :type data: DataFrame
        """
        super(Clean, self).__init__()
        self.que = que
        self.out = out
        self.start()

    def to_grey(self, image):
        self.width, self.height = image.size
        data = np.asarray(image).reshape((self.width * self.height, 3))
        new_image = []
        for point in data:
            gp = (15 * point[0] + 75 * point[1] + 38 * point[2]) >> 7        #     
            new_image.append(0 if gp < 180 else 255)                      #     （      ，      180）
        return np.asarray(new_image).reshape((self.height, self.width))  #      

    def clean_point(self, grey0, degree=200):
        grey = grey0.copy()                                   #                      
        for x in range(1, len(grey) - 1):
            for y in range(1, len(grey[x]) - 1):
                mar = grey[x - 1:x + 2, y - 1: y + 2]        #   （x,y）      
                if mar[mar > degree].size > 5:              #     200       5      
                    grey0[x][y] = 255
        return grey0[1:-1, 1:-1]                           #         ，           ，       ，    


    def clean(self, filename, filepath):
        image = Image.open(filepath)
        grey = self.to_grey(image)
        smooth = self.clean_point(grey)


    def run(self):
        while True:
            try:
                filename, filepath = self.que.get(block=False)
                res = self.clean(filename, filepath)
                if res == 0:
                    break
                self.que.task_done()
            except Exception as e:
                print(e)
                break


class Manager:
    def __init__(self, thread=1):
        self.threads = []
        files = os.listdir('./image')      #       
        filepaths = [os.path.join('./image/', file) for file in files] #      
        filenames = [file[6:-4] for file in files]   #            
        self.que = queue.Queue()               #           
        self.out = queue.Queue()
        self.init_work_queue(filenames, filepaths)  #       
        self.init_thread(thread)                     #     
        self.wait_allcomplete()

    def init_thread(self, thread):
        for i in range(thread):
            self.threads.append(Clean(self.que, self.out))

    def init_work_queue(self, filenames, filepaths):
        for i in range(len(filenames)):
            self.que.put((filenames[i], filepaths[i]))

    def wait_allcomplete(self):
        for item in self.threads:
            if item.isAlive():
                item.join()


if __name__ == '__main__':
    manager = Manager(2)    #         2   
    data = pd.DataFrame()
    while True:
        try:
            data = data.append(manager.out.get(block=False), ignore_index=True)
        except:
            break
    data.to_csv('./csv/res.csv')

後続画像分割データ処理は次節で完備