Parts-of-Speech: Viterbi Algorithm

# python
from unittest.mock import call, MagicMock

import math

# pypi
from dotenv import load_dotenv
from expects import (equal, expect)

import numpy
np = numpy

# this stuff
from neurotic.nlp.parts_of_speech import DataLoader, Matrices, TheTrainer

# my other stuff
from graeae import Timer

def initialize(states, tag_counts, A, B, corpus, vocab):
    """Initializes the ``best_probs`` and ``best_paths`` matrices

    Args: 
       states: a list of all possible parts-of-speech
       tag_counts: a dictionary mapping each tag to its respective count
       A: Transition Matrix of dimension (num_tags, num_tags)
       B: Emission Matrix of dimension (num_tags, len(vocab))
       corpus: a sequence of words whose POS is to be identified in a list 
       vocab: a dictionary where keys are words in vocabulary and value is an index

    Returns:
       best_probs: matrix of dimension (num_tags, len(corpus)) of floats
       best_paths: matrix of dimension (num_tags, len(corpus)) of integers
    """
    # Get the total number of unique POS tags
    num_tags = len(tag_counts)

    # Initialize best_probs matrix 
    # POS tags in the rows, number of words in the corpus as the columns
    best_probs = np.zeros((num_tags, len(corpus)))

    # Initialize best_paths matrix
    # POS tags in the rows, number of words in the corpus as columns
    best_paths = np.zeros((num_tags, len(corpus)), dtype=int)

    # Define the start token
    s_idx = states.index("--s--")
    ### START CODE HERE (Replace instances of 'None' with your code) ###

    # Go through each of the POS tags
    for i in range(len(states)): # complete this line
        # Handle the special case when the transition from start token to POS tag i is zero
        if A[s_idx, i] == 0: # complete this line

            # Initialize best_probs at POS tag 'i', column 0, to negative infinity
            best_probs[i,0] = float("-inf")
            print(f"{i}: negitive infinity")

        # For all other cases when transition from start token to POS tag i is non-zero:
        else:
            # Initialize best_probs at POS tag 'i', column 0
            # Check the formula in the instructions above
            best_probs[i,0] = math.log(A[s_idx, i]) + math.log(B[i, vocab[corpus[0]]])
    ### END CODE HERE ### 
    return best_probs, best_paths

states = matrices.tags
tag_counts = trainer.tag_counts
A = matrices.transition
B = matrices.emission
prep = loader.test_words
vocab = loader.vocabulary
best_probs, best_paths = initialize(states, tag_counts, A, B, prep, vocab)

actual = best_probs[0,0]
expected = -22.6098
print(f"best_probs[0,0]: {actual:.4f}")

assert math.isclose(actual, expected, abs_tol=1e-4), (actual, expected)

actual = best_paths[2,3]
expected = 0.0000
print(f"best_paths[2,3]: {actual:.4f}")
assert math.isclose(actual, expected)

def viterbi_forward(A, B, test_corpus, best_probs, best_paths, vocab):
    """The forward training pass

    Args: 
       A, B: The transition and emission matrices respectively
       test_corpus: a list containing a preprocessed corpus
       best_probs: an initilized matrix of dimension (num_tags, len(corpus))
       best_paths: an initilized matrix of dimension (num_tags, len(corpus))
       vocab: a dictionary where keys are words in vocabulary and value is an index 
    Returns: 
       best_probs: a completed matrix of dimension (num_tags, len(corpus))
       best_paths: a completed matrix of dimension (num_tags, len(corpus))
    """
    # Get the number of unique POS tags (which is the num of rows in best_probs)
    num_tags = best_probs.shape[0]

    # Go through every word in the corpus starting from word 1
    # Recall that word 0 was initialized in `initialize()`
    for i in range(1, len(test_corpus)): 

        # Print number of words processed, every 5000 words
        if i % 5000 == 0:
            print("Words processed: {:>8}".format(i))

        ### START CODE HERE (Replace instances of 'None' with your code EXCEPT the first 'best_path_i = None') ###
        # For each unique POS tag that the current word can be
        for j in range(num_tags): # complete this line

            # Initialize best_prob for word i to negative infinity
            best_prob_i = float("-inf")

            # Initialize best_path for current word i to None
            best_path_i = None

            # For each POS tag that the previous word can be:
            for k in range(num_tags): # complete this line

                # Calculate the probability = 
                # best probs of POS tag k, previous word i-1 + 
                # log(prob of transition from POS k to POS j) + 
                # log(prob that emission of POS j is word i)
                prob = best_probs[k, i-1] + math.log(A[k, j]) + math.log(B[j, vocab[test_corpus[i]]])

                # check if this path's probability is greater than
                # the best probability up to and before this point
                if prob > best_prob_i: # complete this line

                    # Keep track of the best probability
                    best_prob_i = prob

                    # keep track of the POS tag of the previous word
                    # that is part of the best path.  
                    # Save the index (integer) associated with 
                    # that previous word's POS tag
                    best_path_i = k

            # Save the best probability for the 
            # given current word's POS tag
            # and the position of the current word inside the corpus
            best_probs[j,i] = best_prob_i

            # Save the unique integer ID of the previous POS tag
            # into best_paths matrix, for the POS tag of the current word
            # and the position of the current word inside the corpus.
            best_paths[j,i] = best_path_i

        ### END CODE HERE ###
    return best_probs, best_paths

with TIMER:
    best_probs, best_paths = viterbi_forward(A, B,
                                             prep,
                                             best_probs,
                                             best_paths,
                                             vocab)

expected = -24.7822
actual = best_probs[0,1]
print(f"best_probs[0,1]: {actual:.4f}")
assert math.isclose(expected, actual, abs_tol=1e-4)

actual = best_probs[0,4]
expected = -49.5601
print(f"best_probs[0,4]: {actual:.4f}")
assert math.isclose(actual, expected, abs_tol=1e-4)

def viterbi_backward(best_probs: numpy.ndarray,
                     best_paths: numpy.ndarray,
                     corpus: list,
                     states: dict) -> list:
    """
    This function returns the best path.
    """
    # Get the number of words in the corpus
    # which is also the number of columns in best_probs, best_paths
    m = best_paths.shape[1] 

    # Initialize array z, same length as the corpus
    z = [None] * m

    # Get the number of unique POS tags
    num_tags = best_probs.shape[0]

    # Initialize the best probability for the last word
    best_prob_for_last_word = float('-inf')

    # Initialize pred array, same length as corpus
    pred = [None] * m

    ### START CODE HERE (Replace instances of 'None' with your code) ###
    ## Step 1 ##

    # Go through each POS tag for the last word (last column of best_probs)
    # in order to find the row (POS tag integer ID) 
    # with highest probability for the last word
    for k in range(num_tags): # complete this line

        # If the probability of POS tag at row k 
        # is better than the previously best probability for the last word:
        if best_probs[k, -1] > best_prob_for_last_word: # complete this line

            # Store the new best probability for the lsat word
            best_prob_for_last_word = best_probs[k, -1]

            # Store the unique integer ID of the POS tag
            # which is also the row number in best_probs
            z[m - 1] = k

    # Convert the last word's predicted POS tag
    # from its unique integer ID into the string representation
    # using the 'states' dictionary
    # store this in the 'pred' array for the last word
    pred[m - 1] = states[z[m - 1]]

    ## Step 2 ##
    # Find the best POS tags by walking backward through the best_paths
    # From the last word in the corpus to the 0th word in the corpus
    for i in range(m - 1, 0, -1): # complete this line

        # Retrieve the unique integer ID of
        # the POS tag for the word at position 'i' in the corpus
        pos_tag_for_word_i = z[i]

        # In best_paths, go to the row representing the POS tag of word i
        # and the column representing the word's position in the corpus
        # to retrieve the predicted POS for the word at position i-1 in the corpus
        z[i - 1] = best_paths[pos_tag_for_word_i, i]

        # Get the previous word's POS tag in string form
        # Use the 'states' dictionary, 
        # where the key is the unique integer ID of the POS tag,
        # and the value is the string representation of that POS tag
        pred[i - 1] = states[z[i - 1]]

     ### END CODE HERE ###
    return pred

states = matrices.tags
pred = viterbi_backward(best_probs, best_paths, corpus=prep, states=states)
m=len(pred)
actual_prep = prep[-7:m-1]
actual_pred = pred[-7:m-1]

expected_prep =  ['see', 'them', 'here', 'with', 'us', '.']  
print('The prediction for pred[-7:m-1] is: \n', actual_prep, "\n", actual_pred, "\n")
print('The prediction for pred[0:7] is: \n', pred[0:7], "\n", prep[0:7])

# python
from collections import Counter

import math

# pypi
import attr
import numpy

# this project
from .preprocessing import DataLoader, Empty
from .training import TheTrainer
from .matrices import Matrices

class AlgorithmError(Exception):
    """Called when the methods are called out of order"""

@attr.s(auto_attribs=True)
class HiddenMarkov:
    """A Hidden Markov Model Class

    Args:
     loader: a DataLoader
     trainer: A TheTrainer object
     matrices: A Matrices object
    """
    loader: DataLoader
    trainer: TheTrainer
    matrices: Matrices
    best_probabilities: numpy.ndarray=None
    best_paths: numpy.ndarray=None
    _states: list=None
    _tag_counts: Counter=None
    _tag_count: int=None
    _transition_matrix: numpy.ndarray=None
    _emission_matrix: numpy.ndarray=None
    _test_words: list=None
    _test_word_count: int=None
    _vocabulary: dict=None
    _start_token_index: int=None
    _negative_infinity: float = None

from neurotic.nlp.parts_of_speech import HiddenMarkov

model = HiddenMarkov(loader, trainer, matrices)
assert all(original == other for original, other in zip(states, model.states))
assert all(value == model.tag_counts[key] for key, value in tag_counts.items())
assert numpy.array_equal(A, model.transition_matrix)
assert numpy.array_equal(B, model.emission_matrix)
assert all(original == new for original, new in zip(prep, model.test_words))
assert all(value == model.vocabulary[key] for key, value in vocab.items())

model.initialize_matrices()

assert model.best_probabilities.shape == best_probs.shape
assert model.best_paths.shape == best_paths.shape

actual = model.best_probabilities[0,0]
expected = -22.6098
assert math.isclose(actual, expected, abs_tol=1e-4), (actual, expected)

actual = model.best_paths[2,3]
expected = 0.0000
assert math.isclose(actual, expected)

model.viterbi_forward()
expected = -24.7822
actual = model.best_probabilities[0,1]
assert math.isclose(expected, actual, abs_tol=1e-4)

actual = model.best_probabilities[0,4]
expected = -49.5601
assert math.isclose(actual, expected, abs_tol=1e-4)

model.viterbi_backward()

actual = pred[0:7]
expected = ['DT', 'NN', 'POS', 'NN', 'MD', 'VB', 'VBN']
assert all((a==e for a,e in zip(actual, expected)))