text:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# This is a sample Project for using MediaPipe in Game.
# And this is for teaching, so this is only demo, not a real game.
import random

import cv2
import numpy
import time
from PIL import Image, ImageDraw, ImageFont
import mediapipe_hand
import queue
import os

# 強制設定遊戲視窗大小
FRAME_WIDTH = 640
FRAME_HEIGHT = 480

# 設定遊玩次數
game_total_count = 10
# 設定每次活動秒數
game_play_time = 7

# 得分紀錄
total_score = 0
# 紀錄遊戲過程的 queue
game_movie_queue = None
# 錄影存檔的檔名
game_save_file_name = None

# 框之大小，用一半的量，比較好計算左上右下
rect_box_width_half = 75
rect_box_height_half = 100

# 右手相關資訊
right_hand_line_color = (255, 255, 0)
right_hand_fill_color = (80, 80, 0)
right_hand_width = 10
right_hand_mark = "右"
# 右邊框的起始位置
right_box_x = 480
right_box_y = 240
# 右手是否在右邊框
right_hand_in_box = False

# 左手相關資訊
left_hand_line_color = (0, 255, 255)
left_hand_fill_color = (0, 80, 80)
left_hand_width = 10
left_hand_mark = "左"
# 左邊框的起始位置
left_box_x = 160
left_box_y = 240
# 左手是否在左邊框
left_hand_in_box = False

# 不填滿框的內容
no_fill_color = None

# 中文字型檔
font_text_24 = ImageFont.truetype("jf-openhuninn-1.1.ttf", 24, encoding="utf-8")
font_text_36 = ImageFont.truetype("jf-openhuninn-1.1.ttf", 36, encoding="utf-8")
font_text_48 = ImageFont.truetype("jf-openhuninn-1.1.ttf", 48, encoding="utf-8")

# 顯示中文字之副程式
def add_chinese_font_to_image(img, text, left, top, text_color=(0, 255, 0), font_text=font_text_24):

    # 判斷圖片格式，若為 OpenCV 格式，就做陣列與 BGR->RGB 的轉變
    if isinstance(img, numpy.ndarray):
        img = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))

    # 將 img 切換成 PIL 格式
    draw = ImageDraw.Draw(img)

    # 利用 draw 進行文字繪製
    draw.text((left, top), text, text_color, font=font_text)

    # 返回繪製好的 圖片+文字
    return cv2.cvtColor(numpy.asarray(img), cv2.COLOR_RGB2BGR)

# 畫左右兩框
def draw_rect_box(image, in_r_rect, in_l_rect):

    # 判斷圖片格式，若為 OpenCV 格式，就做陣列與 BGR->RGB 的轉變
    if isinstance(image, numpy.ndarray):
        image = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))

    # 將 image 切換成 pil 格式
    draw = ImageDraw.Draw(image)

    # 計算 r-xy 座標
    r_up_x = int(right_box_x - rect_box_width_half)
    r_up_y = int(right_box_y - rect_box_height_half)
    r_down_x = int(right_box_x + rect_box_width_half)
    r_down_y = int(right_box_y + rect_box_height_half)

    # 計算 l-xy 座標
    l_up_x = int(left_box_x - rect_box_width_half)
    l_up_y = int(left_box_y - rect_box_height_half)
    l_down_x = int(left_box_x + rect_box_width_half)
    l_down_y = int(left_box_y + rect_box_height_half)

    # 繪製 right 的框
    if in_r_rect:
        draw.rectangle((r_up_x, r_up_y, r_down_x, r_down_y), outline=right_hand_line_color, fill=no_fill_color, width=7)
    else:
        draw.rectangle((r_up_x, r_up_y, r_down_x, r_down_y), outline=right_hand_line_color, fill=no_fill_color, width=2)

    # 標上 right 文字
    draw.text((r_up_x+10, r_up_y+10), right_hand_mark, right_hand_line_color, font=font_text_36)

    # 繪製 left 的框
    if in_l_rect:
        draw.rectangle((l_up_x, l_up_y, l_down_x, l_down_y), outline=left_hand_line_color, fill=no_fill_color, width=7)
    else:
        draw.rectangle((l_up_x, l_up_y, l_down_x, l_down_y), outline=left_hand_line_color, fill=no_fill_color, width=2)

    # 標上 left 文字
    draw.text((l_up_x+10, l_up_y+10), left_hand_mark, left_hand_line_color, font=font_text_36)

    # 返回繪製好的 圖片+框
    return cv2.cvtColor(numpy.asarray(image), cv2.COLOR_RGB2BGR)

# 使用 Random 隨機定義左右手框的位置
def random_get_box_xy():

    random.seed()

    # 中心點的隨機值，必須要符合框能劃入的大小
    a_right_box_x = random.randint(rect_box_width_half, FRAME_WIDTH - rect_box_width_half)
    a_right_box_y = random.randint(rect_box_height_half, FRAME_HEIGHT - rect_box_height_half)
    a_left_box_x = random.randint(rect_box_width_half, FRAME_WIDTH - rect_box_width_half)
    a_left_box_y = random.randint(rect_box_height_half, FRAME_HEIGHT - rect_box_height_half)

    return a_right_box_x, a_right_box_y, a_left_box_x, a_left_box_y

# 確認存檔檔名
def decide_file_name():

    global game_save_file_name

    # 依序找看看檔銘是否存在，若不存在則用此檔名
    origin_index = 0
    while True:
        temp_filename = "H" + ("%05d" % origin_index) + ".mp4"

        if os.path.isfile(temp_filename):
            origin_index += 1
        else:
            game_save_file_name = temp_filename
            break

# 遊戲的入口點
def play_game():

    global right_box_x, right_box_y, left_box_x, left_box_y, total_score, game_movie_queue

    # 是否從一般模式進入 Game Mode
    is_first_in_game_mode = True
    # 紀錄目前是否為遊戲模式，還是只顯示畫面
    is_game_mode = False
    # 紀錄目前已經經歷了幾次
    play_loop_count = 0
    # 記錄每一個 loop 開始的時間
    begin_time = 0
    # 記錄每一個畫面當下的時間
    now_time = 0
    # 建立存放預先錄影用 image 的 queue
    game_movie_queue = queue.Queue(100)
    # 錄影格式
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    # 錄影串流
    out_movie = None

    # 開啟攝影機
    cap = cv2.VideoCapture(0, cv2.CAP_DSHOW)

    # 設定攝影機尺寸為固定大小，以便標示方框與文字
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, FRAME_WIDTH)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, FRAME_HEIGHT)

    # 進行不斷循環的撥放
    while True:
        # 擷取影像，在 play_game 這一層中， image 都要保持 OpenCV 格式
        # 其他副程式若用到 Pillow，則自行轉換，但返回值仍要保持 OpenCV(BGR) 格式
        success, image = cap.read()

        # 若有攫取到畫面，就在視窗中顯示
        if success:

            # 用 WebCam 拍自己，所以先做水平翻轉
            image = cv2.flip(image, 1)

            # 若在遊戲模式中
            if is_game_mode:

                # 第一次開始到 game mode 時要作的事
                if is_first_in_game_mode:
                    # 關閉第一次的標記
                    is_first_in_game_mode = False
                    # 開始錄影 - 決定檔名
                    decide_file_name()
                    # 取得影像的高與寬
                    image_height, image_width, _ = image.shape
                    # 開啟影片檔
                    out_movie = cv2.VideoWriter(game_save_file_name, fourcc, 30.0, (image_width, image_height))
                    # 先將 Queue 中所有的影像寫到檔案中
                    while not game_movie_queue.empty():
                        frame = game_movie_queue.get_nowait()
                        out_movie.write(frame)

                # 每一個 game loop 的開始
                if begin_time == 0:
                    begin_time = time.time()
                    now_time = time.time()
                    right_box_x, right_box_y, left_box_x, left_box_y = random_get_box_xy()

                # 取得時間與計算間隔秒數
                time_interval = round(now_time - begin_time, 1)
                now_time = time.time()

                # 先進行手部位置評估
                image, in_r_rect, in_l_rect = mediapipe_hand.hands_estimation(image, right_box_x, right_box_y, left_box_x, left_box_y)

                # 依照左右手位置的條件來畫框
                image = draw_rect_box(image, in_r_rect, in_l_rect)

                # 顯示目前為第幾輪的遊戲，倒數時間與分數
                image = add_chinese_font_to_image(image, "手忙腳亂：第" + str(play_loop_count + 1) + "輪 已得分=" + str(total_score) + "分", 10, 10, font_text=font_text_36)
                image = add_chinese_font_to_image(image, "時間倒數 " + str(int(game_play_time - time_interval)) + " 秒", 10, 70, font_text=font_text_36)

                # 當秒數為 game_play_time 時，更新次數與回覆時間值
                if time_interval >= game_play_time:
                    begin_time = 0
                    play_loop_count += 1
                    # 此時若左右手都在框中，就算各得一分
                    if in_r_rect:
                        total_score += 1
                    if in_l_rect:
                        total_score += 1
                    right_box_x, right_box_y, left_box_x, left_box_y = random_get_box_xy()

                # 當已經跑 total 次時就回到一般模式
                if play_loop_count == game_total_count:
                    is_game_mode = False
                    # 右邊框的起始位置
                    right_box_x = 480
                    right_box_y = 240
                    # 左邊框的起始位置
                    left_box_x = 160
                    left_box_y = 240

            else:
                # 關閉錄影串流
                if out_movie is not None:
                    out_movie.release()
                    out_movie = None

                # 若在一般模式之下，顯示啟動畫面
                image = add_chinese_font_to_image(image, "手忙腳亂：請將左右手掌放於框中開始", 10, 10, font_text=font_text_36)
                image = add_chinese_font_to_image(image, "上次得分：" + str(total_score) + "分", 150, 50, font_text=font_text_48)

                # 先進行手部位置評估
                image, in_r_rect, in_l_rect = mediapipe_hand.hands_estimation(image, right_box_x, right_box_y, left_box_x, left_box_y)

                # 依照左右手位置的條件來畫框
                image = draw_rect_box(image, in_r_rect, in_l_rect)

                # 在一般模式下，將這兒設為 True，以便下一次開始遊戲時，要作一些事
                is_first_in_game_mode = True

                # 若左手此時都在框中，就進入遊戲模式
                is_game_mode = in_r_rect and in_l_rect
                if is_game_mode:
                    play_loop_count = 0
                    total_score = 0

            # 存放到 Queue 中，若 Queue 已滿，就先取出一個舊的，加入一個新的
            if game_movie_queue.full():
                game_movie_queue.get_nowait()
            if not is_game_mode:
                game_movie_queue.put(image)

            # 每 image 錄影
            if out_movie is not None:
                out_movie.write(image)

            cv2.imshow('手忙腳亂', cv2.resize(image, (1024, 768), interpolation=cv2.INTER_AREA))

        # 需有 waitKey，畫面才會顯示內容
        press_key = cv2.waitKey(5) & 0xFF

        # 保留按下 ESC 結束整個畫面的功能
        if press_key == 27:
            break


text:
import numpy as np
import os
import urllib.request
import gzip

class Ann_with_np:
    def __init__(self):
        dir = os.getcwd()
        urllib.request.urlretrieve('http://yann.lecun.com/exdb/mnist/' + 'train-images-idx3-ubyte.gz',
                                   dir + '/train-images-idx3-ubyte.gz')
        urllib.request.urlretrieve('http://yann.lecun.com/exdb/mnist/' + 'train-labels-idx1-ubyte.gz',
                                   dir + '/train-labels-idx1-ubyte.gz')
        urllib.request.urlretrieve('http://yann.lecun.com/exdb/mnist/' + 't10k-images-idx3-ubyte.gz',
                                   dir + '/t10k-images-idx3-ubyte.gz')
        urllib.request.urlretrieve('http://yann.lecun.com/exdb/mnist/' + 't10k-labels-idx1-ubyte.gz',
                                   dir + '/t10k-labels-idx1-ubyte.gz')
        self.__xdata = self.__extractdata('train-images-idx3-ubyte.gz',16,784,60000).reshape(60000,784)/256
        self.__ydata = self.__onehot(self.__extractdata('train-labels-idx1-ubyte.gz',8,1,60000))
        self.__xdata_test = self.__extractdata('t10k-images-idx3-ubyte.gz',16,784,10000).reshape(10000,784)/256
        self.__ydata_test =  self.__onehot(self.__extractdata('t10k-labels-idx1-ubyte.gz',8,1,10000))
        self.__xdata,self.__ydata = self.__batching(self.__xdata,self.__ydata)
        self.w1 = np.random.randn(784,256) / np.sqrt(392)
        self.w2 = np.random.randn(256,32) / np.sqrt(128)
        self.w3 = np.random.randn(32,10) / np.sqrt(16)
        self.b1 = np.random.randn(256) / np.sqrt(128)
        self.b2 = np.random.randn(32) / np.sqrt(16)
        self.b3 = np.random.randn(10) / np.sqrt(5)
        print("model is created")

    def __extractdata(self,filename, readdata, datasize, num):
        with gzip.open(filename) as bytestream:
            bytestream.read(readdata)
            buf = bytestream.read(datasize * num)
            data = np.frombuffer(buf, dtype=np.uint8).astype(np.float32)
            return data

    def __onehot(self,ydata):
        output = np.zeros(shape=(len(ydata), 10))
        for i in range(len(ydata)):
            output[i][int(ydata[i])] = 1
        return output

    def __batching(self,xdata,ydata,number_of_input = 60000,batch=200):
        total_batch = np.int(number_of_input / batch)
        x_batch = []
        y_batch = []
        for i in range(total_batch):
            x_batch.append(xdata[i * batch:(i + 1) * batch])
            y_batch.append(ydata[i * batch:(i + 1) * batch])
        return x_batch,y_batch

    @property
    def xdata(self):
        return self.__xdata

    @property
    def ydata(self):
        return self.__ydata

    @property
    def xdata_test(self):
        return self.__xdata_test

    @property
    def ydata_test(self):
        return self.__ydata_test

    def learning(self,xdata,ydata,num_epoch = 15,learning_rate = 0.01):
        for epoch in range(num_epoch):
            for batch in np.arange(len(xdata)):
                ######## forward ###########################
                x_batch = xdata[batch]
                y_batch = ydata[batch]
                beforeactivation1 = np.dot(x_batch, self.w1) + self.b1
                hidden1 = np.maximum(beforeactivation1, 0.1 * beforeactivation1)
                beforeactivation2 = np.dot(hidden1, self.w2) + self.b2
                hidden2 = np.maximum(beforeactivation2, 0.1 * beforeactivation2)
                output = np.dot(hidden2, self.w3) + self.b3
                softmax = self.__applysoftmax(output)
                loss = np.sum(-y_batch*np.log(softmax),axis = 1)

                ######### backpropagation ##########
                error3 = softmax - y_batch
                d_loss_w3 = np.zeros([200,32,10])
                for i in range(200):
                    d_loss_w3[i] = hidden2[i].reshape(32,1) * error3[i].reshape(1,10)
                d_loss_b3 = error3
                d_activation2 = self.__d_LeakyRelu(np.dot(hidden1, self.w2))

                error2 = d_activation2 * np.matmul(error3,self.w3.T)
                d_loss_w2 = np.zeros([200, 256, 32])
                for i in range(200):
                    d_loss_w2[i] = hidden1[i].reshape(256,1) * error2[i].reshape(1,32)
                d_loss_b2 = error2
                d_activation1 = self.__d_LeakyRelu(np.dot(x_batch, self.w1))

                error1 = d_activation1 * np.matmul(error2,self.w2.T)
                d_loss_w1 = np.zeros([200, 784, 256])
                for i in range(200):
                    d_loss_w1[i] = x_batch[i].reshape(784,1) * error1[i].reshape(1,256)
                d_loss_b1 = error1

                #####updating################################################
                self.w3 = self.w3 - learning_rate*np.mean(d_loss_w3,axis=0)
                self.w2 = self.w2 - learning_rate*np.mean(d_loss_w2,axis=0)
                self.w1 = self.w1 - learning_rate*np.mean(d_loss_w1,axis=0)
                self.b3 = self.b3 - learning_rate*np.mean(d_loss_b3,axis=0)
                self.b2 = self.b2 - learning_rate*np.mean(d_loss_b2,axis=0)
                self.b1 = self.b1 - learning_rate*np.mean(d_loss_b1,axis=0)

                if batch %10 == 0:
                    print("epoch : {}, batch : {}, loss: {}".format(epoch,batch,np.mean(loss,axis = 0)))
        print("learning ends")

    def __applysoftmax(self,array):
        z = array - np.max(array)
        exp_output = np.exp(z)
        return exp_output / np.sum(exp_output,axis = -1).reshape(-1,1)

    def __d_LeakyRelu(self,array):
        array[array>0] = 1
        array[array<0] = 0.1
        return array

    def accuracy(self,xdata_test,ydata_test):
        correct = 0
        for i in np.arange(len(xdata_test)):
            hidden1 = np.maximum(np.dot(xdata_test[i], self.w1) + self.b1, 0.1 * np.dot(xdata_test[i], self.w1) + self.b1)
            hidden2 = np.maximum(np.dot(hidden1, self.w2) + self.b2, 0.1 * np.dot(hidden1, self.w2) + self.b2)
            output = np.matmul(hidden2, self.w3) + self.b3
            softmax = self.__applysoftmax(output)
            if (np.where(softmax[0] == softmax[0].max()) == np.where(ydata_test[i] == 1)):
               correct += 1

        print("accuracy: ", correct / len(ydata_test))

def main():
    nn = Ann_with_np()
    xdata, ydata, xdata_test, ydata_test = nn.xdata, nn.ydata, nn.xdata_test, nn.ydata_test
    nn.learning(xdata,ydata)
    nn.accuracy(xdata_test,ydata_test)

if __name__ == "__main__":
    main()

ext: py

reason: import numpy


text:
import os
import unittest
from unittest import mock
import re
import warnings
import json
from io import StringIO

import pandas as pd
import numpy as np

from dataprofiler.data_readers.csv_data import CSVData
from dataprofiler.data_readers.csv_data import JSONData
from dataprofiler.data_readers.csv_data import ParquetData
from dataprofiler.data_readers.csv_data import AVROData

from dataprofiler.labelers import data_processing, DataLabeler, \
    UnstructuredDataLabeler
from dataprofiler.labelers.character_level_cnn_model import \
    CharacterLevelCnnModel

test_root_path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))

data_labeler_parameters = {
    'model': {
        'class': 'CharacterLevelCnnModel',
        'parameters': {}
    },
    'label_mapping': {
        'PAD': 0,
        'CITY': 1,  # SAME AS UNKNOWN
        'UNKNOWN': 1,
        'ADDRESS': 2,
        'PERSON': 3,
    },
    'preprocessor': {
        'class': 'CharPreprocessor'
    },
    'postprocessor': {
        'class': 'CharPostprocessor'
    },
}

preprocessor_parameters = {
    'flatten_split': 0,
    'flatten_separator': ' ',
    'is_separate_at_max_len': True,

}

postprocessor_parameters = {
    'use_word_level_argmax': True,
    'output_format': 'character_argmax',
    'separators': (' ', ',', ';', "'", '"', ':', '\n', '\t', "."),
    'word_level_min_percent': 0.75,
}

def mock_open(filename, *args):
    if filename.find('data_labeler_parameters') >= 0:
        return StringIO(json.dumps(data_labeler_parameters))
    elif filename.find('preprocessor_parameters') >= 0:
        return StringIO(json.dumps(preprocessor_parameters))
    elif filename.find('postprocessor_parameters') >= 0:
        return StringIO(json.dumps(postprocessor_parameters))

def setup_save_mock_open(mock_open):
    mock_file = StringIO()
    mock_file.close = lambda: None
    mock_open.side_effect = lambda *args: mock_file
    return mock_file

@mock.patch('dataprofiler.labelers.data_processing.BaseDataProcessor')
@mock.patch.object(CharacterLevelCnnModel, 'load_from_disk')
@mock.patch("builtins.open", side_effect=mock_open)
class TestDataLabeler(unittest.TestCase):

    @staticmethod
    def _setup_mock_load_model(mock_load_model):
        model_mock = mock.Mock(spec=CharacterLevelCnnModel)
        model_mock.set_num_labels = mock.Mock()
        mock_load_model.return_value = model_mock
        model_mock.requires_zero_mapping = True
        model_mock.labels = ['PAD', 'UNKNOWN', 'ADDRESS', 'PERSON']
        model_mock.label_mapping = {
            'PAD': 0,
            'CITY': 1,  # SAME AS UNKNOWN
            'UNKNOWN': 1,
            'ADDRESS': 2,
            'PERSON': 3,
        }
        model_mock.reverse_label_mapping = {
            0: 'PAD',
            1: 'UNKNOWN',
            2: 'ADDRESS',
            3: 'PERSON',
        }

    @staticmethod
    def _setup_mock_load_processor(mock_base_processor):
        def side_effect(arg):
            processor = {
                'CharPreprocessor': mock.Mock(
                    spec=data_processing.CharPreprocessor),
                'CharPostprocessor': mock.Mock(
                    spec=data_processing.CharPostprocessor),
            }[arg]
            processor.load_from_disk.return_value = processor
            return processor

        mock_base_processor.get_class.side_effect = side_effect

    def test_has_public_functions(self, *args):
        public_functions = [
            "predict",
            "set_preprocessor",
            "set_model",
            "set_postprocessor",
            "save_to_disk",
            "load_from_disk",
            "load_from_library",
            "check_pipeline"
        ]

        for func in public_functions:
            self.assertTrue(hasattr(UnstructuredDataLabeler, func))

    def test_load_parameters(self, mock_open, mock_load_model,
                             mock_base_processor):

        # test not model class
        with self.assertRaisesRegex(TypeError,
                                    '`model_class` must be a BaseModel'):
            load_options = dict(model_class='fake')
            UnstructuredDataLabeler._load_parameters('test/path', load_options)

        # test not correct model class
        with self.assertRaisesRegex(ValueError,
                                    'The load_options model class does not '
                                    'match the required DataLabeler model.\n '
                                    '.* != .*'):
            mocked_model = mock.Mock(spec=CharacterLevelCnnModel)
            mocked_model.__name__ = 'FakeClassName'
            load_options = dict(model_class=mocked_model)
            UnstructuredDataLabeler._load_parameters('test/path', load_options)

        # test not preprocessor class
        with self.assertRaisesRegex(TypeError,
                                    '`preprocessor_class` must be a '
                                    'BaseDataPreprocessor'):
            load_options = dict(preprocessor_class='fake')
            UnstructuredDataLabeler._load_parameters('test/path', load_options)

        # test not correct preprocessor class
        with self.assertRaisesRegex(ValueError,
                                    'The load_options preprocessor class does '
                                    'not match the required DataLabeler '
                                    'preprocessor.\n .* != .*'):
            mocked_preprocessor = mock.Mock(
                spec=data_processing.BaseDataPreprocessor)
            mocked_preprocessor.__name__ = 'FakeProcessorName'
            load_options = dict(preprocessor_class=mocked_preprocessor)
            UnstructuredDataLabeler._load_parameters('test/path', load_options)

        # test not postprocessor class
        with self.assertRaisesRegex(TypeError,
                                    '`postprocessor_class` must be a '
                                    'BaseDataPostprocessor'):
            load_options = dict(postprocessor_class='fake')
            UnstructuredDataLabeler._load_parameters('test/path', load_options)

        # test not correct postprocessor class
        with self.assertRaisesRegex(ValueError,
                                    'The load_options postprocessor class does '
                                    'not match the required DataLabeler '
                                    'postprocessor.\n .* != .*'):
            mocked_postprocessor = mock.Mock(
                spec=data_processing.BaseDataPostprocessor)
            mocked_postprocessor.__name__ = 'FakeProcessorName'
            load_options = dict(postprocessor_class=mocked_postprocessor)
            UnstructuredDataLabeler._load_parameters('test/path', load_options)

    def test_load_labeler(self, mock_open, mock_load_model,
                          mock_base_processor):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_base_processor)

        # load default
        data_labeler = UnstructuredDataLabeler()

        self.assertDictEqual(data_labeler.label_mapping,
                             data_labeler_parameters['label_mapping'])
        self.assertListEqual(
            data_labeler.labels, ['PAD', 'UNKNOWN', 'ADDRESS', 'PERSON'])
        self.assertIsInstance(data_labeler.preprocessor,
                              data_processing.BaseDataPreprocessor)
        self.assertIsInstance(data_labeler.postprocessor,
                              data_processing.BaseDataPostprocessor)

    def test_load_from_library(self, mock_open, mock_load_model,
                               mock_base_processor):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_base_processor)

        # load default
        data_labeler = UnstructuredDataLabeler.load_from_library('default')

        self.assertDictEqual(data_labeler.label_mapping,
                             data_labeler_parameters['label_mapping'])
        self.assertListEqual(
            data_labeler.labels,
            ['PAD', 'UNKNOWN', 'ADDRESS', 'PERSON'])
        self.assertIsInstance(
            data_labeler.preprocessor, data_processing.BaseDataPreprocessor)
        self.assertIsInstance(
            data_labeler.postprocessor, data_processing.BaseDataPostprocessor)

    def test_load_from_disk(self, mock_open, mock_load_model,
                            mock_base_processor):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_base_processor)

        # load default
        data_labeler = UnstructuredDataLabeler.load_from_disk('fake/path')

        self.assertDictEqual(data_labeler.label_mapping,
                             data_labeler_parameters['label_mapping'])
        self.assertListEqual(
            data_labeler.labels,
            ['PAD', 'UNKNOWN', 'ADDRESS', 'PERSON'])
        self.assertIsInstance(
            data_labeler.preprocessor, data_processing.BaseDataPreprocessor)
        self.assertIsInstance(
            data_labeler.postprocessor, data_processing.BaseDataPostprocessor)

    def test_reverse_label_mappings(self, mock_open, mock_load_model,
                                    mock_load_processor, *mocks):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_load_processor)

        # load default
        data_labeler = UnstructuredDataLabeler()

        reverse_label_mapping = {
            0: 'PAD',
            1: 'UNKNOWN',
            2: 'ADDRESS',
            3: 'PERSON',
        }

        self.assertDictEqual(reverse_label_mapping,
                             data_labeler.reverse_label_mapping)

    def test_labels(self, mock_open, mock_load_model, mock_load_processor,
                    *mocks):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_load_processor)

        # load default
        labels = ['PAD', 'UNKNOWN', 'ADDRESS', 'PERSON']
        data_labeler = UnstructuredDataLabeler()

        self.assertListEqual(labels, data_labeler.labels)

    def test_invalid_data_formats(self, *mocks):

        def _invalid_check(data):
            with self.assertRaisesRegex(TypeError,
                                        "Data must be imported using the "
                                        "data_readers, pd.DataFrames, "
                                        "np.ndarrays, or lists."):
                UnstructuredDataLabeler._check_and_return_valid_data_format(
                    data)

        invalid_data = ["string", 1, None, dict()]
        print("\nInvalid Data Checks:")
        for data in invalid_data:
            print("\tChecking data format: {}".format(str(type(data))))
            _invalid_check(data)

        # cannot predict dict
        _invalid_check(data=None)

        # cannot predict dict
        _invalid_check(data={})

    def test_valid_fit_data_formats(self, *mocks):

        def _valid_check(data):
            try:
                print("\tChecking data format: {}".format(str(type(data))))
                data = UnstructuredDataLabeler.\
                    _check_and_return_valid_data_format(
                        data, fit_or_predict='fit')
            except Exception as e:
                self.fail("Exception raised on input of accepted types.")
            return data

        valid_data = [
            CSVData(data=pd.DataFrame([])),
            JSONData(data=pd.DataFrame([])),
            ParquetData(data=pd.DataFrame([])),
            AVROData(data=pd.DataFrame([])),
            pd.DataFrame([]),
            list(),
            np.array([])
        ]
        print("\nValid Data Fit Checks:")
        for data in valid_data:
            data = _valid_check(data)
            self.assertIsInstance(data, np.ndarray)

    def test_valid_predict_data_formats(self, *mocks):

        def _valid_check(data):
            try:
                print("\tChecking data format: {}".format(str(type(data))))
                data = UnstructuredDataLabeler.\
                    _check_and_return_valid_data_format(
                        data, fit_or_predict='predict')
            except Exception as e:
                self.fail("Exception raised on input of accepted types.")
            return data

        valid_data = [
            CSVData(data=pd.DataFrame([])),
            JSONData(data=pd.DataFrame([])),
            ParquetData(data=pd.DataFrame([])),
            AVROData(data=pd.DataFrame([])),
            pd.DataFrame([]),
            list(),
            np.array([]),
            pd.Series([], dtype=object)
        ]
        print("\nValid Data Predict Checks:")
        for data in valid_data:
            data = _valid_check(data)
            self.assertTrue(
                isinstance(data, np.ndarray) or
                isinstance(data, pd.Series) or
                isinstance(data, pd.DataFrame)
            )

    def test_set_params(self, mock_open, mock_load_model, mock_base_processor):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_base_processor)

        # load default
        data_labeler = UnstructuredDataLabeler()

        # check empty sent
        with self.assertRaisesRegex(ValueError, re.escape(
                                    'The params dict must have the following '
                                    'format:\nparams=dict(preprocessor=dict(...'
                                    '), model=dict(...), postprocessor=dict(...'
                                    ')), where each sub-dict contains '
                                    'parameters of the specified data_labeler '
                                    'pipeline components.')):
            data_labeler.set_params(None)

        with self.assertRaisesRegex(ValueError, re.escape(
                                    'The params dict must have the following '
                                    'format:\nparams=dict(preprocessor=dict(...'
                                    '), model=dict(...), postprocessor=dict(...'
                                    ')), where each sub-dict contains '
                                    'parameters of the specified data_labeler '
                                    'pipeline components.')):
            data_labeler.set_params({})

        # test if invalid key sent
        with self.assertRaisesRegex(ValueError, re.escape(
                                    'The params dict must have the following '
                                    'format:\nparams=dict(preprocessor=dict(...'
                                    '), model=dict(...), postprocessor=dict(...'
                                    ')), where each sub-dict contains '
                                    'parameters of the specified data_labeler '
                                    'pipeline components.')):
            data_labeler.set_params({'bad key': None})

        # validate no errors occur when correct params are sent
        data_labeler._preprocessor.get_parameters.return_value = dict()
        data_labeler._model.get_parameters.return_value = dict()
        data_labeler._postprocessor.get_parameters.return_value = dict()

        data_labeler.set_params({'preprocessor': {'test': 1},
                                 'model': {'test': 1},
                                 'postprocessor': {'test2': 3}})

        # validate warning on overlaps.
        # here we presume parameters are set as dict(test=1), dict(test=2)
        data_labeler._preprocessor.get_parameters.return_value = dict(test=1)
        data_labeler._model.get_parameters.return_value = dict(test=2)
        with self.assertWarnsRegex(RuntimeWarning,
                                   'Model and preprocessor value for `test` do '
                                   'not match. 2 != 1'):
            data_labeler.set_params({'preprocessor': {'test': 1},
                                     'model': {'test': 2}})

        # check if param sent for missing pipeline component
        data_labeler._preprocessor = None
        with self.assertRaisesRegex(ValueError,
                                    'Parameters for the preprocessor, model, or'
                                    ' postprocessor were specified when one or '
                                    'more of these were not set in the '
                                    'DataLabeler.'):
            data_labeler.set_params({'preprocessor': {'test': 1}})

        data_labeler._model = None
        with self.assertRaisesRegex(ValueError,
                                    'Parameters for the preprocessor, model, or'
                                    ' postprocessor were specified when one or '
                                    'more of these were not set in the '
                                    'DataLabeler.'):
            data_labeler.set_params({'model': {'test': 1}})

        data_labeler._postprocessor = None
        with self.assertRaisesRegex(ValueError,
                                    'Parameters for the preprocessor, model, or'
                                    ' postprocessor were specified when one or '
                                    'more of these were not set in the '
                                    'DataLabeler.'):
            data_labeler.set_params({'postprocessor': {'test': 1}})

    def test_set_labels(self, mock_open, mock_load_model,
                        mock_load_processor, *mocks):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_load_processor)

        data_labeler = UnstructuredDataLabeler()
        # test label list to label_mapping
        labels = ['a', 'b', 'd', 'c']
        data_labeler.set_labels(labels)
        mock_load_model.return_value.set_label_mapping.assert_called_with(
            label_mapping= ['a', 'b', 'd', 'c'])

        # test label dict to label_mapping
        labels = dict(b=1, c=2, d=3, e=4)
        data_labeler.set_labels(labels)
        mock_load_model.return_value.set_label_mapping.assert_called_with(
            label_mapping=dict(b=1, c=2, d=3, e=4))

    def test_set_model(self, mock_open, mock_load_model, mock_load_processor,
                       *mocks):
        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_load_processor)

        # load default
        data_labeler = UnstructuredDataLabeler()

        # test setting model
        model_mock = mock.Mock(spec=CharacterLevelCnnModel)
        data_labeler.set_model(model_mock)
        self.assertEqual(model_mock, data_labeler.model)

        # test failure bc not model object
        with self.assertRaisesRegex(TypeError,
                                    'The specified model was not of the correct'
                                    ' type, `BaseModel`.'):
            data_labeler.set_model(1)

    def test_set_preprocessor(self, mock_open, mock_load_model,
                              mock_load_processor, *mocks):
        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_load_processor)

        # load default
        data_labeler = UnstructuredDataLabeler()

        # test setting preprocessor
        processor_mock = mock.Mock(
            spec=data_processing.CharPreprocessor)
        data_labeler.set_preprocessor(processor_mock)
        self.assertEqual(processor_mock, data_labeler.preprocessor)

        # test failure bc not processing object
        with self.assertRaisesRegex(TypeError,
                                    'The specified preprocessor was not of the '
                                    'correct type, `DataProcessing`.'):
            data_labeler.set_preprocessor(1)

    def test_set_postprocessor(self, mock_open, mock_load_model,
                               mock_load_processor, *mocks):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_load_processor)

        # load default
        data_labeler = UnstructuredDataLabeler()

        # test setting postprocessor
        processor_mock = mock.Mock(
            spec=data_processing.CharPostprocessor)
        data_labeler.set_postprocessor(processor_mock)
        self.assertEqual(processor_mock, data_labeler.postprocessor)

        # test failure bc not processing object
        with self.assertRaisesRegex(TypeError,
                                    'The specified postprocessor was not of '
                                    'the correct type, `DataProcessing`.'
                                    ):
            data_labeler.set_postprocessor(1)

    def test_save_to_disk(self, mock_open, mock_load_model,
                          mock_load_processor, *mocks):

        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_load_processor)

        # call func
        data_labeler = UnstructuredDataLabeler()

        # setup save mock
        mock_file = setup_save_mock_open(mock_open)

        # save and test
        data_labeler.save_to_disk('test/path')
        self.assertEqual(
            '{"model": {"class": "CharacterLevelCnnModel"}, '
            '"preprocessor": {"class": "CharPreprocessor"}, '
            '"postprocessor": {"class": "CharPostprocessor"}}',
            mock_file.getvalue())

        # close mock
        StringIO.close(mock_file)

    def test_check_pipeline(self, mock_open, mock_load_model,
                            mock_base_processor):
        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_base_processor)

        data_labeler = UnstructuredDataLabeler()

        # check pipeline with no errors w overlap
        data_labeler._model = mock.Mock()
        data_labeler._model.get_parameters.return_value = dict(a=1)
        data_labeler._model.get_parameters.return_value = dict(a=1)
        data_labeler._preprocessor.get_parameters.return_value = dict(a=1)
        data_labeler._postprocessor.get_parameters.return_value = dict(a=1)
        with warnings.catch_warnings(record=True) as w:
            data_labeler.check_pipeline()
        self.assertEqual(0, len(w))  # assert no warnings raised

        # invalid pipeline, model != preprocessor
        data_labeler._model.get_parameters.return_value = dict(a=1)
        data_labeler.preprocessor.get_parameters.return_value = dict(a=2)
        with self.assertWarnsRegex(RuntimeWarning,
                                   'Model and preprocessor value for `a` do '
                                   'not match. 1 != 2'):
            data_labeler.check_pipeline()

        # invalid pipeline, model != postprocessor
        data_labeler.preprocessor.get_parameters.return_value = dict(a=1)
        data_labeler.postprocessor.get_parameters.return_value = dict(a=2)
        with self.assertWarnsRegex(RuntimeWarning,
                                   'Model and postprocessor value for `a` do '
                                   'not match. 1 != 2'):
            data_labeler.check_pipeline()

        # invalid pipeline, preprocessor != postprocessor
        data_labeler._model = mock.Mock()
        data_labeler._model.get_parameters.return_value = dict(a=1)
        data_labeler.preprocessor.get_parameters.return_value = dict(a=1, b=1)
        data_labeler.postprocessor.get_parameters.return_value = dict(a=1, b=2)
        with self.assertWarnsRegex(RuntimeWarning,
                                   'Preprocessor and postprocessor value for '
                                   '`b` do not match. 1 != 2'):
            data_labeler.check_pipeline()

        # valid pipeline, preprocessor != postprocessor but skips processor
        data_labeler._model = mock.Mock()
        data_labeler._model.get_parameters.return_value = dict(a=1)
        data_labeler.preprocessor.get_parameters.return_value = dict(a=1, b=1)
        data_labeler.postprocessor.get_parameters.return_value = dict(a=1, b=2)
        with warnings.catch_warnings(record=True) as w:
            data_labeler.check_pipeline(skip_postprocessor=True)
        self.assertEqual(0, len(w))

        # assert raises error instead of warning
        with self.assertRaisesRegex(RuntimeError,
                                    'Preprocessor and postprocessor value for '
                                    '`b` do not match. 1 != 2'):
            data_labeler.check_pipeline(error_on_mismatch=True)

    def test_fit(self, mock_open, mock_load_model, mock_base_processor):
        self._setup_mock_load_model(mock_load_model)
        self._setup_mock_load_processor(mock_base_processor)

        def reset_mocks(dl):
            dl.preprocessor.process.reset_mock()
            dl._model.set_label_mapping.reset_mock()
            dl._model.reset_weights.reset_mock()
            dl._model.fit.reset_mock()

        # setup data
        samples = [['1'], ['2'], ['3'], ['4'], ['5']]
        labels = [[1], [2], [3], [4], [5]]
        data = list(zip(samples, labels))
        fake_results = dict(pred='fake pred', conf='fake conf')
        fake_results_no_conf = dict(pred='fake pred')

        # setup mocks
        data_labeler = DataLabeler(labeler_type='unstructured', trainable=True)
        data_labeler._model = mock.Mock()
        data_labeler._preprocessor.get_parameters.return_value = dict()
        data_labeler._model.get_parameters.return_value = dict()
        data_labeler._model.label_mapping = dict(fake=1)
        data_labeler._model.requires_zero_mapping = True
        data_labeler._model.fit.return_value = 'fake result'

        # test empty data
        with self.assertRaisesRegex(ValueError, 'No data or labels to fit.'):
            data_labeler.fit(x=[], y=[])

        # test default
        output = data_labeler.fit(x=samples, y=labels)
        self.assertEqual(2, data_labeler.preprocessor.process.call_count)
        data_labeler._model.set_label_mapping.assert_not_called()
        data_labeler._model.reset_weights.assert_not_called()
        self.assertEqual(1, data_labeler._model.fit.call_count)
        self.assertEqual(1, len(output))

        # test validation_split = 1
        with self.assertRaisesRegex(ValueError,
                                    "`validation_split` must be >= 0 and less "
                                    "than 1.0"):
            data_labeler.fit(x=samples, y=labels, validation_split=1)

        # test validation_split = 0
        reset_mocks(data_labeler)

        output = data_labeler.fit(x=samples, y=labels, validation_split=0)
        self.assertEqual(1, data_labeler.preprocessor.process.call_count)
        data_labeler._model.set_label_mapping.assert_not_called()
        data_labeler._model.reset_weights.assert_not_called()
        self.assertEqual(1, data_labeler._model.fit.call_count)
        self.assertEqual(1, len(output))

        # test labels
        reset_mocks(data_labeler)

        output = data_labeler.fit(
            x=samples, y=labels, labels=['new', 'labels'])
        self.assertEqual(2, data_labeler.preprocessor.process.call_count)
        data_labeler._model.set_label_mapping.assert_called_with(
            label_mapping=['new', 'labels'])
        data_labeler._model.reset_weights.assert_not_called()
        self.assertEqual(1, data_labeler._model.fit.call_count)
        self.assertEqual(1, len(output))

        # test reset_weights=True
        reset_mocks(data_labeler)

        output = data_labeler.fit(
            x=samples, y=labels, reset_weights=True)
        self.assertEqual(2, data_labeler.preprocessor.process.call_count)
        data_labeler._model.set_label_mapping.assert_not_called()
        data_labeler._model.reset_weights.assert_called_once()
        self.assertEqual(1, data_labeler._model.fit.call_count)
        self.assertEqual(1, len(output))

        # test multiple epochs
        reset_mocks(data_labeler)

        output = data_labeler.fit(x=samples, y=labels, epochs=3)
        self.assertEqual(6, data_labeler.preprocessor.process.call_count)
        data_labeler._model.set_label_mapping.assert_not_called()
        data_labeler._model.reset_weights.assert_not_called()
        self.assertEqual(3, data_labeler._model.fit.call_count)
        self.assertEqual(3, len(output))