<kaggle_start><code># # NLP Disaster Tweets Kaggle Mini-Project
# ## Overview
# This notebook is a practice of utilizing the TensorFlow and Keras to build a Neural Network (NN) to identify false and real emergency alert from the post on Twitter. The practice is based on a Kaggle competition, and the data can be obtained from the competition website at https://www.kaggle.com/competitions/nlp-getting-started/data.
# This notebook can also be found at https://github.com/Lorby04/msds/tree/main/dl/week4 and https://www.kaggle.com/code/lorbybi/nlp-disaster-tweets
# 1. Data preparing
# Downloading data from the source, extract the compressed files to local disk.
# The data includes 3 csv files, one includes the training data, and one includes the testing data, another one is a sample file.
# We will use the training data to train the model and then make estimation on the testing data.
# import required libraries
import pathlib
import os
import sys
import concurrent.futures
import pandas as pd
import numpy as np
import random as rn
import tensorflow as tf
import tensorflow_datasets as tfds
import matplotlib.pyplot as plt
from timeit import default_timer as timer
import re
import shutil
import string


class Constants:
    QUICK_TEST = False
    MAX_FILES = sys.maxsize
    TARGET_SIZE = [96, 96]
    BATCH_SIZE = 32
    RETRAIN_MODEL = False
    MAX_FEATURES = 1000
    SEQUENCE_LEN = 10


class Config:
    def __init__(self):
        self.dataset_url = (
            "https://www.kaggle.com/competitions/histopathologic-cancer-detection/data"
        )
        self.data_root_dir = "/kaggle/input/mayo-clinic-strip-ai/"
        self.working_dir = "/kaggle/working/"
        self.temp_dir = "/kaggle/working/temp/"
        if os.path.exists("/kaggle"):
            print("Working in kaggle notebook enviorment")
        else:
            print("Working locally")
            self.data_root_dir = "./nlp-getting-started/"
            self.working_dir = self.data_root_dir
            self.temp_dir = "./nlp-getting-started/"
        self.temp_train_dir = self.temp_dir + "train/"
        self.temp_test_dir = self.temp_dir + "test/"
        self.data_dir = self.data_root_dir
        self.train_csv = self.data_dir + "train.csv"
        self.test_csv = self.data_dir + "test.csv"
        self.origin_train_dir = "././nlp-getting-started/train/"
        self.origin_test_dir = "././nlp-getting-started/test/"
        self.train_dir = self.temp_train_dir  # self.data_dir + "train/"
        self.test_dir = self.temp_test_dir  # self.data_dir + "test/"
        self.dir_true = self.train_dir + "1/"
        self.dir_false = self.train_dir + "0/"
        self.origin_train_path = pathlib.Path(self.origin_train_dir).with_suffix("")
        self.origin_test_path = pathlib.Path(self.origin_test_dir).with_suffix("")
        self.train_path = pathlib.Path(self.train_dir).with_suffix("")
        self.test_path = pathlib.Path(self.test_dir).with_suffix("")

    # Convert the image from tif to jpg
    # Move train data to subclass directory
    def new_dir(directory):
        cmd = "mkdir " + directory
        os.system(cmd)

    def download_data(self):
        if not os.path.exists(self.data_dir):
            cmd = "pip install opendatasets"
            os.system(cmd)
            import opendatasets as od

            od.download(self.dataset_url)
            new_dir(data_dir)
            new_dir(train_dir)
            new_dir(test_dir)
            new_dir(dir_true)
            new_dir(dir_false)


# Download data, create pandas dataframe from the csv files.
class Df:
    def __init__(self, cfg: Config):
        self.train_df = pd.read_csv(cfg.train_csv)
        self.test_df = None if cfg.test_csv == None else pd.read_csv(cfg.test_csv)

    def combine_train_set(self):
        pass

    def glance_at(self, df):
        print("\nQuick view of {} data set".format(df))
        if df == "train":
            print(self.train_df)
            self.train_df.info()
            print("Target values:")
            vc = self.train_df["target"].value_counts()
            print(vc)
            labels = ["Fake", "Real"]
            plt.pie(vc, autopct="%1.1f%%", labels=labels)
            plt.legend(vc.index, loc="best")
            plt.show()
        elif df == "test" and self.test_df is not None:
            print(self.test_df)
            self.test_df.info()
        else:
            pass

    def glance(self):
        self.glance_at("train")
        self.glance_at("test")


config = Config()
config.download_data()
df = Df(config)


# ## Exploratory Data Analysis (EDA)
# Firstly, have a glance on the training data. It includes 7613 observations and 5 features. Among these features, most of the observations of 'keywords' and 'location' are not available. So these 2 features will not be considered in the training. The 'id' field is for indexing only, it is not needed for training.
# The 'text' is the input data and the 'target' field is output data.
# Further check on the values of the feature 'target', it has 2 unique values: 0 means false alert and 1 means true alert.
# There are 3263 rows of data for testing.
# From the statistics information, 57% of the training data are false alerts and the other 43% are true alerts.
df.glance_at("train")
df.train_df[df.train_df["target"] == 1]["text"].values[100]
df.train_df[df.train_df["target"] != 1]["text"].values[100]
df.glance_at("test")

# The tensorflow dataset API text_dataset_from_directory requires the training data to be organized in a hierarchical structure with the sundir name being the name of the category. To follow the requirement of the API, the 'text' field in the dataframe are copied to txt file respectively.
# Cross check the raw data with the files in the traning directory. The data matches with the csv file.
list(config.train_path.iterdir())


def check_create_dir(path):
    if not os.path.exists(path):
        os.makedirs(path)
    else:
        pass


class TextRawDataInDisk:
    def __init__(self, df, to_path, quick_test=Constants.QUICK_TEST):
        self.df = df  # List of df
        self.to_path = to_path
        self.max_files = min(len(df), Constants.MAX_FILES)
        self.quick_test = quick_test
        if quick_test:
            self.max_files = min(4, len(df))
        self.loaded_files = 0
        self.skipped_files = 0
        self.is_training = "target" in df.columns


class TextLoaderToDisk:
    def build(df, to_path, quick_test=Constants.QUICK_TEST):
        owner = TextRawDataInDisk(df, to_path, quick_test)
        check_create_dir(to_path)
        loader = TextLoaderToDisk(owner)
        loader.load()
        return loader

    def __init__(self, owner=None):
        self._owner = owner

    def reset_owner(self, owner):
        self._owner = owner

    def owner(self):
        return self._owner

    def load(self):
        start = timer()
        self.owner().loaded_files = 0
        loop_start = timer()
        to_dir = self.owner().to_path
        for row in self.owner().df.itertuples(index=False):
            txt = "/" + str(row.id) + ".txt"
            if self.owner().is_training:
                label = str(row.target)
                to_dir = self.owner().to_path + label + "/"
            else:
                assert to_dir == self.owner().to_path
            to_file = to_dir + str(row.id) + ".txt"
            if os.path.exists(to_file):
                self.owner().skipped_files += 1
                self.owner().loaded_files += 1
                continue
            check_create_dir(to_dir)
            with open(to_file, "w") as f:
                f.write(row.text)
            self.owner().loaded_files += 1
            if self.owner().loaded_files % 50 == 0:
                print(".", end=" ")
        elapsed = timer() - start
        print(
            "{} files are stored to {} in {} seconds, {} are skipped among which.".format(
                self.owner().loaded_files,
                self.owner().to_path,
                elapsed,
                self.owner().skipped_files,
            )
        )

    def statistics(self):
        print(
            "{} file are stored to {} , {} are skipped among which.".format(
                self.owner().loaded_files,
                self.owner().to_path,
                self.owner().skipped_files,
            )
        )


class DatasetFromDisk:
    def __init__(self, td, training_ratio=0.8):
        self.ds_from = td.to_path
        assert self.ds_from != None
        self.is_training = td.is_training
        self.training_ratio = (
            training_ratio if training_ratio > 0 and training_ratio <= 1 else 0.8
        )
        self.train_ds_ = None
        self.val_ds_ = None
        self.test_ds_ = None
        self.is_preprocessed = False


class DatasetBuilderFromDisk:
    def build(td, training_ratio=0.8):
        owner = DatasetFromDisk(td, training_ratio)
        builder = DatasetBuilderFromDisk(owner)
        builder.build_dataset()
        return builder

    def __init__(self, owner=None):  # MUST set owner before using
        self._owner = owner

    def reset_owner(self, owner):
        self._owner = owner

    def owner(self):
        return self._owner

    def build_dataset(self):
        if self.owner().is_training:
            self.owner().train_ds_ = tf.keras.utils.text_dataset_from_directory(
                self.owner().ds_from,
                validation_split=0.2,
                subset="training",
                seed=123,
                batch_size=Constants.BATCH_SIZE,
            )
            self.owner().val_ds_ = tf.keras.utils.text_dataset_from_directory(
                self.owner().ds_from,
                validation_split=1 - self.owner().training_ratio,
                subset="validation",
                seed=123,
                batch_size=Constants.BATCH_SIZE,
            )
        else:
            pass
        self.pre_process()

    def pre_process(self):
        return  # Don't perform preprocess

    def train_ds(self):
        return self.owner().train_ds_

    def val_ds(self):
        return self.owner().val_ds_

    def test_ds(self):
        return self.owner().test_ds_

    def show_train_texts(self, ds, number):
        assert ds != None and number >= 1
        i = 0
        for txts, labels in ds:
            # print(imgs)
            # print(labels)
            for txt in txts:
                print(txt)
                i += 1
                if i >= number:
                    return

    def show_test_texts(self, ds, number):
        i = 0
        for b in ds:
            for t in b:
                print(t)
                i += 1
                if i >= number:
                    return

    def show_texts(self):
        if self.owner().train_ds_ != None:
            self.show_train_texts(self.owner().train_ds_, 3)
            assert self.owner().val_ds_ != None
            self.show_train_texts(self.owner().val_ds_, 3)
        else:
            assert self.owner().test_ds_ != None
            self.show_test_texts(self.owner().test_ds_, 3)

    def statistics(self):
        if self.owner().train_ds_ != None:
            for i, target in enumerate(self.owner().train_ds_.class_names):
                print("Index:", i, "corresponds to:", target)
            for text_batch, target_batch in self.owner().train_ds_.take(1):
                for i in range(3):
                    print("Statement: ", text_batch.numpy()[i])
                    print("Target:", target_batch.numpy()[i])
            for text_batch, target_batch in self.owner().val_ds_.take(1):
                for i in range(3):
                    print("Statement: ", text_batch.numpy()[i])
                    print("Target:", target_batch.numpy()[i])
        else:
            for text_batch in self.owner().test_ds_.take(1):
                for i in range(4):
                    print("Statement: ", text_batch.numpy()[i])


# Use alias to try different options
TextData = TextRawDataInDisk
TextLoader = TextLoaderToDisk
Dataset = DatasetFromDisk
DatasetBuilder = DatasetBuilderFromDisk


# Generate training and validation dataset with text_dataset_from_directory
train_texts = TextLoader.build(df.train_df, config.temp_train_dir).owner()
train_text_loader = TextLoader(train_texts)
ds = DatasetBuilder.build(train_texts).owner()
ds_loader = DatasetBuilder(ds)
ds_loader.show_texts()

# Check the dataset is as expected.
ds_loader.statistics()

# ## Model Architecture
# The high level architecture of the training network are referring to the tensorflow training material for text classification.
# The model includes the following layers:
# 1. Input layer
# 2. Vectorization layer: This layer is to standardize, tokenize, and vectorize the data. The vectorization layer is from the tf.keras.layers.TextVectorization
# 3. Hidden network layers, including embedding pooling and dense layer for respective model.
# 4. SparseCategoricalCrossentropy is used as loss function
# 5. Optimize with Adam function
# 6. Accuracy as the metric parameter
# 7. Early stop if validation accuracy is not increasing for continuous 3 epochs
# Two models, int vectorization base and binary vectorization base, are tried. The hyper-parameters are adjusted with multiple running to avoid overfitting and underfitting with reasonable accuracy.
#
from tensorflow.keras.optimizers.legacy import Adam


class Model:
    def __init__(self, ds):
        self.ds = ds
        self.history = None
        self.opt = Adam()
        self.loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
        self.metrics = ["accuracy"]
        self.callbacks = [
            tf.keras.callbacks.EarlyStopping(monitor="val_accuracy", patience=5)
        ]
        self.model = self.build_model()

    def fit(self, ds=None, epochs=10):
        if ds == None:
            ds = self.ds
        self.history = self.model.fit(
            ds.train_ds_,
            validation_data=ds.val_ds_,
            epochs=epochs,
            callbacks=self.callbacks,
        )

    def build_model(self):
        import tensorflow.strings as tfs
        from tensorflow.strings import regex_replace as rr

        # Prepare the vectorization layer
        normalization = lambda s: rr(
            rr(tfs.lower(s), "<br />", " "), "[%s]" % re.escape(string.punctuation), ""
        )
        extraction = lambda x, y: x
        vectorize_layer = tf.keras.layers.TextVectorization(
            standardize=normalization,
            max_tokens=Constants.MAX_FEATURES,
            output_mode="binary",
        )
        texts = ds.train_ds_.map(extraction)
        vectorize_layer.adapt(texts)
        model = tf.keras.Sequential(
            [vectorize_layer, tf.keras.layers.Dense(len(self.ds.train_ds_.class_names))]
        )
        model.compile(loss=self.loss, optimizer=self.opt, metrics=["accuracy"])
        return model


class ModelVisualization:
    def __init__(self, model: Model):
        self.model = model

    def show_history(self):
        history = self.model.history.history
        fig = plt.figure(figsize=(6, 10))
        fig_width = 1
        fig_height = 2
        ax = fig.subplots(fig_height, fig_width)
        ax[0].plot(history["accuracy"], "*-")
        ax[0].plot(history["val_accuracy"], "x-")
        ax[0].legend(["train", "validation"])
        ax[0].set_xlabel("epoch")
        ax[0].set_ylabel("accuracy")
        ax[0].set_title("Accuracies")
        ax[1].plot(history["loss"], "*-")
        ax[1].plot(history["val_loss"], "x-")
        ax[1].legend(["train", "validation"])
        ax[1].set_title("Losses")
        ax[1].set_xlabel("epoch")
        ax[1].set_ylabel("loss")
        plt.show()

    def summary(self):
        self.model.model.summary()
        self.model.model.get_metrics_result()


m = Model(ds)
m.fit(epochs=100)
vm = ModelVisualization(m)
vm.show_history()
vm.summary()


# ## Test
# From the training result, the binary model has higher validation accuracy and there is no significant indication of overfitting.
# So the binary model will be used for the testing.
# Before testing, a softmax layer is added to the model to make the probability more obvious.
# The test images are from the previously prepared test directory.
# To make the result clearer, a pandas dataframe is created for the record.
class Predictor:
    def __init__(self, model: tf.keras.Model, test_df):
        self.model = model
        # self.model.add(tf.keras.layers.Softmax())
        self.test_df = test_df
        self.df = None
        self.pred = self.predict()
        self.pred_df = None
        self.build_result_df()
        self.submit()

    def predict(self):
        y_pred_prob = self.model.model.predict(np.array(self.test_df.text))
        return tf.keras.activations.softmax(tf.convert_to_tensor(y_pred_prob))

    def build_result_df(self):
        self.pred_df = pd.DataFrame(self.test_df["id"])
        self.pred_df["target"] = pd.DataFrame(self.pred).apply(np.argmax, axis=1)

    def submit(self):
        self.pred_df.to_csv("submission.csv", index=False)


class PredictorDisplay:
    def __init__(self, owner):
        self.owner = owner

    def dump(self):
        print("Prediction possibilities:")
        print(self.owner.pred_df)
        print(self.owner.pred_df.target.value_counts())

    def show_result(self):
        labels = ["Fake", "Real"]
        plt.pie(
            self.owner.pred_df["target"].value_counts(),
            autopct="%1.1f%%",
            labels=labels,
        )
        plt.show()


pred = Predictor(m, df.test_df)
disp = PredictorDisplay(pred)
disp.dump()
disp.show_result()
labels = ["Fake", "Real"]
plt.pie(df.train_df["target"].value_counts(), autopct="%1.1f%%", labels=labels)
plt.show()
<|endoftext|>
[{"Id": 121063973, "ScriptId": 35176022, "ParentScriptVersionId": NaN, "ScriptLanguageId": 9, "AuthorUserId": 11726793, "CreationDate": "03/04/2023 21:45:31", "VersionNumber": 2.0, "Title": "NLP Disaster Tweets", "EvaluationDate": "03/04/2023", "IsChange": true, "TotalLines": 563.0, "LinesInsertedFromPrevious": 485.0, "LinesChangedFromPrevious": 0.0, "LinesUnchangedFromPrevious": 78.0, "LinesInsertedFromFork": NaN, "LinesDeletedFromFork": NaN, "LinesChangedFromFork": NaN, "LinesUnchangedFromFork": NaN, "TotalVotes": 0}]
# # NLP Disaster Tweets Kaggle Mini-Project
# ## Overview
# This notebook is a practice of utilizing the TensorFlow and Keras to build a Neural Network (NN) to identify false and real emergency alert from the post on Twitter. The practice is based on a Kaggle competition, and the data can be obtained from the competition website at https://www.kaggle.com/competitions/nlp-getting-started/data.
# This notebook can also be found at https://github.com/Lorby04/msds/tree/main/dl/week4 and https://www.kaggle.com/code/lorbybi/nlp-disaster-tweets
# 1. Data preparing
# Downloading data from the source, extract the compressed files to local disk.
# The data includes 3 csv files, one includes the training data, and one includes the testing data, another one is a sample file.
# We will use the training data to train the model and then make estimation on the testing data.
# import required libraries
import pathlib
import os
import sys
import concurrent.futures
import pandas as pd
import numpy as np
import random as rn
import tensorflow as tf
import tensorflow_datasets as tfds
import matplotlib.pyplot as plt
from timeit import default_timer as timer
import re
import shutil
import string


class Constants:
    QUICK_TEST = False
    MAX_FILES = sys.maxsize
    TARGET_SIZE = [96, 96]
    BATCH_SIZE = 32
    RETRAIN_MODEL = False
    MAX_FEATURES = 1000
    SEQUENCE_LEN = 10


class Config:
    def __init__(self):
        self.dataset_url = (
            "https://www.kaggle.com/competitions/histopathologic-cancer-detection/data"
        )
        self.data_root_dir = "/kaggle/input/mayo-clinic-strip-ai/"
        self.working_dir = "/kaggle/working/"
        self.temp_dir = "/kaggle/working/temp/"
        if os.path.exists("/kaggle"):
            print("Working in kaggle notebook enviorment")
        else:
            print("Working locally")
            self.data_root_dir = "./nlp-getting-started/"
            self.working_dir = self.data_root_dir
            self.temp_dir = "./nlp-getting-started/"
        self.temp_train_dir = self.temp_dir + "train/"
        self.temp_test_dir = self.temp_dir + "test/"
        self.data_dir = self.data_root_dir
        self.train_csv = self.data_dir + "train.csv"
        self.test_csv = self.data_dir + "test.csv"
        self.origin_train_dir = "././nlp-getting-started/train/"
        self.origin_test_dir = "././nlp-getting-started/test/"
        self.train_dir = self.temp_train_dir  # self.data_dir + "train/"
        self.test_dir = self.temp_test_dir  # self.data_dir + "test/"
        self.dir_true = self.train_dir + "1/"
        self.dir_false = self.train_dir + "0/"
        self.origin_train_path = pathlib.Path(self.origin_train_dir).with_suffix("")
        self.origin_test_path = pathlib.Path(self.origin_test_dir).with_suffix("")
        self.train_path = pathlib.Path(self.train_dir).with_suffix("")
        self.test_path = pathlib.Path(self.test_dir).with_suffix("")

    # Convert the image from tif to jpg
    # Move train data to subclass directory
    def new_dir(directory):
        cmd = "mkdir " + directory
        os.system(cmd)

    def download_data(self):
        if not os.path.exists(self.data_dir):
            cmd = "pip install opendatasets"
            os.system(cmd)
            import opendatasets as od

            od.download(self.dataset_url)
            new_dir(data_dir)
            new_dir(train_dir)
            new_dir(test_dir)
            new_dir(dir_true)
            new_dir(dir_false)


# Download data, create pandas dataframe from the csv files.
class Df:
    def __init__(self, cfg: Config):
        self.train_df = pd.read_csv(cfg.train_csv)
        self.test_df = None if cfg.test_csv == None else pd.read_csv(cfg.test_csv)

    def combine_train_set(self):
        pass

    def glance_at(self, df):
        print("\nQuick view of {} data set".format(df))
        if df == "train":
            print(self.train_df)
            self.train_df.info()
            print("Target values:")
            vc = self.train_df["target"].value_counts()
            print(vc)
            labels = ["Fake", "Real"]
            plt.pie(vc, autopct="%1.1f%%", labels=labels)
            plt.legend(vc.index, loc="best")
            plt.show()
        elif df == "test" and self.test_df is not None:
            print(self.test_df)
            self.test_df.info()
        else:
            pass

    def glance(self):
        self.glance_at("train")
        self.glance_at("test")


config = Config()
config.download_data()
df = Df(config)


# ## Exploratory Data Analysis (EDA)
# Firstly, have a glance on the training data. It includes 7613 observations and 5 features. Among these features, most of the observations of 'keywords' and 'location' are not available. So these 2 features will not be considered in the training. The 'id' field is for indexing only, it is not needed for training.
# The 'text' is the input data and the 'target' field is output data.
# Further check on the values of the feature 'target', it has 2 unique values: 0 means false alert and 1 means true alert.
# There are 3263 rows of data for testing.
# From the statistics information, 57% of the training data are false alerts and the other 43% are true alerts.
df.glance_at("train")
df.train_df[df.train_df["target"] == 1]["text"].values[100]
df.train_df[df.train_df["target"] != 1]["text"].values[100]
df.glance_at("test")

# The tensorflow dataset API text_dataset_from_directory requires the training data to be organized in a hierarchical structure with the sundir name being the name of the category. To follow the requirement of the API, the 'text' field in the dataframe are copied to txt file respectively.
# Cross check the raw data with the files in the traning directory. The data matches with the csv file.
list(config.train_path.iterdir())


def check_create_dir(path):
    if not os.path.exists(path):
        os.makedirs(path)
    else:
        pass


class TextRawDataInDisk:
    def __init__(self, df, to_path, quick_test=Constants.QUICK_TEST):
        self.df = df  # List of df
        self.to_path = to_path
        self.max_files = min(len(df), Constants.MAX_FILES)
        self.quick_test = quick_test
        if quick_test:
            self.max_files = min(4, len(df))
        self.loaded_files = 0
        self.skipped_files = 0
        self.is_training = "target" in df.columns


class TextLoaderToDisk:
    def build(df, to_path, quick_test=Constants.QUICK_TEST):
        owner = TextRawDataInDisk(df, to_path, quick_test)
        check_create_dir(to_path)
        loader = TextLoaderToDisk(owner)
        loader.load()
        return loader

    def __init__(self, owner=None):
        self._owner = owner

    def reset_owner(self, owner):
        self._owner = owner

    def owner(self):
        return self._owner

    def load(self):
        start = timer()
        self.owner().loaded_files = 0
        loop_start = timer()
        to_dir = self.owner().to_path
        for row in self.owner().df.itertuples(index=False):
            txt = "/" + str(row.id) + ".txt"
            if self.owner().is_training:
                label = str(row.target)
                to_dir = self.owner().to_path + label + "/"
            else:
                assert to_dir == self.owner().to_path
            to_file = to_dir + str(row.id) + ".txt"
            if os.path.exists(to_file):
                self.owner().skipped_files += 1
                self.owner().loaded_files += 1
                continue
            check_create_dir(to_dir)
            with open(to_file, "w") as f:
                f.write(row.text)
            self.owner().loaded_files += 1
            if self.owner().loaded_files % 50 == 0:
                print(".", end=" ")
        elapsed = timer() - start
        print(
            "{} files are stored to {} in {} seconds, {} are skipped among which.".format(
                self.owner().loaded_files,
                self.owner().to_path,
                elapsed,
                self.owner().skipped_files,
            )
        )

    def statistics(self):
        print(
            "{} file are stored to {} , {} are skipped among which.".format(
                self.owner().loaded_files,
                self.owner().to_path,
                self.owner().skipped_files,
            )
        )


class DatasetFromDisk:
    def __init__(self, td, training_ratio=0.8):
        self.ds_from = td.to_path
        assert self.ds_from != None
        self.is_training = td.is_training
        self.training_ratio = (
            training_ratio if training_ratio > 0 and training_ratio <= 1 else 0.8
        )
        self.train_ds_ = None
        self.val_ds_ = None
        self.test_ds_ = None
        self.is_preprocessed = False


class DatasetBuilderFromDisk:
    def build(td, training_ratio=0.8):
        owner = DatasetFromDisk(td, training_ratio)
        builder = DatasetBuilderFromDisk(owner)
        builder.build_dataset()
        return builder

    def __init__(self, owner=None):  # MUST set owner before using
        self._owner = owner

    def reset_owner(self, owner):
        self._owner = owner

    def owner(self):
        return self._owner

    def build_dataset(self):
        if self.owner().is_training:
            self.owner().train_ds_ = tf.keras.utils.text_dataset_from_directory(
                self.owner().ds_from,
                validation_split=0.2,
                subset="training",
                seed=123,
                batch_size=Constants.BATCH_SIZE,
            )
            self.owner().val_ds_ = tf.keras.utils.text_dataset_from_directory(
                self.owner().ds_from,
                validation_split=1 - self.owner().training_ratio,
                subset="validation",
                seed=123,
                batch_size=Constants.BATCH_SIZE,
            )
        else:
            pass
        self.pre_process()

    def pre_process(self):
        return  # Don't perform preprocess

    def train_ds(self):
        return self.owner().train_ds_

    def val_ds(self):
        return self.owner().val_ds_

    def test_ds(self):
        return self.owner().test_ds_

    def show_train_texts(self, ds, number):
        assert ds != None and number >= 1
        i = 0
        for txts, labels in ds:
            # print(imgs)
            # print(labels)
            for txt in txts:
                print(txt)
                i += 1
                if i >= number:
                    return

    def show_test_texts(self, ds, number):
        i = 0
        for b in ds:
            for t in b:
                print(t)
                i += 1
                if i >= number:
                    return

    def show_texts(self):
        if self.owner().train_ds_ != None:
            self.show_train_texts(self.owner().train_ds_, 3)
            assert self.owner().val_ds_ != None
            self.show_train_texts(self.owner().val_ds_, 3)
        else:
            assert self.owner().test_ds_ != None
            self.show_test_texts(self.owner().test_ds_, 3)

    def statistics(self):
        if self.owner().train_ds_ != None:
            for i, target in enumerate(self.owner().train_ds_.class_names):
                print("Index:", i, "corresponds to:", target)
            for text_batch, target_batch in self.owner().train_ds_.take(1):
                for i in range(3):
                    print("Statement: ", text_batch.numpy()[i])
                    print("Target:", target_batch.numpy()[i])
            for text_batch, target_batch in self.owner().val_ds_.take(1):
                for i in range(3):
                    print("Statement: ", text_batch.numpy()[i])
                    print("Target:", target_batch.numpy()[i])
        else:
            for text_batch in self.owner().test_ds_.take(1):
                for i in range(4):
                    print("Statement: ", text_batch.numpy()[i])


# Use alias to try different options
TextData = TextRawDataInDisk
TextLoader = TextLoaderToDisk
Dataset = DatasetFromDisk
DatasetBuilder = DatasetBuilderFromDisk


# Generate training and validation dataset with text_dataset_from_directory
train_texts = TextLoader.build(df.train_df, config.temp_train_dir).owner()
train_text_loader = TextLoader(train_texts)
ds = DatasetBuilder.build(train_texts).owner()
ds_loader = DatasetBuilder(ds)
ds_loader.show_texts()

# Check the dataset is as expected.
ds_loader.statistics()

# ## Model Architecture
# The high level architecture of the training network are referring to the tensorflow training material for text classification.
# The model includes the following layers:
# 1. Input layer
# 2. Vectorization layer: This layer is to standardize, tokenize, and vectorize the data. The vectorization layer is from the tf.keras.layers.TextVectorization
# 3. Hidden network layers, including embedding pooling and dense layer for respective model.
# 4. SparseCategoricalCrossentropy is used as loss function
# 5. Optimize with Adam function
# 6. Accuracy as the metric parameter
# 7. Early stop if validation accuracy is not increasing for continuous 3 epochs
# Two models, int vectorization base and binary vectorization base, are tried. The hyper-parameters are adjusted with multiple running to avoid overfitting and underfitting with reasonable accuracy.
#
from tensorflow.keras.optimizers.legacy import Adam


class Model:
    def __init__(self, ds):
        self.ds = ds
        self.history = None
        self.opt = Adam()
        self.loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
        self.metrics = ["accuracy"]
        self.callbacks = [
            tf.keras.callbacks.EarlyStopping(monitor="val_accuracy", patience=5)
        ]
        self.model = self.build_model()

    def fit(self, ds=None, epochs=10):
        if ds == None:
            ds = self.ds
        self.history = self.model.fit(
            ds.train_ds_,
            validation_data=ds.val_ds_,
            epochs=epochs,
            callbacks=self.callbacks,
        )

    def build_model(self):
        import tensorflow.strings as tfs
        from tensorflow.strings import regex_replace as rr

        # Prepare the vectorization layer
        normalization = lambda s: rr(
            rr(tfs.lower(s), "<br />", " "), "[%s]" % re.escape(string.punctuation), ""
        )
        extraction = lambda x, y: x
        vectorize_layer = tf.keras.layers.TextVectorization(
            standardize=normalization,
            max_tokens=Constants.MAX_FEATURES,
            output_mode="binary",
        )
        texts = ds.train_ds_.map(extraction)
        vectorize_layer.adapt(texts)
        model = tf.keras.Sequential(
            [vectorize_layer, tf.keras.layers.Dense(len(self.ds.train_ds_.class_names))]
        )
        model.compile(loss=self.loss, optimizer=self.opt, metrics=["accuracy"])
        return model


class ModelVisualization:
    def __init__(self, model: Model):
        self.model = model

    def show_history(self):
        history = self.model.history.history
        fig = plt.figure(figsize=(6, 10))
        fig_width = 1
        fig_height = 2
        ax = fig.subplots(fig_height, fig_width)
        ax[0].plot(history["accuracy"], "*-")
        ax[0].plot(history["val_accuracy"], "x-")
        ax[0].legend(["train", "validation"])
        ax[0].set_xlabel("epoch")
        ax[0].set_ylabel("accuracy")
        ax[0].set_title("Accuracies")
        ax[1].plot(history["loss"], "*-")
        ax[1].plot(history["val_loss"], "x-")
        ax[1].legend(["train", "validation"])
        ax[1].set_title("Losses")
        ax[1].set_xlabel("epoch")
        ax[1].set_ylabel("loss")
        plt.show()

    def summary(self):
        self.model.model.summary()
        self.model.model.get_metrics_result()


m = Model(ds)
m.fit(epochs=100)
vm = ModelVisualization(m)
vm.show_history()
vm.summary()


# ## Test
# From the training result, the binary model has higher validation accuracy and there is no significant indication of overfitting.
# So the binary model will be used for the testing.
# Before testing, a softmax layer is added to the model to make the probability more obvious.
# The test images are from the previously prepared test directory.
# To make the result clearer, a pandas dataframe is created for the record.
class Predictor:
    def __init__(self, model: tf.keras.Model, test_df):
        self.model = model
        # self.model.add(tf.keras.layers.Softmax())
        self.test_df = test_df
        self.df = None
        self.pred = self.predict()
        self.pred_df = None
        self.build_result_df()
        self.submit()

    def predict(self):
        y_pred_prob = self.model.model.predict(np.array(self.test_df.text))
        return tf.keras.activations.softmax(tf.convert_to_tensor(y_pred_prob))

    def build_result_df(self):
        self.pred_df = pd.DataFrame(self.test_df["id"])
        self.pred_df["target"] = pd.DataFrame(self.pred).apply(np.argmax, axis=1)

    def submit(self):
        self.pred_df.to_csv("submission.csv", index=False)


class PredictorDisplay:
    def __init__(self, owner):
        self.owner = owner

    def dump(self):
        print("Prediction possibilities:")
        print(self.owner.pred_df)
        print(self.owner.pred_df.target.value_counts())

    def show_result(self):
        labels = ["Fake", "Real"]
        plt.pie(
            self.owner.pred_df["target"].value_counts(),
            autopct="%1.1f%%",
            labels=labels,
        )
        plt.show()


pred = Predictor(m, df.test_df)
disp = PredictorDisplay(pred)
disp.dump()
disp.show_result()
labels = ["Fake", "Real"]
plt.pie(df.train_df["target"].value_counts(), autopct="%1.1f%%", labels=labels)
plt.show()
