Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_prefix_matcher_hmm(pattern):
model = Model(name="Prefix Matcher HMM Model")
insert_distribution = DiscreteDistribution({'A': 0.25, 'C': 0.25, 'G': 0.25, 'T': 0.25})
insert_states = []
match_states = []
delete_states = []
hmm_name = 'prefix'
for i in range(len(pattern) + 1):
insert_states.append(State(insert_distribution, name='I%s_%s' % (i, hmm_name)))
for i in range(len(pattern)):
distribution_map = dict({'A': 0.01, 'C': 0.01, 'G': 0.01, 'T': 0.01})
distribution_map[pattern[i]] = 0.97
match_states.append(State(DiscreteDistribution(distribution_map), name='M%s_%s' % (str(i + 1), hmm_name)))
for i in range(len(pattern)):
delete_states.append(State(None, name='D%s_%s' % (str(i + 1), hmm_name)))
unit_start = State(None, name='prefix_start_%s' % hmm_name)
unit_end = State(None, name='prefix_end_%s' % hmm_name)
model.add_states(insert_states + match_states + delete_states + [unit_start, unit_end])
last = len(delete_states)-1
model.add_transition(model.start, unit_start, 1)
model.add_transition(unit_end, model.end, 1)
insert_error = settings.MAX_ERROR_RATE * 2 / 5
delete_error = settings.MAX_ERROR_RATE * 1 / 5
model.add_transition(unit_start, match_states[0], 1 - insert_error - delete_error)
def update_hmm(self):
num_states = self.num_states
start_prob = self.start_prob
num_emissions = self.num_emissions
hmm = HiddenMarkovModel('hmm')
dist = [DiscreteDistribution(dict(zip(range(num_emissions), self.emissions[i]))) for i in range(num_states)]
states = [State(dist[i], 's' + str(i).zfill(2)) for i in range(num_states)]
hmm.add_states(states)
for i in range(num_states):
s_i = states[i]
hmm.add_transition(hmm.start, s_i, start_prob[i])
for j in range(num_states):
s_j = states[j]
p = self.transitions[i, j]
hmm.add_transition(s_i, s_j, p)
self.hmm = hmm
self.hmm.bake()
def test_example_pomegranate(self):
"""
This example is taken from https://pomegranate.readthedocs.io/en/latest/HiddenMarkovModel.html
"""
from pomegranate import DiscreteDistribution, State, HiddenMarkovModel
d1 = DiscreteDistribution({'A': 0.35, 'C': 0.20, 'G': 0.05, 'T': 0.40})
d2 = DiscreteDistribution({'A': 0.25, 'C': 0.25, 'G': 0.25, 'T': 0.25})
d3 = DiscreteDistribution({'A': 0.10, 'C': 0.40, 'G': 0.40, 'T': 0.10})
s1 = State(d1, name="s1")
s2 = State(d2, name="s2")
s3 = State(d3, name="s3")
model = HiddenMarkovModel(name='example')
model.add_states([s1, s2, s3])
model.add_transition(model.start, s1, 0.90)
model.add_transition(model.start, s2, 0.10)
model.add_transition(s1, s1, 0.80)
model.add_transition(s1, s2, 0.20)
model.add_transition(s2, s2, 0.90)
model.add_transition(s2, s3, 0.10)
model.add_transition(s3, s3, 0.70)
model.add_transition(s3, model.end, 0.30)
def test_dense_transition_matrix(self):
"""
This test checks the computation of the dense_transition_matrix method
"""
from pomegranate import DiscreteDistribution as pome_DiscreteDistribution
from pomegranate import State as pome_State
from pomegranate import HiddenMarkovModel as pome_HiddenMarkovModel
d1 = pome_DiscreteDistribution({'A': 0.35, 'C': 0.20, 'G': 0.05, 'T': 0.40})
d2 = pome_DiscreteDistribution({'A': 0.25, 'C': 0.25, 'G': 0.25, 'T': 0.25})
d3 = pome_DiscreteDistribution({'A': 0.10, 'C': 0.40, 'G': 0.40, 'T': 0.10})
s1 = pome_State(d1, name="s1")
s2 = pome_State(d2, name="s2")
s3 = pome_State(d3, name="s3")
pome_model = pome_HiddenMarkovModel(name='example')
pome_model.add_states([s1, s2, s3])
pome_model.add_transition(pome_model.start, s1, 0.90)
pome_model.add_transition(pome_model.start, s2, 0.10)
pome_model.add_transition(s1, s1, 0.80)
pome_model.add_transition(s1, s2, 0.20)
pome_model.add_transition(s2, s2, 0.90)
pome_model.add_transition(s2, s3, 0.10)
pome_model.add_transition(s3, s3, 0.70)
pome_model.add_transition(s3, pome_model.end, 0.30)
model = Model(name="Repeating Pattern Matcher HMM Model")
if vpaths:
alignment = get_multiple_alignment_of_repeats_from_reads(vpaths)
transitions, emissions = build_profile_hmm_pseudocounts_for_alignment(settings.MAX_ERROR_RATE, alignment)
else:
transitions, emissions = build_profile_hmm_for_repeats(patterns, settings.MAX_ERROR_RATE)
matches = [m for m in emissions.keys() if m.startswith('M')]
last_end = None
for repeat in range(copies):
insert_states = []
match_states = []
delete_states = []
for i in range(len(matches) + 1):
insert_distribution = DiscreteDistribution(emissions['I%s' % i])
insert_states.append(State(insert_distribution, name='I%s_%s' % (i, repeat)))
for i in range(1, len(matches) + 1):
match_distribution = DiscreteDistribution(emissions['M%s' % i])
match_states.append(State(match_distribution, name='M%s_%s' % (str(i), repeat)))
for i in range(1, len(matches) + 1):
delete_states.append(State(None, name='D%s_%s' % (str(i), repeat)))
unit_start = State(None, name='unit_start_%s' % repeat)
unit_end = State(None, name='unit_end_%s' % repeat)
model.add_states(insert_states + match_states + delete_states + [unit_start, unit_end])
n = len(delete_states)-1
if repeat > 0:
model.add_transition(last_end, unit_start, 1)
def __init__(self):
A = DiscreteDistribution({'1': 1./3, '2': 1./3, '3': 1./3})
B = ConditionalProbabilityTable(
[['1','1',0.5],
['1','2',0.5],
['1','3',0],
['2','1',0],
['2','2',0.5],
['2','3',0.5],
['3','1',0.5],
['3','2',0],
['3','3',0.5],
],[A])
C = ConditionalProbabilityTable(
[['1','1',0.5],
['1','2',0.5],
['1','3',0],
['2','1',0],
def __init__(self):
Pollution = DiscreteDistribution({'F': 0.9, 'T': 0.1})
Smoker = DiscreteDistribution({'T': 0.3, 'F': 0.7})
print(Smoker)
Cancer = ConditionalProbabilityTable(
[['T','T','T',0.05],
['T','T','F',0.95],
['T','F','T',0.02],
['T','F','F',0.98],
['F','T','T',0.03],
['F','T','F',0.97],
['F','F','T',0.001],
['F','F','F',0.999],
],[Pollution,Smoker])
print(Cancer)
XRay = ConditionalProbabilityTable(
[['T','T',0.9],
['T','F',0.1],
def test_example_pomegranate(self):
"""
This example is taken from https://pomegranate.readthedocs.io/en/latest/HiddenMarkovModel.html
"""
from pomegranate import DiscreteDistribution, State, HiddenMarkovModel
d1 = DiscreteDistribution({'A': 0.35, 'C': 0.20, 'G': 0.05, 'T': 0.40})
d2 = DiscreteDistribution({'A': 0.25, 'C': 0.25, 'G': 0.25, 'T': 0.25})
d3 = DiscreteDistribution({'A': 0.10, 'C': 0.40, 'G': 0.40, 'T': 0.10})
s1 = State(d1, name="s1")
s2 = State(d2, name="s2")
s3 = State(d3, name="s3")
model = HiddenMarkovModel(name='example')
model.add_states([s1, s2, s3])
model.add_transition(model.start, s1, 0.90)
model.add_transition(model.start, s2, 0.10)
model.add_transition(s1, s1, 0.80)
model.add_transition(s1, s2, 0.20)
model.add_transition(s2, s2, 0.90)
model.add_transition(s2, s3, 0.10)
model.add_transition(s3, s3, 0.70)
model.add_transition(s3, model.end, 0.30)
model.bake()
def make_hmm_model(emission_mat, transition_probs):
model = pomegranate.HiddenMarkovModel('ndf')
ictal_emissions = {i:emission_mat[1,i] for i in range(emission_mat.shape[1])}
baseline_emissions = {i:emission_mat[0,i] for i in range(emission_mat.shape[1])}
ictal = pomegranate.State(pomegranate.DiscreteDistribution(ictal_emissions ), name = '1')
baseline = pomegranate.State(pomegranate.DiscreteDistribution(baseline_emissions), name = '0')
model.add_state(ictal)
model.add_state(baseline)
model.add_transition( model.start, ictal, 0.05 )
model.add_transition( model.start, baseline, 99.95)
model.add_transition( baseline, baseline, transition_probs[0,0] )
model.add_transition( baseline, ictal, transition_probs[0,1] )
model.add_transition( ictal, ictal , transition_probs[1,1] )
model.add_transition( ictal, baseline, transition_probs[1,0] )
model.bake(verbose=False )
return model
def get_suffix_matcher_hmm(pattern):
model = Model(name="Suffix Matcher HMM Model")
insert_distribution = DiscreteDistribution({'A': 0.25, 'C': 0.25, 'G': 0.25, 'T': 0.25})
insert_states = []
match_states = []
delete_states = []
hmm_name = 'suffix'
for i in range(len(pattern) + 1):
insert_states.append(State(insert_distribution, name='I%s_%s' % (i, hmm_name)))
for i in range(len(pattern)):
distribution_map = dict({'A': 0.01, 'C': 0.01, 'G': 0.01, 'T': 0.01})
distribution_map[pattern[i]] = 0.97
match_states.append(State(DiscreteDistribution(distribution_map), name='M%s_%s' % (str(i + 1), hmm_name)))
for i in range(len(pattern)):
delete_states.append(State(None, name='D%s_%s' % (str(i + 1), hmm_name)))
unit_start = State(None, name='suffix_start_%s' % hmm_name)