probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit#
Classes#
Enum where members are also (and must be) ints |
|
Class for all units of a probabilistic circuit. |
|
Class for Leaf units. |
|
Class for inner units |
|
Class for inner units |
|
Decomposable Product Units for Probabilistic Circuits |
|
Probabilistic Circuits as a directed, rooted, acyclic graph. |
|
class for PC in shallow form, sum unit as root followed by product units which only have leafs as children. |
|
Class for Leaf units. |
|
Class for Leaf units. |
|
Class for Leaf units. |
Functions#
|
Factory that creates the correct leaf from a distribution. |
Module Contents#
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.PlotAlignment#
Bases:
enum.IntEnumEnum where members are also (and must be) ints
- HORIZONTAL = 0#
- VERTICAL = 1#
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.Unit#
Bases:
random_events.utils.SubclassJSONSerializer,probabilistic_model.interfaces.drawio.drawio.DrawIOInterface,abc.ABCClass for all units of a probabilistic circuit.
This class should not be used by users directly. Use
ProbabilisticCircuitas interface to users.- probabilistic_circuit: typing_extensions.Optional[ProbabilisticCircuit] = None#
The circuit this component is part of.
- result_of_current_query: typing_extensions.Any = None#
The result of the current query.
- index: typing_extensions.Optional[int] = None#
The index of the node in the graph of its circuit.
- __post_init__()#
- abstract support()#
Calculate the support of this unit.
- abstract property is_leaf#
- Returns:
If this unit is a leaf unit.
- property variables: sortedcontainers.SortedSet#
- Abstractmethod:
- property leaves: typing_extensions.List[LeafUnit]#
- Returns:
The leaves of the circuit that are descendants of this node.
- update_variables(new_variables: random_events.product_algebra.VariableMap)#
Update the variables of this unit and its descendants.
- Parameters:
new_variables – A map that maps the variables that should be replaced to their new variable.
- connect_incoming_edges_to(other: Unit)#
Connect all incoming edges to this unit to another unit.
- Parameters:
other – The other unit to connect the incoming edges to.
- filter_variable_map_by_self(variable_map: random_events.product_algebra.VariableMap)#
Filter a variable map by the variables of this unit.
- Parameters:
variable_map – The map to filter
- Returns:
The map filtered by the variables of this unit.
- property impossible_condition_result: typing_extensions.Tuple[typing_extensions.Optional[Unit], float]#
- Returns:
The result of an impossible truncated query.
- abstract log_mode()#
- __hash__()#
- copy_without_graph()#
- empty_copy() typing_extensions.Self#
Creat a copy of this circuit without any subcircuits. Only the parameters should be copied. This is used whenever a new circuit has to be created during inference.
- Returns:
A copy of this circuit without any subcircuits that is not in this units graph.
- abstract simplify()#
Simplify the circuit by removing nodes and redirected edges that have no impact in-place. Essentially, this method transforms the circuit into an alternating order of sum and product units.
- Returns:
The simplified circuit.
- abstract sample(*args, **kwargs)#
Draw samples from the circuit.
For sampling, a node gets requested a number of samples from all his parents. The parents write into the result_of_current_query attribute a tuple describing the beginning index of the sampling and how many samples are requested.
- abstract marginal(*args, **kwargs) typing_extensions.Optional[typing_extensions.Self]#
Remove nodes that are not part of the marginal distribution.
- abstract moment(*args, **kwargs)#
- property drawio_style: typing_extensions.Dict[str, typing_extensions.Any]#
The style of the object.
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.LeafUnit#
Bases:
UnitClass for Leaf units.
- distribution: typing_extensions.Optional[probabilistic_model.probabilistic_model.ProbabilisticModel]#
The distribution contained in this leaf unit.
- __repr__()#
- property drawio_label#
The label of the object as a drawio compatible string.
- property drawio_style: typing_extensions.Dict[str, typing_extensions.Any]#
The style of the object.
- property variables: typing_extensions.Iterable[random_events.variable.Variable]#
- property is_leaf#
- Returns:
If this unit is a leaf unit.
- property leaves: typing_extensions.List[LeafUnit]#
- Returns:
The leaves of the circuit that are descendants of this node.
- log_likelihood(events: numpy.array)#
- cdf(events: numpy.array)#
- probability_of_simple_event(event: random_events.product_algebra.SimpleEvent)#
- support()#
Calculate the support of this unit.
- simplify()#
Simplify the circuit by removing nodes and redirected edges that have no impact in-place. Essentially, this method transforms the circuit into an alternating order of sum and product units.
- Returns:
The simplified circuit.
- log_truncated_of_simple_event_in_place(event: random_events.product_algebra.SimpleEvent)#
- moment(order, center, variable_to_index_map)#
- sample(samples: numpy.array, variable_to_index_map: typing_extensions.Dict[random_events.variable.Variable, int])#
Sample from the distribution and write the samples into the samples array. :param samples: The array to write the samples into. :param variable_to_index_map: The map from variables to column indices in the samples array.
- marginal(variables: typing_extensions.Iterable[random_events.variable.Variable]) typing_extensions.Optional[typing_extensions.Self]#
Remove nodes that are not part of the marginal distribution.
- log_mode()#
- to_json()#
- classmethod _from_json(data: typing_extensions.Dict[str, typing_extensions.Any]) typing_extensions.Self#
Create a variable from a json dict. This method is called from the from_json method after the correct subclass is determined and should be overwritten by the respective subclass.
- Parameters:
data – The json dict
- Returns:
The deserialized object
- log_conditional_in_place(point: typing_extensions.Dict[random_events.variable.Variable, typing_extensions.Any])#
- copy_without_graph()#
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.InnerUnit#
Bases:
UnitClass for inner units
- property is_leaf#
- Returns:
If this unit is a leaf unit.
- abstract forward(*args, **kwargs)#
- abstract log_forward(*args, **kwargs)#
- marginal(*args, **kwargs) typing_extensions.Optional[typing_extensions.Self]#
Remove nodes that are not part of the marginal distribution.
- classmethod _from_json(data: typing_extensions.Dict[str, typing_extensions.Any]) typing_extensions.Self#
Create a variable from a json dict. This method is called from the from_json method after the correct subclass is determined and should be overwritten by the respective subclass.
- Parameters:
data – The json dict
- Returns:
The deserialized object
- add_subcircuit(subcircuit: Unit, log_weight: float = None)#
Add a subcircuit to the subcircuits of this unit.
Note
This method does not normalize the edges to the subcircuits.
- Parameters:
subcircuit – The subcircuit to add.
log_weight – The logarithmic weight of the subcircuit.
Only needed if this is a sum unit
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.SumUnit#
Bases:
InnerUnitClass for inner units
- _latent_variable: typing_extensions.Optional[random_events.variable.Symbolic] = None#
The latent variable of this unit. This has to be here due to the rvalue/lvalue problem in random events.
TODO remove this when RE is fixed
- __repr__()#
- __hash__#
- property representation: str#
- property drawio_label: str#
The label of the object as a drawio compatible string.
- property log_weighted_subcircuits: typing_extensions.List[typing_extensions.Tuple[float, Unit]]#
- Returns:
The weighted subcircuits of this unit.
- property variables: sortedcontainers.SortedSet#
- property latent_variable: random_events.variable.Symbolic#
- forward(*args, **kwargs)#
- log_forward(*args, **kwargs)#
- moment#
- log_forward_conditioning(*args, **kwargs)#
- support()#
Calculate the support of this unit.
- property log_weights: numpy.array#
- Returns:
The log_weights of the subcircuits.
- sample(*args, **kwargs) numpy.array#
Draw samples from the circuit.
For sampling, a node gets requested a number of samples from all his parents. The parents write into the result_of_current_query attribute a tuple describing the beginning index of the sampling and how many samples are requested.
- classmethod _from_json(data: typing_extensions.Dict[str, typing_extensions.Any]) typing_extensions.Self#
Create a variable from a json dict. This method is called from the from_json method after the correct subclass is determined and should be overwritten by the respective subclass.
- Parameters:
data – The json dict
- Returns:
The deserialized object
- mount_with_interaction_terms(other: typing_extensions.Self, interaction_model: probabilistic_model.probabilistic_model.ProbabilisticModel)#
Create a distribution that factorizes as follows:
\[p(self.latent\_variable) \cdot p(self.variables | self.latent\_variable) \cdot p(other.latent\_variable | self.latent\_variable) \cdot p(other.variables | other.latent\_variable)\]where self.latent_variable and other.latent_variable are the results of the latent variable interpretation of mixture models.
- Parameters:
other – The other distribution to mount at this distribution children level.
interaction_model – The interaction probabilities between both latent variables
- mount_from_bayesian_network(other: SumUnit)#
Mount a distribution from tge to_probabilistic_circuit method in bayesian networks. The distribution is mounted as follows:
- Parameters:
other – The other distribution to mount at this distribution children level.
- Returns:
- simplify()#
Simplify the circuit by removing nodes and redirected edges that have no impact in-place. Essentially, this method transforms the circuit into an alternating order of sum and product units.
- Returns:
The simplified circuit.
- normalize()#
Normalize the log_weights of the subcircuits such that they sum up to 1 inplace.
- is_deterministic() bool#
- Returns:
If this unit is deterministic or not.
- log_mode()#
- subcircuit_index_of_samples(samples: numpy.array) numpy.array#
- Returns:
the index of the subcircuit where p(sample) > 0 and None if p(sample) = 0 for all subcircuits.
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.ProductUnit#
Bases:
InnerUnitDecomposable Product Units for Probabilistic Circuits
- representation = '×'#
- __hash__#
- property drawio_label: str#
The label of the object as a drawio compatible string.
- __repr__()#
- forward(*args, **kwargs)#
- log_forward(*args, **kwargs)#
- moment#
- property variables: sortedcontainers.SortedSet#
- support()#
Calculate the support of this unit.
- is_decomposable()#
- log_mode()#
- __copy__()#
- simplify()#
Simplify the circuit by removing nodes and redirected edges that have no impact in-place. Essentially, this method transforms the circuit into an alternating order of sum and product units.
- Returns:
The simplified circuit.
- sample(*args, **kwargs)#
Draw samples from the circuit.
For sampling, a node gets requested a number of samples from all his parents. The parents write into the result_of_current_query attribute a tuple describing the beginning index of the sampling and how many samples are requested.
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.ProbabilisticCircuit#
Bases:
probabilistic_model.probabilistic_model.ProbabilisticModel,random_events.utils.SubclassJSONSerializerProbabilistic Circuits as a directed, rooted, acyclic graph.
The nodes of the graph are the units of the circuit. The edges of the graph indicate how the units are connected. The outgoing edges of a sum unit contain the log-log_weights of the subcircuits.
- __len__()#
Return the number of nodes in the graph.
- Returns:
The number of nodes in the graph.
- __iter__()#
Return an iterator over the nodes in the graph.
- Returns:
An iterator over the nodes in the graph.
- classmethod from_other(other: typing_extensions.Self) typing_extensions.Self#
- property variables: sortedcontainers.SortedSet#
- Returns:
The variables of the model.
- property variable_to_index_map: typing_extensions.Dict[random_events.variable.Variable, int]#
- is_valid() bool#
Check if this graph is:
acyclic
connected
- Returns:
True if the graph is valid, False otherwise.
- add_edges_from(edges: typing_extensions.Iterable[typing_extensions.Union[typing_extensions.Tuple[Unit, Unit], typing_extensions.Tuple[Unit, Unit, float]]])#
- in_edges(unit: Unit) typing_extensions.List[typing_extensions.Tuple[Unit, Unit, typing_extensions.Optional[float]]]#
- add_from_subgraph(subgraph: rustworkx.PyDAG[Unit]) typing_extensions.Dict[int, Unit]#
Add nodes and edges from a subgraph to this circuit.
- Parameters:
subgraph – The subgraph to add nodes from.
- Returns:
A dictionary mapping the node indices in the subgraph to the new units in this circuit.
- nodes() typing_extensions.List[Unit]#
Return an iterator over the nodes.
- Returns:
An iterator over the nodes.
- edges() typing_extensions.List[typing_extensions.Tuple[Unit, Unit, typing_extensions.Optional[float]]]#
- property root: Unit#
The root of the circuit is the node with in-degree 0. This is the output node, that will perform the final computation.
- Returns:
The root of the circuit.
- log_likelihood(events: numpy.array) numpy.array#
Calculate the log-likelihood of an event.
Check the documentation of likelihood for more information.
- Parameters:
events – The full evidence event with shape (#events, #variables)
- Returns:
The log-likelihood of the event with shape (#events).
- cdf(events: numpy.array) numpy.array#
Calculate the cumulative distribution function of an event-array.
The event belongs to the class of full evidence queries.
..Note:: The cdf only exists if all variables are continuous or integers.
- Parameters:
events – The array of full evidence events. The shape of the array has to be (n, len(self.variables)).
- Returns:
The cumulative distribution function of the event as an array of shape (n,).
- probability_of_simple_event(event: random_events.product_algebra.SimpleEvent) float#
Calculate the probability of a simple event.
The event belongs to the class of marginal queries.
Note
You can read more about queries of this class in Definition 11 in [CVVdB20] or watch the video tutorial. [US20]
- Parameters:
event – The event.
- Returns:
The probability of the event.
- log_mode(check_determinism: bool = True) typing_extensions.Tuple[random_events.product_algebra.Event, float]#
Calculate the mode of the model.
Check the documentation of mode for more information.
- Returns:
The mode and its log-likelihood.
- log_truncated_of_simple_event_in_place(simple_event: random_events.product_algebra.SimpleEvent) typing_extensions.Tuple[typing_extensions.Optional[typing_extensions.Self], float]#
Construct the truncated circuit from a simple event.
- Parameters:
simple_event – The simple event to condition on.
- Returns:
The truncated circuit and the log-probability of the event
- log_truncated_in_place(event: random_events.product_algebra.Event) typing_extensions.Tuple[typing_extensions.Optional[typing_extensions.Self], float]#
Efficiently compute the truncated for an Event, batching as much as possible.
- log_truncated(event: random_events.product_algebra.Event) typing_extensions.Tuple[typing_extensions.Optional[typing_extensions.Self], float]#
Calculate the truncated distribution P(*| event) and the probability of the event.
Check the documentation of truncated for more information.
- Parameters:
event – The event to condition on.
- Returns:
The truncated distribution and the log-probability of the event.
- marginal_in_place(variables: typing_extensions.Iterable[random_events.variable.Variable]) typing_extensions.Optional[typing_extensions.Self]#
- log_conditional_in_place(point: typing_extensions.Dict[random_events.variable.Variable, typing_extensions.Any]) typing_extensions.Tuple[typing_extensions.Optional[typing_extensions.Self], float]#
- log_conditional(point: typing_extensions.Dict[random_events.variable.Variable, typing_extensions.Any]) typing_extensions.Tuple[typing_extensions.Optional[typing_extensions.Self], float]#
Calculate the truncated distribution P(*| point) and the probability of the event. Check the documentation of conditional for more information.
- Parameters:
point – A partial point to calculate the truncated distribution on.
- Returns:
The truncated distribution and the log-probability of the point.
- marginal(variables: typing_extensions.Iterable[random_events.variable.Variable]) typing_extensions.Optional[typing_extensions.Self]#
Calculate the marginal distribution of a set of variables.
- Parameters:
variables – The variables to calculate the marginal distribution on.
- Returns:
The marginal distribution over the variables.
- sample(amount: int) numpy.array#
Sample from the model.
- Parameters:
amount – The number of samples to draw.
- Returns:
The samples.
- moment(order: probabilistic_model.probabilistic_model.OrderType, center: probabilistic_model.probabilistic_model.CenterType) probabilistic_model.probabilistic_model.MomentType#
Calculate the (centralized) moment of the distribution.
\[\int_{-\infty}^{\infty} (x - center)^{order} pdf(x) dx\]Note
You can read more about queries of this class in Definition 22 in :cite:p:`choi2020probabilistic`_. [US20]
- Parameters:
order – The orders of the moment as a variable map for every continuous and integer variable.
center – The center of the moment as a variable map for every continuous and integer variable.
- Returns:
The moments of the variables in order.
- simplify() typing_extensions.Self#
Simplify the circuit inplace.
- property support: random_events.product_algebra.Event#
- Returns:
The support of the model.
- is_decomposable() bool#
Check if the whole circuit is decomposed.
A circuit is decomposed if all its product units are decomposed.
- Returns:
if the whole circuit is decomposed
- abstract __eq__(other: typing_extensions.Self)#
- empty_copy() typing_extensions.Self#
Create a copy of this circuit without any nodes. Only the parameters should be copied. This is used whenever a new circuit has to be created during inference.
- Returns:
A copy of this circuit without any subcircuits that is not in this units graph.
- __deepcopy__(memo=None) typing_extensions.Self#
Deep copy of the circuit.
- Parameters:
memo – A dictionary that is used to keep track of objects that have already been copied.
- Returns:
A deep copy of the circuit.
- to_json() typing_extensions.Dict[str, typing_extensions.Any]#
- classmethod parameters_from_json(data: typing_extensions.Dict[str, typing_extensions.Any]) typing_extensions.Self#
- classmethod _from_json(data: typing_extensions.Dict[str, typing_extensions.Any]) typing_extensions.Self#
Create a variable from a json dict. This method is called from the from_json method after the correct subclass is determined and should be overwritten by the respective subclass.
- Parameters:
data – The json dict
- Returns:
The deserialized object
- update_variables(new_variables: random_events.product_algebra.VariableMap)#
Update the variables of this unit and its descendants.
- Parameters:
new_variables – The new variables to set.
- is_deterministic() bool#
- Returns:
Whether, this circuit is deterministic or not.
- normalize()#
Normalize every sum node of this circuit in-place.
- add_edges_and_nodes_from_circuit(other: typing_extensions.Self)#
Add all edges and nodes from another circuit to this circuit.
- Parameters:
other – The other circuit to add.
- add_weighted_edges_from(ebunch_to_add, weight='log_weight', **attr)#
- subgraph_of(node: Unit) typing_extensions.Self#
Create a subgraph with a node as root.
- Parameters:
node – The root of the subgraph.
- Returns:
The subgraph.
- fill_node_colors(node_colors: typing_extensions.Dict[Unit, str])#
Fill the node colors for the structure plot.
- Parameters:
node_colors – The node colors to fill.
- bfs_layout(scale: float = 1.0, align: PlotAlignment = PlotAlignment.VERTICAL) typing_extensions.Dict[int, numpy.array]#
Generate a bfs layout for this circuit.
- Returns:
A dict mapping the node indices to 2d coordinates.
- plot_structure(node_colors: typing_extensions.Optional[typing_extensions.Dict[Unit, str]] = None, variable_name_offset=0.2, plot_inference=False, inference_representation: typing_extensions.Callable = lambda node: ..., inference_result_offset: float = -0.25)#
Plot the structure of the circuit using matplotlib.
- Parameters:
node_colors – Optionally specified colors of the node.
If nodes are not specified in the dictionary, they will be black. :param node_size: The size of the nodes :param variable_name_offset: The offset to the right of the variable names. :param plot_inference: If the results of the inference should be plotted. :param inference_representation: The representation of the inference results as a function from node to string. :param inference_result_offset: The vertical offset of the inference results.
- nodes_weights() dict#
- Returns:
dict with keys as nodes and values as list of all the log_weights for the node.
- replace_discrete_distribution_with_deterministic_sum()#
splits the distribution into sum unit with all the discrete possibilities as leaf.
- translate(translation: typing_extensions.Dict[random_events.variable.Variable, float])#
Translate the model in-place. Translation is done by adding the translation to the variable location influencing values. The translation can be viewed as what happens when you shift the numeric variables of the model by a constant vector.
- Parameters:
translation – The variable value pairs to translate the model by.
- scale(scale: typing_extensions.Dict[random_events.variable.Variable, float])#
Scale the model in-place. Scaling is done by multiplying the variable location influencing values. The scaling can be viewed as what happens when you multiply the numeric variables of the model by a constant vector.
- Parameters:
scaling – The variable value pairs to scale the model by.
- mount(other: Unit) typing_extensions.Dict[int, Unit]#
Mount another unit including its descendants. There will be no edge from self to other. This will also remove the nodes in other and their descendants from their circuit.
- Parameters:
other – The other unit to mount.
- __repr__()#
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.ShallowProbabilisticCircuit#
Bases:
ProbabilisticCircuitclass for PC in shallow form, sum unit as root followed by product units which only have leafs as children.
- classmethod from_probabilistic_circuit(probabilistic_circuit: ProbabilisticCircuit)#
Initialization function, to input a PC to create its shallow version.
- shallowing(node: Unit, presucc: Unit | None)#
This function transforms the PC into it shallow form, in place. This function uses recursion and need to be called on the root of the PC. :node: the Node in focus to be shallowed :presucc: the predecessor of the node of before shallowing.
- events_of_higher_density_product(other: typing_extensions.Self, own_pro_unit, other_pro_unit, tolerance: float = 1e-07)#
Construct E_p of a product unit in a shallow context. :own_pro_unit: product unit which is part of E_p :other: other product unit which is part of E_p :tolerance: float as how close to zero is zero, because of imprecision.
- events_of_higher_density_sum(other: typing_extensions.Self, tolerance: float = 1e-07)#
Construct E_p of a sum unit in a shallow context. :other: the other Root shallow PC node to create the E_p :tolerance: float as how close to zero is zero, because of imprecisions
- l1(other: typing_extensions.Self, tolerance: float = 1e-07) float#
The L1 metric between shallow Circuits are calculated. It is important, that before the shallowing the PC replace_discrete_distribution_with_deterministic_sum called on.´ :other: the other shallow PC which the L1 metric is calculated :tolerance: float as how close to zero is zero, because of imprecision for the Creation of E_p.
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.UnivariateLeaf#
Bases:
LeafUnitClass for Leaf units.
- property variable: random_events.variable.Variable#
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.UnivariateContinuousLeaf#
Bases:
UnivariateLeafClass for Leaf units.
- distribution: typing_extensions.Optional[probabilistic_model.distributions.ContinuousDistribution]#
The distribution contained in this leaf unit.
- __hash__#
- log_truncated_of_simple_event_in_place(event: random_events.product_algebra.SimpleEvent)#
- univariate_log_truncated_of_simple_event_in_place(event: random_events.interval.Interval)#
Condition this distribution on a simple event in-place but use sum units to create conditions on composite intervals. :param event: The simple event to condition on.
- class probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.UnivariateDiscreteLeaf#
Bases:
UnivariateLeafClass for Leaf units.
- distribution: typing_extensions.Optional[probabilistic_model.distributions.DiscreteDistribution]#
The distribution contained in this leaf unit.
- __hash__#
- as_deterministic_sum() SumUnit#
Convert this distribution to a deterministic sum unit that encodes the same distribution in-place. The result has as many children as the probability dictionary of this distribution. Each child encodes the value of the variable.
- Returns:
The deterministic sum unit that encodes the same distribution.
- classmethod from_mixture(mixture: ProbabilisticCircuit)#
Create a discrete distribution from a univariate mixture.
- Parameters:
mixture – The mixture to create the distribution from.
- Returns:
The discrete distribution.
- probabilistic_model.probabilistic_circuit.rx.probabilistic_circuit.leaf(distribution: probabilistic_model.distributions.UnivariateDistribution, probabilistic_circuit: typing_extensions.Optional[ProbabilisticCircuit] = None) UnivariateLeaf#
Factory that creates the correct leaf from a distribution.
- Returns:
The leaf.