minus_log parameterized function (#265)

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

"""iMars3D normalization module."""

import logging

import param

# package imports

from imars3d.backend.util.functions import clamp_max_workers

# third party imports

import numpy as np

import param

from tomopy.prep.normalize import minus_log as tm_minus_log

# standard imports

import logging

logger = logging.getLogger(__name__)

    max_workers = param.Integer(

        default=0,

        bounds=(0, None),

        doc="Maximum number of processes allowed during loading",

        doc="Maximum number of processes allowed during execution",

    )

    def __call__(self, **params):

        _ = self.instance(**params)

        # sanitize arguments

        params = param.ParamOverrides(self, params)

        # import pdb; pdb.set_trace()

        # type validation is done, now replacing max_worker with an actual integer

        self.max_workers = clamp_max_workers(params.max_workers)

        logger.debug(f"max_worker={self.max_workers}")

        logger.info("FINISHED Executing Filter: Normalization")

        return arrays_normalized

class minus_log(param.ParameterizedFunction):

    r"""Computation of the minus natural log of a given array.

    Calls `tomopy.prep.normalize.minus_log`.

    Parameters

    ----------

    arrays : np.ndarray

        any multidimensional array with values greater than one.

    max_workers : int

        number of cores to use for parallel processing, default is 0, which means using all available cores.

    Returns

    -------

        np.ndarray

    Raises

    ------

    ValueError

        Any entry in the input array is equal or smaller than zero

    """

    arrays = param.Array(doc="any multidimensional array with values greater than one.", default=None)

    max_workers = param.Integer(

        default=0,

        bounds=(0, None),

        doc="Maximum number of processes allowed during execution",

    )

    def __call__(self, **params):

        """Perform minus_log via tomopy."""

        logger.info("Executing Filter: minus_log")

        # type*bounds check via Parameter

        _ = self.instance(**params)

        # sanitize arguments

        params = param.ParamOverrides(self, params)

        # type validation is done, now replacing max_worker with an actual integer

        self.max_workers = clamp_max_workers(params.max_workers)

        logger.debug(f"max_worker={self.max_workers}")

        if not np.all(params.arrays > 0.0):

            raise ValueError("'minus_log' cannot be applied to arrays containing elements equal or smaller than zero")

        arrays_normalized = tm_minus_log(params.arrays, ncore=self.max_workers)

        logger.info("FINISHED Executing Filter: minus_log")

        return arrays_normalized

#!/usr/bin/env python3

# package imports

from imars3d.backend.preparation.normalization import minus_log, normalization

from imars3d.backend.workflow.engine import WorkflowEngineAuto

# third party imports

import numpy as np

import pytest

import skimage

from scipy.ndimage import gaussian_filter

from imars3d.backend.preparation.normalization import normalization

import skimage

# standard library

import json

from pathlib import Path

def generate_fake_darkfield(

    assert diff < 0.01

class TestMinusLog:

    @pytest.mark.parametrize("ncore", [1, 2])

    def test_execution(self, ncore: int) -> None:

        r"""Invoke minus_log on a simple array with different number of processing cores"""

        arrays = 42 * np.ones(6).reshape((1, 2, 3))

        arrays_normalized = minus_log(arrays=arrays, max_workers=ncore)

        np.testing.assert_allclose(arrays_normalized, -np.log(arrays), atol=1.0e-3)

    def test_dryrun(self, JSON_DIR: Path) -> None:

        r"""Validate a JSON file containing a minus_log task"""

        task = json.loads(

            """

        {

            "name": "minus_log",

            "function": "imars3d.backend.preparation.normalization.minus_log",

            "inputs": {"arrays": "ct", "max_workers": 2},

            "outputs": ["ct"]

        }"""

        )

        config = json.load(open(JSON_DIR / "good_non_interactive_full.json"))

        config["tasks"].insert(6, task)  # insert minus_log task after normalization

        workflow = WorkflowEngineAuto(config)

        workflow._dryrun()

    def test_exception(self) -> None:

        r"""Applying minus_log to an array with elements equal or smaller than zero should raise ValueError"""

        with pytest.raises(ValueError) as e:

            minus_log(arrays=1.0 * np.arange(42), max_workers=1)

        assert "'minus_log' cannot be applied" in str(e.value)

if __name__ == "__main__":

    pytest.main([__file__])