<jupyter_start><jupyter_text>SUTD 2021 50.007 Machine Learning HMM Project Parts 1-3> Group 4:> - <NAME> (1004519)> - <NAME> (1004103)> - <NAME> (1004555)  Initial Workspace Setup<jupyter_code># Setup and install dependencies
!pip3 install numpy

# Import libraries
import os
import numpy as np
from collections import Counter

# Enable floating-point underflow warnings and disable divide-by-zero warnings
np.seterr(under="warn", divide="ignore")

np.set_printoptions(precision=3, suppress=True)

# Set OS-independent paths, relative to current directory
es_train_path = os.path.join("data", "ES", "train")
es_dev_in_path = os.path.join("data", "ES", "dev.in")
es_dev_out_path = os.path.join("data", "ES", "dev.out")
es_dev_p1_out_path = os.path.join("data", "ES", "dev.p1.out")
es_dev_p2_out_path = os.path.join("data", "ES", "dev.p2.out")
es_dev_p3_out_path = os.path.join("data", "ES", "dev.p3.out")
ru_train_path = os.path.join("data", "RU", "train")
ru_dev_in_path = os.path.join("data", "RU", "dev.in")
ru_dev_out_path = os.path.join("data", "RU", "dev.out")
ru_dev_p1_out_path = os.path.join("data", "RU", "dev.p1.out")
ru_dev_p2_out_path = os.path.join("data", "RU", "dev.p2.out")
ru_dev_p3_out_path = os.path.join("data", "RU", "dev.p3.out")

# Define constant variables
N = 7
START, O, BPOS, IPOS, BNEU, INEU, BNEG, INEG, END = 0, 1, 2, 3, 4, 5, 6, 7, 8
labels = {"START": START,
          "O": O,
          "B-positive": BPOS,
          "I-positive": IPOS,
          "B-neutral": BNEU,
          "I-neutral": INEU,
          "B-negative": BNEG,
          "I-negative": INEG,
          "END": END}
labels_list = ["START", "O", "B-positive", "I-positive", "B-neutral", "I-neutral", "B-negative", "I-negative", "END"]

# Initialise a random number generator with a fixed seed for reproducible results and deterministic behavior
rng = np.random.default_rng(1004519 + 1004103 + 1004555)
# Read dev.in data
def read_dev_in_data(filepath):
    results = []
    with open(filepath, "r", encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            results.append(line.strip())
    return results

# Read dev.out data
def read_dev_out_data(filepath):
    results = []
    with open(filepath, "r", encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            if len(line.strip().rsplit(" ", 1)) == 2:
                token, label = line.strip().rsplit(" ", 1)
                results.append((token, labels[label]))
            else:
                continue
    return results

# Read dev.out data with line ending
def read_dev_out_data_w_end(filepath):
    results = []
    with open(filepath, "r", encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            if len(line.strip().rsplit(" ", 1)) == 2:
                token, label = line.strip().rsplit(" ", 1)
                results.append((token, labels[label]))
            else:
                results.append(('', END))
                results.append(('', START))
    return results

# Read training data
def read_training_data(filepath):
    results = []
    with open(filepath, "r", encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            if len(line.strip().rsplit(" ", 1)) == 2:
                token, label = line.strip().rsplit(" ", 1)
                results.append((token, labels[label]))
            else:
                continue
    return results

# Read training data with line ending
def read_training_data_w_end(filepath):
    results = []
    with open(filepath, "r", encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            if len(line.strip().rsplit(" ", 1)) == 2:
                token, label = line.strip().rsplit(" ", 1)
                results.append((token, labels[label]))
            else:
                results.append(('', END))
                results.append(('', START))
    return results<jupyter_output><empty_output><jupyter_text>Part 1<jupyter_code>def calculate_number_of_labels(input_data):
    return Counter(elem[1] for elem in input_data)

def get_all_unique_tokens(input_data):
    # Might want to somehow ensure that this order stays consistent between runs
    return list(set(item[0] for item in input_data))

# For the return value, we follow the matrix format defined in the slides accordingly
def calculate_emission_parameters(input_data, all_unique_tokens, k=1.0):
    # Final index is for #UNK# tokens
    emission_counts = np.zeros((N, len(all_unique_tokens) + 1), dtype=np.longdouble)

    label_counts = np.array(list(val[1] for val in sorted(calculate_number_of_labels(input_data).items())))

    for token, label in input_data:
        emission_counts[label - 1][all_unique_tokens.index(token)] += 1

    # This is for the other case of #UNK# tokens
    emission_counts[:, -1] = np.full((1, N), k)[0]

    emission_parameters = np.empty((N, len(all_unique_tokens) + 1), dtype=np.longdouble)

    for index, _ in enumerate(emission_counts):
        emission_parameters[index] = emission_counts[index] / (label_counts[index] + k)

    # Do some assertion checks
    for row in emission_parameters:
        assert np.absolute(1.0 - np.sum(row)) <= np.finfo(np.longdouble).eps

    return emission_parameters
# Get tag from word
def get_label_from_token(input_word, all_unique_tokens, emission_parameters):
    if input_word not in all_unique_tokens:
        column_to_consider = emission_parameters[:, -1]
    else:
        column_to_consider = emission_parameters[:, all_unique_tokens.index(input_word)]

    # Randomly choose the index if there is more than one argmax value
    x = rng.choice(np.argwhere(np.isclose(column_to_consider, column_to_consider.max())).flatten()) + 1
    return labels_list[x]

def write_prediction_output_to_file(language):
    if language == "ES":
        # Conduct training/supervised learning (M-Step)
        train_data = read_training_data(es_train_path)
        all_unique_tokens = get_all_unique_tokens(train_data)
        emission_parameters = calculate_emission_parameters(train_data, all_unique_tokens)

        # Execute testing/decoding (E-Step)
        predicted_results = []
        test_data = read_dev_in_data(es_dev_in_path)
        for token in test_data:
            if token:
                predicted_results.append(token + " " + get_label_from_token(token, all_unique_tokens, emission_parameters))
            else:
                predicted_results.append("")
        with open(es_dev_p1_out_path, "w+", encoding="utf-8") as file:
            for line in predicted_results:
                file.write(line + "\n")

    elif language == "RU":
        # Conduct training/supervised learning (M-Step)
        train_data = read_training_data(ru_train_path)
        all_unique_tokens = get_all_unique_tokens(train_data)
        emission_parameters = calculate_emission_parameters(train_data, all_unique_tokens)

        # Execute testing/decoding (E-Step)
        predicted_results = []
        test_data = read_dev_in_data(ru_dev_in_path)
        for token in test_data:
            if token:
                predicted_results.append(token + " " + get_label_from_token(token, all_unique_tokens, emission_parameters))
            else:
                predicted_results.append("")
        with open(ru_dev_p1_out_path, "w+", encoding="utf-8") as file:
            for line in predicted_results:
                file.write(line + "\n")
for language in ["ES", "RU"]:
    write_prediction_output_to_file(language)<jupyter_output><empty_output><jupyter_text>Part 2<jupyter_code>def get_emission_from_token(input_word, all_unique_tokens, emission_parameters):
    if input_word not in all_unique_tokens:
        b = emission_parameters[:, -1]

    else:
        b = emission_parameters[:, all_unique_tokens.index(input_word)]

    return np.reshape(np.pad(b, 1), (-1, 1))

def get_parents(res):
    # Cannot use np.isclose because the scores are really small
    return [rng.choice(np.argwhere(res[1:-1, i] == res[1:-1, i].max(axis=0)).flatten()) + 1 for i in range(res.shape[1])]

# For the return value, we follow the matrix format defined in the slides accordingly
# However, for the convenience of computation later, both END and START appear in both u and v
def calculate_transition_parameters(input_data):
    transition_counts = np.zeros([N+2, N+2])
    priori_counts = np.zeros([N+2, 1])
    prev = START
    for pair in input_data:
        curr = pair[1]
        priori_counts[prev] += 1
        transition_counts[prev, curr] += 1
        prev = curr
    transition = transition_counts / priori_counts
    transition[-1, 0] = 0  # Ignore transition from END to START
    # print(transition)
    return transition

def viterbi_log(transition, emission, all_unique_tokens, input_data):
    # Initial step
    log_transition = np.log(transition)
    pi = [np.log(np.zeros([transition.shape[0], 1]))]
    pi[0][0] = 0
    parents = []
    j = 0
    predicted_results = []
    
    for x in input_data:
        if x != '':  # Propagate with Viterbi
            b = np.log(get_emission_from_token(x, all_unique_tokens, emission))
            res = np.matmul(pi[j], np.transpose(np.ones(b.shape))) + np.matmul(np.ones(b.shape), np.transpose(b)) + log_transition
            # print(res.max(axis=0))
            # print(np.matmul(pi[j], np.transpose(np.zeros(b.shape))))
            pi.append(np.reshape(res.max(axis=0), (-1, 1)))
            parents.append(get_parents(res))
            # parents.append(np.argmax(res, axis = 0))
            j += 1
        else:  # Final step
            # print("new sentence")
            res = pi[j] + log_transition[:,-1:]
            output = get_parents(res)
            # debug = [pi[j][output[0]]]

            # Output, trace back for the sequence
            while j > 1:  # Trace until second (first is START)
                j -= 1
                # debug.insert(0, pi[j][output[0]])
                output.insert(0, parents[j][output[0]])
            for i in output:
                predicted_results.append(labels_list[i])
            # for i in range(len(output)):
            #     predicted_results.append(labels_list[output[i]] + ' ' + str(debug[i]))
            predicted_results.append('')

            # Reset
            pi = [np.log(np.zeros([transition.shape[0], 1]))]
            pi[0][0] = 0
            parents = []
            j = 0
            # break

    return predicted_results

def write_viterbi_output_to_file(language):
    if language == "ES":
        train_data = read_training_data(es_train_path)
        train_data_w_end = read_training_data_w_end(es_train_path)
        test_data = read_dev_in_data(es_dev_in_path)
        output_path = es_dev_p2_out_path

    elif language == "RU":
        train_data = read_training_data(ru_train_path)
        train_data_w_end = read_training_data_w_end(ru_train_path)
        test_data = read_dev_in_data(ru_dev_in_path)
        output_path = ru_dev_p2_out_path

    # Conduct training/supervised learning (M-Step)
    all_unique_tokens = get_all_unique_tokens(train_data)
    emission_parameters = calculate_emission_parameters(train_data, all_unique_tokens)
    transition_parameters = calculate_transition_parameters(train_data_w_end)

    # Execute testing/decoding with Viterbi Algorithm (E-Step)
    predicted_results = viterbi_log(transition_parameters, emission_parameters, all_unique_tokens, test_data)
    with open(output_path, "w+", encoding="utf-8") as file:
        for i in range(len(test_data)):
            if test_data[i] and predicted_results[i]:
                file.write("{} {}\n".format(test_data[i], predicted_results[i]))
            else:
                file.write("\n")

for language in ["ES", "RU"]:
    write_viterbi_output_to_file(language)<jupyter_output><empty_output><jupyter_text>Part 3<jupyter_code>def viterbi_best_five_log(transition, emission, all_unique_tokens, input_data):
    # Initial step
    log_transition = np.log(transition)
    pi = [np.log(np.zeros([transition.shape[0], 5]))]
    for i in range(5):
        pi[0][0, i] = 0
    parents = []
    j = 0
    predicted_results = [[],[],[],[],[]]
    for x in input_data:
        if x != '':  # Propagate with Viterbi
            b = np.log(get_emission_from_token(x, all_unique_tokens, emission))

            # Get top 5
            res_top=[]
            for i in range(5):
                res = np.matmul(pi[j][:,i][np.newaxis].transpose(), np.transpose(np.ones(b.shape))) + np.matmul(np.ones(b.shape), np.transpose(b)) + log_transition
                res_top.append(res)
            
            res_final = np.concatenate((res_top[0], res_top[1], res_top[2], res_top[3], res_top[4]), axis=0)
            # START
            if j==0:
                res_index_sorted=(-res_final).argsort(axis=0, kind="stable")

                # Sort the res matrix 
                res_final=np.take_along_axis(res_final, res_index_sorted, axis=0)
                pi.append(res_final[0:5].transpose())
                # Change to top 5
                parent_candidates=res_index_sorted[0:5]%transition.shape[0]
                parents.append(parent_candidates)
            else:
                start_indices=list(range(0, res_final.shape[0], transition.shape[0]))
                end_indices=[x+transition.shape[0]-1 for x in start_indices]
                res_final=np.delete(res_final, start_indices + end_indices , axis=0)
                res_index_sorted=(-res_final).argsort(axis=0, kind="stable")

                # Sort the res matrix 
                res_final=np.take_along_axis(res_final, res_index_sorted, axis=0)
                pi.append(res_final[0:5].transpose())
                # Change to top 5
                parent_candidates=res_index_sorted[0:5]%(transition.shape[0]-2)+1
                parents.append(parent_candidates)
            j += 1
        else:  # Final step

            # Get top 5
            res_top=[]
            for i in range(5):
                res = pi[j][:,i][np.newaxis].transpose() + log_transition[:,-1:]
                res_top.append(res)

            res_final = np.concatenate((res_top[0],res_top[1],res_top[2],res_top[3],res_top[4]),axis=0)

            start_indices=list(range(0, res_final.shape[0], transition.shape[0]))
            end_indices=[x+transition.shape[0]-1 for x in start_indices]
            res_final=np.delete(res_final, start_indices + end_indices , axis=0)
            res_index_sorted=(-res_final).argsort(axis=0, kind="stable")

            # Sort the res matrix 
            res_final=np.take_along_axis(res_final, res_index_sorted, axis=0)
            pi.append(res_final[0:5].transpose())
            # print(pi)
            # Change to top 5
            parent_candidates=res_index_sorted[0:5]%(transition.shape[0]-2)+1

            output = [parent_candidates]
            while j > 1:  # Trace until second (first is START)
                j -= 1
                all_predecessor_candidates=[]
                all_predecessor_scores=[]

                for index, output_candidate in enumerate(output[0]):
                    # Get top 5
                    res_top=[]
                    for i in range(5):
                        # print("j",j,len(pi))
                        res = pi[j][:,i][np.newaxis].transpose() + log_transition
                        res_top.append(res)
                    res_final = np.concatenate((res_top[0], res_top[1], res_top[2], res_top[3], res_top[4]), axis=0)

                    predecessor_candidates=parents[j][:,output_candidate]
                    predecessor_scores=[]

                    candidate_count=np.zeros([transition.shape[0]])
                    for cand in predecessor_candidates:
                        candidate_id=candidate_count[cand]
                        predecessor_scores.append(res_final[int(candidate_id*transition.shape[1]+cand)])
                        candidate_count[cand]+=1
                    all_predecessor_candidates.append(np.array(predecessor_candidates))
                    all_predecessor_scores.append(np.array(predecessor_scores))
                predec_score_array=np.concatenate((all_predecessor_scores[0],all_predecessor_scores[1],all_predecessor_scores[2],all_predecessor_scores[3],all_predecessor_scores[4]),axis=0)

                predec_cand_array=np.concatenate((all_predecessor_candidates[0],all_predecessor_candidates[1],all_predecessor_candidates[2],all_predecessor_candidates[3],all_predecessor_candidates[4]),axis=0)
                sorting_array=np.argsort(predec_score_array,kind="stable")

                prefec_cand_array = np.take_along_axis(predec_cand_array, sorting_array, axis=0)

                predec_cand_array=predec_cand_array[0:5]

                output.insert(0,predec_cand_array)

            for best_order in range(5):
                for i in range(len(output)):
                    predicted_results[best_order].append(labels_list[int(output[i][best_order]) % transition.shape[1]])

                predicted_results[best_order].append('')

            # Reset
            pi = [np.zeros([transition.shape[0], 5])]
            for i in range(5):
                pi[0][0,i] = 1
            parents = []
            j = 0
            # break

    return predicted_results

def write_viterbi_output_to_file_best_five(language):
    if language == "ES":
        train_data = read_training_data(es_train_path)
        train_data_w_end = read_training_data_w_end(es_train_path)
        test_data = read_dev_in_data(es_dev_in_path)
        output_path = es_dev_p3_out_path

    elif language == "RU":
        train_data = read_training_data(ru_train_path)
        train_data_w_end = read_training_data_w_end(ru_train_path)
        test_data = read_dev_in_data(ru_dev_in_path)
        output_path = ru_dev_p3_out_path

    # Conduct training/supervised learning (M-Step)
    all_unique_tokens = get_all_unique_tokens(train_data)
    emission_parameters = calculate_emission_parameters(train_data, all_unique_tokens)
    transition_parameters = calculate_transition_parameters(train_data_w_end)

    # Execute testing/decoding with Viterbi Algorithm (E-Step)
    predicted_results = viterbi_best_five_log(transition_parameters, emission_parameters, all_unique_tokens, test_data)
    with open(output_path, "w+", encoding="utf-8") as file:
        for i in range(len(test_data)):
            if i < len(test_data) and i < len(predicted_results[0]) and test_data[i] and predicted_results[0][i]:
                file.write("{} {}\n".format(test_data[i], predicted_results[4][i]))
            else:
                file.write("\n")
for language in ["ES", "RU"]:
    write_viterbi_output_to_file_best_five(language)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11933390082996383, 'fraction_numerical': 0.014843583741221536, 'mean_word_length': 2.7347506457381283, 'pattern_counts': {'":': 15, '<': 23, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '376784', 'license': 'MIT', 'n_tokens_mistral': 6272, 'n_tokens_neox': 5963, 'n_words': 1191, 'path': 'parts_1_to_3.ipynb', 'repo_name': 'jamestiotio/humu_humu', 'size': '18822'}