fglib2.graphs

Module Contents

Classes

Node	Abstract Class for a nodes in a factor graph.
VariableNode	Variable node in a factor graph.
FactorNode	Factor node in a factor graph.
Edge	Edge in a factor graph.
FactorGraph	A factor graph.

class fglib2.graphs.Node

Bases: abc.ABC

Abstract Class for a nodes in a factor graph.

abstract sum_product(messages: List[fglib2.distributions.Multinomial]) → fglib2.distributions.Multinomial

Calculate the sum product algorithms step at this node.

Parameters:

messages – The input messages

Returns:

The output message

class fglib2.graphs.VariableNode(variable: random_events.variables.Discrete)

Bases: Node

Variable node in a factor graph.

variable: random_events.variables.Discrete

The variable asserted with this node.

unity() → fglib2.distributions.Multinomial

Create a uniform distribution over the variable of this node. :return:

__repr__()

Return repr(self).

sum_product(messages: List[fglib2.distributions.Multinomial]) → fglib2.distributions.Multinomial

Apply the sum product algorithm of a variable node.

All the input messages are multiplied together.

Parameters:

messages – The incoming messages.

Returns:

The product of all input messages.

class fglib2.graphs.FactorNode(distribution: fglib2.distributions.Multinomial)

Bases: Node

Factor node in a factor graph.

property variables: List[random_events.variables.Discrete]

__repr__()

Return repr(self).

sum_product(messages: List[fglib2.distributions.Multinomial]) → fglib2.distributions.Multinomial

Apply the sum product algorithm of a factor node.

The product of all incoming messages is calculated and multiplied with the distribution of this factor node.

Parameters:

messages – The incoming messages.

Returns:

The resulting, multivariate distribution.

__mul__(other: FactorNode) → FactorGraph

Multiply factor nodes to get a factor graph.

Parameters:

other – The other factor node.

Returns:

The resulting factor graph.

max_message(variable) → fglib2.distributions.Multinomial

Construct a message that contains the maximum likelihood for each value of the variable.

Note

The message is not normalized. The reason is the purpose of a max message. In every entry of the probabilities array is the maximum possible likelihood for the corresponding event. Therefore, this message should not be normalized.

Parameters:

variable – The variable to construct it over.

Returns:

A not normalized distribution over the variable with the maximum likelihood for each value.

class fglib2.graphs.Edge(source: Node, target: Node)

Edge in a factor graph.

Edges always have to be directed from a variable node to a factor node or vice versa.

property variable_node

Get the variable node of this edge.

property factor_node

Get the factor node of this edge.

property variable_to_factor

Get the message from variable to factor.

property factor_to_variable

Get the message from factor to variable.

__str__()

Return str(self).

__repr__()

Return repr(self).

class fglib2.graphs.FactorGraph

Bases: networkx.Graph, probabilistic_model.probabilistic_model.ProbabilisticModel

A factor graph.

A factor graph is a bipartite graph representing the factorization of a function. In probability theory and its applications, factor graphs are used to represent factorization of a probability distribution function, enabling efficient computations, such as the computation of marginal distributions through the sum–product algorithm.

Note

Only factor trees are efficient to compute with the sum/max product algorithm. Other graphs rely on approximate inference.

property variables: List[random_events.variables.Discrete]

Return a list of all variables in the factor graph.

property variable_nodes: List[VariableNode]

Return a list of all variable nodes in the factor graph.

property factor_nodes: List[FactorNode]

Return a list of all factor nodes in the factor graph.

add_edge(u_of_edge: Node, v_of_edge: Node, **attr)

Add an edge between u and v. :param u_of_edge: Source node. :param v_of_edge: target node. :param attr: edge attributes.

add_edges_from(edges: Iterable[Tuple[Node, Node]], **attr)

Add all the edges in edges. :param edges: List of edges. :param attr: Edge attributes.

sum_product()

Apply the sum product algorithm to the factor graph. The messages of the edges are set in place.

max_product() → random_events.events.Event

Apply the max product algorithm to the factor graph. The messages of the edges are set in place.

Returns:

The mode of the joint distribution as an Event.

node_of(variable: random_events.variables.Discrete) → VariableNode

Return the variable node of a variable. :param variable: The variable. :return: The variable node.

factor_of(variables: Iterable[random_events.variables.Discrete]) → FactorNode

Return the factor node of a set of factors. :param variables: The variables of the factor. :return: The factor node.

belief(variable: random_events.variables.Discrete) → fglib2.distributions.Multinomial

Compute the belief of a variable. :param variable: The variable :return: The distribution over the variable.

__mul__(other: FactorNode) → FactorGraph

Add a factor to the graph.

The variables that are not yet in the graph are added and the required edges are created. :param other: The factor to add.

Returns:

The factor graph with the added factor.

to_latex_equation() → str

Returns:

a latex representation of the equation represented by this factor graph.

brute_force_joint_distribution() → fglib2.distributions.Multinomial

Compute the joint distribution of the factor graphs variables by brute force.

Warning

This method is only feasible for a small number of variables as it has exponential runtime.

Returns:

A Multinomial distribution over all the variables.

reset()

Clear all messages in the graph in place.

abstract _likelihood(event: Iterable) → float

abstract _probability(event: random_events.events.EncodedEvent) → float

abstract _mode() → Tuple[Iterable[random_events.events.EncodedEvent], float]

abstract _conditional(event: random_events.events.EncodedEvent) → Tuple[typing_extensions.Self | None, float]

abstract marginal(variables: Iterable[random_events.variables.Variable]) → typing_extensions.Self | None