Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
triangulated_graph = self.triangulate()
# Find maximal cliques in the chordal graph
cliques = list(map(tuple, nx.find_cliques(triangulated_graph)))
# If there is only 1 clique, then the junction tree formed is just a
# clique tree with that single clique as the node
if len(cliques) == 1:
clique_trees = JunctionTree()
clique_trees.add_node(cliques[0])
# Else if the number of cliques is more than 1 then create a complete
# graph with all the cliques as nodes and weight of the edges being
# the length of sepset between two cliques
elif len(cliques) >= 2:
complete_graph = UndirectedGraph()
edges = list(itertools.combinations(cliques, 2))
weights = list(map(lambda x: len(set(x[0]).intersection(set(x[1]))), edges))
for edge, weight in zip(edges, weights):
complete_graph.add_edge(*edge, weight=-weight)
# Create clique trees by minimum (or maximum) spanning tree method
clique_trees = JunctionTree(
nx.minimum_spanning_tree(complete_graph).edges()
)
# Check whether the factors are defined for all the random variables or not
all_vars = itertools.chain(*[factor.scope() for factor in self.factors])
if set(all_vars) != set(self.nodes()):
ValueError("DiscreteFactor for all the random variables not specified")
# Dictionary stating whether the factor is used to create clique
def __init__(self):
super(UndirectedGraph, self).__init__()
self._pull_status = False
self._push_status = False
import networkx as nx
from pgmpy.base import UndirectedGraph
from pgmpy.factors import Factor
class JunctionTree(UndirectedGraph):
"""
This class is meant to represent junction trees (Called as clique trees popularly).
It will contain a lot of functionalities to work on junction trees and to run
inference algorithms on JunctionTrees,
"""
def __init__(self):
super(UndirectedGraph, self).__init__()
self._pull_status = False
self._push_status = False
def add_jt_edges(self):
"""
This adds appropriate edges to the junction tree graph. Given a junction tree
with all the nodes containing cliques of the MarkovModel, running the function
on the junction tree will add all edges to the JT and then remove edges as
nodes = list(nodes)
if isinstance(independencies, Independencies):
def is_independent(X, Y, Zs):
return IndependenceAssertion(X, Y, Zs) in independencies
elif callable(independencies):
is_independent = independencies
else:
raise ValueError(
"'independencies' must be either Independencies-instance "
+ "or a ternary function that decides independencies."
)
graph = UndirectedGraph(combinations(nodes, 2))
lim_neighbors = 0
separating_sets = dict()
while not all(
[len(list(graph.neighbors(node))) < lim_neighbors for node in nodes]
):
for node in nodes:
for neighbor in list(graph.neighbors(node)):
# search if there is a set of neighbors (of size lim_neighbors)
# that makes X and Y independent:
for separating_set in combinations(
set(graph.neighbors(node)) - set([neighbor]), lim_neighbors
):
if is_independent(node, neighbor, separating_set):
separating_sets[
frozenset((node, neighbor))
] = separating_set
#!/usr/bin/env python3
import itertools
from collections import defaultdict
import numpy as np
from networkx.algorithms import bipartite
from pgmpy.models.MarkovModel import MarkovModel
from pgmpy.base import UndirectedGraph
from pgmpy.factors.discrete import DiscreteFactor
from pgmpy.factors import factor_product
class FactorGraph(UndirectedGraph):
"""
Class for representing factor graph.
DiscreteFactor graph is a bipartite graph representing factorization of a function.
They allow efficient computation of marginal distributions through sum-product
algorithm.
A factor graph contains two types of nodes. One type corresponds to random
variables whereas the second type corresponds to factors over these variables.
The graph only contains edges between variables and factor nodes. Each factor
node is associated with one factor whose scope is the set of variables that
are its neighbors.
Parameters
----------
data: input graph
# Backward Phase
for neigh in neighbors[node]:
other_neighbors = [n for n in neighbors[node] if n != neigh]
for sep_set in powerset(other_neighbors):
if self.test_conditional_independence(node, neigh, sep_set):
neighbors[node].remove(neigh)
break
# correct for false positives
for node in nodes:
for neigh in neighbors[node]:
if node not in neighbors[neigh]:
neighbors[node].remove(neigh)
skel = UndirectedGraph()
skel.add_nodes_from(nodes)
for node in nodes:
skel.add_edges_from([(node, neigh) for neigh in neighbors[node]])
return skel
#!/usr/bin/env python3
import itertools
from collections import defaultdict
import networkx as nx
import numpy as np
from pgmpy.base import UndirectedGraph
from pgmpy.factors.discrete import DiscreteFactor
from pgmpy.factors import factor_product
from pgmpy.independencies import Independencies
class MarkovModel(UndirectedGraph):
"""
Base class for markov model.
A MarkovModel stores nodes and edges with potentials
MarkovModel holds undirected edges.
Parameters
----------
data : input graph
Data to initialize graph. If data=None (default) an empty
graph is created. The data can be an edge list, or any
NetworkX graph object.
Examples
--------