Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
chain_data = []
for path in posterior_predictive:
parsed_output = _read_output(path)
for sample, *_ in parsed_output:
chain_data.append(sample)
data = _unpack_dataframes(chain_data)
else:
if isinstance(posterior_predictive, str):
posterior_predictive = [posterior_predictive]
posterior_predictive_cols = [
col
for col in columns
if any(item == col.split(".")[0] for item in posterior_predictive)
]
data = _unpack_dataframes([item[posterior_predictive_cols] for item in self.posterior])
return dict_to_dataset(data, coords=self.coords, dims=self.dims)
def prior_predictive_to_xarray(self):
"""Convert prior_predictive samples to xarray."""
prior_predictive = self.prior_predictive
columns = self.prior.column_names
if isinstance(prior_predictive, str):
prior_predictive = [prior_predictive]
valid_cols = [col for col in columns if col.split(".")[0] in set(prior_predictive)]
data = _unpack_frame(self.prior.sample, columns, valid_cols)
return dict_to_dataset(data, library=self.cmdstanpy, coords=self.coords, dims=self.dims)
def posterior_predictive_to_xarray(self):
"""Convert posterior_predictive samples to xarray."""
posterior_predictive = self.posterior_predictive
columns = self.posterior.column_names
if isinstance(posterior_predictive, str):
posterior_predictive = [posterior_predictive]
valid_cols = [col for col in columns if col.split(".")[0] in set(posterior_predictive)]
data = _unpack_frame(self.posterior.sample, columns, valid_cols)
return dict_to_dataset(data, library=self.cmdstanpy, coords=self.coords, dims=self.dims)
def prior_to_xarray(self):
"""Convert prior samples to xarray."""
prior = self.prior
# filter posterior_predictive and log_likelihood
prior_predictive = self.prior_predictive
if prior_predictive is None:
prior_predictive = []
elif isinstance(prior_predictive, str):
prior_predictive = [prior_predictive]
ignore = prior_predictive + ["lp__"]
data = get_draws(prior, ignore=ignore)
return dict_to_dataset(data, library=self.pystan, coords=self.coords, dims=self.dims)
else: # pystan or pystan3
return from_pystan(**kwargs)
elif obj.__class__.__name__ == "MultiTrace": # ugly, but doesn't make PyMC3 a requirement
return from_pymc3(trace=kwargs.pop(group), **kwargs)
elif obj.__class__.__name__ == "EnsembleSampler": # ugly, but doesn't make emcee a requirement
return from_emcee(sampler=kwargs.pop(group), **kwargs)
elif obj.__class__.__name__ == "MCMC" and obj.__class__.__module__.startswith("pyro"):
return from_pyro(posterior=kwargs.pop(group), **kwargs)
elif obj.__class__.__name__ == "MCMC" and obj.__class__.__module__.startswith("numpyro"):
return from_numpyro(posterior=kwargs.pop(group), **kwargs)
# Cases that convert to xarray
if isinstance(obj, xr.Dataset):
dataset = obj
elif isinstance(obj, dict):
dataset = dict_to_dataset(obj, coords=coords, dims=dims)
elif isinstance(obj, np.ndarray):
dataset = dict_to_dataset({"x": obj}, coords=coords, dims=dims)
elif isinstance(obj, (list, tuple)) and isinstance(obj[0], str) and obj[0].endswith(".csv"):
if group == "sample_stats":
kwargs["posterior"] = kwargs.pop(group)
elif group == "sample_stats_prior":
kwargs["prior"] = kwargs.pop(group)
return from_cmdstan(**kwargs)
else:
allowable_types = (
"xarray dataset",
"dict",
"netcdf filename",
"numpy array",
"pystan fit",
"pymc3 trace",
def sample_stats_to_xarray(self):
"""Convert sample_stats samples to xarray."""
data = self.sample_stats
if not isinstance(data, dict):
raise TypeError("DictConverter.sample_stats is not a dictionary")
return dict_to_dataset(data, library=None, coords=self.coords, dims=self.dims)
def prior_predictive_to_xarray(self):
"""Convert prior_predictive samples to xarray."""
prior = self.prior
prior_model = self.prior_model
prior_predictive = self.prior_predictive
data = get_draws_stan3(prior, model=prior_model, variables=prior_predictive)
return dict_to_dataset(data, library=self.stan, coords=self.coords, dims=self.dims)
posterior_predictive = self.posterior_predictive
if posterior_predictive is None:
posterior_predictive = []
elif isinstance(posterior_predictive, str):
posterior_predictive = [posterior_predictive]
log_likelihood = self.log_likelihood
if not isinstance(log_likelihood, str):
log_likelihood = []
else:
log_likelihood = [log_likelihood]
ignore = posterior_predictive + log_likelihood
data = get_draws_stan3(posterior, model=posterior_model, ignore=ignore)
return dict_to_dataset(data, library=self.stan, coords=self.coords, dims=self.dims)
if self.prior is None:
return {"prior": None, "prior_predictive": None}
if self.posterior is not None:
prior_vars = list(self._samples.keys())
prior_predictive_vars = [key for key in self.prior.keys() if key not in prior_vars]
else:
prior_vars = self.prior.keys()
prior_predictive_vars = None
priors_dict = {}
for group, var_names in zip(
("prior", "prior_predictive"), (prior_vars, prior_predictive_vars)
):
priors_dict[group] = (
None
if var_names is None
else dict_to_dataset(
{k: utils.expand_dims(self.prior[k]) for k in var_names},
library=self.numpyro,
coords=self.coords,
dims=self.dims,
)
)
return priors_dict
def prior_predictive_to_xarray(self):
"""Convert prior_predictive samples to xarray."""
prior = self.prior
prior_predictive = self.prior_predictive
data = get_draws(prior, variables=prior_predictive)
return dict_to_dataset(data, library=self.pystan, coords=self.coords, dims=self.dims)