Skip to content
Snippets Groups Projects
Commit a6d76f4c authored by Hendrik Buschmeier's avatar Hendrik Buschmeier
Browse files

Refactoring: Moved individual network modules to networks.py.

parent b668f8e1
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
primo/core/BayesianDecisionNetwork.py deleted 100644 → 0
+ 0
− 105
View file @ b668f8e1
# -*- coding: utf-8 -*-
import networkx as nx
from primo.core import BayesNet
from primo.core import Node
from primo.decision import DecisionNode
from primo.decision import UtilityNode
from primo.reasoning import DiscreteNode
class BayesianDecisionNetwork(BayesNet):
def __init__(self):
super(BayesianDecisionNetwork, self).__init__()
self.partialOrdering = []
self.random_nodes = []
self.decision_nodes = []
self.utility_nodes = []
def is_valid(self):
'''Check if graph structure is valid.
Returns true if graph is directed, acyclic and if there is a path that connects every decision node(consistency check),
false otherwise'''
if self.graph.number_of_selfloops() > 0:
return False
for node in self.graph.nodes():
if self.has_loop(node):
return False
decisionNodeList = []
for node in self.get_all_nodes():
if isinstance(node, DecisionNode):
decisionNodeList.append(node)
return all([nx.has_path(self.graph, x, y) == True for x in decisionNodeList for y in decisionNodeList])
def add_node(self, node):
if isinstance(node, Node):
if node.name in self.node_lookup.keys():
raise Exception("Node name already exists in Bayesnet: "+node.name)
if isinstance(node, DiscreteNode):
self.random_nodes.append(node)
elif isinstance(node, UtilityNode):
self.utility_nodes.append(node)
elif isinstance(node, DecisionNode):
self.decision_nodes.append(node)
else:
raise Exception("Tried to add a node which the Bayesian Decision Network can not work with")
self.node_lookup[node.name]=node
self.graph.add_node(node)
else:
raise Exception("Can only add 'Node' and its subclasses as nodes into the BayesNet")
def get_all_nodes(self):
'''Returns all RandomNodes'''
return self.random_nodes
def get_all_decision_nodes(self):
return self.decision_nodes
def get_all_utility_nodes(self):
return self.utility_nodes
def add_edge(self, node_from, node_to):
"""
Adds an edge between two nodes. It is impossible to create en edge between two decision nodes and between two
utility nodes.
keyword arguments:
node_from -- Node from where the edge shall begin
node_to -- Node where the edge shall end
"""
if isinstance(node_from, DecisionNode) and isinstance(node_to, DecisionNode):
raise Exception("Tried to add an edge from a DecisionNode to a DecisionNode")
if isinstance(node_from, UtilityNode) and isinstance(node_to, UtilityNode):
raise Exception("Tried to add an edge from a UtilityNode to a UtilityNode")
if node_from in self.graph.nodes() and node_to in self.graph.nodes():
self.graph.add_edge(node_from, node_to)
node_to.announce_parent(node_from)
else:
raise Exception("Tried to add an Edge between two Nodes of which at least one was not contained in the Bayesnet")
def reset_Decisions(self):
"""
Resets all decisions in the Bayesian Decision Network
"""
for node in self.decision_nodes:
node.set_state(None)
def get_partialOrdering(self):
"""
Getter for the partial ordere
"""
return self.partialOrdering
def set_partialOrdering(self, partialOrder):
"""
Sets the partial ordering for this Bayesian Decision Network
partialOrder -- ordered list of RandomNodes and Decision Nodes
example: [decisionNode1, [randomNode1,randomNode2], decisionNode2, [randomNode3]]
"""
self.partialOrdering = partialOrder
\ No newline at end of file
primo/core/DynamicBayesNet.py deleted 100644 → 0
+ 0
− 67
View file @ b668f8e1
# -*- coding: utf-8 -*-
from primo.core import BayesNet
from primo.core import TwoTBN
class DynamicBayesNet(BayesNet):
''' This is the implementation of a dynamic Bayesian network (also called
temporal Bayesian network).
Definition: DBN is a pair (B0, TwoTBN), where B0 is a BN over X(0),
representing the initial distribution over states, and TwoTBN is a
2-TBN for the process.
See Koller, Friedman - "Probabilistic Graphical Models" (p. 204)
Properties: Markov property, stationary, directed, discrete,
acyclic (within a slice)
'''
def __init__(self):
super(DynamicBayesNet, self).__init__()
self._B0 = BayesNet()
self._twoTBN = TwoTBN()
@property
def B0(self):
''' Get the Bayesian network representing the initial distribution.'''
return self._B0
@B0.setter
def B0(self, value):
''' Set the Bayesian network representing the initial distribution.'''
if isinstance(value, BayesNet):
if not value.is_valid():
raise Exception("BayesNet is not valid.")
self._B0 = value
else:
raise Exception("Can only set 'BayesNet' and its subclasses as " +
"B0 of a DBN.")
@property
def twoTBN(self):
''' Get the 2-time-slice Bayesian network.'''
return self._twoTBN
@twoTBN.setter
def twoTBN(self, value):
''' Set the 2-time-slice Bayesian network.'''
if isinstance(value, TwoTBN):
if not value.is_valid():
raise Exception("BayesNet is not valid.")
self._twoTBN = value
else:
raise Exception("Can only set 'TwoTBN' and its subclasses as " +
"twoTBN of a DBN.")
def is_valid(self):
'''Check if graph structure is valid. And if there is a same-named
inital node in towTBN for every node in BO.
Returns true if graph is directed and acyclic, false otherwiese'''
for node in self._B0.get_nodes():
if not self._twoTBN.has_initial_node_by_name(node.name):
print("Node with name " + str(node.name) +
" not found in TwoTBN!")
return False;
return super(DynamicBayesNet, self).is_valid()
\ No newline at end of file
primo/core/TwoTBN.py deleted 100644 → 0
+ 0
− 81
View file @ b668f8e1
# -*- coding: utf-8 -*-
from primo.core import BayesNet
from primo.reasoning.density import ProbabilityTable
class TwoTBN(BayesNet):
''' This is the implementation of a 2-time-slice Bayesian network (2-TBN).
'''
def __init__(self, bayesnet=None):
BayesNet.__init__(self)
if bayesnet:
if not isinstance(bayesnet, BayesNet):
raise Exception("Parameter 'bayesnet' is not a instance of class BayesNet.")
self.graph = bayesnet.graph
self.node_lookup = bayesnet.node_lookup
self.__initial_nodes = []
def create_timeslice(self, state, initial=False):
'''
Set all initial nodes to the value of their corresponding nodes
in state (previous time slice).
Keyword arguments:
state -- Current state of the network (previous time slice).
initial -- Set initial to true if this will be the first time slice
and state only contains nodes of the initial distribution.
Returns this instance with all initial nodes set to their
new value.
'''
for (node, node_t) in self.__initial_nodes:
cpd = ProbabilityTable()
cpd.add_variable(node)
node.set_cpd(cpd)
if not initial:
node.set_probability(1., [(node, state[node_t])])
else:
for node0 in state:
if node0.name == node.name:
node.set_probability(1., [(node, state[node0])])
continue
return self
def add_node(self, node, initial=False, node_t=None):
'''
Add a node to the TwoTBN.
Keyword arguments:
node -- Node to be added.
initial -- If true node is marked as initial node.
node_t -- If initial is true this is the corresponding node in the time slice.
'''
super(TwoTBN, self).add_node(node)
if initial:
self.set_initial_node(node, node_t)
def set_initial_node(self, node_name, node_name_t):
'''
Mark a node as initial node.
Keyword arguments:
node_name -- Name of the initial node.
node_name_t -- Name of the corresponding node in the time slice.
'''
node0 = self.get_node(node_name)
node1 = self.get_node(node_name_t)
self.__initial_nodes.append((node0, node1))
def has_initial_node_by_name(self, node_name):
'''
Check if this instance has an inital node with name node_name.
Returns true on success, false otherwise.
'''
for (node, node_t) in self.__initial_nodes:
if node.name == node_name:
return True
return False
\ No newline at end of file
primo/core/__init__.py deleted 100644 → 0
+ 0
− 5
View file @ b668f8e1
from Node import Node
from BayesNet import BayesNet
from BayesianDecisionNetwork import BayesianDecisionNetwork
from TwoTBN import TwoTBN
from DynamicBayesNet import DynamicBayesNet
primo/core/BayesNet.py → primo/networks.py
+ 374
− 0
View file @ a6d76f4c
import networkx as nx
from primo.core import Node
import primo.nodes
class BayesNet(object):
......@@ -10,7 +9,7 @@ class BayesNet(object):
self.node_lookup = {}
def add_node(self, node):
if isinstance(node, Node):
if isinstance(node, primo.nodes.Node):
if node.name in self.node_lookup.keys():
raise Exception("Node name already exists in Bayesnet: "+node.name)
self.node_lookup[node.name]=node
......@@ -135,4 +134,241 @@ class BayesNet(object):
def __len__(self):
'''Return the number of nodes in the graph.'''
return len(self.graph)
\ No newline at end of file
return len(self.graph)
class BayesianDecisionNetwork(BayesNet):
def __init__(self):
super(BayesianDecisionNetwork, self).__init__()
self.partialOrdering = []
self.random_nodes = []
self.decision_nodes = []
self.utility_nodes = []
def is_valid(self):
'''Check if graph structure is valid.
Returns true if graph is directed, acyclic and if there is a path that connects every decision node(consistency check),
false otherwise'''
if self.graph.number_of_selfloops() > 0:
return False
for node in self.graph.nodes():
if self.has_loop(node):
return False
decisionNodeList = []
for node in self.get_all_nodes():
if isinstance(node, DecisionNode):
decisionNodeList.append(node)
return all([nx.has_path(self.graph, x, y) == True for x in decisionNodeList for y in decisionNodeList])
def add_node(self, node):
if isinstance(node, Node):
if node.name in self.node_lookup.keys():
raise Exception("Node name already exists in Bayesnet: "+node.name)
if isinstance(node, DiscreteNode):
self.random_nodes.append(node)
elif isinstance(node, UtilityNode):
self.utility_nodes.append(node)
elif isinstance(node, DecisionNode):
self.decision_nodes.append(node)
else:
raise Exception("Tried to add a node which the Bayesian Decision Network can not work with")
self.node_lookup[node.name]=node
self.graph.add_node(node)
else:
raise Exception("Can only add 'Node' and its subclasses as nodes into the BayesNet")
def get_all_nodes(self):
'''Returns all RandomNodes'''
return self.random_nodes
def get_all_decision_nodes(self):
return self.decision_nodes
def get_all_utility_nodes(self):
return self.utility_nodes
def add_edge(self, node_from, node_to):
"""
Adds an edge between two nodes. It is impossible to create en edge between two decision nodes and between two
utility nodes.
keyword arguments:
node_from -- Node from where the edge shall begin
node_to -- Node where the edge shall end
"""
if isinstance(node_from, DecisionNode) and isinstance(node_to, DecisionNode):
raise Exception("Tried to add an edge from a DecisionNode to a DecisionNode")
if isinstance(node_from, UtilityNode) and isinstance(node_to, UtilityNode):
raise Exception("Tried to add an edge from a UtilityNode to a UtilityNode")
if node_from in self.graph.nodes() and node_to in self.graph.nodes():
self.graph.add_edge(node_from, node_to)
node_to.announce_parent(node_from)
else:
raise Exception("Tried to add an Edge between two Nodes of which at least one was not contained in the Bayesnet")
def reset_Decisions(self):
"""
Resets all decisions in the Bayesian Decision Network
"""
for node in self.decision_nodes:
node.set_state(None)
def get_partialOrdering(self):
"""
Getter for the partial ordere
"""
return self.partialOrdering
def set_partialOrdering(self, partialOrder):
"""
Sets the partial ordering for this Bayesian Decision Network
partialOrder -- ordered list of RandomNodes and Decision Nodes
example: [decisionNode1, [randomNode1,randomNode2], decisionNode2, [randomNode3]]
"""
self.partialOrdering = partialOrder
class DynamicBayesNet(BayesNet):
''' This is the implementation of a dynamic Bayesian network (also called
temporal Bayesian network).
Definition: DBN is a pair (B0, TwoTBN), where B0 is a BN over X(0),
representing the initial distribution over states, and TwoTBN is a
2-TBN for the process.
See Koller, Friedman - "Probabilistic Graphical Models" (p. 204)
Properties: Markov property, stationary, directed, discrete,
acyclic (within a slice)
'''
def __init__(self):
super(DynamicBayesNet, self).__init__()
self._B0 = BayesNet()
self._twoTBN = TwoTBN()
@property
def B0(self):
''' Get the Bayesian network representing the initial distribution.'''
return self._B0
@B0.setter
def B0(self, value):
''' Set the Bayesian network representing the initial distribution.'''
if isinstance(value, BayesNet):
if not value.is_valid():
raise Exception("BayesNet is not valid.")
self._B0 = value
else:
raise Exception("Can only set 'BayesNet' and its subclasses as " +
"B0 of a DBN.")
@property
def twoTBN(self):
''' Get the 2-time-slice Bayesian network.'''
return self._twoTBN
@twoTBN.setter
def twoTBN(self, value):
''' Set the 2-time-slice Bayesian network.'''
if isinstance(value, TwoTBN):
if not value.is_valid():
raise Exception("BayesNet is not valid.")
self._twoTBN = value
else:
raise Exception("Can only set 'TwoTBN' and its subclasses as " +
"twoTBN of a DBN.")
def is_valid(self):
'''Check if graph structure is valid. And if there is a same-named
inital node in towTBN for every node in BO.
Returns true if graph is directed and acyclic, false otherwiese'''
for node in self._B0.get_nodes():
if not self._twoTBN.has_initial_node_by_name(node.name):
print("Node with name " + str(node.name) +
" not found in TwoTBN!")
return False;
return super(DynamicBayesNet, self).is_valid()
class TwoTBN(BayesNet):
''' This is the implementation of a 2-time-slice Bayesian network (2-TBN).
'''
def __init__(self, bayesnet=None):
BayesNet.__init__(self)
if bayesnet:
if not isinstance(bayesnet, BayesNet):
raise Exception("Parameter 'bayesnet' is not a instance of class BayesNet.")
self.graph = bayesnet.graph
self.node_lookup = bayesnet.node_lookup
self.__initial_nodes = []
def create_timeslice(self, state, initial=False):
'''
Set all initial nodes to the value of their corresponding nodes
in state (previous time slice).
Keyword arguments:
state -- Current state of the network (previous time slice).
initial -- Set initial to true if this will be the first time slice
and state only contains nodes of the initial distribution.
Returns this instance with all initial nodes set to their
new value.
'''
for (node, node_t) in self.__initial_nodes:
cpd = ProbabilityTable()
cpd.add_variable(node)
node.set_cpd(cpd)
if not initial:
node.set_probability(1., [(node, state[node_t])])
else:
for node0 in state:
if node0.name == node.name:
node.set_probability(1., [(node, state[node0])])
continue
return self
def add_node(self, node, initial=False, node_t=None):
'''
Add a node to the TwoTBN.
Keyword arguments:
node -- Node to be added.
initial -- If true node is marked as initial node.
node_t -- If initial is true this is the corresponding node in the time slice.
'''
super(TwoTBN, self).add_node(node)
if initial:
self.set_initial_node(node, node_t)
def set_initial_node(self, node_name, node_name_t):
'''
Mark a node as initial node.
Keyword arguments:
node_name -- Name of the initial node.
node_name_t -- Name of the corresponding node in the time slice.
'''
node0 = self.get_node(node_name)
node1 = self.get_node(node_name_t)
self.__initial_nodes.append((node0, node1))
def has_initial_node_by_name(self, node_name):
'''
Check if this instance has an inital node with name node_name.
Returns true on success, false otherwise.
'''
for (node, node_t) in self.__initial_nodes:
if node.name == node_name:
return True
return False
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment