def propose()

in src/lic/ppl/inference/proposer/abstract_single_site_single_step_proposer.py [0:0]
32 lines of code
3 McCabe index (conditional complexity)

    def propose(self, node: RVIdentifier, world: World) -> Tuple[Tensor, Tensor, Dict]:
        """
        Proposes a new value for the node.

        :param node: the node for which we'll need to propose a new value for.
        :param world: the world in which we'll propose a new value for node.
        :returns: a new proposed value for the node and the -ve log probability of
        proposing this new value and dict of auxiliary variables that needs to
        be passed to post process.
        """
        node_var = world.get_node_in_world_raise_error(node, False)

        (
            proposal_distribution_struct,
            auxiliary_variables,
        ) = self.get_proposal_distribution(node, node_var, world, {})
        node_var.proposal_distribution = proposal_distribution_struct
        proposal_distribution_struct = node_var.proposal_distribution
        proposal_distribution = proposal_distribution_struct.proposal_distribution
        requires_transform = proposal_distribution_struct.requires_transform
        requires_reshape = proposal_distribution_struct.requires_reshape

        LOGGER_PROPOSER.log(
            LogLevel.DEBUG_PROPOSER.value,
            "- Distribution: {pt}\n".format(pt=str(proposal_distribution_struct))
            + "- Auxiliary params: {pa}\n".format(pa=str(auxiliary_variables)),
        )

        new_value = proposal_distribution.sample()
        negative_proposal_log_update = (
            -1.0 * proposal_distribution.log_prob(new_value).sum()
        )

        if requires_reshape:
            new_value = new_value.reshape(node_var.transformed_value.shape)

        if requires_transform:
            new_value = node_var.inverse_transform_value(new_value)
            negative_proposal_log_update = (
                negative_proposal_log_update - node_var.jacobian
            )

        # pyre-fixme[7]: Expected Tuple[Tensor, Tensor, Dict[typing.Any,
        # typing.Any]] but got Tuple[typing.Any, typing.Union[Tensor, float],
        # Dict[typing.Any, typing.Any]].
        return (
            new_value,
            negative_proposal_log_update,
            auxiliary_variables,
        )