core

navis.core.NeuronProcessor

Apply function across all neurons of a neuronlist.

This assumes that the first argument for the function accepts a single neuron.

Source code in navis/core/core_utils.py

class NeuronProcessor:
    """Apply function across all neurons of a neuronlist.

    This assumes that the first argument for the function accepts a single
    neuron.
    """

    def __init__(self,
                 nl: 'core.NeuronList',
                 function: Callable,
                 parallel: bool = False,
                 n_cores: int = os.cpu_count() // 2,
                 chunksize: int = 1,
                 progress: bool = True,
                 warn_inplace: bool = True,
                 omit_failures: bool = False,
                 exclude_zip: list = [],
                 desc: Optional[str] = None):
        if utils.is_iterable(function):
            if len(function) != len(nl):
                raise ValueError('Number of functions must match neurons.')
            self.funcs = function
            self.function = function[0]
        elif callable(function):
            self.funcs = [function] * len(nl)
            self.function = function
        else:
            raise TypeError('Expected `function` to be callable or list '
                            f'thereof,  got "{type(function)}"')

        self.nl = nl
        self.desc = desc
        self.parallel = parallel
        self.n_cores = n_cores
        self.chunksize = chunksize
        self.progress = progress
        self.warn_inplace = warn_inplace
        self.exclude_zip = exclude_zip
        self.omit_failures = omit_failures

        # This makes sure that help and name match the functions being called
        functools.update_wrapper(self, self.function)

    def __call__(self, *args, **kwargs):
        # Explicitly providing these parameters overwrites defaults
        parallel = kwargs.pop('parallel', self.parallel)
        n_cores = kwargs.pop('n_cores', self.n_cores)

        # We will check, for each argument, if it matches the number of
        # functions to run. If they it does, we will zip the values
        # with the neurons
        parsed_args = []
        parsed_kwargs = []

        for i, n in enumerate(self.nl):
            parsed_args.append([])
            parsed_kwargs.append({})
            for k, a in enumerate(args):
                if k in self.exclude_zip:
                    parsed_args[i].append(a)
                elif not utils.is_iterable(a) or len(a) != len(self.nl):
                    parsed_args[i].append(a)
                else:
                    parsed_args[i].append(a[i])

            for k, v in kwargs.items():
                if k in self.exclude_zip:
                    parsed_kwargs[i][k] = v
                elif not utils.is_iterable(v) or len(v) != len(self.nl):
                    parsed_kwargs[i][k] = v
                else:
                    parsed_kwargs[i][k] = v[i]

        # Silence loggers (except Errors)
        level = logger.getEffectiveLevel()

        if level < 30:
            logger.setLevel('WARNING')

        # Apply function
        if parallel:
            if not ProcessingPool:
                raise ModuleNotFoundError(
                    'navis relies on pathos for multiprocessing!'
                    'Please install pathos and try again:\n'
                    '  pip3 install pathos -U'
                    )

            if self.warn_inplace and kwargs.get('inplace', False):
                logger.warning('`inplace=True` does not work with '
                               'multiprocessing ')

            with ProcessingPool(n_cores) as pool:
                combinations = list(zip(self.funcs,
                                        parsed_args,
                                        parsed_kwargs))
                chunksize = kwargs.pop('chunksize', self.chunksize)  # max(int(len(combinations) / 100), 1)

                if not self.omit_failures:
                    wrapper = _call
                else:
                    wrapper = _try_call

                res = list(config.tqdm(pool.imap(wrapper,
                                                 combinations,
                                                 chunksize=chunksize),
                                       total=len(combinations),
                                       desc=self.desc,
                                       disable=config.pbar_hide or not self.progress,
                                       leave=config.pbar_leave))
        else:
            res = []
            for i, n in enumerate(config.tqdm(self.nl, desc=self.desc,
                                              disable=(config.pbar_hide
                                                       or not self.progress
                                                       or len(self.nl) <= 1),
                                              leave=config.pbar_leave)):
                try:
                    res.append(self.funcs[i](*parsed_args[i], **parsed_kwargs[i]))
                except BaseException as e:
                    if self.omit_failures:
                        res.append(FailedRun(func=self.funcs[i],
                                             args=parsed_args[i],
                                             kwargs=parsed_kwargs[i],
                                             exception=e))
                    else:
                        raise

        # Reset logger level to previous state
        logger.setLevel(level)

        failed = np.array([isinstance(r, FailedRun) for r in res])
        res = [r for r in res if not isinstance(r, FailedRun)]
        if any(failed):
            logger.warn(f'{sum(failed)} of {len(self.funcs)} runs failed. '
                        'Set logging to debug (`navis.set_loggers("DEBUG")`) '
                        'or repeat with `omit_failures=False` for details.')
            failed_ids = self.nl.id[np.where(failed)].astype(str)
            logger.debug(f'The following IDs failed to complete: {", ".join(failed_ids)}')

        # If result is a list of neurons, combine them back into a single list
        is_neuron = [isinstance(r, (core.NeuronList, core.BaseNeuron)) for r in res]
        if all(is_neuron):
            return self.nl.__class__(utils.unpack_neurons(res))
        # If results are all None return nothing instead of a list of [None, ..]
        if np.all([r is None for r in res]):
            res = None
        # If not all neurons simply return results and let user deal with it
        return res

navis.core.to_neuron_space

Convert units to match neuron space.

Note that trying to convert units for non-isometric neurons will fail.

PARAMETER	DESCRIPTION
units	The units to convert to neuron units. Simple numbers are just passed through. TYPE: number | str | pint.Quantity | pint.Units
neuron	A single neuron. TYPE: Neuron
on_error	What to do if an error occurs (e.g. because `neuron` does not have units specified). If "ignore" will simply return `units` unchanged. TYPE: "raise" | "ignore" DEFAULT: 'raise'

RETURNS	DESCRIPTION
float	The units in neuron space. Note that this number may be rounded to avoid ugly floating point precision issues such as 0.124999999999999 instead of 0.125.

Examples:

>>> import navis
>>> # Example neurons are in 8x8x8nm voxel space
>>> n = navis.example_neurons(1)
>>> navis.core.to_neuron_space('1 nm', n)
0.125
>>> # Alternatively use the neuron method
>>> n.map_units('1 nm')
0.125
>>> # Numbers are passed-through
>>> n.map_units(1)
1
>>> # For neuronlists
>>> nl = navis.example_neurons(3)
>>> nl.map_units(1)
[1, 1, 1]
>>> nl.map_units('1 nanometer')
[0.125, 0.125, 0.125]

Source code in navis/core/core_utils.py

def to_neuron_space(units: Union[int, float, pint.Quantity, pint.Unit],
                    neuron: core.BaseNeuron,
                    on_error: Union[Literal['ignore'],
                                    Literal['raise']] = 'raise'):
    """Convert units to match neuron space.

    Note that trying to convert units for non-isometric neurons will fail.

    Parameters
    ----------
    units :     number | str | pint.Quantity | pint.Units
                The units to convert to neuron units. Simple numbers are just
                passed through.
    neuron :    Neuron
                A single neuron.
    on_error :  "raise" | "ignore"
                What to do if an error occurs (e.g. because `neuron` does not
                have units specified). If "ignore" will simply return `units`
                unchanged.

    Returns
    -------
    float
                The units in neuron space. Note that this number may be rounded
                to avoid ugly floating point precision issues such as
                0.124999999999999 instead of 0.125.

    Examples
    --------
    >>> import navis
    >>> # Example neurons are in 8x8x8nm voxel space
    >>> n = navis.example_neurons(1)
    >>> navis.core.to_neuron_space('1 nm', n)
    0.125
    >>> # Alternatively use the neuron method
    >>> n.map_units('1 nm')
    0.125
    >>> # Numbers are passed-through
    >>> n.map_units(1)
    1
    >>> # For neuronlists
    >>> nl = navis.example_neurons(3)
    >>> nl.map_units(1)
    [1, 1, 1]
    >>> nl.map_units('1 nanometer')
    [0.125, 0.125, 0.125]

    """
    utils.eval_param(on_error, name='on_error',
                     allowed_values=('ignore', 'raise'))
    utils.eval_param(neuron, name='neuron', allowed_types=(core.BaseNeuron, ))

    # If string, convert to units
    if isinstance(units, str):
        units = pint.Quantity(units)
    # If not a pint object (i.e. just a number)
    elif not isinstance(units, (pint.Quantity, pint.Unit)):
        return units

    if neuron.units.dimensionless:
        if on_error == 'raise':
            raise ValueError(f'Unable to convert "{str(units)}": Neuron units '
                             'unknown or dimensionless.')
        else:
            return units

    if not neuron.is_isometric:
        if on_error == 'raise':
            raise ValueError(f'Unable to convert "{str(units)}": neuron is not '
                             'isometric ({neuron.units}).')
        else:
            return units

    # If input was e.g. `units="1"`
    if units.dimensionless:
        return units.magnitude

    # First convert to same unit as neuron units
    units = units.to(neuron.units)

    # Now convert magnitude
    mag = units.magnitude / neuron.units.magnitude

    # Rounding may not be exactly kosher but it avoids floating point issues
    # like 124.9999999999999 instead of 125
    # I hope that in practice it won't screw things up:
    # even if asking for
    return utils.round_smart(mag)