Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
start_state: dict or array-like iterable object
Dict (or list) of tuples representing the starting states of the variables.
Examples
--------
>>> from pgmpy.models import MarkovChain as MC
>>> from pgmpy.factors.discrete import State
>>> model = MC(['a', 'b'], [2, 2])
>>> model.set_start_state([State('a', 0), State('b', 1)])
"""
if start_state is not None:
if not hasattr(start_state, "__iter__") or isinstance(start_state, str):
raise ValueError("start_state must be a non-string iterable.")
# Must be an array-like iterable. Reorder according to self.variables.
state_dict = {var: st for var, st in start_state}
start_state = [State(var, state_dict[var]) for var in self.variables]
if start_state is None or self._check_state(start_state):
self.state = start_state
"""
if start_state is None:
if self.state is None:
self.state = self.random_state()
# else use previously-set state
else:
self.set_start_state(start_state)
# sampled.loc[0] = [self.state[var] for var in self.variables]
for i in range(size):
for j, (var, st) in enumerate(self.state):
next_st = sample_discrete(
list(self.transition_models[var][st].keys()),
list(self.transition_models[var][st].values()),
)[0]
self.state[j] = State(var, next_st)
yield self.state[:]
"""
Generates a random state of the Markov Chain.
Returns
-------
List of namedtuples, representing a random assignment to all variables of the model.
Examples
--------
>>> from pgmpy.models import MarkovChain as MC
>>> model = MC(['intel', 'diff'], [2, 3])
>>> model.random_state()
[State('diff', 2), State('intel', 1)]
"""
return [
State(var, np.random.randint(self.cardinalities[var]))
for var in self.variables
]
var_states = defaultdict(dict)
var_values = defaultdict(dict)
samples = defaultdict(dict)
for var in self.transition_models.keys():
for st in self.transition_models[var]:
var_states[var][st] = list(self.transition_models[var][st].keys())
var_values[var][st] = list(self.transition_models[var][st].values())
samples[var][st] = sample_discrete(
var_states[var][st], var_values[var][st], size=size
)
for i in range(size - 1):
for j, (var, st) in enumerate(self.state):
next_st = samples[var][st][i]
self.state[j] = State(var, next_st)
sampled.loc[i + 1] = [st for var, st in self.state]
return sampled
transition_mat = np.array(
[
np.array(list(self.transition_models[k][i].values()))
for i in self.transition_models[k].keys()
],
dtype=np.float,
)
S, U = eig(transition_mat.T)
stationary = np.array(U[:, np.where(np.abs(S - 1.0) < 1e-8)[0][0]].flat)
stationary = (stationary / np.sum(stationary)).real
probabilites = []
window_size = 10000 if sample is None else len(sample)
for i in range(0, transition_mat.shape[0]):
probabilites.extend(
self.prob_from_sample([State(k, i)], window_size=window_size)
)
if any(
np.abs(i) > tolerance for i in np.subtract(probabilites, stationary)
):
return_val = return_val and False
else:
return_val = return_val and True
return return_val