Parts-of-Speech Tagging: Hidden Markov Model

load_dotenv("posts/nlp/.env")
loader = DataLoader()
trainer = TheTrainer(loader.processed_training)

SLUG = "parts-of-speech-tagging-hidden-markov-model"
FOLDER = f"files/posts/nlp/{SLUG}/" 
Embed = partial(EmbedHoloviews, folder_path=FOLDER)

Plot = Namespace(
    width=990,
    height=780,
    fontscale=2,
    tan="#ddb377",
    blue="#4687b7",
    red="#ce7b6d",
    color_map="OrRd",
    path_color_map="blues",
)

def create_transition_matrix(alpha: float, tag_counts: dict,
                             transition_counts: dict) -> numpy.ndarray:
    """Transition Matrix for the Hidden Markov Model

    Args: 
      ``alpha``: number used for smoothing
      ``tag_counts``: a dictionary mapping each tag to its respective count
      ``transition_counts``: transition count for the previous word and tag

    Returns:
      ``A``: matrix of dimension (``num_tags``,``num_tags``)
    """
    # Get a sorted list of unique POS tags
    all_tags = sorted(tag_counts.keys())

    # Count the number of unique POS tags
    num_tags = len(all_tags)

    # Initialize the transition matrix 'A'
    A = numpy.zeros((num_tags,num_tags))

    # Get the unique transition tuples (previous POS, current POS)
    trans_keys = set(transition_counts.keys())

    # Go through each row of the transition matrix A
    for i in range(num_tags):

        # Go through each column of the transition matrix A
        for j in range(num_tags):

            # Initialize the count of the (prev POS, current POS) to zero
            count = 0

            # Define the tuple (prev POS, current POS)
            # Get the tag at position i and tag at position j (from the all_tags list)
            key = (all_tags[i], all_tags[j])

            # Check if the (prev POS, current POS) tuple 
            # exists in the transition counts dictionary
            if key in transition_counts: #complete this line

                # Get count from the transition_counts dictionary 
                # for the (prev POS, current POS) tuple
                count = transition_counts[key]

            # Get the count of the previous tag (index position i) from tag_counts
            count_prev_tag = tag_counts[all_tags[i]]

            # Apply smoothing using count of the tuple, alpha, 
            # count of previous tag, alpha, and total number of tags
            A[i,j] = (count + alpha)/(count_prev_tag + alpha * num_tags)
    return A

# setup some values
alpha = 0.001
states = sorted(trainer.tag_counts.keys())

A = create_transition_matrix(alpha,
                             trainer.tag_counts,
                             trainer.transition_counts)
# Testing your function
expected = 0.000007040
actual = A[0,0]

print(f"A at row 0, col 0: {actual:.9f}")
assert math.isclose(expected, actual, abs_tol=1e-6), (expected, actual)

expected = 0.1691
actual = A[3,1]
print(f"A at row 3, col 1: {actual:.4f}")
assert math.isclose(expected, actual, abs_tol=1e-4)

print("View a subset of transition matrix A")
actual = A[30:35,30:35]
A_sub = pandas.DataFrame(actual, index=states[30:35], columns = states[30:35] )
print(A_sub)

expected = numpy.array([
 [2.217069e-06, 2.217069e-06, 2.217069e-06, 0.008870, 2.217069e-06],
 [3.756509e-07, 7.516775e-04, 3.756509e-07, 0.051089, 3.756509e-07],
 [1.722772e-05, 1.722772e-05, 1.722772e-05, 0.000017, 1.722772e-05],
 [4.477336e-05, 4.472863e-08, 4.472863e-08, 0.000090, 4.477336e-05],
 [1.030439e-05, 1.030439e-05, 1.030439e-05, 0.061837, 3.092348e-02],
])

assert numpy.allclose(expected, actual, atol=1e-5)

plotter = pandas.DataFrame(A, index=states, columns=states)[::-1]
plot = plotter.hvplot.heatmap(cmap=Plot.color_map).opts(
    title="Emission Matrix A",
    width=Plot.width, height=Plot.height, fontscale=Plot.fontscale,
    xrotation=90,
)
#plot *= holoviews.Labels(plot)
outcome = Embed(plot=plot, file_name="emission_matrix_a")()

print(outcome)

URL = "https://www.ling.upenn.edu/courses/Fall_2003/ling001/penn_treebank_pos.html"
data = pandas.read_html(URL, header=0)[0]

TRANSLATOR = {row.Tag:row.Description for row in data.itertuples()}

print("|Tag| Description|")
print("|-+-|")
for tag in states:
    print(f"|{tag}|{TRANSLATOR.get(tag, 'unknown')}|")

def create_emission_matrix(alpha: float,
                           tag_counts: dict,
                           emission_counts: dict,
                           vocab: dict) -> numpy.ndarray:
    """Create Matrix B

    Args: 
       ``alpha``: tuning parameter used in smoothing 
       ``tag_counts``: a dictionary mapping each tag to its respective count
       ``emission_counts``: a dictionary where the keys are (tag, word) and the values are the counts
       ``vocab``: a dictionary where keys are words in vocabulary and value is an index.
              within the function it'll be treated as a list
    Returns:
       ``B``: a matrix of dimension ``(num_tags, len(vocab))``
    """

    # get the number of POS tag
    num_tags = len(tag_counts)

    # Get a list of all POS tags
    all_tags = sorted(tag_counts.keys())

    # Get the total number of unique words in the vocabulary
    num_words = len(vocab)

    # Initialize the emission matrix B with places for
    # tags in the rows and words in the columns
    B = numpy.zeros((num_tags, num_words))

    # Get a set of all (POS, word) tuples 
    # from the keys of the emission_counts dictionary
    emis_keys = set(list(emission_counts.keys()))

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

    # Go through each row (POS tags)
    for i in range(num_tags): # complete this line

        # Go through each column (words)
        for j in range(num_words): # complete this line

            # Initialize the emission count for the (POS tag, word) to zero
            count = 0

            # Define the (POS tag, word) tuple for this row and column
            key =  (all_tags[i], vocab[j])

            # check if the (POS tag, word) tuple exists as a key in emission counts
            if key in emission_counts: # complete this line

                # Get the count of (POS tag, word) from the emission_counts d
                count = emission_counts[key]

            # Get the count of the POS tag
            count_tag = tag_counts[all_tags[i]]

            # Apply smoothing and store the smoothed value 
            # into the emission matrix B for this row and column
            B[i,j] = (count + alpha)/(count_tag + alpha * num_words)

    ### END CODE HERE ###
    return B

# creating your emission probability matrix. this takes a few minutes to run.
vocab = loader.vocabulary
B = create_emission_matrix(alpha,
                           trainer.tag_counts,
                           trainer.emission_counts,
                           list(vocab))

actual = B[0,0]
expected = 0.000006032
print(f"View Matrix position at row 0, column 0: {actual:.9f}")
assert math.isclose(actual, expected, abs_tol=1e-6)

actual = B[3,1]
expected = 0.000000720
print(f"View Matrix position at row 3, column 1: {actual:.9f}")
assert math.isclose(actual, expected, abs_tol=1e-7)

# Try viewing emissions for a few words in a sample dataframe
cidx  = ['725','adroitly','engineers', 'promoted', 'synergy']

# Get the integer ID for each word
cols = [vocab[a] for a in cidx]

# Choose POS tags to show in a sample dataframe
rvals =['CD','NN','NNS', 'VB','RB','RP']

# For each POS tag, get the row number from the 'states' list
rows = [states.index(a) for a in rvals]

# Get the emissions for the sample of words, and the sample of POS tags
actual = B[numpy.ix_(rows,cols)]
B_sub = pandas.DataFrame(actual, index=rvals, columns = cidx )
print(B_sub)
expected = ([
 [8.201296e-05, 2.732854e-08, 2.732854e-08, 2.732854e-08, 2.732854e-08],
 [7.521128e-09, 7.521128e-09, 7.521128e-09, 7.521128e-09, 2.257091e-05],
 [1.670013e-08, 1.670013e-08, 4.676203e-04, 1.670013e-08, 1.670013e-08],
 [3.779036e-08, 3.779036e-08, 3.779036e-08, 3.779036e-08, 3.779036e-08],
 [3.226454e-08, 6.456135e-05, 3.226454e-08, 3.226454e-08, 3.226454e-08],
 [3.723317e-07, 3.723317e-07, 3.723317e-07, 3.723317e-07, 3.723317e-07],
])

assert numpy.allclose(expected, actual, atol=1e-5)

plotter = B_sub[::-1]
plot = plotter.hvplot.heatmap(cmap=Plot.color_map).opts(
    title="Emission Matrix B",
    width=Plot.width, height=Plot.height, fontscale=Plot.fontscale,
    xrotation=90,
)
outcome = Embed(plot=plot, file_name="emission_matrix_b")()

print(outcome)

# pypi
import attr
import numpy

@attr.s(auto_attribs=True)
class Matrices:
    """The matrices for the hidden markov model

    Args:
     ``transition_counts``: dictionary of counts of adjacent POS tags
     ``emission_counts``: dictionary of (word, POS) counts
     ``tag_counts``: dictionary of POS tag-counts
     ``words``: list of words in the vocabulary
     ``alpha``: The smoothing value
    """
    transition_counts: dict
    emission_counts: dict
    tag_counts: dict
    # all the lists have to be sorted for the matrices to match
    words: list=attr.ib(converter=sorted)
    alpha: float=0.001
    _tags: list=None
    _tag_count: int=None
    _word_count: int=None
    _transition: numpy.ndarray=None
    _emission: numpy.ndarray=None

@property
def tags(self) -> list:
    """Sorted list of the POS tags"""
    if self._tags is None:
        self._tags = sorted(self.tag_counts)
    return self._tags

@property
def tag_count(self) -> int:
    """Number of tags"""
    if self._tag_count is None:
        self._tag_count = len(self.tags)
    return self._tag_count

@property
def word_count(self) -> int:
    """Number of words in the vocabulary"""
    if self._word_count is None:
        self._word_count = len(self.words)
    return self._word_count

@property
def transition(self) -> numpy.ndarray:
    """The Transition Matrix"""
    if self._transition is None:
        self._transition = numpy.zeros((self.tag_count, self.tag_count))
        for row in range(self.tag_count):
            for column in range(self.tag_count):
                key = (self.tags[row], self.tags[column])
                count = self.transition_counts[key] if key in self.transition_counts else 0
                previous_tag_count = self.tag_counts[self.tags[row]]
                self._transition[row, column] = (
                    (count + self.alpha)
                    /(previous_tag_count + self.alpha * self.tag_count))
    return self._transition

@property
def emission(self) -> numpy.ndarray:
    """The Emission Matrix"""
    if self._emission is None:
        self._emission = numpy.zeros((self.tag_count, self.word_count))
        for row in range(self.tag_count):
            for column in range(self.word_count):
                key = (self.tags[row], self.words[column])
                emission_count = self.emission_counts[key] if key in self.emission_counts else 0
                tag_count = self.tag_counts[self.tags[row]]
                self._emission[row, column] = (
                    (emission_count + self.alpha)
                    /(tag_count + self.alpha * self.word_count)                
                )
    return self._emission

transition = matrices.transition
expected = 0.000007040
actual = transition_matrix[0,0]

print(f"Transition Matrix at row 0, col 0: {actual:.9f}")
assert math.isclose(expected, actual, abs_tol=1e-6), (expected, actual)

expected = 0.1691
actual = transition_matrix[3,1]
print(f"Transition Matrix at row 3, col 1: {actual:.4f}")
assert math.isclose(expected, actual, abs_tol=1e-4)

print("View a subset of the transition matrix")
actual = transition_matrix[30:35,30:35]
A_sub = pandas.DataFrame(actual, index=states[30:35], columns = states[30:35] )
print(A_sub)

expected = numpy.array([
 [2.217069e-06, 2.217069e-06, 2.217069e-06, 0.008870, 2.217069e-06],
 [3.756509e-07, 7.516775e-04, 3.756509e-07, 0.051089, 3.756509e-07],
 [1.722772e-05, 1.722772e-05, 1.722772e-05, 0.000017, 1.722772e-05],
 [4.477336e-05, 4.472863e-08, 4.472863e-08, 0.000090, 4.477336e-05],
 [1.030439e-05, 1.030439e-05, 1.030439e-05, 0.061837, 3.092348e-02],
])

assert numpy.allclose(expected, actual, atol=1e-5)

matrices = Matrices(transition_counts=trainer.transition_counts,
                    emission_counts=trainer.emission_counts,
                    words=loader.vocabulary_words,
                    tag_counts=trainer.tag_counts,
                    alpha=0.001)
emission = matrices.emission

actual = emission[0,0]
expected = 0.000006032
print(f"Emission Matrix position at row 0, column 0: {actual:.9f}")
assert math.isclose(actual, expected, abs_tol=1e-6)

actual = emission[3,1]
expected = 0.000000720
print(f"Emission Matrix position at row 3, column 1: {actual:.9f}")
assert math.isclose(actual, expected, abs_tol=1e-7)

# Try viewing emissions for a few words in a sample dataframe
columns  = ['725','adroitly','engineers', 'promoted', 'synergy']

# Get the integer ID for each word
column_ids = [loader.vocabulary[column] for column in columns]

# Choose POS tags to show in a sample dataframe
rows =['CD','NN','NNS', 'VB','RB','RP']

# For each POS tag, get the row number from the 'states' list
row_numbers = [matrices.tags.index(a) for a in rows]

# Get the emissions for the sample of words, and the sample of POS tags
actual = emission[numpy.ix_(row_numbers,column_ids)]
B_sub = pandas.DataFrame(actual, index=rows, columns = columns)
print(B_sub)
expected = ([
 [8.201296e-05, 2.732854e-08, 2.732854e-08, 2.732854e-08, 2.732854e-08],
 [7.521128e-09, 7.521128e-09, 7.521128e-09, 7.521128e-09, 2.257091e-05],
 [1.670013e-08, 1.670013e-08, 4.676203e-04, 1.670013e-08, 1.670013e-08],
 [3.779036e-08, 3.779036e-08, 3.779036e-08, 3.779036e-08, 3.779036e-08],
 [3.226454e-08, 6.456135e-05, 3.226454e-08, 3.226454e-08, 3.226454e-08],
 [3.723317e-07, 3.723317e-07, 3.723317e-07, 3.723317e-07, 3.723317e-07],
])

assert numpy.allclose(expected, actual, atol=1e-5)