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

Refactoring: Moved individual node modules to nodes.py

parent cc7f651d
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
primo/core/Node.py deleted 100644 → 0
+ 0
− 30
View file @ cc7f651d
import abc
import re
class Node(object):
__metaclass__ = abc.ABCMeta
name = "UninitializedName"
position = (0, 0)
def __init__(self, node_name):
# Remove all special characters and replace " " with "_"
name = re.sub(r"[^a-zA-Z_0-9 ]*", "", node_name)
self.name = name.replace(" ", "_")
# for visual illustration
self.pos = (0, 0)
# @abc.abstractmethod
# def announce_parent(self, node):
# """This method will be called by the graph-management to inform nodes
# which just became children of other nodes, so they can adapt themselves
# (e.g. their cpt)"""
# return
def __str__(self):
print self.name
return self.name
primo/reasoning/ContinuousNode.py → primo/nodes.py
+ 310
− 0
View file @ 70a59a25
# -*- coding: utf-8 -*- import abc
from primo.reasoning import RandomNode
import random import random
import re
import scipy import scipy
from primo.reasoning.density import Beta
from primo.reasoning.density import Exponential
from primo.reasoning.density import Gauss
from primo.reasoning.density import ProbabilityTable
class Node(object):
__metaclass__ = abc.ABCMeta
name = "UninitializedName"
position = (0, 0)
def __init__(self, node_name):
# Remove all special characters and replace " " with "_"
name = re.sub(r"[^a-zA-Z_0-9 ]*", "", node_name)
self.name = name.replace(" ", "_")
# for visual illustration
self.pos = (0, 0)
def __str__(self):
return self.name
class RandomNode(Node):
'''Represents a random variable. There should be subclasses of this for
different kinds of data. There are currently DiscreteNode for
discrete-valued random variables and ContinuousNode for random Variables
with R or an Intervall in R as domain.
At a later point in time there may be structural nodes too.
'''
#The Continditional Propability Distribution of this random variable
cpd = None
def __init__(self, name):
super(RandomNode, self).__init__(name)
#value_range defines the domain of this random variable
self.value_range=None
def set_cpd(self, cpd):
self.cpd = cpd
def get_cpd(self):
return self.cpd
def announce_parent(self, node):
'''
Adjust the cpd so a new node is incorporated as dependency.
'''
self.cpd.add_variable(node)
def get_cpd_reduced(self, evidence):
'''
Return a reduced version of the cpd of this node. This reduced version
is constructed according to some evidence.
@param evidence: A List of (Node,Value) pairs.
'''
return self.cpd.reduction(evidence)
def get_value_range(self):
return self.value_range
def sample_gobal(self, x, evidence=None):
'''
This method can be used to sample from this local distribution.
@param state: A Dict from Node-objects to values. You can specify the
values of this nodes parents in this dict and the conditional
probability density will be adjusted accordingly.
'''
raise Exception("Called unimplemented Method")
def sample_local(self, x, evidence=None):
'''
This method can be used to do a random walk in the domain of this node.
@param x: The spot around which the next sample shall be generated.
@param evidence: Evidence which is to be concerned when new samples are
being generated. I am not entirely sure that this belongs here or is
correct in theory...
'''
raise Exception("Called unimplemented Method")
def is_valid(self):
raise Exception("Called an unimplemented function")
class DiscreteNode(RandomNode):
'''#TODO: write doc'''
def __init__(self, name, value_range):
super(DiscreteNode, self).__init__(name)
self.value_range = value_range
self.cpd = ProbabilityTable()
self.cpd.add_variable(self)
def __str__(self):
return self.name
def __repr__(self):
return "DiscreteNode("+self.name+")"
def set_probability(self, value, node_value_pairs):
self.cpd.set_probability(value, node_value_pairs)
def get_probability(self, value, node_value_pairs):
return self.cpd.get_probability([(self,value)] + node_value_pairs)
def set_probability_table(self, table, nodes):
self.cpd.set_probability_table(table, nodes)
def is_valid(self):
return self.cpd.is_normalized_as_cpt(self)
def sample_global(self, state, evidence):
if evidence==None or not self in evidence.keys():
compatibles=self.value_range
else:
compatibles=[]
for v in self.value_range:
if evidence[self].is_compatible(v):
compatibles.append(v)
return self.cpd.sample_global(state,self,compatibles)
def sample_local(self, x, evidence=None):
if evidence==None or not self in evidence.keys():
compatibles=self.value_range
else:
compatibles=[]
for v in self.value_range:
if evidence[self].is_compatible(v):
compatibles.append(v)
return random.choice(compatibles), 1.0
class ContinuousNode(RandomNode): class ContinuousNode(RandomNode):
''' '''
Represents a random-variable with a real-valued domain. Can only be defined Represents a random-variable with a real-valued domain. Can only be defined
...@@ -124,4 +264,47 @@ class ContinuousNode(RandomNode): ...@@ -124,4 +264,47 @@ class ContinuousNode(RandomNode):
all variables from this nodes markov-blanket. all variables from this nodes markov-blanket.
''' '''
return self.cpd.get_probability(value, state) return self.cpd.get_probability(value, state)
class ContinuousNodeFactory(object):
'''This class offers methods for generating ContinuousNodes'''
def __init__(self):
pass
def createGaussNode(self, name):
'''
Create a LinearGaussNode with linear dependencies on parents.
@param name: The name of the node.
'''
return self.createContinuousNode(name,(-float("Inf"),float("Inf")),Gauss)
def createExponentialNode(self, name):
'''
Create a LinearExponentialNode with linear dependencies on parents.
@param name: The name of the node.
'''
return self.createContinuousNode(name,(0,float("Inf")),Exponential)
def createBetaNode(self, name):
'''
Create a LinearBetaNode with linear dependencies on parents.
@param name: The name of the node.
'''
return self.createContinuousNode(name,(0,1),Beta)
def createContinuousNode(self,name,value_range,DensityClass):
'''
Create a ContinuousNode. This method should only be invoked from
outside this class if no specialized method is available.
@param name: The name of the node.
@param value_range: A 2-tuple which represents the interval that is the
domain of the variable.
@param DensityClass: A class from primo.reasoning.density that shall be
the node's pdf
'''
return ContinuousNode(name,value_range,DensityClass)
primo/reasoning/ContinuousNodeFactory.py deleted 100644 → 0
+ 0
− 47
View file @ cc7f651d
from primo.reasoning import ContinuousNode
from primo.reasoning.density import Gauss
from primo.reasoning.density import Exponential
from primo.reasoning.density import Beta
class ContinuousNodeFactory(object):
'''This class offers methods for generating ContinuousNodes'''
def __init__(self):
pass
def createGaussNode(self, name):
'''
Create a LinearGaussNode with linear dependencies on parents.
@param name: The name of the node.
'''
return self.createContinuousNode(name,(-float("Inf"),float("Inf")),Gauss)
def createExponentialNode(self, name):
'''
Create a LinearExponentialNode with linear dependencies on parents.
@param name: The name of the node.
'''
return self.createContinuousNode(name,(0,float("Inf")),Exponential)
def createBetaNode(self, name):
'''
Create a LinearBetaNode with linear dependencies on parents.
@param name: The name of the node.
'''
return self.createContinuousNode(name,(0,1),Beta)
def createContinuousNode(self,name,value_range,DensityClass):
'''
Create a ContinuousNode. This method should only be invoked from
outside this class if no specialized method is available.
@param name: The name of the node.
@param value_range: A 2-tuple which represents the interval that is the
domain of the variable.
@param DensityClass: A class from primo.reasoning.density that shall be
the node's pdf
'''
return ContinuousNode(name,value_range,DensityClass)
primo/reasoning/DiscreteNode.py deleted 100644 → 0
+ 0
− 56
View file @ cc7f651d
# -*- coding: utf-8 -*-
from primo.reasoning import RandomNode
from primo.reasoning.density import ProbabilityTable
import random
class DiscreteNode(RandomNode):
'''#TODO: write doc'''
def __init__(self, name, value_range):
super(DiscreteNode, self).__init__(name)
self.value_range = value_range
self.cpd = ProbabilityTable()
self.cpd.add_variable(self)
def __str__(self):
return self.name
def __repr__(self):
return "DiscreteNode("+self.name+")"
def set_probability(self, value, node_value_pairs):
self.cpd.set_probability(value, node_value_pairs)
def get_probability(self, value, node_value_pairs):
return self.cpd.get_probability([(self,value)] + node_value_pairs)
def set_probability_table(self, table, nodes):
self.cpd.set_probability_table(table, nodes)
def is_valid(self):
return self.cpd.is_normalized_as_cpt(self)
def sample_global(self, state, evidence):
if evidence==None or not self in evidence.keys():
compatibles=self.value_range
else:
compatibles=[]
for v in self.value_range:
if evidence[self].is_compatible(v):
compatibles.append(v)
return self.cpd.sample_global(state,self,compatibles)
def sample_local(self, x, evidence=None):
if evidence==None or not self in evidence.keys():
compatibles=self.value_range
else:
compatibles=[]
for v in self.value_range:
if evidence[self].is_compatible(v):
compatibles.append(v)
return random.choice(compatibles),1.0
primo/reasoning/RandomNode.py deleted 100644 → 0
+ 0
− 72
View file @ cc7f651d
# -*- coding: utf-8 -*-
from primo.core import Node
#from primo.reasoning.density import Density
class RandomNode(Node):
'''Represents a random variable. There should be subclasses of this for
different kinds of data. There are currently DiscreteNode for discrete-valued
random variables and ContinuousNode for random Variables with R or an Intervall
in R as domain.
At a later point in time there may be structural nodes too.
'''
#The Continditional Propability Distribution of this random variable
cpd = None
def __init__(self, name):
super(RandomNode, self).__init__(name)
#value_range defines the domain of this random variable
self.value_range=None
def set_cpd(self, cpd):
self.cpd = cpd
def get_cpd(self):
return self.cpd
def announce_parent(self, node):
'''
Adjust the cpd so a new node is incorporated as dependency.
'''
self.cpd.add_variable(node)
def get_cpd_reduced(self, evidence):
'''
Return a reduced version of the cpd of this node. This reduced version
is constructed according to some evidence.
@param evidence: A List of (Node,Value) pairs.
'''
return self.cpd.reduction(evidence)
def get_value_range(self):
return self.value_range
def sample_gobal(self, x, evidence=None):
'''
This method can be used to sample from this local distribution.
@param state: A Dict from Node-objects to values. You can specify the
values of this nodes parents in this dict and the conditional
probability density will be adjusted accordingly.
'''
raise Exception("Called unimplemented Method")
def sample_local(self, x, evidence=None):
'''
This method can be used to do a random walk in the domain of this node.
@param x: The spot around which the next sample shall be generated.
@param evidence: Evidence which is to be concerned when new samples are
being generated. I am not entirely sure that this belongs here or is
correct in theory...
'''
raise Exception("Called unimplemented Method")
def is_valid(self):
raise Exception("Called an unimplemented function")
primo/reasoning/Untitled Document 1 deleted 100644 → 0
+ 0
− 13
View file @ cc7f651d
from primo.reasoning import ContinuousNode
from primo.reasoning import LinearGaussNode
class ContinuousNodeFactory(object):
def __init__(self):
pass
def createLinearGaussNode(self, name):
return self.createContinuousNode(name,(-float("Inf"),float("Inf")),LinearGauss)
def createContinuousNode(self,name,value_range,DensityClass):
return ContinuousNode(name,value_range,DensityClass)
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