<s> import numpy as np <EOL> import sequence_classification_decoder as scd <EOL> import pdb <EOL> class SequenceClassifier ( ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , observation_labels , state_labels ) : <EOL> '''<STR_LIT>''' <EOL> self . decoder = scd . SequenceClassificationDecoder ( ) <EOL> self . observation_labels = observation_labels <EOL> self . state_labels = state_labels <EOL> self . trained = False <EOL> def get_num_states ( self ) : <EOL> '''<STR_LIT>''' <EOL> return len ( self . state_labels ) <EOL> def get_num_observations ( self ) : <EOL> '''<STR_LIT>''' <EOL> return len ( self . observation_labels ) <EOL> def train_supervised ( self , sequence_list ) : <EOL> '''<STR_LIT>''' <EOL> raise NotImplementedError <EOL> def compute_scores ( self , sequence ) : <EOL> '''<STR_LIT>''' <EOL> raise NotImplementedError <EOL> def compute_output_score ( self , states , initial_scores , transition_scores , <EOL> final_scores , emission_scores ) : <EOL> length = np . size ( emission_scores , <NUM_LIT:0> ) <EOL> score = <NUM_LIT:0.0> <EOL> score += initial_scores [ states [ <NUM_LIT:0> ] ] <EOL> for pos in xrange ( length ) : <EOL> score += emission_scores [ pos , states [ pos ] ] <EOL> if pos > <NUM_LIT:0> : <EOL> score += transition_scores [ pos - <NUM_LIT:1> , states [ pos ] , states [ pos - <NUM_LIT:1> ] ] <EOL> score += final_scores [ states [ length - <NUM_LIT:1> ] ] <EOL> return score <EOL> def compute_posteriors ( self , initial_scores , transition_scores , <EOL> final_scores , emission_scores ) : <EOL> '''<STR_LIT>''' <EOL> length = np . size ( emission_scores , <NUM_LIT:0> ) <EOL> num_states = np . size ( emission_scores , <NUM_LIT:1> ) <EOL> log_likelihood , forward = self . decoder . run_forward ( initial_scores , <EOL> transition_scores , <EOL> final_scores , <EOL> emission_scores ) <EOL> log_likelihood , backward = self . decoder . run_backward ( initial_scores , <EOL> transition_scores , <EOL> final_scores , <EOL> emission_scores ) <EOL> state_posteriors = np . zeros ( [ length , num_states ] ) <EOL> for pos in xrange ( length ) : <EOL> state_posteriors [ pos , : ] = forward [ pos , : ] + backward [ pos , : ] <EOL> state_posteriors [ pos , : ] -= log_likelihood <EOL> transition_posteriors = np . zeros ( [ length - <NUM_LIT:1> , num_states , num_states ] ) <EOL> for pos in xrange ( length - <NUM_LIT:1> ) : <EOL> for prev_state in xrange ( num_states ) : <EOL> for state in xrange ( num_states ) : <EOL> transition_posteriors [ pos , state , prev_state ] = forward [ pos , prev_state ] + transition_scores [ pos , state , prev_state ] + emission_scores [ pos + <NUM_LIT:1> , state ] + backward [ pos + <NUM_LIT:1> , state ] <EOL> transition_posteriors [ pos , state , prev_state ] -= log_likelihood <EOL> state_posteriors = np . exp ( state_posteriors ) <EOL> transition_posteriors = np . exp ( transition_posteriors ) <EOL> return state_posteriors , transition_posteriors , log_likelihood <EOL> def posterior_decode ( self , sequence ) : <EOL> '''<STR_LIT>''' <EOL> initial_scores , transition_scores , final_scores , emission_scores = self . compute_scores ( sequence ) <EOL> state_posteriors , _ , _ = self . compute_posteriors ( initial_scores , <EOL> transition_scores , <EOL> final_scores , <EOL> emission_scores ) <EOL> best_states = np . argmax ( state_posteriors , axis = <NUM_LIT:1> ) <EOL> predicted_sequence = sequence . copy_sequence ( ) <EOL> predicted_sequence . y = best_states <EOL> return predicted_sequence <EOL> def posterior_decode_corpus ( self , dataset ) : <EOL> '''<STR_LIT>''' <EOL> predictions = [ ] <EOL> for sequence in dataset . seq_list : <EOL> predictions . append ( self . posterior_decode ( sequence ) ) <EOL> return predictions <EOL> def viterbi_decode ( self , sequence ) : <EOL> '''<STR_LIT>''' <EOL> initial_scores , transition_scores , final_scores , emission_scores = self . compute_scores ( sequence ) <EOL> best_states , total_score = self . decoder . run_viterbi ( initial_scores , <EOL> transition_scores , <EOL> final_scores , <EOL> emission_scores ) <EOL> predicted_sequence = sequence . copy_sequence ( ) <EOL> predicted_sequence . y = best_states <EOL> return predicted_sequence , total_score <EOL> def viterbi_decode_corpus ( self , dataset ) : <EOL> '''<STR_LIT>''' <EOL> predictions = [ ] <EOL> for sequence in dataset . seq_list : <EOL> predicted_sequence , _ = self . viterbi_decode ( sequence ) <EOL> predictions . append ( predicted_sequence ) <EOL> return predictions <EOL> def evaluate_corpus ( self , dataset , predictions ) : <EOL> '''<STR_LIT>''' <EOL> total = <NUM_LIT:0.0> <EOL> correct = <NUM_LIT:0.0> <EOL> for i , sequence in enumerate ( dataset . seq_list ) : <EOL> pred = predictions [ i ] <EOL> for j , y_hat in enumerate ( pred . y ) : <EOL> if ( sequence . y [ j ] == y_hat ) : <EOL> correct += <NUM_LIT:1> <EOL> total += <NUM_LIT:1> <EOL> return correct / total </s>
