elicito

A Python package for learning prior distributions based on expert knowledge

Modules:

Name	Description
elicit	setting-up the elicitation method with Elicit
exceptions	Exceptions that are used throughout
initialization	Hyperparameter initialization for parametric prior
losses	Built-in loss functions
networks	setup network argument of Elicit class
optimization	Defines the optimization algorithm
plots	plotting helpers
simulations	Simulations from prior and model
targets	Specification of target quantities and elicited statistics
types	specification of custom types
utils	helper functions for setting up the Elicit object

Classes:

Name	Description
Elicit	Configure the elicitation method

Functions:

Name	Description
hyper	Specify prior hyperparameters.
initializer	Initialize hyperparameter values
model	Specify the generative model.
optimizer	Specify optimizer and its settings for SGD.
parameter	Specify model parameters.
target	Specify target quantity and corresponding elicitation technique.
trainer	Specify training settings for learning the prior distribution(s).

Elicit

Configure the elicitation method

Methods:

Name	Description
__init__	Specify the elicitation method
__repr__	Return a readable representation of the object.
__str__	Return a readable summary of the object.
fit	Fit the eliobj and learn prior distributions.
save	Save data on disk
update	Update attributes of Elicit object
workflow	Build the main workflow of the prior elicitation method.

Source code in src/elicito/__init__.py

class Elicit:
    """
    Configure the elicitation method
    """

    def __init__(  # noqa: PLR0913
        self,
        model: dict[str, Any],
        parameters: list[Parameter],
        targets: list[Target],
        expert: ExpertDict,
        trainer: Trainer,
        optimizer: dict[str, Any],
        network: NFDict | None = None,
        initializer: Initializer | None = None,
        meta_settings: MetaSettings = meta_settings(),
    ):
        """
        Specify the elicitation method

        Parameters
        ----------
        model
            specification of generative model using [`model`][elicito.elicit.model].

        parameters
            list of model parameters specified with [`parameter`][elicito.elicit.parameter].

        targets
            list of target quantities specified with [`target`][elicito.elicit.target].

        expert
            provide input data from expert or simulate data from oracle with
            either the ``data`` or ``simulator`` method of the
            [`Expert`][elicito.elicit.Expert] module.

        trainer
            specification of training settings and meta-information for
            workflow using [`trainer`][elicito.elicit.trainer].

        optimizer
            specification of SGD optimizer and its settings using
            [`optimizer`][elicito.elicit.optimizer].

        network
            specification of neural network using a method implemented in
            [`networks`][elicito.networks].
            Only required for ``deep_prior`` method.

        initializer
            specification of initialization settings using
            [`initializer`][elicito.elicit.initializer].
            Only required for ``parametric_prior`` method.

        meta_settings
            dictionary of meta settings for the elicitation workflow. See
            [`meta_settings`][elicito.types.MetaSettings] for available options.

        Returns
        -------
        eliobj :
            specification of all settings to run the elicitation workflow and
            fit the eliobj.

        Raises
        ------
        AssertionError
            ``expert`` data are not in the required format. Correct specification of
            keys can be checked using
            [`get_expert_datformat`][elicito.utils.get_expert_datformat]

            Dimensionality of ``ground_truth`` for simulating expert data, must be
            the same as the number of model parameters.

        ValueError
            if ``method = "deep_prior"``, ``network`` can't be None and ``initialization``
            should be None.

            if ``method="deep_prior"``, ``num_params`` as specified in the ``network_specs``
            argument (section: network) does not match the number of parameters
            specified in the parameters section.

            if ``method="parametric_prior"``, ``network`` should be None and
            ``initialization`` can't be None.

            if ``method ="parametric_prior" and multiple hyperparameter have
            the same name but are not shared by setting ``shared = True``."

            if ``hyperparams`` is specified in section ``initializer`` and a
            hyperparameter name (key in hyperparams dict) does not match any
            hyperparameter name specified in [`hyper`][elicito.elicit.hyper].

        NotImplementedError
            [network] Currently only the standard normal distribution is
            implemented as base distribution. See
            [GitHub issue #35](https://github.com/florence-bockting/prior_elicitation/issues/35).

        """  # noqa: E501
        _checks.check_elicit(
            model,
            parameters,
            targets,
            expert,
            trainer,
            optimizer,
            network,
            initializer,
            meta_settings,
        )

        self.model = model
        self.parameters = parameters
        self.targets = targets
        self.expert = expert
        self.trainer = trainer
        self.optimizer = optimizer
        self.network = network
        self.initializer = initializer
        self.meta_settings = meta_settings

        self.temp_history: list[dict[str, Any]] = []
        self.temp_results: list[dict[str, Any]] = []

        # helper for subsequent checks
        self.dry_run = self.meta_settings["dry_run"]
        # overwrite global seed
        globals()["SEED"] = self.trainer["seed"]

        # set seed
        tf.random.set_seed(SEED)

        if self.dry_run:
            (
                self.dry_elicits,
                self.dry_priors,
                self.dry_modelsims,
                self.dry_targets,
                self.dry_prior_model,
            ) = utils.dry_run(
                self.model,
                self.parameters,
                self.targets,
                self.trainer,
                self.initializer,  # type: ignore
                self.network,
            )

    def __str__(self) -> str:  # noqa: PLR0912
        """Return a readable summary of the object."""
        # fitted eliobj with shape information
        try:
            self.results  # type: ignore
        except AttributeError:
            if len(self.temp_results) != 0:
                targets_str = "\n".join(
                    f"  - {k1} {tuple(self.temp_results[0]['target_quantities'][k1].shape)} -> "  # noqa: E501
                    f"{k2} {tuple(self.temp_results[0]['elicited_statistics'][k2].shape)}"  # noqa: E501
                    for k1, k2 in zip(
                        self.temp_results[0]["target_quantities"],
                        self.temp_results[0]["elicited_statistics"],
                    )
                )
            # unfitted eliobj with shape information due to dry run
            elif self.dry_run:
                targets_str = "\n".join(
                    f"  - {k1} {tuple(self.dry_targets[k1].shape)} -> "
                    f"{k2} {tuple(self.dry_elicits[k2].shape)}"
                    for k1, k2 in zip(self.dry_targets, self.dry_elicits)
                )
            # unfitted eliobj without shape information
            else:
                targets_str = "\n".join(
                    f"  - {self.targets[tar]['name']} -> {eli}"
                    for tar, eli in zip(
                        range(len(self.targets)),
                        utils.get_expert_datformat(self.targets),
                    )
                )
        else:
            target_list = list(self.results.target_quantity.data_vars.keys())  # type: ignore
            elicit_list = list(self.results.elicited_summary.data_vars.keys())  # type: ignore

            targets_str = "\n".join(
                f"  - {k1} {self.results.target_quantity[k1].shape[1:]} -> "  # type: ignore
                f"{k2} {self.results.elicited_summary[k2].shape[1:]}"  # type: ignore
                for k1, k2 in zip(target_list, elicit_list)
            )

        opt_name = self.optimizer["optimizer"].__name__
        opt_lr = self.optimizer["learning_rate"]

        get_num_hyperpar: int | str
        try:
            self.results  # type: ignore
        except AttributeError:
            pass
        else:
            if self.trainer["method"] == "deep_prior":
                get_num_hyperpar = utils.compute_num_weights(
                    self.results[0]["num_NN_weights"]  # type: ignore
                )

        if (self.trainer["method"] == "deep_prior") and (self.dry_run):
            trainable_vars = self.dry_prior_model.init_priors.trainable_variables
            num_NN_weights = [
                trainable_vars[i].shape for i in range(len(trainable_vars))
            ]
            get_num_hyperpar = utils.compute_num_weights(num_NN_weights)

        elif self.trainer["method"] == "parametric_prior":
            get_num_hyperpar = sum(
                [
                    len(self.parameters[i]["hyperparams"])
                    for i in range(len(self.parameters))
                ]
            )
        else:
            get_num_hyperpar = "?"
            print("Number of hyperparameter in model can't be computed.")

        summary = (
            f"Model hyperparameters: {get_num_hyperpar}\n"
            f"Model parameters: {len(self.parameters)}\n"
            "Targets -> Elicited summaries (loss components)"
            f"{': ' + str(len(self.dry_elicits)) if self.dry_run else ''}\n"
            f"{targets_str}\n"
            f"Prior samples: {self.trainer['num_samples']}"
            f"{' ' + str(tuple(self.dry_priors.shape)) if self.dry_run else ''}\n"
            f"Batch size: {self.trainer['B']}\n"
            f"Epochs: {self.trainer['epochs']}\n"
            f"Method: {self.trainer['method']}\n"
            f"Seed: {self.trainer['seed']}\n"
            f"Optimizer: {opt_name}(lr={opt_lr})\n"
        )
        if self.trainer["method"] == "parametric_prior":
            if self.initializer is not None:
                summary += (
                    f"Initializer: (method: {self.initializer['method']}, "
                    f"iterations: {self.initializer['iterations']})\n"
                )
            else:
                summary += "Initializer: None\n"
        elif self.network is not None:
            summary += f"Network: {self.network['inference_network'].__name__}\n"
        else:
            summary += "Network: None\n"

        return summary

    def __repr__(self) -> str:
        """Return a readable representation of the object."""
        return self.__str__()

    def fit(
        self,
        overwrite: bool = False,
        parallel: Parallel | None = None,
    ) -> None:
        """
        Fit the eliobj and learn prior distributions.

        Parameters
        ----------
        overwrite
            If the eliobj was already fitted and the user wants to refit it,
            the user is asked whether they want to overwrite the previous
            fitting results. Setting ``overwrite=True`` allows the user to
            force overfitting without being prompted.

        parallel
            specify parallelization settings if multiple trainings should run
            in parallel. See [`parallel`][elicito.utils.parallel].

        Examples
        --------
        >>> eliobj.fit()  # doctest: +SKIP

        >>> eliobj.fit(overwrite=True)  # doctest: +SKIP

        >>> eliobj.fit(parallel=el.utils.parallel(runs=4))  # doctest: +SKIP

        """
        # set seed
        tf.random.set_seed(self.trainer["seed"])

        # check whether elicit object is already fitted
        try:
            self.results  # type: ignore
        except AttributeError:
            # run single time if no parallelization is required
            if parallel is None:
                self.temp_results = []
                self.temp_history = []
                self.results = xr.DataTree()

                results, history = self.workflow(self.trainer["seed"])
                # include seed information into results
                results["seed"] = self.trainer["seed"]
                # save results in list attribute
                self.temp_history.append(history)
                self.temp_results.append(results)

                results = _outputs.create_datatree(
                    self.temp_history,
                    self.temp_results,
                    self.trainer,
                    self.parameters,
                    self.expert,
                )

                self.results.update(results)
                delattr(self, "temp_history")
                delattr(self, "temp_results")

            # run multiple replications
            if parallel is not None:
                self.temp_results = []
                self.temp_history = []
                self.results = xr.DataTree()

                # create a list of seeds if not provided
                if parallel["seeds"] is None:
                    # generate seeds
                    seeds = [
                        int(s) for s in tfd.Uniform(0, 999999).sample(parallel["runs"])
                    ]
                else:
                    seeds = parallel["seeds"]

                # run training simultaneously for multiple seeds
                (*res,) = joblib.Parallel(n_jobs=parallel["cores"])(
                    joblib.delayed(self.workflow)(seed) for seed in seeds
                )

                for i, seed in enumerate(seeds):
                    self.temp_results.append(res[i][0])
                    self.temp_history.append(res[i][1])
                    self.temp_results[i]["seed"] = seed

                results = _outputs.create_datatree(
                    self.temp_history,
                    self.temp_results,
                    self.trainer,
                    self.parameters,
                    self.expert,
                )

                self.results = results
                delattr(self, "temp_history")
                delattr(self, "temp_results")
        else:
            if not overwrite:
                user_answ = input(
                    "eliobj is already fitted."
                    + " Do you want to fit it again and overwrite the results?"
                    + " Press 'n' to stop process and 'y' to continue fitting."
                )

                if user_answ not in ["y", "n"]:
                    raise ValueError("Invalid input. Please use 'y' or 'n'.")  # noqa: TRY003

                if user_answ == "n":
                    print("Process aborded; eliobj is not re-fitted.")

    def save(
        self,
        name: str | None = None,
        file: str | None = None,
        overwrite: bool = False,
    ) -> None:
        """
        Save data on disk

        Parameters
        ----------
        name
            file name used to store the eliobj. Saving is done
            according to the following rule: ``./{method}/{name}_{seed}.pkl``
            with 'method' and 'seed' being arguments of
            [`trainer`][elicito.elicit.trainer].

        file
            user-specific path for saving the eliobj. If file is specified
            **name** must be ``None``.

        overwrite
            If already a fitted object exists in the same path, the user is
            asked whether the eliobj should be refitted and the results
            overwritten.
            With the ``overwrite`` argument, you can disable this
            behavior. In this case the results are automatically overwritten
            without prompting the user.

        Raises
        ------
        AssertionError
            ``name`` and ``file`` can't be specified simultaneously.

        Examples
        --------
        >>> eliobj.save(name="toymodel")  # doctest: +SKIP

        >>> eliobj.save(file="res/toymodel", overwrite=True)  # doctest: +SKIP

        """
        # check that either name or file is specified
        if not (name is None) ^ (file is None):
            msg = (
                "Name and file cannot be both None or both specified. "
                + "Either one has to be None.",
            )
            raise AssertionError(msg)

        # add a saving path
        return utils.save(self, name=name, file=file, overwrite=overwrite)

    def update(self, **kwargs: dict[Any, Any]) -> None:
        """
        Update attributes of Elicit object

        Method for updating the attributes of the Elicit class. Updating
        an eliobj leads to an automatic reset of results.

        Parameters
        ----------
        **kwargs
            keyword argument used for updating an attribute of Elicit class.
            Key must correspond to one attribute of the class and value refers
            to the updated value.

        Raises
        ------
        ValueError
            key of provided keyword argument is not an eliobj attribute. Please
            check `dir(eliobj)`.

        Examples
        --------
        >>> eliobj.update(parameter=updated_parameter_dict)  # doctest: +SKIP

        """
        # check that arguments exist as eliobj attributes
        for key in kwargs:
            if str(key) not in [
                "model",
                "parameters",
                "targets",
                "expert",
                "trainer",
                "optimizer",
                "network",
                "initializer",
            ]:
                msg = (
                    f"{key=} is not an eliobj attribute. "
                    + "Use dir() to check for attributes.",
                )
                raise ValueError(msg)

        # create first test variables
        test = SimpleNamespace(
            model=self.model,
            parameters=self.parameters,
            targets=self.targets,
            expert=self.expert,
            trainer=self.trainer,
            optimizer=self.optimizer,
            network=self.network,
            initializer=self.initializer,
            meta_settings=self.meta_settings,
        )

        for key, value in kwargs.items():
            setattr(test, key, value)

        _checks.check_elicit(
            test.model,
            test.parameters,
            test.targets,
            test.expert,
            test.trainer,
            test.optimizer,
            test.network,
            test.initializer,
            test.meta_settings,
        )

        # only if checks pass update variables of Elicit
        for i, key in enumerate(kwargs):
            setattr(self, key, kwargs[key])
            # reset results
            try:
                self.results
            except AttributeError:
                pass
            else:
                delattr(self, "results")
            self.temp_results = list()
            self.temp_history = list()
            if i == 0:
                # inform user about reset of results
                print("INFO: Results have been reset.")

    def workflow(self, seed: int) -> tuple[Any, ...]:
        """
        Build the main workflow of the prior elicitation method.

        Get expert data, initialize method, run optimization.
        Results are returned for further post-processing.

        Parameters
        ----------
        seed
            seed information used for reproducing results.

        Returns
        -------
        :
            results and history object of the optimization process.

        """
        # overwrite global seed
        # TODO test correct seed usage for parallel processing
        globals()["SEED"] = seed

        self.trainer["seed_chain"] = seed
        # get expert data; use trainer seed
        # (and not seed from list)
        expert_elicits, expert_prior = utils.get_expert_data(
            self.trainer,
            self.model,
            self.targets,
            self.expert,
            self.parameters,
            self.network,
            self.trainer["seed"],
        )

        # initialization of hyperparameter
        (init_prior_model, loss_list, init_prior_obj, init_matrix) = (
            initialization.init_prior(
                expert_elicits,
                self.initializer,
                self.parameters,
                self.trainer,
                self.model,
                self.targets,
                self.network,
                self.expert,
                seed,
                self.trainer["progress"],
            )
        )
        # run dag with optimal set of initial values
        # save results in corresp. attributes

        history, results = optimization.sgd_training(
            expert_elicits,
            init_prior_model,
            self.trainer,
            self.optimizer,
            self.model,
            self.targets,
            self.parameters,
            seed,
            self.trainer["progress"],
        )
        # add some additional results
        results["expert_elicited_statistics"] = expert_elicits
        try:
            self.expert["ground_truth"]
        except KeyError:
            pass
        else:
            results["expert_prior_samples"] = expert_prior

        if self.trainer["method"] == "parametric_prior":
            results["init_loss_list"] = loss_list
            results["init_matrix"] = init_matrix

        return tuple((results, history))

__init__

__init__(
    model: dict[str, Any],
    parameters: list[Parameter],
    targets: list[Target],
    expert: ExpertDict,
    trainer: Trainer,
    optimizer: dict[str, Any],
    network: NFDict | None = None,
    initializer: Initializer | None = None,
    meta_settings: MetaSettings = meta_settings(),
)

Specify the elicitation method

Parameters:

Name	Type	Description	Default
model	dict[str, Any]	specification of generative model using model.	required
parameters	list[Parameter]	list of model parameters specified with parameter.	required
targets	list[Target]	list of target quantities specified with target.	required
expert	ExpertDict	provide input data from expert or simulate data from oracle with either the data or simulator method of the Expert module.	required
trainer	Trainer	specification of training settings and meta-information for workflow using trainer.	required
optimizer	dict[str, Any]	specification of SGD optimizer and its settings using optimizer.	required
network	NFDict | None	specification of neural network using a method implemented in networks. Only required for deep_prior method.	None
initializer	Initializer | None	specification of initialization settings using initializer. Only required for parametric_prior method.	None
meta_settings	MetaSettings	dictionary of meta settings for the elicitation workflow. See meta_settings for available options.	meta_settings()

Returns:

Name	Type	Description
eliobj		specification of all settings to run the elicitation workflow and fit the eliobj.

Raises:

Type	Description
AssertionError	expert data are not in the required format. Correct specification of keys can be checked using get_expert_datformat Dimensionality of ground_truth for simulating expert data, must be the same as the number of model parameters.
ValueError	if method = "deep_prior", network can't be None and initialization should be None. if method="deep_prior", num_params as specified in the network_specs argument (section: network) does not match the number of parameters specified in the parameters section. if method="parametric_prior", network should be None and initialization can't be None. if method ="parametric_prior" and multiple hyperparameter have the same name but are not shared by settingshared = True``." if hyperparams is specified in section initializer and a hyperparameter name (key in hyperparams dict) does not match any hyperparameter name specified in hyper.
NotImplementedError	[network] Currently only the standard normal distribution is implemented as base distribution. See GitHub issue #35.

Source code in src/elicito/__init__.py

def __init__(  # noqa: PLR0913
    self,
    model: dict[str, Any],
    parameters: list[Parameter],
    targets: list[Target],
    expert: ExpertDict,
    trainer: Trainer,
    optimizer: dict[str, Any],
    network: NFDict | None = None,
    initializer: Initializer | None = None,
    meta_settings: MetaSettings = meta_settings(),
):
    """
    Specify the elicitation method

    Parameters
    ----------
    model
        specification of generative model using [`model`][elicito.elicit.model].

    parameters
        list of model parameters specified with [`parameter`][elicito.elicit.parameter].

    targets
        list of target quantities specified with [`target`][elicito.elicit.target].

    expert
        provide input data from expert or simulate data from oracle with
        either the ``data`` or ``simulator`` method of the
        [`Expert`][elicito.elicit.Expert] module.

    trainer
        specification of training settings and meta-information for
        workflow using [`trainer`][elicito.elicit.trainer].

    optimizer
        specification of SGD optimizer and its settings using
        [`optimizer`][elicito.elicit.optimizer].

    network
        specification of neural network using a method implemented in
        [`networks`][elicito.networks].
        Only required for ``deep_prior`` method.

    initializer
        specification of initialization settings using
        [`initializer`][elicito.elicit.initializer].
        Only required for ``parametric_prior`` method.

    meta_settings
        dictionary of meta settings for the elicitation workflow. See
        [`meta_settings`][elicito.types.MetaSettings] for available options.

    Returns
    -------
    eliobj :
        specification of all settings to run the elicitation workflow and
        fit the eliobj.

    Raises
    ------
    AssertionError
        ``expert`` data are not in the required format. Correct specification of
        keys can be checked using
        [`get_expert_datformat`][elicito.utils.get_expert_datformat]

        Dimensionality of ``ground_truth`` for simulating expert data, must be
        the same as the number of model parameters.

    ValueError
        if ``method = "deep_prior"``, ``network`` can't be None and ``initialization``
        should be None.

        if ``method="deep_prior"``, ``num_params`` as specified in the ``network_specs``
        argument (section: network) does not match the number of parameters
        specified in the parameters section.

        if ``method="parametric_prior"``, ``network`` should be None and
        ``initialization`` can't be None.

        if ``method ="parametric_prior" and multiple hyperparameter have
        the same name but are not shared by setting ``shared = True``."

        if ``hyperparams`` is specified in section ``initializer`` and a
        hyperparameter name (key in hyperparams dict) does not match any
        hyperparameter name specified in [`hyper`][elicito.elicit.hyper].

    NotImplementedError
        [network] Currently only the standard normal distribution is
        implemented as base distribution. See
        [GitHub issue #35](https://github.com/florence-bockting/prior_elicitation/issues/35).

    """  # noqa: E501
    _checks.check_elicit(
        model,
        parameters,
        targets,
        expert,
        trainer,
        optimizer,
        network,
        initializer,
        meta_settings,
    )

    self.model = model
    self.parameters = parameters
    self.targets = targets
    self.expert = expert
    self.trainer = trainer
    self.optimizer = optimizer
    self.network = network
    self.initializer = initializer
    self.meta_settings = meta_settings

    self.temp_history: list[dict[str, Any]] = []
    self.temp_results: list[dict[str, Any]] = []

    # helper for subsequent checks
    self.dry_run = self.meta_settings["dry_run"]
    # overwrite global seed
    globals()["SEED"] = self.trainer["seed"]

    # set seed
    tf.random.set_seed(SEED)

    if self.dry_run:
        (
            self.dry_elicits,
            self.dry_priors,
            self.dry_modelsims,
            self.dry_targets,
            self.dry_prior_model,
        ) = utils.dry_run(
            self.model,
            self.parameters,
            self.targets,
            self.trainer,
            self.initializer,  # type: ignore
            self.network,
        )

__repr__

__repr__() -> str

Return a readable representation of the object.

Source code in src/elicito/__init__.py

def __repr__(self) -> str:
    """Return a readable representation of the object."""
    return self.__str__()

__str__

__str__() -> str

Return a readable summary of the object.

Source code in src/elicito/__init__.py

def __str__(self) -> str:  # noqa: PLR0912
    """Return a readable summary of the object."""
    # fitted eliobj with shape information
    try:
        self.results  # type: ignore
    except AttributeError:
        if len(self.temp_results) != 0:
            targets_str = "\n".join(
                f"  - {k1} {tuple(self.temp_results[0]['target_quantities'][k1].shape)} -> "  # noqa: E501
                f"{k2} {tuple(self.temp_results[0]['elicited_statistics'][k2].shape)}"  # noqa: E501
                for k1, k2 in zip(
                    self.temp_results[0]["target_quantities"],
                    self.temp_results[0]["elicited_statistics"],
                )
            )
        # unfitted eliobj with shape information due to dry run
        elif self.dry_run:
            targets_str = "\n".join(
                f"  - {k1} {tuple(self.dry_targets[k1].shape)} -> "
                f"{k2} {tuple(self.dry_elicits[k2].shape)}"
                for k1, k2 in zip(self.dry_targets, self.dry_elicits)
            )
        # unfitted eliobj without shape information
        else:
            targets_str = "\n".join(
                f"  - {self.targets[tar]['name']} -> {eli}"
                for tar, eli in zip(
                    range(len(self.targets)),
                    utils.get_expert_datformat(self.targets),
                )
            )
    else:
        target_list = list(self.results.target_quantity.data_vars.keys())  # type: ignore
        elicit_list = list(self.results.elicited_summary.data_vars.keys())  # type: ignore

        targets_str = "\n".join(
            f"  - {k1} {self.results.target_quantity[k1].shape[1:]} -> "  # type: ignore
            f"{k2} {self.results.elicited_summary[k2].shape[1:]}"  # type: ignore
            for k1, k2 in zip(target_list, elicit_list)
        )

    opt_name = self.optimizer["optimizer"].__name__
    opt_lr = self.optimizer["learning_rate"]

    get_num_hyperpar: int | str
    try:
        self.results  # type: ignore
    except AttributeError:
        pass
    else:
        if self.trainer["method"] == "deep_prior":
            get_num_hyperpar = utils.compute_num_weights(
                self.results[0]["num_NN_weights"]  # type: ignore
            )

    if (self.trainer["method"] == "deep_prior") and (self.dry_run):
        trainable_vars = self.dry_prior_model.init_priors.trainable_variables
        num_NN_weights = [
            trainable_vars[i].shape for i in range(len(trainable_vars))
        ]
        get_num_hyperpar = utils.compute_num_weights(num_NN_weights)

    elif self.trainer["method"] == "parametric_prior":
        get_num_hyperpar = sum(
            [
                len(self.parameters[i]["hyperparams"])
                for i in range(len(self.parameters))
            ]
        )
    else:
        get_num_hyperpar = "?"
        print("Number of hyperparameter in model can't be computed.")

    summary = (
        f"Model hyperparameters: {get_num_hyperpar}\n"
        f"Model parameters: {len(self.parameters)}\n"
        "Targets -> Elicited summaries (loss components)"
        f"{': ' + str(len(self.dry_elicits)) if self.dry_run else ''}\n"
        f"{targets_str}\n"
        f"Prior samples: {self.trainer['num_samples']}"
        f"{' ' + str(tuple(self.dry_priors.shape)) if self.dry_run else ''}\n"
        f"Batch size: {self.trainer['B']}\n"
        f"Epochs: {self.trainer['epochs']}\n"
        f"Method: {self.trainer['method']}\n"
        f"Seed: {self.trainer['seed']}\n"
        f"Optimizer: {opt_name}(lr={opt_lr})\n"
    )
    if self.trainer["method"] == "parametric_prior":
        if self.initializer is not None:
            summary += (
                f"Initializer: (method: {self.initializer['method']}, "
                f"iterations: {self.initializer['iterations']})\n"
            )
        else:
            summary += "Initializer: None\n"
    elif self.network is not None:
        summary += f"Network: {self.network['inference_network'].__name__}\n"
    else:
        summary += "Network: None\n"

    return summary

fit

fit(
    overwrite: bool = False,
    parallel: Parallel | None = None,
) -> None

Fit the eliobj and learn prior distributions.

Parameters:

Name	Type	Description	Default
overwrite	bool	If the eliobj was already fitted and the user wants to refit it, the user is asked whether they want to overwrite the previous fitting results. Setting overwrite=True allows the user to force overfitting without being prompted.	False
parallel	Parallel | None	specify parallelization settings if multiple trainings should run in parallel. See parallel.	None

Examples:

>>> eliobj.fit()

>>> eliobj.fit(overwrite=True)

>>> eliobj.fit(parallel=el.utils.parallel(runs=4))

Source code in src/elicito/__init__.py

def fit(
    self,
    overwrite: bool = False,
    parallel: Parallel | None = None,
) -> None:
    """
    Fit the eliobj and learn prior distributions.

    Parameters
    ----------
    overwrite
        If the eliobj was already fitted and the user wants to refit it,
        the user is asked whether they want to overwrite the previous
        fitting results. Setting ``overwrite=True`` allows the user to
        force overfitting without being prompted.

    parallel
        specify parallelization settings if multiple trainings should run
        in parallel. See [`parallel`][elicito.utils.parallel].

    Examples
    --------
    >>> eliobj.fit()  # doctest: +SKIP

    >>> eliobj.fit(overwrite=True)  # doctest: +SKIP

    >>> eliobj.fit(parallel=el.utils.parallel(runs=4))  # doctest: +SKIP

    """
    # set seed
    tf.random.set_seed(self.trainer["seed"])

    # check whether elicit object is already fitted
    try:
        self.results  # type: ignore
    except AttributeError:
        # run single time if no parallelization is required
        if parallel is None:
            self.temp_results = []
            self.temp_history = []
            self.results = xr.DataTree()

            results, history = self.workflow(self.trainer["seed"])
            # include seed information into results
            results["seed"] = self.trainer["seed"]
            # save results in list attribute
            self.temp_history.append(history)
            self.temp_results.append(results)

            results = _outputs.create_datatree(
                self.temp_history,
                self.temp_results,
                self.trainer,
                self.parameters,
                self.expert,
            )

            self.results.update(results)
            delattr(self, "temp_history")
            delattr(self, "temp_results")

        # run multiple replications
        if parallel is not None:
            self.temp_results = []
            self.temp_history = []
            self.results = xr.DataTree()

            # create a list of seeds if not provided
            if parallel["seeds"] is None:
                # generate seeds
                seeds = [
                    int(s) for s in tfd.Uniform(0, 999999).sample(parallel["runs"])
                ]
            else:
                seeds = parallel["seeds"]

            # run training simultaneously for multiple seeds
            (*res,) = joblib.Parallel(n_jobs=parallel["cores"])(
                joblib.delayed(self.workflow)(seed) for seed in seeds
            )

            for i, seed in enumerate(seeds):
                self.temp_results.append(res[i][0])
                self.temp_history.append(res[i][1])
                self.temp_results[i]["seed"] = seed

            results = _outputs.create_datatree(
                self.temp_history,
                self.temp_results,
                self.trainer,
                self.parameters,
                self.expert,
            )

            self.results = results
            delattr(self, "temp_history")
            delattr(self, "temp_results")
    else:
        if not overwrite:
            user_answ = input(
                "eliobj is already fitted."
                + " Do you want to fit it again and overwrite the results?"
                + " Press 'n' to stop process and 'y' to continue fitting."
            )

            if user_answ not in ["y", "n"]:
                raise ValueError("Invalid input. Please use 'y' or 'n'.")  # noqa: TRY003

            if user_answ == "n":
                print("Process aborded; eliobj is not re-fitted.")

save

save(
    name: str | None = None,
    file: str | None = None,
    overwrite: bool = False,
) -> None

Save data on disk

Parameters:

Name	Type	Description	Default
name	str | None	file name used to store the eliobj. Saving is done according to the following rule: ./{method}/{name}_{seed}.pkl with 'method' and 'seed' being arguments of trainer.	None
file	str | None	user-specific path for saving the eliobj. If file is specified name must be None.	None
overwrite	bool	If already a fitted object exists in the same path, the user is asked whether the eliobj should be refitted and the results overwritten. With the overwrite argument, you can disable this behavior. In this case the results are automatically overwritten without prompting the user.	False

Raises:

Type	Description
AssertionError	name and file can't be specified simultaneously.

Examples:

>>> eliobj.save(name="toymodel")

>>> eliobj.save(file="res/toymodel", overwrite=True)

Source code in src/elicito/__init__.py

def save(
    self,
    name: str | None = None,
    file: str | None = None,
    overwrite: bool = False,
) -> None:
    """
    Save data on disk

    Parameters
    ----------
    name
        file name used to store the eliobj. Saving is done
        according to the following rule: ``./{method}/{name}_{seed}.pkl``
        with 'method' and 'seed' being arguments of
        [`trainer`][elicito.elicit.trainer].

    file
        user-specific path for saving the eliobj. If file is specified
        **name** must be ``None``.

    overwrite
        If already a fitted object exists in the same path, the user is
        asked whether the eliobj should be refitted and the results
        overwritten.
        With the ``overwrite`` argument, you can disable this
        behavior. In this case the results are automatically overwritten
        without prompting the user.

    Raises
    ------
    AssertionError
        ``name`` and ``file`` can't be specified simultaneously.

    Examples
    --------
    >>> eliobj.save(name="toymodel")  # doctest: +SKIP

    >>> eliobj.save(file="res/toymodel", overwrite=True)  # doctest: +SKIP

    """
    # check that either name or file is specified
    if not (name is None) ^ (file is None):
        msg = (
            "Name and file cannot be both None or both specified. "
            + "Either one has to be None.",
        )
        raise AssertionError(msg)

    # add a saving path
    return utils.save(self, name=name, file=file, overwrite=overwrite)

update

update(**kwargs: dict[Any, Any]) -> None

Update attributes of Elicit object

Method for updating the attributes of the Elicit class. Updating an eliobj leads to an automatic reset of results.

Parameters:

Name	Type	Description	Default
**kwargs	dict[Any, Any]	keyword argument used for updating an attribute of Elicit class. Key must correspond to one attribute of the class and value refers to the updated value.	{}

Raises:

Type	Description
ValueError	key of provided keyword argument is not an eliobj attribute. Please check dir(eliobj).

Examples:

>>> eliobj.update(parameter=updated_parameter_dict)

Source code in src/elicito/__init__.py

def update(self, **kwargs: dict[Any, Any]) -> None:
    """
    Update attributes of Elicit object

    Method for updating the attributes of the Elicit class. Updating
    an eliobj leads to an automatic reset of results.

    Parameters
    ----------
    **kwargs
        keyword argument used for updating an attribute of Elicit class.
        Key must correspond to one attribute of the class and value refers
        to the updated value.

    Raises
    ------
    ValueError
        key of provided keyword argument is not an eliobj attribute. Please
        check `dir(eliobj)`.

    Examples
    --------
    >>> eliobj.update(parameter=updated_parameter_dict)  # doctest: +SKIP

    """
    # check that arguments exist as eliobj attributes
    for key in kwargs:
        if str(key) not in [
            "model",
            "parameters",
            "targets",
            "expert",
            "trainer",
            "optimizer",
            "network",
            "initializer",
        ]:
            msg = (
                f"{key=} is not an eliobj attribute. "
                + "Use dir() to check for attributes.",
            )
            raise ValueError(msg)

    # create first test variables
    test = SimpleNamespace(
        model=self.model,
        parameters=self.parameters,
        targets=self.targets,
        expert=self.expert,
        trainer=self.trainer,
        optimizer=self.optimizer,
        network=self.network,
        initializer=self.initializer,
        meta_settings=self.meta_settings,
    )

    for key, value in kwargs.items():
        setattr(test, key, value)

    _checks.check_elicit(
        test.model,
        test.parameters,
        test.targets,
        test.expert,
        test.trainer,
        test.optimizer,
        test.network,
        test.initializer,
        test.meta_settings,
    )

    # only if checks pass update variables of Elicit
    for i, key in enumerate(kwargs):
        setattr(self, key, kwargs[key])
        # reset results
        try:
            self.results
        except AttributeError:
            pass
        else:
            delattr(self, "results")
        self.temp_results = list()
        self.temp_history = list()
        if i == 0:
            # inform user about reset of results
            print("INFO: Results have been reset.")

workflow

workflow(seed: int) -> tuple[Any, ...]

Build the main workflow of the prior elicitation method.

Get expert data, initialize method, run optimization. Results are returned for further post-processing.

Parameters:

Name	Type	Description	Default
seed	int	seed information used for reproducing results.	required

Returns:

Type	Description
tuple[Any, ...]	results and history object of the optimization process.

Source code in src/elicito/__init__.py

def workflow(self, seed: int) -> tuple[Any, ...]:
    """
    Build the main workflow of the prior elicitation method.

    Get expert data, initialize method, run optimization.
    Results are returned for further post-processing.

    Parameters
    ----------
    seed
        seed information used for reproducing results.

    Returns
    -------
    :
        results and history object of the optimization process.

    """
    # overwrite global seed
    # TODO test correct seed usage for parallel processing
    globals()["SEED"] = seed

    self.trainer["seed_chain"] = seed
    # get expert data; use trainer seed
    # (and not seed from list)
    expert_elicits, expert_prior = utils.get_expert_data(
        self.trainer,
        self.model,
        self.targets,
        self.expert,
        self.parameters,
        self.network,
        self.trainer["seed"],
    )

    # initialization of hyperparameter
    (init_prior_model, loss_list, init_prior_obj, init_matrix) = (
        initialization.init_prior(
            expert_elicits,
            self.initializer,
            self.parameters,
            self.trainer,
            self.model,
            self.targets,
            self.network,
            self.expert,
            seed,
            self.trainer["progress"],
        )
    )
    # run dag with optimal set of initial values
    # save results in corresp. attributes

    history, results = optimization.sgd_training(
        expert_elicits,
        init_prior_model,
        self.trainer,
        self.optimizer,
        self.model,
        self.targets,
        self.parameters,
        seed,
        self.trainer["progress"],
    )
    # add some additional results
    results["expert_elicited_statistics"] = expert_elicits
    try:
        self.expert["ground_truth"]
    except KeyError:
        pass
    else:
        results["expert_prior_samples"] = expert_prior

    if self.trainer["method"] == "parametric_prior":
        results["init_loss_list"] = loss_list
        results["init_matrix"] = init_matrix

    return tuple((results, history))

hyper

hyper(
    name: str,
    lower: float = float("-inf"),
    upper: float = float("inf"),
    vtype: VariableType = real,
    dim: int = 1,
    shared: bool = False,
) -> Hyper

Specify prior hyperparameters.

Parameters:

Name	Type	Description	Default
name	str	Custom name of hyperparameter.	required
lower	float	Lower bound of hyperparameter.	float('-inf')
upper	float	Upper bound of hyperparameter.	float('inf')
vtype	VariableType	Hyperparameter type. Either "real", "array", "cov" or "cov2tril" (lower triangular of covariance matrix realised via cholesky(cov))	real
dim	int	Dimensionality of variable. Only required if vtype = "array".	1
shared	bool	Shared hyperparameter between model parameters.	False

Returns:

Name	Type	Description
hyppar_dict	Hyper	Dictionary including all hyperparameter settings.

Raises:

Type	Description
ValueError	lower, upper take only values that are float or "-inf"or "inf". lower value should not be higher than upper value. vtype value can only be either 'real', 'array', 'cov', or 'cov2tril' dim value can't be '1' if 'vtype="array"'

Examples:

>>> # sigma hyperparameter of a parametric distribution
>>> el.hyper(name="sigma0", lower=0)

>>> # shared hyperparameter
>>> el.hyper(name="sigma", lower=0, shared=True)

Source code in src/elicito/elicit.py

def hyper(  # noqa: PLR0913
    name: str,
    lower: float = float("-inf"),
    upper: float = float("inf"),
    vtype: VariableType = VariableType.real,
    dim: int = 1,
    shared: bool = False,
) -> Hyper:
    """
    Specify prior hyperparameters.

    Parameters
    ----------
    name
        Custom name of hyperparameter.

    lower
        Lower bound of hyperparameter.

    upper
        Upper bound of hyperparameter.

    vtype
        Hyperparameter type. Either "real", "array",
        "cov" or "cov2tril" (lower triangular of
        covariance matrix realised via cholesky(cov))

    dim
        Dimensionality of variable.
        Only required if `vtype = "array"`.

    shared
        Shared hyperparameter between model parameters.

    Returns
    -------
    hyppar_dict :
        Dictionary including all hyperparameter settings.

    Raises
    ------
    ValueError
        ``lower``, ``upper`` take only values that are float
        or `"-inf"`or `"inf"`.

        ``lower`` value should not be higher than ``upper`` value.

        ``vtype`` value can only be either 'real', 'array', 'cov', or 'cov2tril'

        ``dim`` value can't be '1' if 'vtype="array"'

    Examples
    --------
    >>> # sigma hyperparameter of a parametric distribution
    >>> el.hyper(name="sigma0", lower=0)  # doctest: +SKIP

    >>> # shared hyperparameter
    >>> el.hyper(name="sigma", lower=0, shared=True)  # doctest: +SKIP

    """
    # check correct value for lower
    if lower == "-inf":  # type: ignore
        lower = float("-inf")

    if (type(lower) is str) and (lower != "-inf"):  # type: ignore
        msg = "Lower must be either '-inf' or a float." "Other strings are not allowed."
        raise ValueError(msg)

    # check correct value for upper
    if upper == "inf":  # type: ignore
        upper = float("inf")
    if (type(upper) is str) and (upper != "inf"):  # type: ignore
        msg = "Upper must be either 'inf' or a float." "Other strings are not allowed."
        raise ValueError(msg)

    if lower > upper:
        msg = "The value for 'lower' must be smaller than the value for 'upper'."
        raise ValueError(msg)

    # check values for vtype are implemented
    if vtype not in ["real", "array", "cov", "cov2tril"]:
        msg = (
            "vtype must be either 'real', 'array', 'cov', 'cov2tril'. "
            f"You provided {vtype=}."
        )
        raise ValueError(msg)

    # check that dimensionality is adapted when "array" is chosen
    if (vtype == "array") and dim == 1:
        msg = "For vtype='array', the 'dim' argument must have a value greater 1."
        raise ValueError(msg)

    # constraints
    # only lower bound
    if (lower != float("-inf")) and (upper == float("inf")):
        lower_bound = LowerBound(lower=lower)
        transform = lower_bound.inverse
        constraint_name = "softplusL"
    # only upper bound
    elif (upper != float("inf")) and (lower == float("-inf")):
        upper_bound = UpperBound(upper=upper)
        transform = upper_bound.inverse
        constraint_name = "softplusU"
    # upper and lower bound
    elif (upper != float("inf")) and (lower != float("-inf")):
        double_bound = DoubleBound(lower=lower, upper=upper)
        transform = double_bound.inverse  # type: ignore
        constraint_name = "invlogit"
    # unbounded
    else:
        transform = identity  # type: ignore
        constraint_name = "identity"

    # value type
    dtype_dim = Dtype(vtype, dim)

    hyper_dict: Hyper = dict(
        name=name,
        constraint=transform,
        constraint_name=constraint_name,
        vtype=dtype_dim,
        dim=dim,
        shared=shared,
    )

    return hyper_dict

initializer

initializer(
    method: Optional[SamplingMethod] = None,
    distribution: Optional[Uniform] = None,
    iterations: Optional[int] = None,
    hyperparams: Optional[dict[str, Any]] = None,
) -> Initializer

Initialize hyperparameter values

Only necessary for method parametric_prior. Two approaches are currently possible:

1. Specify specific initial values for each hyperparameter.
2. Use one of the implemented sampling approaches to draw initial values from one of the provided initialization distributions

In (2) initial values for each hyperparameter are drawn from a uniform distribution ranging from mean - radius to mean + radius.

Parameters:

Name	Type	Description	Default
method	Optional[SamplingMethod]	Name of initialization method. Currently supported are "random", "lhs", and "sobol".	None
distribution	Optional[Uniform]	Specification of initialization distribution. Currently implemented methods: uniform	None
iterations	Optional[int]	Number of samples drawn from the initialization distribution.	None
hyperparams	Optional[dict[str, Any]]	Dictionary with specific initial values per hyperparameter. Note: Initial values are considered to be on the unconstrained scale. Use the forward method of LowerBound, UpperBound and DoubleBound for transforming a constrained hyperparameter into an unconstrained one. In hyperparams dictionary, keys refer to hyperparameter names, as specified in hyper and values to the respective initial values.	None

Returns:

Name	Type	Description
init_dict	Initializer	Dictionary specifying the initialization method.

Raises:

Type	Description
ValueError	method can only take the values "random", "sobol", or "lhs" loss_quantile must be a probability ranging between 0 and 1. Either method or hyperparams has to be specified.

Examples:

>>> el.initializer(
>>>     method="lhs",
>>>     iterations=32,
>>>     distribution=el.initialization.uniform(
>>>         radius=1,
>>>         mean=0
>>>         )
>>>     )

>>> el.initializer(
>>>     hyperparams = dict(
>>>         mu0=0.,
>>>         sigma0=el.utils.LowerBound(lower=0.).forward(0.3),
>>>         mu1=1.,
>>>         sigma1=el.utils.LowerBound(lower=0.).forward(0.5),
>>>         sigma2=el.utils.LowerBound(lower=0.).forward(0.4)
>>>         )
>>>     )

Source code in src/elicito/elicit.py

def initializer(
    method: Optional[SamplingMethod] = None,
    distribution: Optional[Uniform] = None,
    iterations: Optional[int] = None,
    hyperparams: Optional[dict[str, Any]] = None,
) -> Initializer:
    """
    Initialize hyperparameter values

    Only necessary for method ``parametric_prior``.
    Two approaches are currently possible:

    1. Specify specific initial values for each hyperparameter.
    2. Use one of the implemented sampling approaches to draw initial
       values from one of the provided initialization distributions

    In (2) initial values for each hyperparameter are drawn from a uniform
    distribution ranging from ``mean - radius`` to ``mean + radius``.

    Parameters
    ----------
    method
        Name of initialization method.
        Currently supported are "random", "lhs", and "sobol".

    distribution
        Specification of initialization distribution.
        Currently implemented methods: [`uniform`][elicito.initialization.uniform]

    iterations
        Number of samples drawn from the initialization distribution.

    hyperparams
        Dictionary with specific initial values per hyperparameter.
        **Note:** Initial values are considered to be on the *unconstrained
        scale*. Use  the ``forward`` method of [`LowerBound`][elicito.utils.LowerBound],
        [`UpperBound`][elicito.utils.UpperBound] and
        [`DoubleBound`][elicito.utils.DoubleBound]
        for transforming a constrained hyperparameter into an
        unconstrained one. In hyperparams dictionary, *keys* refer to
        hyperparameter names, as specified in [`hyper`][elicito.elicit.hyper]
        and *values* to the respective initial values.

    Returns
    -------
    init_dict :
        Dictionary specifying the initialization method.

    Raises
    ------
    ValueError
        ``method`` can only take the values "random", "sobol", or "lhs"

        ``loss_quantile`` must be a probability ranging between 0 and 1.

        Either ``method`` or ``hyperparams`` has to be specified.

    Examples
    --------
    >>> el.initializer(  # doctest: +SKIP
    >>>     method="lhs",  # doctest: +SKIP
    >>>     iterations=32,  # doctest: +SKIP
    >>>     distribution=el.initialization.uniform(  # doctest: +SKIP
    >>>         radius=1,  # doctest: +SKIP
    >>>         mean=0   # doctest: +SKIP
    >>>         )  # doctest: +SKIP
    >>>     )  # doctest: +SKIP

    >>> el.initializer(  # doctest: +SKIP
    >>>     hyperparams = dict(  # doctest: +SKIP
    >>>         mu0=0.,  # doctest: +SKIP
    >>>         sigma0=el.utils.LowerBound(lower=0.).forward(0.3),  # doctest: +SKIP
    >>>         mu1=1.,  # doctest: +SKIP
    >>>         sigma1=el.utils.LowerBound(lower=0.).forward(0.5),  # doctest: +SKIP
    >>>         sigma2=el.utils.LowerBound(lower=0.).forward(0.4)  # doctest: +SKIP
    >>>         )  # doctest: +SKIP
    >>>     )  # doctest: +SKIP
    """
    # check that method is implemented

    if method is None:
        args = {"distribution": distribution, "iterations": iterations}

        for name, value in args.items():
            if value is not None:
                raise ValueError(f"If method is None, '{name}' must also be None.")  # noqa: TRY003

        if hyperparams is None:
            msg = (
                "Either 'method' or 'hyperparams' has"
                "to be specified. Use method for sampling from an"
                "initialization distribution and 'hyperparams' for"
                "specifying exact initial values per hyperparameter."
            )
            raise ValueError(msg)

        # hardcode loss_quantile as it was rather meant for experimental purposes
        # however results suggest that loss_quantile different from zero are not
        # really reasonable
        loss_quantile = 0.0

        quantile_perc = loss_quantile

    else:
        args = {"distribution": distribution, "iterations": iterations}

        for name, value in args.items():
            if value is None:
                msg = f"If '{name}' is None, then 'method' must also be None."
                raise ValueError(msg)

        # hardcode loss_quantile as it was rather meant for experimental purposes
        # however results suggest that loss_quantile different from zero are not
        # really reasonable
        loss_quantile = 0.0

        # compute percentage from probability
        if loss_quantile is not None:
            quantile_perc = int(loss_quantile * 100)
        # ensure that iterations is an integer
        if iterations is not None:
            iterations = int(iterations)

        if method not in ["random", "lhs", "sobol"]:
            msg = (
                "Currently implemented initialization "
                f"methods are 'random', 'sobol', and 'lhs', but got {method=}"
                " as input."
            )
            raise ValueError(msg)

    init_dict: Initializer = dict(
        method=method,
        distribution=distribution,
        loss_quantile=quantile_perc,
        iterations=iterations,
        hyperparams=hyperparams,
    )

    return init_dict

model

model(
    obj: Callable[[str], Tensor], **kwargs: dict[Any, Any]
) -> dict[str, Any]

Specify the generative model.

Parameters:

Name	Type	Description	Default
obj	Callable[[str], Tensor]	Generative model class as defined by the user.	required
**kwargs	dict[Any, Any]	additional keyword arguments expected by obj.	{}

Returns:

Name	Type	Description
generator_dict	dict[str, Any]	Dictionary including all generative model settings.

Raises:

Type	Description
ValueError	generative model in obj requires the input argument 'prior_samples', but argument has not been found. optional argument(s) of the generative model specified in obj are not specified

Examples:

>>> # specify the generative model class
>>> class ToyModel:
>>>     def __call__(self, prior_samples, design_matrix):
>>> # linear predictor
>>>         epred = tf.matmul(prior_samples, design_matrix,
>>>                           transpose_b=True)
>>> # data-generating model
>>>         likelihood = tfd.Normal(
>>>             loc=epred,
>>>             scale=tf.expand_dims(prior_samples[:, :, -1], -1)
>>>             )
>>> # prior predictive distribution
>>>         ypred = likelihood.sample()
>>>
>>>         return dict(
>>>             likelihood=likelihood,
>>>             ypred=ypred, epred=epred,
>>>             prior_samples=prior_samples
>>>             )

>>> # specify the model category in the elicit object
>>> el.model(obj=ToyModel,
>>>          design_matrix=design_matrix
>>>          )

Source code in src/elicito/elicit.py

def model(obj: Callable[[str], tf.Tensor], **kwargs: dict[Any, Any]) -> dict[str, Any]:
    """
    Specify the generative model.

    Parameters
    ----------
    obj
        Generative model class as defined by the user.

    **kwargs
        additional keyword arguments expected by `obj`.

    Returns
    -------
    generator_dict :
        Dictionary including all generative model settings.

    Raises
    ------
    ValueError
        generative model in `obj` requires the input argument
        'prior_samples', but argument has not been found.

        optional argument(s) of the generative model specified in `obj` are
        not specified

    Examples
    --------
    >>> # specify the generative model class
    >>> class ToyModel:  # doctest: +SKIP
    >>>     def __call__(self, prior_samples, design_matrix):  # doctest: +SKIP
    >>> # linear predictor
    >>>         epred = tf.matmul(prior_samples, design_matrix,  # doctest: +SKIP
    >>>                           transpose_b=True)  # doctest: +SKIP
    >>> # data-generating model
    >>>         likelihood = tfd.Normal(  # doctest: +SKIP
    >>>             loc=epred,  # doctest: +SKIP
    >>>             scale=tf.expand_dims(prior_samples[:, :, -1], -1)  # doctest: +SKIP
    >>>             )  # doctest: +SKIP
    >>> # prior predictive distribution
    >>>         ypred = likelihood.sample()  # doctest: +SKIP
    >>>
    >>>         return dict(  # doctest: +SKIP
    >>>             likelihood=likelihood,  # doctest: +SKIP
    >>>             ypred=ypred, epred=epred,  # doctest: +SKIP
    >>>             prior_samples=prior_samples  # doctest: +SKIP
    >>>             )  # doctest: +SKIP

    >>> # specify the model category in the elicit object
    >>> el.model(obj=ToyModel,  # doctest: +SKIP
    >>>          design_matrix=design_matrix  # doctest: +SKIP
    >>>          )  # doctest: +SKIP
    """
    # get input arguments of generative model class
    input_args = inspect.getfullargspec(obj.__call__)[0]  # type: ignore
    # check correct input form of generative model class
    if "prior_samples" not in input_args:
        msg = (
            "The generative model class 'obj' requires the"
            " input variable 'prior_samples' but argument has not been found"
            " in 'obj'."
        )
        raise ValueError(msg)

    # check that all optional arguments have been provided by the user
    optional_args = set(input_args).difference({"prior_samples", "self"})
    for arg in optional_args:
        if arg not in list(kwargs.keys()):
            msg = (
                f"The argument {arg=} required by the"
                "generative model class 'obj' is missing."
            )
            raise ValueError(msg)

    generator_dict = dict(obj=obj)

    for key in kwargs:  # noqa: PLC0206
        generator_dict[key] = kwargs[key]  # type: ignore

    return generator_dict

optimizer

optimizer(
    optimizer: Any = Adam(), **kwargs: dict[Any, Any]
) -> dict[str, Any]

Specify optimizer and its settings for SGD.

Parameters:

Name	Type	Description	Default
optimizer	Any	Optimizer used for SGD implemented. Must be an object implemented in tf.keras.optimizers	Adam()
**kwargs	dict[Any, Any]	Additional keyword arguments expected by optimizer.	{}

Returns:

Name	Type	Description
optimizer_dict	dict[str, Any]	Dictionary specifying the SGD optimizer and its additional settings.

Raises:

Type	Description
TypeError	optimizer is not a tf.keras.optimizers object
ValueError	optimizer could not be found in tf.keras.optimizers

Examples:

>>> optimizer = el.optimizer(
>>>     optimizer=tf.keras.optimizers.Adam,
>>>     learning_rate=0.1,
>>>     clipnorm=1.0
>>> )

Source code in src/elicito/elicit.py

def optimizer(
    optimizer: Any = tf.keras.optimizers.Adam(), **kwargs: dict[Any, Any]
) -> dict[str, Any]:
    """
    Specify optimizer and its settings for SGD.

    Parameters
    ----------
    optimizer
        Optimizer used for SGD implemented.
        Must be an object implemented in [`tf.keras.optimizers`](https://www.tensorflow.org/api_docs/python/tf/keras/optimizers)

    **kwargs
        Additional keyword arguments expected by **optimizer**.

    Returns
    -------
    optimizer_dict :
        Dictionary specifying the SGD optimizer and its additional settings.

    Raises
    ------
    TypeError
        ``optimizer`` is not a tf.keras.optimizers object
    ValueError
        ``optimizer`` could not be found in tf.keras.optimizers

    Examples
    --------
    >>> optimizer = el.optimizer(  # doctest: +SKIP
    >>>     optimizer=tf.keras.optimizers.Adam,  # doctest: +SKIP
    >>>     learning_rate=0.1,  # doctest: +SKIP
    >>>     clipnorm=1.0  # doctest: +SKIP
    >>> )  # doctest: +SKIP
    """
    optimizer_dict = dict(optimizer=optimizer)

    for key in kwargs:  # noqa: PLC0206
        optimizer_dict[key] = kwargs[key]

    return optimizer_dict

parameter

parameter(
    name: str,
    family: Optional[Any] = None,
    hyperparams: Optional[dict[str, Hyper]] = None,
    lower: float = float("-inf"),
    upper: float = float("inf"),
) -> Parameter

Specify model parameters.

Parameters:

Name	Type	Description	Default
name	str	Custom name of parameter.	required
family	Optional[Any]	Prior distribution family for model parameter. Only required for parametric_prior method. Must be a member of tfp.distributions.	None
hyperparams	Optional[dict[str, Hyper]]	Hyperparameters of distribution as specified in family. Only required for parametric_prior method. Structure of dictionary: keys must match arguments of tfp.distributions object and values have to be specified using the hyper method.	None
lower	float	Only used if method = "deep_prior". Lower bound of parameter.	float('-inf')
upper	float	Only used if method = "deep_prior". Upper bound of parameter.	float('inf')

Returns:

Name	Type	Description
param_dict	dict	Dictionary including all model (hyper)parameter settings.

Raises:

Type	Description
ValueError	hyperparams value is a dict with keys corresponding to arguments of tfp.distributions object in 'family'. Raises error if key does not correspond to any argument of distribution.

Examples:

>>> el.parameter(name="beta0",
>>>              family=tfd.Normal,
>>>              hyperparams=dict(loc=el.hyper("mu0"),
>>>                               scale=el.hyper("sigma0", lower=0)
>>>                               )
>>>              )

Source code in src/elicito/elicit.py

def parameter(
    name: str,
    family: Optional[Any] = None,
    hyperparams: Optional[dict[str, Hyper]] = None,
    lower: float = float("-inf"),
    upper: float = float("inf"),
) -> Parameter:
    """
    Specify model parameters.

    Parameters
    ----------
    name
        Custom name of parameter.

    family
        Prior distribution family for model parameter.
        Only required for ``parametric_prior`` method.
        Must be a member of [`tfp.distributions`](https://www.tensorflow.org/probability/api_docs/python/tfp/distributions).

    hyperparams
        Hyperparameters of distribution as specified in **family**.
        Only required for ``parametric_prior`` method.
        Structure of dictionary: *keys* must match arguments of
        [`tfp.distributions`](https://www.tensorflow.org/probability/api_docs/python/tfp/distributions)
        object and *values* have to be specified using the [`hyper`][elicito.elicit.hyper]
        method.

    lower
        Only used if ``method = "deep_prior"``.
        Lower bound of parameter.

    upper
        Only used if ``method = "deep_prior"``.
        Upper bound of parameter.

    Returns
    -------
    param_dict : dict
        Dictionary including all model (hyper)parameter settings.

    Raises
    ------
    ValueError
        ``hyperparams`` value is a dict with keys corresponding to arguments of
        tfp.distributions object in 'family'. Raises error if key does not
        correspond to any argument of distribution.

    Examples
    --------
    >>> el.parameter(name="beta0",  # doctest: +SKIP
    >>>              family=tfd.Normal,  # doctest: +SKIP
    >>>              hyperparams=dict(loc=el.hyper("mu0"),  # doctest: +SKIP
    >>>                               scale=el.hyper("sigma0", lower=0)  # doctest: +SKIP
    >>>                               )  # doctest: +SKIP
    >>>              )  # doctest: +SKIP

    """  # noqa: E501
    # check whether keys of hyperparams dict correspond to arguments of family
    if hyperparams is not None:
        for key in hyperparams:
            if key not in inspect.getfullargspec(family)[0]:
                raise ValueError(  # noqa: TRY003
                    f"'{family.__module__.split('.')[-1]}'"
                    f" family has no argument '{key}'. Check keys of "
                    "'hyperparams' dict."
                )

    # constraints
    # only lower bound
    if (lower != float("-inf")) and (upper == float("inf")):
        lower_bound = LowerBound(lower)
        transform = lower_bound.inverse
        constraint_name: str = "softplusL"
    # only upper bound
    elif (upper != float("inf")) and (lower == float("-inf")):
        upper_bound = UpperBound(upper)
        transform = upper_bound.inverse
        constraint_name = "softplusU"
    # upper and lower bound
    elif (upper != float("inf")) and (lower != float("-inf")):
        double_bound = DoubleBound(lower, upper)
        transform = double_bound.inverse  # type: ignore
        constraint_name = "invlogit"
    # unbounded
    else:
        transform = identity  # type: ignore
        constraint_name = "identity"

    return Parameter(
        name=name,
        family=family,
        hyperparams=hyperparams,
        constraint_name=constraint_name,
        constraint=transform,  # type: ignore
    )

target

target(
    name: str,
    loss: Callable[[Any], Any],
    query: QueriesDict,
    target_method: Optional[Callable[[Any], Any]] = None,
    weight: float = 1.0,
) -> Target

Specify target quantity and corresponding elicitation technique.

Parameters:

Name	Type	Description	Default
name	str	Name of the target quantity. Two approaches are possible: (1) Target quantity is identical to an output from the generative model: The name must match the output variable name. (2) Custom target quantity is computed using the target_method argument.	required
query	QueriesDict	Specify the elicitation technique by using one of the methods implemented in Queries.	required
loss	Callable[[Any], Any]	Lossfunction for computing the discrepancy between expert data and model simulations. See losses.	required
target_method	Optional[Callable[[Any], Any]]	Custom method for computing a target quantity. Note: This method hasn't been implemented yet and will raise an NotImplementedError. See GitHub issue #34.	None
weight	float	Weight of the corresponding elicited quantity in the total loss.	1.0

Returns:

Name	Type	Description
target_dict	Target	Dictionary including all settings regarding the target quantity and corresponding elicitation technique.

Examples:

>>> el.target(name="y_X0",
>>>           query=el.queries.quantiles(
>>>                 (.05, .25, .50, .75, .95)),
>>>           loss=el.losses.MMD2(kernel="energy"),
>>>           weight=1.0
>>>           )

>>> el.target(name="correlation",
>>>           query=el.queries.correlation(),
>>>           loss=el.losses.L2,
>>>           weight=1.0
>>>           )

Source code in src/elicito/elicit.py

def target(
    name: str,
    loss: Callable[[Any], Any],
    query: QueriesDict,
    target_method: Optional[Callable[[Any], Any]] = None,
    weight: float = 1.0,
) -> Target:
    """
    Specify target quantity and corresponding elicitation technique.

    Parameters
    ----------
    name
        Name of the target quantity. Two approaches are possible:
        (1) Target quantity is identical to an output from the generative
        model: The name must match the output variable name. (2) Custom target
        quantity is computed using the `target_method` argument.

    query
        Specify the elicitation technique by using one of the methods
        implemented in [`Queries`][elicito.elicit.Queries].

    loss
        Lossfunction for computing the discrepancy between expert data and
        model simulations. See [`losses`][elicito.losses].

    target_method
        Custom method for computing a target quantity.
        Note: This method hasn't been implemented yet and will raise an
        ``NotImplementedError``. See
        [GitHub issue #34](https://github.com/florence-bockting/prior_elicitation/issues/34).

    weight
        Weight of the corresponding elicited quantity in the total loss.

    Returns
    -------
    target_dict :
        Dictionary including all settings regarding the target quantity and
        corresponding elicitation technique.

    Examples
    --------
    >>> el.target(name="y_X0",  # doctest: +SKIP
    >>>           query=el.queries.quantiles(  # doctest: +SKIP
    >>>                 (.05, .25, .50, .75, .95)),  # doctest: +SKIP
    >>>           loss=el.losses.MMD2(kernel="energy"),  # doctest: +SKIP
    >>>           weight=1.0  # doctest: +SKIP
    >>>           )  # doctest: +SKIP

    >>> el.target(name="correlation",  # doctest: +SKIP
    >>>           query=el.queries.correlation(),  # doctest: +SKIP
    >>>           loss=el.losses.L2,  # doctest: +SKIP
    >>>           weight=1.0  # doctest: +SKIP
    >>>           )  # doctest: +SKIP
    """
    # create instance of loss class
    loss_instance = loss

    return Target(
        name=name,
        query=query,
        target_method=target_method,
        loss=loss_instance,
        weight=weight,
    )

trainer

trainer(
    method: PriorMethods,
    seed: int,
    epochs: int,
    B: int = 128,
    num_samples: int = 200,
    progress: ProgressMethod = SHOW_PROGRESS,
) -> Trainer

Specify training settings for learning the prior distribution(s).

Parameters:

Name	Type	Description	Default
method	PriorMethods	Method for learning the prior distribution. Available is either parametric_prior for learning independent parametric priors or deep_prior for learning a joint non-parameteric prior.	required
seed	int	Seed used for learning.	required
epochs	int	Number of iterations until training is stopped.	required
B	int	Batch size.	128
num_samples	int	Number of samples from the prior(s).	200
progress	ProgressMethod	whether training progress should be printed. Progress is shown if progress=1 and muted if progress=0.	SHOW_PROGRESS

Returns:

Name	Type	Description
train_dict	Trainer	dictionary specifying the training settings for learning the prior distribution(s).

Raises:

Type	Description
ValueError	method can only take the value "parametric_prior" or "deep_prior" epochs can only take positive integers. Minimum number of epochs is 1. progress can only be 0 (mute progress) or 1 (print progress)

Examples:

>>> el.trainer(
>>>     method="parametric_prior",
>>>     seed=0,
>>>     epochs=400,
>>>     B=128,
>>>     num_samples=200
>>> )

Source code in src/elicito/elicit.py

def trainer(  # noqa: PLR0913
    method: PriorMethods,
    seed: int,
    epochs: int,
    B: int = 128,
    num_samples: int = 200,
    progress: ProgressMethod = ProgressMethod.SHOW_PROGRESS,
) -> Trainer:
    """
    Specify training settings for learning the prior distribution(s).

    Parameters
    ----------
    method
        Method for learning the prior distribution. Available is either
        ``parametric_prior`` for learning independent parametric priors
        or ``deep_prior`` for learning a joint non-parameteric prior.

    seed
        Seed used for learning.

    epochs
        Number of iterations until training is stopped.

    B
        Batch size.

    num_samples
        Number of samples from the prior(s).

    progress
        whether training progress should be printed. Progress is shown if
        `progress=1` and muted if `progress=0`.

    Returns
    -------
    train_dict :
        dictionary specifying the training settings for learning the prior
        distribution(s).

    Raises
    ------
    ValueError
        ``method`` can only take the value "parametric_prior" or "deep_prior"

        ``epochs`` can only take positive integers. Minimum number of epochs
        is 1.

        `progress` can only be 0 (mute progress) or 1 (print progress)

    Examples
    --------
    >>> el.trainer(  # doctest: +SKIP
    >>>     method="parametric_prior",  # doctest: +SKIP
    >>>     seed=0,  # doctest: +SKIP
    >>>     epochs=400,  # doctest: +SKIP
    >>>     B=128,  # doctest: +SKIP
    >>>     num_samples=200  # doctest: +SKIP
    >>> )  # doctest: +SKIP
    """
    # check that progress is either 0 or 1
    if progress not in [0, 1]:
        raise ValueError(f"Progress has to be either 0 or 1. Got {progress=}.")  # noqa: TRY003
    # check that epochs are positive numbers
    if epochs <= 0:
        msg = "The number of epochs has to be greater 0." f" Got {epochs=}."
        raise ValueError(msg)

    # check that method is implemented
    if method not in ["parametric_prior", "deep_prior"]:
        msg = (
            "Currently only the methods 'deep_prior' and"
            f"'parametric prior' are implemented but got {method=}."
        )
        raise ValueError(msg)

    train_dict: Trainer = dict(
        method=method,
        seed=int(seed),
        B=int(B),
        num_samples=int(num_samples),
        epochs=int(epochs),
        progress=progress,
    )
    return train_dict