Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _convert_axes_to_int(axes):
if isinstance(axes, ZfitSpace):
axes = axes.axes
else:
axes = convert_to_container(axes, container=tuple)
return axes
def __init__(self, pdfs, name="BaseFunctor", **kwargs):
self.pdfs = convert_to_container(pdfs)
super().__init__(models=self.pdfs, name=name, **kwargs)
self._set_norm_range_from_daugthers()
def extract_extended_pdfs(pdfs: Union[Iterable[ZfitPDF], ZfitPDF]) -> List[ZfitPDF]:
"""Return all extended pdfs that are daughters.
Args:
pdfs (Iterable[pdfs]):
Returns:
List[pdfs]:
"""
from ..models.functor import BaseFunctor
pdfs = convert_to_container(pdfs)
indep_pdfs = []
for pdf in pdfs:
if not pdf.is_extended:
continue
elif isinstance(pdf, BaseFunctor):
if all(pdf.pdfs_extended):
indep_pdfs.extend(extract_extended_pdfs(pdfs=pdf.pdfs))
elif not any(pdf.pdfs_extended):
indep_pdfs.append(pdf)
else:
assert False, "Should not reach this point, wrong assumptions. Please report bug."
else: # extended, but not a functor
indep_pdfs.append(pdf)
return indep_pdfs
def _sort_value(self, value, obs: Tuple[str]):
obs = convert_to_container(value=obs, container=tuple)
perm_indices = self.space.axes if self.space.axes != tuple(range(value.shape[-1])) else False
# permutate = perm_indices is not None
if obs:
if not frozenset(obs) <= frozenset(self.obs):
raise ValueError("The observable(s) {} are not contained in the dataset. "
"Only the following are: {}".format(frozenset(obs) - frozenset(self.obs),
self.obs))
perm_indices = self.space.get_axes(obs=obs)
# values = list(values[self.obs.index(o)] for o in obs if o in self.obs)
if perm_indices:
value = ztf.unstack_x(value)
value = list(value[i] for i in perm_indices)
value = ztf.stack_x(value)
return value
def __init__(self, funcs, name="BaseFunctorFunc", params=None, **kwargs):
funcs = convert_to_container(funcs)
if params is None:
params = {}
# for func in funcs:
# params.update(func.params)
self.funcs = funcs
super().__init__(name=name, models=self.funcs, params=params, **kwargs)
def _check_convert_obs(self, obs):
obs = convert_to_container(obs)
wrong_type_obs = tuple(o for o in obs if not isinstance(o, str))
if wrong_type_obs:
raise TypeError(f"The observable(s) {wrong_type_obs} are not strings.")
return obs
def extract_limits_from_dict(limits_dict, obs=None, axes=None):
if (obs is None) and (axes is None):
raise ValueError("Need to specify at least one, obs or axes.")
elif (obs is not None) and (axes is not None):
axes = None # obs has precedency
if obs is None:
obs_in_use = False
coords_to_extract = axes
else:
obs_in_use = True
coords_to_extract = obs
coords_to_extract = convert_to_container(coords_to_extract)
coords_to_extract = set(coords_to_extract)
limits_to_eval = {}
limit_dict = limits_dict['obs' if obs_in_use else 'axes'].items()
keys_sorted = sorted(limit_dict, key=lambda x: len(x[0]), reverse=True)
for key_coords, limit in keys_sorted:
coord_intersec = frozenset(key_coords).intersection(coords_to_extract)
if not coord_intersec: # this limit does not contain any requested obs
continue
if coord_intersec == frozenset(key_coords):
if isinstance(limit, ZfitOrderableDimensional): # drop coordinates if given
if obs_in_use:
limit = limit.with_axes(None)
else:
limit = limit.with_obs(None)
limits_to_eval[key_coords] = limit
params (list(:py:class:`~zfit.Parameter`)): The parameters to calculate the
errors error. If None, use all parameters.
sigma (float): Errors are calculated with respect to `sigma` std deviations.
rtol (float, default=0.01): relative tolerance between the computed and the exact roots
method (str, defautl='hybr'): type of solver, `method` argument of `scipy.optimize.root_
covariance_method (str or Callable): The method to use to calculate the correlation matrix. Valid choices are
{'minuit_hesse', 'hesse_np'} or a Callable.
Returns:
`OrderedDict`: A `OrderedDict` containing as keys the parameter and as value a `dict` which
contains two keys 'lower' and 'upper', holding the calculated errors.
Example: result[par1]['upper'] -> the asymmetric upper error of 'par1'
`FitResult` or `None`: a fit result is returned when a new minimum is found during the loss scan
"""
params = convert_to_container(params)
new_result = None
all_params = list(result.params.keys())
loss = result.loss
errordef = loss.errordef
fmin = result.fmin
rtol *= errordef
minimizer = result.minimizer
covariance = result.covariance(method=covariance_method, as_dict=True)
set_values(all_params, result)
try:
to_return = {}
for param in params:
def _check_convert_input_axes(self, axes: ztyping.AxesTypeInput,
allow_none: bool = False) -> ztyping.AxesTypeReturn:
if axes is None:
if allow_none:
return None
else:
raise AxesNotSpecifiedError("TODO: Cannot be None")
if isinstance(axes, ZfitSpace):
obs = axes.obs
else:
axes = convert_to_container(value=axes, container=tuple) # TODO(Mayou36): extend like _check_obs?
return axes
def convert_to_obs_str(obs, container=tuple):
"""Convert `obs` to the list of obs, also if it is a :py:class:`~ZfitSpace`. Return None if obs is None.
Raises:
TypeError: if the observable is not a string
"""
if obs is None:
return obs
if isinstance(obs, ZfitDimensional):
new_obs = obs.obs
else:
obs = convert_to_container(value=obs, container=container)
new_obs = []
for ob in obs:
if not isinstance(ob, str):
raise TypeError(f"Observables have to be string, not {ob} as in {obs}")
else:
new_obs.append(ob)
return container(new_obs)