# Copyright 2020 The PyMC Developers

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import functools

import threading

import types

import warnings

from sys import modules

from typing import (

TYPE_CHECKING,

Any,

Callable,

Dict,

List,

Optional,

Sequence,

Tuple,

Type,

TypeVar,

Union,

cast,

)

import aesara

import aesara.sparse as sparse

import aesara.tensor as at

import numpy as np

import scipy.sparse as sps

from aesara.compile.sharedvalue import SharedVariable

from aesara.graph.basic import Constant, Variable, graph_inputs

from aesara.graph.fg import FunctionGraph

from aesara.scalar import Cast

from aesara.tensor.elemwise import Elemwise

from aesara.tensor.random.op import RandomVariable

from aesara.tensor.random.rewriting import local_subtensor_rv_lift

from aesara.tensor.sharedvar import ScalarSharedVariable

from aesara.tensor.var import TensorConstant, TensorVariable

from pymc.aesaraf import (

PointFunc,

SeedSequenceSeed,

compile_pymc,

convert_observed_data,

gradient,

hessian,

inputvars,

replace_rvs_by_values,

)

from pymc.blocking import DictToArrayBijection, RaveledVars

from pymc.data import GenTensorVariable, Minibatch

from pymc.distributions.logprob import _joint_logp

from pymc.distributions.transforms import _default_transform

from pymc.exceptions import ImputationWarning, SamplingError, ShapeError, ShapeWarning

from pymc.initial_point import make_initial_point_fn

from pymc.util import (

UNSET,

WithMemoization,

_add_future_warning_tag,

get_transformed_name,

get_value_vars_from_user_vars,

get_var_name,

treedict,

treelist,

)

from pymc.vartypes import continuous_types, discrete_types, typefilter

__all__ = [

"Model",

"modelcontext",

"Deterministic",

"Potential",

"set_data",

"Point",

"compile_fn",

]

class InstanceMethod:

"""Class for hiding references to instance methods so they can be pickled.

>>> self.method = InstanceMethod(some_object, 'method_name')

"""

def __init__(self, obj, method_name):

self.obj = obj

self.method_name = method_name

def __call__(self, *args, **kwargs):

return getattr(self.obj, self.method_name)(*args, **kwargs)

def incorporate_methods(source, destination, methods, wrapper=None, override=False):

"""

Add attributes to a destination object which point to

methods from from a source object.

Parameters

----------

source: object

The source object containing the methods.

destination: object

The destination object for the methods.

methods: list of str

Names of methods to incorporate.

wrapper: function

An optional function to allow the source method to be

wrapped. Should take the form my_wrapper(source, method_name)

and return a single value.

override: bool

If the destination object already has a method/attribute

an AttributeError will be raised if override is False (the default).

"""

for method in methods:

if hasattr(destination, method) and not override:

raise AttributeError(

f"Cannot add method {method!r}" + "to destination object as it already exists. "

"To prevent this error set 'override=True'."

)

if hasattr(source, method):

if wrapper is None:

setattr(destination, method, getattr(source, method))

else:

setattr(destination, method, wrapper(source, method))

else:

setattr(destination, method, None)

T = TypeVar("T", bound="ContextMeta")

class ContextMeta(type):

"""Functionality for objects that put themselves in a context using

the `with` statement.

"""

def __new__(cls, name, bases, dct, **kwargs): # pylint: disable=unused-argument

"Add __enter__ and __exit__ methods to the class."

def __enter__(self):

self.__class__.context_class.get_contexts().append(self)

# self._aesara_config is set in Model.__new__

self._config_context = None

if hasattr(self, "_aesara_config"):

self._config_context = aesara.config.change_flags(**self._aesara_config)

self._config_context.__enter__()

return self

def __exit__(self, typ, value, traceback): # pylint: disable=unused-argument

self.__class__.context_class.get_contexts().pop()

# self._aesara_config is set in Model.__new__

if self._config_context:

self._config_context.__exit__(typ, value, traceback)

dct[__enter__.__name__] = __enter__

dct[__exit__.__name__] = __exit__

# We strip off keyword args, per the warning from

# StackExchange:

# DO NOT send "**kwargs" to "type.__new__". It won't catch them and

# you'll get a "TypeError: type() takes 1 or 3 arguments" exception.

return super().__new__(cls, name, bases, dct)

# FIXME: is there a more elegant way to automatically add methods to the class that

# are instance methods instead of class methods?

def __init__(

cls, name, bases, nmspc, context_class: Optional[Type] = None, **kwargs

): # pylint: disable=unused-argument

"""Add ``__enter__`` and ``__exit__`` methods to the new class automatically."""

if context_class is not None:

cls._context_class = context_class

super().__init__(name, bases, nmspc)

def get_context(cls, error_if_none=True) -> Optional[T]:

"""Return the most recently pushed context object of type ``cls``

on the stack, or ``None``. If ``error_if_none`` is True (default),

raise a ``TypeError`` instead of returning ``None``."""

try:

candidate: Optional[T] = cls.get_contexts()[-1]

except IndexError as e:

# Calling code expects to get a TypeError if the entity

# is unfound, and there's too much to fix.

if error_if_none:

raise TypeError(f"No {cls} on context stack")

return None

return candidate

def get_contexts(cls) -> List[T]:

"""Return a stack of context instances for the ``context_class``

of ``cls``."""

# This lazily creates the context class's contexts

# thread-local object, as needed. This seems inelegant to me,

# but since the context class is not guaranteed to exist when

# the metaclass is being instantiated, I couldn't figure out a

# better way. [2019/10/11:rpg]

# no race-condition here, contexts is a thread-local object

# be sure not to override contexts in a subclass however!

context_class = cls.context_class

assert isinstance(

context_class, type

), f"Name of context class, {context_class} was not resolvable to a class"

if not hasattr(context_class, "contexts"):

context_class.contexts = threading.local()

contexts = context_class.contexts

if not hasattr(contexts, "stack"):

contexts.stack = []

return contexts.stack

# the following complex property accessor is necessary because the

# context_class may not have been created at the point it is

# specified, so the context_class may be a class *name* rather

# than a class.

@property

def context_class(cls) -> Type:

def resolve_type(c: Union[Type, str]) -> Type:

if isinstance(c, str):

c = getattr(modules[cls.__module__], c)

if isinstance(c, type):

return c

raise ValueError(f"Cannot resolve context class {c}")

assert cls is not None

if isinstance(cls._context_class, str):

cls._context_class = resolve_type(cls._context_class)

if not isinstance(cls._context_class, (str, type)):

raise ValueError(

f"Context class for {cls.__name__}, {cls._context_class}, is not of the right type"

)

return cls._context_class

# Inherit context class from parent

def __init_subclass__(cls, **kwargs):

super().__init_subclass__(**kwargs)

cls.context_class = super().context_class

# Initialize object in its own context...

# Merged from InitContextMeta in the original.

def __call__(cls, *args, **kwargs):

instance = cls.__new__(cls, *args, **kwargs)

with instance: # appends context

instance.__init__(*args, **kwargs)

return instance

def modelcontext(model: Optional["Model"]) -> "Model":

"""

Return the given model or, if none was supplied, try to find one in

the context stack.

"""

if model is None:

model = Model.get_context(error_if_none=False)

if model is None:

# TODO: This should be a ValueError, but that breaks

# ArviZ (and others?), so might need a deprecation.

raise TypeError("No model on context stack.")

return model

class ValueGradFunction:

"""Create an Aesara function that computes a value and its gradient.

Parameters

----------

costs: list of Aesara variables

We compute the weighted sum of the specified Aesara values, and the gradient

of that sum. The weights can be specified with `ValueGradFunction.set_weights`.

grad_vars: list of named Aesara variables or None

The arguments with respect to which the gradient is computed.

extra_vars_and_values: dict of Aesara variables and their initial values

Other arguments of the function that are assumed constant and their

values. They are stored in shared variables and can be set using

`set_extra_values`.

dtype: str, default=aesara.config.floatX

The dtype of the arrays.

casting: {'no', 'equiv', 'save', 'same_kind', 'unsafe'}, default='no'

Casting rule for casting `grad_args` to the array dtype.

See `numpy.can_cast` for a description of the options.

Keep in mind that we cast the variables to the array *and*

back from the array dtype to the variable dtype.

compute_grads: bool, default=True

If False, return only the logp, not the gradient.

kwargs

Extra arguments are passed on to `aesara.function`.

Attributes

----------

profile: Aesara profiling object or None

The profiling object of the Aesara function that computes value and

gradient. This is None unless `profile=True` was set in the

kwargs.

"""

def __init__(

self,

costs,

grad_vars,

extra_vars_and_values=None,

*,

dtype=None,

casting="no",

compute_grads=True,

**kwargs,

):

if extra_vars_and_values is None:

extra_vars_and_values = {}

names = [arg.name for arg in grad_vars + list(extra_vars_and_values.keys())]

if any(name is None for name in names):

raise ValueError("Arguments must be named.")

if len(set(names)) != len(names):

raise ValueError("Names of the arguments are not unique.")

self._grad_vars = grad_vars

self._extra_vars = list(extra_vars_and_values.keys())

self._extra_var_names = {var.name for var in extra_vars_and_values.keys()}

if dtype is None:

dtype = aesara.config.floatX

self.dtype = dtype

self._n_costs = len(costs)

if self._n_costs == 0:

raise ValueError("At least one cost is required.")

weights = np.ones(self._n_costs - 1, dtype=self.dtype)

self._weights = aesara.shared(weights, "__weights")

cost = costs[0]

for i, val in enumerate(costs[1:]):

if cost.ndim > 0 or val.ndim > 0:

raise ValueError("All costs must be scalar.")

cost = cost + self._weights[i] * val

self._extra_are_set = False

for var in self._grad_vars:

if not np.can_cast(var.dtype, self.dtype, casting):

raise TypeError(

f"Invalid dtype for variable {var.name}. Can not "

f"cast to {self.dtype} with casting rule {casting}."

)

if not np.issubdtype(var.dtype, np.floating):

raise TypeError(

f"Invalid dtype for variable {var.name}. Must be "

f"floating point but is {var.dtype}."

)

givens = []

self._extra_vars_shared = {}

for var, value in extra_vars_and_values.items():

shared = aesara.shared(

value, var.name + "_shared__", shape=[1 if s == 1 else None for s in value.shape]

)

self._extra_vars_shared[var.name] = shared

givens.append((var, shared))

if compute_grads:

grads = aesara.grad(cost, grad_vars, disconnected_inputs="ignore")

for grad_wrt, var in zip(grads, grad_vars):

grad_wrt.name = f"{var.name}_grad"

outputs = [cost] + grads

else:

outputs = [cost]

inputs = grad_vars

self._aesara_function = compile_pymc(inputs, outputs, givens=givens, **kwargs)

def set_weights(self, values):

if values.shape != (self._n_costs - 1,):

raise ValueError("Invalid shape. Must be (n_costs - 1,).")

self._weights.set_value(values)

def set_extra_values(self, extra_vars):

self._extra_are_set = True

for var in self._extra_vars:

self._extra_vars_shared[var.name].set_value(extra_vars[var.name])

def get_extra_values(self):

if not self._extra_are_set:

raise ValueError("Extra values are not set.")

return {var.name: self._extra_vars_shared[var.name].get_value() for var in self._extra_vars}

def __call__(self, grad_vars, grad_out=None, extra_vars=None):

if extra_vars is not None:

self.set_extra_values(extra_vars)

if not self._extra_are_set:

raise ValueError("Extra values are not set.")

if isinstance(grad_vars, RaveledVars):

grad_vars = list(DictToArrayBijection.rmap(grad_vars).values())

cost, *grads = self._aesara_function(*grad_vars)

if grads:

grads_raveled = DictToArrayBijection.map(

{v.name: gv for v, gv in zip(self._grad_vars, grads)}

)

if grad_out is None:

return cost, grads_raveled.data

else:

np.copyto(grad_out, grads_raveled.data)

return cost

else:

return cost

@property

def profile(self):

"""Profiling information of the underlying Aesara function."""

return self._aesara_function.profile

class Model(WithMemoization, metaclass=ContextMeta):

"""Encapsulates the variables and likelihood factors of a model.

Model class can be used for creating class based models. To create

a class based model you should inherit from :class:`~pymc.Model` and

override the `__init__` method with arbitrary definitions (do not

forget to call base class :meth:`pymc.Model.__init__` first).

Parameters

----------

name: str

name that will be used as prefix for names of all random

variables defined within model

check_bounds: bool

Ensure that input parameters to distributions are in a valid

range. If your model is built in a way where you know your

parameters can only take on valid values you can set this to

False for increased speed. This should not be used if your model

contains discrete variables.

Examples

--------

How to define a custom model

.. code-block:: python

class CustomModel(Model):

# 1) override init

def __init__(self, mean=0, sigma=1, name=''):

# 2) call super's init first, passing model and name

# to it name will be prefix for all variables here if

# no name specified for model there will be no prefix

super().__init__(name, model)

# now you are in the context of instance,

# `modelcontext` will return self you can define

# variables in several ways note, that all variables

# will get model's name prefix

# 3) you can create variables with the register_rv method

self.register_rv(Normal.dist(mu=mean, sigma=sigma), 'v1', initval=1)

# this will create variable named like '{name::}v1'

# and assign attribute 'v1' to instance created

# variable can be accessed with self.v1 or self['v1']

# 4) this syntax will also work as we are in the

# context of instance itself, names are given as usual

Normal('v2', mu=mean, sigma=sigma)

# something more complex is allowed, too

half_cauchy = HalfCauchy('sigma', beta=10, initval=1.)

Normal('v3', mu=mean, sigma=half_cauchy)

# Deterministic variables can be used in usual way

Deterministic('v3_sq', self.v3 ** 2)

# Potentials too

Potential('p1', at.constant(1))

# After defining a class CustomModel you can use it in several

# ways

# I:

# state the model within a context

with Model() as model:

CustomModel()

# arbitrary actions

# II:

# use new class as entering point in context

with CustomModel() as model:

Normal('new_normal_var', mu=1, sigma=0)

# III:

# just get model instance with all that was defined in it

model = CustomModel()

# IV:

# use many custom models within one context

with Model() as model:

CustomModel(mean=1, name='first')

CustomModel(mean=2, name='second')

# variables inside both scopes will be named like `first::*`, `second::*`

"""

if TYPE_CHECKING:

def __enter__(self: "Model") -> "Model":

...

def __exit__(self: "Model", *exc: Any) -> bool:

...

def __new__(cls, *args, **kwargs):

# resolves the parent instance

instance = super().__new__(cls)

if kwargs.get("model") is not None:

instance._parent = kwargs.get("model")

else:

instance._parent = cls.get_context(error_if_none=False)

instance._aesara_config = kwargs.get("aesara_config", {})

return instance

@staticmethod

def _validate_name(name):

if name.endswith(":"):

raise KeyError("name should not end with `:`")

return name

def __init__(

self,

name="",

coords=None,

check_bounds=True,

*,

aesara_config=None,

model=None,

):

del aesara_config, model # used in __new__

self.name = self._validate_name(name)

self.check_bounds = check_bounds

if self.parent is not None:

self.named_vars = treedict(parent=self.parent.named_vars)

self.named_vars_to_dims = treedict(parent=self.parent.named_vars_to_dims)

self.values_to_rvs = treedict(parent=self.parent.values_to_rvs)

self.rvs_to_values = treedict(parent=self.parent.rvs_to_values)

self.rvs_to_transforms = treedict(parent=self.parent.rvs_to_transforms)

self.rvs_to_total_sizes = treedict(parent=self.parent.rvs_to_total_sizes)

self.rvs_to_initial_values = treedict(parent=self.parent.rvs_to_initial_values)

self.free_RVs = treelist(parent=self.parent.free_RVs)

self.observed_RVs = treelist(parent=self.parent.observed_RVs)

self.deterministics = treelist(parent=self.parent.deterministics)

self.potentials = treelist(parent=self.parent.potentials)

self._coords = self.parent._coords

self._dim_lengths = self.parent._dim_lengths

else:

self.named_vars = treedict()

self.named_vars_to_dims = treedict()

self.values_to_rvs = treedict()

self.rvs_to_values = treedict()

self.rvs_to_transforms = treedict()

self.rvs_to_total_sizes = treedict()

self.rvs_to_initial_values = treedict()

self.free_RVs = treelist()

self.observed_RVs = treelist()

self.deterministics = treelist()

self.potentials = treelist()

self._coords = {}

self._dim_lengths = {}

self.add_coords(coords)

from pymc.printing import str_for_model

self.str_repr = types.MethodType(str_for_model, self)

self._repr_latex_ = types.MethodType(

functools.partial(str_for_model, formatting="latex"), self

)

@property

def model(self):

return self

@property

def parent(self):

return self._parent

@property

def root(self):

model = self

while not model.isroot:

model = model.parent

return model

@property

def isroot(self):

return self.parent is None

def logp_dlogp_function(self, grad_vars=None, tempered=False, **kwargs):

"""Compile an Aesara function that computes logp and gradient.

Parameters

----------

grad_vars: list of random variables, optional

Compute the gradient with respect to those variables. If None,

use all free random variables of this model.

tempered: bool

Compute the tempered logp `free_logp + alpha * observed_logp`.

`alpha` can be changed using `ValueGradFunction.set_weights([alpha])`.

"""

if grad_vars is None:

grad_vars = self.continuous_value_vars

else:

grad_vars = get_value_vars_from_user_vars(grad_vars, self)

for i, var in enumerate(grad_vars):

if var.dtype not in continuous_types:

raise ValueError(f"Can only compute the gradient of continuous types: {var}")

if tempered:

costs = [self.varlogp, self.datalogp]

else:

costs = [self.logp()]

input_vars = {i for i in graph_inputs(costs) if not isinstance(i, Constant)}

ip = self.initial_point(0)

extra_vars_and_values = {

var: ip[var.name]

for var in self.value_vars

if var in input_vars and var not in grad_vars

}

return ValueGradFunction(costs, grad_vars, extra_vars_and_values, **kwargs)

def compile_logp(

self,

vars: Optional[Union[Variable, Sequence[Variable]]] = None,

jacobian: bool = True,

sum: bool = True,

) -> PointFunc:

"""Compiled log probability density function.

Parameters

----------

vars: list of random variables or potential terms, optional

Compute the gradient with respect to those variables. If None, use all

free and observed random variables, as well as potential terms in model.

jacobian:

Whether to include jacobian terms in logprob graph. Defaults to True.

sum:

Whether to sum all logp terms or return elemwise logp for each variable.

Defaults to True.

"""

return self.model.compile_fn(self.logp(vars=vars, jacobian=jacobian, sum=sum))

def compile_dlogp(

self,

vars: Optional[Union[Variable, Sequence[Variable]]] = None,

jacobian: bool = True,

) -> PointFunc:

"""Compiled log probability density gradient function.

Parameters

----------

vars: list of random variables or potential terms, optional

Compute the gradient with respect to those variables. If None, use all

free and observed random variables, as well as potential terms in model.

jacobian:

Whether to include jacobian terms in logprob graph. Defaults to True.

"""

return self.model.compile_fn(self.dlogp(vars=vars, jacobian=jacobian))

def compile_d2logp(

self,

vars: Optional[Union[Variable, Sequence[Variable]]] = None,

jacobian: bool = True,

) -> PointFunc:

"""Compiled log probability density hessian function.

Parameters

----------

vars: list of random variables or potential terms, optional

Compute the gradient with respect to those variables. If None, use all

free and observed random variables, as well as potential terms in model.

jacobian:

Whether to include jacobian terms in logprob graph. Defaults to True.

"""

return self.model.compile_fn(self.d2logp(vars=vars, jacobian=jacobian))

def logp(

self,

vars: Optional[Union[Variable, Sequence[Variable]]] = None,

jacobian: bool = True,

sum: bool = True,

) -> Union[Variable, List[Variable]]:

"""Elemwise log-probability of the model.

Parameters

----------

vars: list of random variables or potential terms, optional

Compute the gradient with respect to those variables. If None, use all

free and observed random variables, as well as potential terms in model.

jacobian:

Whether to include jacobian terms in logprob graph. Defaults to True.

sum:

Whether to sum all logp terms or return elemwise logp for each variable.

Defaults to True.

Returns

-------

Logp graph(s)

"""

varlist: List[TensorVariable]

if vars is None:

varlist = self.free_RVs + self.observed_RVs + self.potentials

elif not isinstance(vars, (list, tuple)):

varlist = [vars]

else:

varlist = cast(List[TensorVariable], vars)

# We need to separate random variables from potential terms, and remember their

# original order so that we can merge them together in the same order at the end

rvs = []

potentials = []

rv_order, potential_order = [], []

for i, var in enumerate(varlist):

rv = self.values_to_rvs.get(var, var)

if rv in self.basic_RVs:

rvs.append(rv)

rv_order.append(i)

else:

if var in self.potentials:

potentials.append(var)

potential_order.append(i)

else:

raise ValueError(

f"Requested variable {var} not found among the model variables"

)

rv_logps: List[TensorVariable] = []

if rvs:

rv_logps = _joint_logp(

rvs=rvs,

rvs_to_values=self.rvs_to_values,

rvs_to_transforms=self.rvs_to_transforms,

rvs_to_total_sizes=self.rvs_to_total_sizes,

jacobian=jacobian,

)

assert isinstance(rv_logps, list)

# Replace random variables by their value variables in potential terms

potential_logps = []

if potentials:

potential_logps = self.replace_rvs_by_values(potentials)

logp_factors = [None] * len(varlist)

for logp_order, logp in zip((rv_order + potential_order), (rv_logps + potential_logps)):

logp_factors[logp_order] = logp

if not sum:

return logp_factors

logp_scalar = at.sum([at.sum(factor) for factor in logp_factors])

logp_scalar_name = "__logp" if jacobian else "__logp_nojac"

if self.name:

logp_scalar_name = f"{logp_scalar_name}_{self.name}"

logp_scalar.name = logp_scalar_name

return logp_scalar

def dlogp(

self,

vars: Optional[Union[Variable, Sequence[Variable]]] = None,

jacobian: bool = True,

) -> Variable:

"""Gradient of the models log-probability w.r.t. ``vars``.

Parameters

----------

vars: list of random variables or potential terms, optional

Compute the gradient with respect to those variables. If None, use all

free and observed random variables, as well as potential terms in model.

jacobian:

Whether to include jacobian terms in logprob graph. Defaults to True.

Returns

-------

dlogp graph

"""

if vars is None:

value_vars = None

else:

if not isinstance(vars, (list, tuple)):

vars = [vars]

value_vars = []

for i, var in enumerate(vars):

value_var = self.rvs_to_values.get(var)

if value_var is not None:

value_vars.append(value_var)

else:

raise ValueError(

f"Requested variable {var} not found among the model variables"

)

cost = self.logp(jacobian=jacobian)

return gradient(cost, value_vars)

def d2logp(

self,

vars: Optional[Union[Variable, Sequence[Variable]]] = None,

jacobian: bool = True,

) -> Variable:

"""Hessian of the models log-probability w.r.t. ``vars``.

Parameters

----------

vars: list of random variables or potential terms, optional

Compute the gradient with respect to those variables. If None, use all

free and observed random variables, as well as potential terms in model.

jacobian:

Whether to include jacobian terms in logprob graph. Defaults to True.

Returns

-------

d²logp graph

"""

if vars is None:

value_vars = None

else:

if not isinstance(vars, (list, tuple)):

vars = [vars]

value_vars = []

for i, var in enumerate(vars):

value_var = self.rvs_to_values.get(var)

if value_var is not None:

value_vars.append(value_var)

else:

raise ValueError(

f"Requested variable {var} not found among the model variables"

)

cost = self.logp(jacobian=jacobian)

return hessian(cost, value_vars)

@property

def datalogp(self) -> Variable:

"""Aesara scalar of log-probability of the observed variables and

potential terms"""

return self.observedlogp + self.potentiallogp

@property

def varlogp(self) -> Variable:

"""Aesara scalar of log-probability of the unobserved random variables

(excluding deterministic)."""

return self.logp(vars=self.free_RVs)

@property

def varlogp_nojac(self) -> Variable:

"""Aesara scalar of log-probability of the unobserved random variables

(excluding deterministic) without jacobian term."""

return self.logp(vars=self.free_RVs, jacobian=False)

@property

def observedlogp(self) -> Variable:

"""Aesara scalar of log-probability of the observed variables"""

return self.logp(vars=self.observed_RVs)

@property

def potentiallogp(self) -> Variable:

"""Aesara scalar of log-probability of the Potential terms"""

# Convert random variables in Potential expression into their log-likelihood

# inputs and apply their transforms, if any

potentials = self.replace_rvs_by_values(self.potentials)

if potentials:

return at.sum([at.sum(factor) for factor in potentials])

else:

return at.constant(0.0)

@property

def value_vars(self):

"""List of unobserved random variables used as inputs to the model's

log-likelihood (which excludes deterministics).

"""

return [self.rvs_to_values[v] for v in self.free_RVs]

@property

def unobserved_value_vars(self):

"""List of all random variables (including untransformed projections),

as well as deterministics used as inputs and outputs of the model's

log-likelihood graph

"""

vars = []

transformed_rvs = []

for rv in self.free_RVs:

value_var = self.rvs_to_values[rv]

transform = self.rvs_to_transforms[rv]

if transform is not None:

transformed_rvs.append(rv)

vars.append(value_var)

# Remove rvs from untransformed values graph

untransformed_vars = self.replace_rvs_by_values(transformed_rvs)

# Remove rvs from deterministics graph

deterministics = self.replace_rvs_by_values(self.deterministics)

return vars + untransformed_vars + deterministics

@property

def disc_vars(self):

warnings.warn(

"Model.disc_vars has been deprecated. Use Model.discrete_value_vars instead.",

FutureWarning,

)

return self.discrete_value_vars

@property

def discrete_value_vars(self):

"""All the discrete value variables in the model"""

return list(typefilter(self.value_vars, discrete_types))

@property

def cont_vars(self):

warnings.warn(

"Model.cont_vars has been deprecated. Use Model.continuous_value_vars instead.",

FutureWarning,

)

return self.continuous_value_vars

@property

def continuous_value_vars(self):

"""All the continuous value variables in the model"""

return list(typefilter(self.value_vars, continuous_types))

@property

def basic_RVs(self):

"""List of random variables the model is defined in terms of

(which excludes deterministics).

These are the actual random variable terms that make up the

"sample-space" graph (i.e. you can sample these graphs by compiling them

with `aesara.function`). If you want the corresponding log-likelihood terms,

use `model.value_vars` instead.

"""

return self.free_RVs + self.observed_RVs

@property

def unobserved_RVs(self):

"""List of all random variables, including deterministic ones.

These are the actual random variable terms that make up the

"sample-space" graph (i.e. you can sample these graphs by compiling them

with `aesara.function`). If you want the corresponding log-likelihood terms,

use `var.unobserved_value_vars` instead.

"""

return self.free_RVs + self.deterministics

@property

def RV_dims(self) -> Dict[str, Tuple[Union[str, None], ...]]:

"""Tuples of dimension names for specific model variables.

Entries in the tuples may be ``None``, if the RV dimension was not given a name.

"""

warnings.warn(

"Model.RV_dims is deprecated. User Model.named_vars_to_dims instead.",

FutureWarning,

)

return self.named_vars_to_dims

@property

def coords(self) -> Dict[str, Union[Tuple, None]]:

"""Coordinate values for model dimensions."""

return self._coords

@property

def dim_lengths(self) -> Dict[str, Variable]:

"""The symbolic lengths of dimensions in the model.

The values are typically instances of ``TensorVariable`` or ``ScalarSharedVariable``.

"""

return self._dim_lengths

def shape_from_dims(self, dims):

shape = []

if len(set(dims)) != len(dims):

raise ValueError("Can not contain the same dimension name twice.")

for dim in dims:

if dim not in self.coords:

raise ValueError(

f"Unknown dimension name '{dim}'. All dimension "

"names must be specified in the `coords` "