Neps space

This module provides functionality for resolving NePS spaces, including sampling from domains, resolving pipelines, and handling various resolvable objects.

NepsCompatConverter #

A class to convert between NePS configurations and NEPS-compatible configurations. It provides methods to convert a SamplingResolutionContext to a NEPS-compatible config and to convert a NEPS-compatible config back to a SamplingResolutionContext.

from_neps_config `classmethod` #

from_neps_config(
    config: Mapping[str, Any],
) -> _FromNepsConfigResult

Convert a NEPS-compatible config to a SamplingResolutionContext.

PARAMETER	DESCRIPTION
`config`	A mapping of NEPS-compatible configuration keys to their values. TYPE: `Mapping[str, Any]`

RETURNS	DESCRIPTION
`_FromNepsConfigResult`	A _FromNepsConfigResult containing predefined samplings, environment values, and extra kwargs.

RAISES	DESCRIPTION
`ValueError`	If the config is not a valid NEPS-compatible config.

Source code in neps/space/neps_spaces/neps_space.py

@classmethod
def from_neps_config(
    cls,
    config: Mapping[str, Any],
) -> _FromNepsConfigResult:
    """Convert a NEPS-compatible config to a SamplingResolutionContext.

    Args:
        config: A mapping of NEPS-compatible configuration keys to their values.

    Returns:
        A _FromNepsConfigResult containing predefined samplings,
            environment values, and extra kwargs.

    Raises:
        ValueError: If the config is not a valid NEPS-compatible config.
    """
    predefined_samplings = {}
    environment_values = {}
    extra_kwargs = {}

    for name, value in config.items():
        if name.startswith(cls._SAMPLING_PREFIX):
            sampling_path = name[cls._SAMPLING_PREFIX_LEN :]
            predefined_samplings[sampling_path] = value
        elif name.startswith(cls._ENVIRONMENT_PREFIX):
            env_name = name[cls._ENVIRONMENT_PREFIX_LEN :]
            environment_values[env_name] = value
        else:
            extra_kwargs[name] = value

    return cls._FromNepsConfigResult(
        predefined_samplings=predefined_samplings,
        environment_values=environment_values,
        extra_kwargs=extra_kwargs,
    )

to_neps_config `classmethod` #

to_neps_config(
    resolution_context: SamplingResolutionContext,
) -> Mapping[str, Any]

Convert a SamplingResolutionContext to a NEPS-compatible config.

PARAMETER	DESCRIPTION
`resolution_context`	The SamplingResolutionContext to convert. TYPE: `SamplingResolutionContext`

RETURNS	DESCRIPTION
`Mapping[str, Any]`	A mapping of NEPS-compatible configuration keys to their values.

RAISES	DESCRIPTION
`ValueError`	If the resolution_context is not a SamplingResolutionContext.

Source code in neps/space/neps_spaces/neps_space.py

@classmethod
def to_neps_config(
    cls,
    resolution_context: SamplingResolutionContext,
) -> Mapping[str, Any]:
    """Convert a SamplingResolutionContext to a NEPS-compatible config.

    Args:
        resolution_context: The SamplingResolutionContext to convert.

    Returns:
        A mapping of NEPS-compatible configuration keys to their values.

    Raises:
        ValueError: If the resolution_context is not a SamplingResolutionContext.
    """
    config: dict[str, Any] = {}

    samplings_made = resolution_context.samplings_made
    for sampling_path, value in samplings_made.items():
        config[f"{cls._SAMPLING_PREFIX}{sampling_path}"] = value

    environment_values = resolution_context.environment_values
    for env_name, value in environment_values.items():
        config[f"{cls._ENVIRONMENT_PREFIX}{env_name}"] = value

    return config

SamplingResolutionContext #

SamplingResolutionContext(
    *,
    resolution_root: Resolvable,
    domain_sampler: DomainSampler,
    environment_values: Mapping[str, Any]
)

A context for resolving samplings in a NePS space. It manages the resolution root, domain sampler, environment values, and keeps track of samplings made and resolved objects.

PARAMETER	DESCRIPTION
`resolution_root`	The root of the resolution, which should be a Resolvable object. TYPE: `Resolvable`
`domain_sampler`	The DomainSampler to use for sampling from Domain objects. TYPE: `DomainSampler`
`environment_values`	A mapping of environment values that are fixed and not related to samplings. These values can be used in the resolution process. TYPE: `Mapping[str, Any]`

RAISES	DESCRIPTION
`ValueError`	If the resolution_root is not a Resolvable, or if the domain_sampler is not a DomainSampler, or if the environment_values is not a Mapping.

PARAMETER	DESCRIPTION
`resolution_root`	The root of the resolution, which should be a Resolvable object. TYPE: `Resolvable`
`domain_sampler`	The DomainSampler to use for sampling from Domain objects. TYPE: `DomainSampler`
`environment_values`	A mapping of environment values that are fixed and not related to samplings. These values can be used in the resolution process. TYPE: `Mapping[str, Any]`

RAISES	DESCRIPTION
`ValueError`	If the resolution_root is not a Resolvable, or if the domain_sampler is not a DomainSampler, or if the environment_values is not a Mapping.

Source code in neps/space/neps_spaces/neps_space.py

def __init__(
    self,
    *,
    resolution_root: Resolvable,
    domain_sampler: DomainSampler,
    environment_values: Mapping[str, Any],
):
    """Initialize the SamplingResolutionContext with a resolution root, domain
    sampler, and environment values.

    Args:
        resolution_root: The root of the resolution, which should be a Resolvable
            object.
        domain_sampler: The DomainSampler to use for sampling from Domain objects.
        environment_values: A mapping of environment values that are fixed and not
            related to samplings. These values can be used in the resolution process.

    Raises:
        ValueError: If the resolution_root is not a Resolvable, or if the
            domain_sampler is not a DomainSampler, or if the environment_values is
            not a Mapping.
    """
    if not isinstance(resolution_root, Resolvable):
        raise ValueError(
            "The received `resolution_root` is not a Resolvable:"
            f" {resolution_root!r}."
        )

    if not isinstance(domain_sampler, DomainSampler):
        raise ValueError(
            "The received `domain_sampler` is not a DomainSampler:"
            f" {domain_sampler!r}."
        )

    if not isinstance(environment_values, Mapping):
        raise ValueError(
            "The received `environment_values` is not a Mapping:"
            f" {environment_values!r}."
        )

    # `_resolution_root` stores the root of the resolution.
    self._resolution_root: Resolvable = resolution_root

    # `_domain_sampler` stores the object responsible for sampling from Domain
    # objects.
    self._domain_sampler = domain_sampler

    # # `_environment_values` stores fixed values from outside.
    # # They are not related to samplings and can not be mutated or similar.
    self._environment_values = environment_values

    # `_samplings_made` stores the values we have sampled
    # and can be used later in case we want to redo a resolving.
    self._samplings_made: dict[str, Any] = {}

    # `_resolved_objects` stores the intermediate values to make re-use possible.
    self._resolved_objects: dict[Any, Any] = {}

    # `_current_path_parts` stores the current path we are resolving.
    self._current_path_parts: list[str] = []

environment_values `property` #

environment_values: Mapping[str, Any]

Get the environment values that are fixed and not related to samplings.

RETURNS	DESCRIPTION
`Mapping[str, Any]`	A mapping of environment variable names to their values.

resolution_root `property` #

resolution_root: Resolvable

Get the root of the resolution.

RETURNS	DESCRIPTION
`Resolvable`	The root of the resolution, which should be a Resolvable object.

samplings_made `property` #

samplings_made: Mapping[str, Any]

Get the samplings made during the resolution process.

RETURNS	DESCRIPTION
`Mapping[str, Any]`	A mapping of paths to sampled values.

add_resolved #

add_resolved(original: Any, resolved: Any) -> None

Add a resolved object to the context.

PARAMETER	DESCRIPTION
`original`	The original object that was resolved. TYPE: `Any`
`resolved`	The resolved value of the original object. TYPE: `Any`

RAISES	DESCRIPTION
`ValueError`	If the original object was already resolved or if it is a Resample.

Source code in neps/space/neps_spaces/neps_space.py

def add_resolved(self, original: Any, resolved: Any) -> None:
    """Add a resolved object to the context.

    Args:
        original: The original object that was resolved.
        resolved: The resolved value of the original object.

    Raises:
        ValueError: If the original object was already resolved or if it is a
            Resample.
    """
    if self.was_already_resolved(original):
        raise ValueError(
            f"Original object has already been resolved: {original!r}. "
            + "\nIf you are doing resampling by name, "
            + "make sure you are not forgetting to request resampling also for"
            " related objects." + "\nOtherwise it could lead to infinite recursion."
        )
    if isinstance(original, Resample):
        raise ValueError(
            f"Attempting to add a Resample object to resolved values: {original!r}."
        )
    self._resolved_objects[original] = resolved

get_resolved #

get_resolved(obj: Any) -> Any

Get the resolved value for the given object.

PARAMETER	DESCRIPTION
`obj`	The object for which to get the resolved value. TYPE: `Any`

RETURNS	DESCRIPTION
`Any`	The resolved value of the object.

RAISES	DESCRIPTION
`ValueError`	If the object was not already resolved in the context.

Source code in neps/space/neps_spaces/neps_space.py

def get_resolved(self, obj: Any) -> Any:
    """Get the resolved value for the given object.

    Args:
        obj: The object for which to get the resolved value.

    Returns:
        The resolved value of the object.

    Raises:
        ValueError: If the object was not already resolved in the context.
    """
    try:
        return self._resolved_objects[obj]
    except KeyError as err:
        raise ValueError(
            f"Given object was not already resolved. Please check first: {obj!r}"
        ) from err

get_value_from_environment #

get_value_from_environment(var_name: str) -> Any

Get a value from the environment variables.

PARAMETER	DESCRIPTION
`var_name`	The name of the environment variable to get the value from. TYPE: `str`

RETURNS	DESCRIPTION
`Any`	The value of the environment variable.

RAISES	DESCRIPTION
`ValueError`	If the environment variable is not found in the context.

Source code in neps/space/neps_spaces/neps_space.py

def get_value_from_environment(self, var_name: str) -> Any:
    """Get a value from the environment variables.

    Args:
        var_name: The name of the environment variable to get the value from.

    Returns:
        The value of the environment variable.

    Raises:
        ValueError: If the environment variable is not found in the context.
    """
    try:
        return self._environment_values[var_name]
    except KeyError as err:
        raise ValueError(
            f"No value is available for the environment variable {var_name!r}."
        ) from err

resolving #

resolving(_obj: Any, name: str) -> Generator[None]

Context manager for resolving an object in the current resolution context.

PARAMETER	DESCRIPTION
`_obj`	The object being resolved, can be any type. TYPE: `Any`
`name`	The name of the object being resolved, used for debugging. TYPE: `str`

RAISES	DESCRIPTION
`ValueError`	If the name is not a valid string.

Source code in neps/space/neps_spaces/neps_space.py

@contextlib.contextmanager
def resolving(self, _obj: Any, name: str) -> Generator[None]:
    """Context manager for resolving an object in the current resolution context.

    Args:
        _obj: The object being resolved, can be any type.
        name: The name of the object being resolved, used for debugging.

    Raises:
        ValueError: If the name is not a valid string.
    """
    if not name or not isinstance(name, str):
        raise ValueError(
            f"Given name for what we are resolving is invalid: {name!r}."
        )

    # It is possible that the received object has already been resolved.
    # That is expected and is okay, so no check is made for it.
    # For example, in the case of a Resample we can receive the same object again.

    self._current_path_parts.append(name)
    try:
        yield
    finally:
        self._current_path_parts.pop()

sample_from #

sample_from(domain_obj: Domain) -> Any

Sample a value from the given domain object.

PARAMETER	DESCRIPTION
`domain_obj`	The domain object from which to sample a value. TYPE: `Domain`

RETURNS	DESCRIPTION
`Any`	The sampled value from the domain object.

RAISES	DESCRIPTION
`ValueError`	If the domain object was already resolved or if the path has already been sampled from.

Source code in neps/space/neps_spaces/neps_space.py

def sample_from(self, domain_obj: Domain) -> Any:
    """Sample a value from the given domain object.

    Args:
        domain_obj: The domain object from which to sample a value.

    Returns:
        The sampled value from the domain object.

    Raises:
        ValueError: If the domain object was already resolved or if the path
            has already been sampled from.
    """
    # Each `domain_obj` is only ever sampled from once.
    # This is okay and the expected behavior.
    # For each `domain_obj`, its sampled value is either directly stored itself,
    # or is used in some other Resolvable.
    # In both cases that sampled value is cached for later uses,
    # and so the `domain_obj` will not be re-sampled from again.
    if self.was_already_resolved(domain_obj):
        raise ValueError(
            "We have already sampled a value for the given domain object:"
            f" {domain_obj!r}." + "\nThis should not be happening."
        )

    # Construct the unique sampling path for this domain object
    current_path = construct_sampling_path(
        path_parts=self._current_path_parts,
        domain_obj=domain_obj,
    )

    if current_path in self._samplings_made:
        # We have already sampled a value for this path. This should not happen.
        # Every time we sample a domain, it should have its own different path.
        raise ValueError(
            f"We have already sampled a value for the current path: {current_path!r}."
            + "\nThis should not be happening."
        )

    sampled_value = self._domain_sampler(
        domain_obj=domain_obj,
        current_path=current_path,
    )

    self._samplings_made[current_path] = sampled_value
    return self._samplings_made[current_path]

was_already_resolved #

was_already_resolved(obj: Any) -> bool

Check if the given object was already resolved in the current context.

PARAMETER	DESCRIPTION
`obj`	The object to check if it was already resolved. TYPE: `Any`

RETURNS	DESCRIPTION
`bool`	True if the object was already resolved, False otherwise.

Source code in neps/space/neps_spaces/neps_space.py

def was_already_resolved(self, obj: Any) -> bool:
    """Check if the given object was already resolved in the current context.

    Args:
        obj: The object to check if it was already resolved.

    Returns:
        True if the object was already resolved, False otherwise.
    """
    return obj in self._resolved_objects

SamplingResolver #

A class responsible for resolving samplings in a NePS space. It uses a SamplingResolutionContext to manage the resolution process, and a DomainSampler to sample values from Domain objects.

call #

__call__(
    obj: Resolvable,
    domain_sampler: DomainSampler,
    environment_values: Mapping[str, Any],
) -> tuple[Resolvable, SamplingResolutionContext]

Resolve the given object in the context of the provided domain sampler and environment values.

PARAMETER	DESCRIPTION
`obj`	The Resolvable object to resolve. TYPE: `Resolvable`
`domain_sampler`	The DomainSampler to use for sampling from Domain objects. TYPE: `DomainSampler`
`environment_values`	A mapping of environment values that are fixed and not related to samplings. TYPE: `Mapping[str, Any]`

RETURNS	DESCRIPTION
`tuple[Resolvable, SamplingResolutionContext]`	A tuple containing the resolved object and the SamplingResolutionContext.

RAISES	DESCRIPTION
`ValueError`	If the object is not a Resolvable, or if the domain_sampler is not a DomainSampler, or if the environment_values is not a Mapping.

Source code in neps/space/neps_spaces/neps_space.py

def __call__(
    self,
    obj: Resolvable,
    domain_sampler: DomainSampler,
    environment_values: Mapping[str, Any],
) -> tuple[Resolvable, SamplingResolutionContext]:
    """Resolve the given object in the context of the provided domain sampler and
    environment values.

    Args:
        obj: The Resolvable object to resolve.
        domain_sampler: The DomainSampler to use for sampling from Domain objects.
        environment_values: A mapping of environment values that are fixed and not
            related to samplings.

    Returns:
        A tuple containing the resolved object and the
            SamplingResolutionContext.

    Raises:
        ValueError: If the object is not a Resolvable, or if the domain_sampler
            is not a DomainSampler, or if the environment_values is not a Mapping.
    """
    context = SamplingResolutionContext(
        resolution_root=obj,
        domain_sampler=domain_sampler,
        environment_values=environment_values,
    )
    return self._resolve(obj, "Resolvable", context), context

adjust_evaluation_pipeline_for_neps_space #

adjust_evaluation_pipeline_for_neps_space(
    evaluation_pipeline: Callable[
        ..., EvaluatePipelineReturn
    ],
    pipeline_space: P,
    operation_converter: Callable[
        [Operation], Any
    ] = convert_operation_to_callable,
) -> Callable

Adjust the evaluation pipeline to work with a NePS space. This function wraps the evaluation pipeline to sample from the NePS space and convert the sampled pipeline to a format compatible with the evaluation pipeline.

PARAMETER	DESCRIPTION
`evaluation_pipeline`	The evaluation pipeline to adjust. TYPE: `Callable[..., EvaluatePipelineReturn]`
`pipeline_space`	The NePS pipeline space to sample from. TYPE: `P`
`operation_converter`	A callable to convert Operation objects to a format compatible with the evaluation pipeline. TYPE: `Callable[[Operation], Any]` DEFAULT: `convert_operation_to_callable`

RETURNS	DESCRIPTION
`Callable`	A wrapped evaluation pipeline that samples from the NePS space.

RAISES	DESCRIPTION
`ValueError`	If the evaluation_pipeline is not callable or if the pipeline_space is not a Pipeline object.

Source code in neps/space/neps_spaces/neps_space.py

def adjust_evaluation_pipeline_for_neps_space(
    evaluation_pipeline: Callable[..., EvaluatePipelineReturn],
    pipeline_space: P,
    operation_converter: Callable[[Operation], Any] = convert_operation_to_callable,
) -> Callable:
    """Adjust the evaluation pipeline to work with a NePS space.
    This function wraps the evaluation pipeline to sample from the NePS space
    and convert the sampled pipeline to a format compatible with the evaluation pipeline.

    Args:
        evaluation_pipeline: The evaluation pipeline to adjust.
        pipeline_space: The NePS pipeline space to sample from.
        operation_converter: A callable to convert Operation objects to a format
            compatible with the evaluation pipeline.

    Returns:
        A wrapped evaluation pipeline that samples from the NePS space.

    Raises:
        ValueError: If the evaluation_pipeline is not callable or if the
            pipeline_space is not a Pipeline object.
    """

    @functools.wraps(evaluation_pipeline)
    def inner(*args: Any, **kwargs: Any) -> Any:
        # `kwargs` can contain other things not related to
        # the samplings to make or to environment values.
        # That is not an issue. Those items will be passed through.

        sampled_pipeline_data = NepsCompatConverter.from_neps_config(config=kwargs)

        sampled_pipeline, _resolution_context = resolve(
            pipeline=pipeline_space,
            domain_sampler=OnlyPredefinedValuesSampler(
                predefined_samplings=sampled_pipeline_data.predefined_samplings,
            ),
            environment_values=sampled_pipeline_data.environment_values,
        )

        config = dict(**sampled_pipeline.get_attrs())

        for name, value in config.items():
            if isinstance(value, Operation):
                # If the operator is a not a string, we convert it to a callable.
                if isinstance(value.operator, str):
                    config[name] = value.operator
                else:
                    config[name] = operation_converter(value)

        # So that we still pass the kwargs not related to the config,
        # start with the extra kwargs we passed to the converter.
        new_kwargs = dict(**sampled_pipeline_data.extra_kwargs)
        # Then add all the kwargs from the config.
        new_kwargs.update(config)

        return evaluation_pipeline(*args, **new_kwargs)

    return inner

check_neps_space_compatibility #

check_neps_space_compatibility(
    optimizer_to_check: (
        OptimizerChoice
        | Mapping[str, Any]
        | tuple[OptimizerChoice, Mapping[str, Any]]
        | Callable[
            Concatenate[SearchSpace, ...], AskFunction
        ]
        | Callable[
            Concatenate[PipelineSpace, ...], AskFunction
        ]
        | Callable[
            Concatenate[SearchSpace | PipelineSpace, ...],
            AskFunction,
        ]
        | CustomOptimizer
        | Literal["auto"]
    ) = "auto",
) -> Literal["neps", "classic", "both"]

Check if the given optimizer is compatible with a NePS space. This function checks if the optimizer is a NePS-specific algorithm, a classic algorithm, or a combination of both.

PARAMETER DESCRIPTION

optimizer_to_check

The optimizer to check for compatibility. It can be a NePS-specific algorithm, a classic algorithm, or a combination of both.

RETURNS	DESCRIPTION
`Literal['neps', 'classic', 'both']`	A string indicating the compatibility: - "neps" if the optimizer is a NePS-specific algorithm, - "classic" if the optimizer is a classic algorithm, - "both" if the optimizer is a combination of both.

Source code in neps/space/neps_spaces/neps_space.py

def check_neps_space_compatibility(
    optimizer_to_check: (
        algorithms.OptimizerChoice
        | Mapping[str, Any]
        | tuple[algorithms.OptimizerChoice, Mapping[str, Any]]
        | Callable[
            Concatenate[SearchSpace, ...], optimizer.AskFunction
        ]  # Hack, while we transit
        | Callable[
            Concatenate[PipelineSpace, ...], optimizer.AskFunction
        ]  # from SearchSpace to
        | Callable[
            Concatenate[SearchSpace | PipelineSpace, ...], optimizer.AskFunction
        ]  # Pipeline
        | algorithms.CustomOptimizer
        | Literal["auto"]
    ) = "auto",
) -> Literal["neps", "classic", "both"]:
    """Check if the given optimizer is compatible with a NePS space.
    This function checks if the optimizer is a NePS-specific algorithm,
    a classic algorithm, or a combination of both.

    Args:
        optimizer_to_check: The optimizer to check for compatibility.
            It can be a NePS-specific algorithm, a classic algorithm,
            or a combination of both.

    Returns:
        A string indicating the compatibility:
            - "neps" if the optimizer is a NePS-specific algorithm,
            - "classic" if the optimizer is a classic algorithm,
            - "both" if the optimizer is a combination of both.
    """
    inner_optimizer = None
    if isinstance(optimizer_to_check, partial):
        inner_optimizer = optimizer_to_check.func
        while isinstance(inner_optimizer, partial):
            inner_optimizer = inner_optimizer.func

    only_classic_algorithm = (
        optimizer_to_check in _get_only_classic_algorithms_functions()
        or (
            inner_optimizer
            and inner_optimizer in _get_only_classic_algorithms_functions()
        )
        or (
            optimizer_to_check[0] in ONLY_CLASSIC_ALGORITHMS_NAMES
            if isinstance(optimizer_to_check, tuple)
            else False
        )
        or (
            optimizer_to_check in ONLY_CLASSIC_ALGORITHMS_NAMES
            if isinstance(optimizer_to_check, str)
            else False
        )
    )
    if only_classic_algorithm:
        return "classic"
    neps_and_classic_algorithm = (
        optimizer_to_check in _get_classic_and_neps_algorithms_functions()
        or (
            inner_optimizer
            and inner_optimizer in _get_classic_and_neps_algorithms_functions()
        )
        or optimizer_to_check == "auto"
        or (
            optimizer_to_check[0] in CLASSIC_AND_NEPS_ALGORITHMS_NAMES
            if isinstance(optimizer_to_check, tuple)
            else False
        )
        or (
            optimizer_to_check in CLASSIC_AND_NEPS_ALGORITHMS_NAMES
            if isinstance(optimizer_to_check, str)
            else False
        )
    )
    if neps_and_classic_algorithm:
        return "both"
    return "neps"

construct_sampling_path #

construct_sampling_path(
    path_parts: list[str], domain_obj: Domain
) -> str

Construct a sampling path for a domain object.

The sampling path uniquely identifies a sampled value in the resolution context. It consists of the hierarchical path through the pipeline space and a domain identifier that includes type and range information.

PARAMETER	DESCRIPTION
`path_parts`	The hierarchical path parts (e.g., ["Resolvable", "integer1"]). TYPE: `list[str]`
`domain_obj`	The domain object for which to construct the path. TYPE: `Domain`

RETURNS	DESCRIPTION
`str`	A string representing the full sampling path in the format: "::__" Example: "Resolvable.integer1::integer__0_1_False"

RAISES	DESCRIPTION
`ValueError`	If path_parts is empty or domain_obj is not a Domain.

Source code in neps/space/neps_spaces/neps_space.py

def construct_sampling_path(
    path_parts: list[str],
    domain_obj: Domain,
) -> str:
    """Construct a sampling path for a domain object.

    The sampling path uniquely identifies a sampled value in the resolution context.
    It consists of the hierarchical path through the pipeline space and a domain
    identifier that includes type and range information.

    Args:
        path_parts: The hierarchical path parts (e.g., ["Resolvable", "integer1"]).
        domain_obj: The domain object for which to construct the path.

    Returns:
        A string representing the full sampling path in the format:
        "<path.parts>::<type>__<range_compatibility_identifier>"
        Example: "Resolvable.integer1::integer__0_1_False"

    Raises:
        ValueError: If path_parts is empty or domain_obj is not a Domain.
    """
    if not path_parts:
        raise ValueError("path_parts cannot be empty")
    if not isinstance(domain_obj, Domain):
        raise ValueError(f"domain_obj must be a Domain, got {type(domain_obj)}")

    # Get the domain type name (e.g., "integer", "float", "categorical")
    domain_obj_type_name = type(domain_obj).__name__.lower()

    # Get the range compatibility identifier (e.g., "0_1_False" for
    # Integer(0, 1, log=False))
    range_compatibility_identifier = domain_obj.range_compatibility_identifier

    # Combine type and range: "integer__0_1_False"
    domain_obj_identifier = f"{domain_obj_type_name}__{range_compatibility_identifier}"

    # Join path parts with dots: "Resolvable.integer1"
    current_path = ".".join(path_parts)

    # Append domain identifier: "Resolvable.integer1::integer__0_1_False"
    current_path += "::" + domain_obj_identifier

    return current_path

convert_classic_to_neps_search_space #

convert_classic_to_neps_search_space(
    space: SearchSpace,
) -> PipelineSpace

Convert a classic SearchSpace to a NePS PipelineSpace if possible. This function converts a classic SearchSpace to a NePS PipelineSpace.

PARAMETER	DESCRIPTION
`space`	The classic SearchSpace to convert. TYPE: `SearchSpace`

RETURNS	DESCRIPTION
`PipelineSpace`	A NePS PipelineSpace.

Source code in neps/space/neps_spaces/neps_space.py

def convert_classic_to_neps_search_space(
    space: SearchSpace,
) -> PipelineSpace:
    """Convert a classic SearchSpace to a NePS PipelineSpace if possible.
    This function converts a classic SearchSpace to a NePS PipelineSpace.

    Args:
        space: The classic SearchSpace to convert.

    Returns:
        A NePS PipelineSpace.
    """

    class NEPSSpace(PipelineSpace):
        """A NePS-specific PipelineSpace."""

    for parameter_name, parameter in space.elements.items():
        if isinstance(parameter, neps.HPOCategorical):
            setattr(
                NEPSSpace,
                parameter_name,
                Categorical(
                    choices=tuple(parameter.choices),
                    prior=(
                        parameter.choices.index(parameter.prior)
                        if parameter.prior
                        else _UNSET
                    ),
                    prior_confidence=(
                        parameter.prior_confidence
                        if parameter.prior_confidence
                        else _UNSET
                    ),
                ),
            )
        elif isinstance(parameter, neps.HPOConstant):
            setattr(NEPSSpace, parameter_name, parameter.value)
        elif isinstance(parameter, neps.HPOInteger):
            new_integer = Integer(
                lower=parameter.lower,
                upper=parameter.upper,
                log=parameter.log,
                prior=parameter.prior if parameter.prior else _UNSET,
                prior_confidence=(
                    parameter.prior_confidence if parameter.prior_confidence else _UNSET
                ),
            )
            setattr(
                NEPSSpace,
                parameter_name,
                (Fidelity(domain=new_integer) if parameter.is_fidelity else new_integer),
            )
        elif isinstance(parameter, neps.HPOFloat):
            new_float = Float(
                lower=parameter.lower,
                upper=parameter.upper,
                log=parameter.log,
                prior=parameter.prior if parameter.prior else _UNSET,
                prior_confidence=(
                    parameter.prior_confidence if parameter.prior_confidence else _UNSET
                ),
            )
            setattr(
                NEPSSpace,
                parameter_name,
                (Fidelity(domain=new_float) if parameter.is_fidelity else new_float),
            )

    return NEPSSpace()

convert_neps_to_classic_search_space #

convert_neps_to_classic_search_space(
    space: PipelineSpace,
) -> SearchSpace | None

Convert a NePS space to a classic SearchSpace if possible. This function checks if the NePS space can be converted to a classic SearchSpace by ensuring that it does not contain any complex types like Operation or Resample, and that all choices of Categorical parameters are of basic types (int, str, float). If the checks pass, it converts the NePS space to a classic SearchSpace.

PARAMETER	DESCRIPTION
`space`	The NePS space to convert, which should be a Pipeline object. TYPE: `PipelineSpace`

RETURNS	DESCRIPTION
`SearchSpace \| None`	A classic SearchSpace if the conversion is possible, otherwise None.

Source code in neps/space/neps_spaces/neps_space.py

def convert_neps_to_classic_search_space(space: PipelineSpace) -> SearchSpace | None:
    """Convert a NePS space to a classic SearchSpace if possible.
    This function checks if the NePS space can be converted to a classic SearchSpace
    by ensuring that it does not contain any complex types like Operation or Resample,
    and that all choices of Categorical parameters are of basic types (int, str, float).
    If the checks pass, it converts the NePS space to a classic SearchSpace.

    Args:
        space: The NePS space to convert, which should be a Pipeline object.

    Returns:
        A classic SearchSpace if the conversion is possible, otherwise None.
    """
    # First check: No parameters are of type Operation or Resample
    if not any(
        isinstance(param, Operation | Resample) for param in space.get_attrs().values()
    ):
        # Second check: All choices of all categoricals are of basic
        # types i.e. int, str or float
        categoricals = [
            param
            for param in space.get_attrs().values()
            if isinstance(param, Categorical)
        ]
        if all(
            any(
                all(isinstance(choice, datatype) for choice in list(cat_param.choices))  # type: ignore
                for datatype in [int, float, str]
            )
            for cat_param in categoricals
        ):
            # If both checks pass, convert the space to a classic SearchSpace
            classic_space: dict[str, Any] = {}
            for key, value in space.get_attrs().items():
                if isinstance(value, Categorical):
                    classic_space[key] = neps.HPOCategorical(
                        choices=list(set(value.choices)),  # type: ignore
                        prior=value.choices[value.prior] if value.has_prior else None,  # type: ignore
                        prior_confidence=(
                            value.prior_confidence.value if value.has_prior else "low"
                        ),
                    )
                elif isinstance(value, Integer):
                    classic_space[key] = neps.HPOInteger(
                        lower=value.lower,
                        upper=value.upper,
                        log=value._log if hasattr(value, "_log") else False,
                        prior=value.prior if value.has_prior else None,
                        prior_confidence=(
                            value.prior_confidence.value if value.has_prior else "low"
                        ),
                    )
                elif isinstance(value, Float):
                    classic_space[key] = neps.HPOFloat(
                        lower=value.lower,
                        upper=value.upper,
                        log=value._log if hasattr(value, "_log") else False,
                        prior=value.prior if value.has_prior else None,
                        prior_confidence=(
                            value.prior_confidence.value if value.has_prior else "low"
                        ),
                    )
                elif isinstance(value, Fidelity):
                    if isinstance(value.domain, Integer):
                        classic_space[key] = neps.HPOInteger(
                            lower=value.domain.lower,
                            upper=value.domain.upper,
                            log=(
                                value.domain._log
                                if hasattr(value.domain, "_log")
                                else False
                            ),
                            is_fidelity=True,
                        )
                    elif isinstance(value.domain, Float):
                        classic_space[key] = neps.HPOFloat(
                            lower=value.domain.lower,
                            upper=value.domain.upper,
                            log=(
                                value.domain._log
                                if hasattr(value.domain, "_log")
                                else False
                            ),
                            is_fidelity=True,
                        )
                else:
                    classic_space[key] = neps.HPOConstant(value)
            return convert_mapping(classic_space)
    return None

convert_operation_to_callable #

convert_operation_to_callable(
    operation: Operation,
) -> Callable

Convert an Operation to a callable that can be executed.

PARAMETER	DESCRIPTION
`operation`	The Operation to convert. TYPE: `Operation`

RETURNS	DESCRIPTION
`Callable`	A callable that represents the operation.

RAISES	DESCRIPTION
`ValueError`	If the operation is not a valid Operation object.

Source code in neps/space/neps_spaces/neps_space.py

def convert_operation_to_callable(operation: Operation) -> Callable:
    """Convert an Operation to a callable that can be executed.

    Args:
        operation: The Operation to convert.

    Returns:
        A callable that represents the operation.

    Raises:
        ValueError: If the operation is not a valid Operation object.
    """
    operator = cast(Callable, operation.operator)

    operation_args: list[Any] = []
    for arg in operation.args:
        if isinstance(arg, tuple | list):
            arg_sequence: list[Any] = []
            for a in arg:
                converted_arg = (
                    convert_operation_to_callable(a) if isinstance(a, Operation) else a
                )
                arg_sequence.append(converted_arg)
            if isinstance(arg, tuple):
                operation_args.append(tuple(arg_sequence))
            else:
                operation_args.append(arg_sequence)
        else:
            operation_args.append(
                convert_operation_to_callable(arg) if isinstance(arg, Operation) else arg
            )

    operation_kwargs: dict[str, Any] = {}
    for kwarg_name, kwarg_value in operation.kwargs.items():
        if isinstance(kwarg_value, tuple | list):
            kwarg_sequence: list[Any] = []
            for a in kwarg_value:
                converted_kwarg = (
                    convert_operation_to_callable(a) if isinstance(a, Operation) else a
                )
                kwarg_sequence.append(converted_kwarg)
            if isinstance(kwarg_value, tuple):
                operation_kwargs[kwarg_name] = tuple(kwarg_sequence)
            else:
                operation_kwargs[kwarg_name] = kwarg_sequence
        else:
            operation_kwargs[kwarg_name] = (
                convert_operation_to_callable(kwarg_value)
                if isinstance(kwarg_value, Operation)
                else kwarg_value
            )

    return cast(Callable, operator(*operation_args, **operation_kwargs))

resolve #

resolve(
    pipeline: P,
    domain_sampler: DomainSampler | None = None,
    environment_values: Mapping[str, Any] | None = None,
) -> tuple[P, SamplingResolutionContext]

Resolve a NePS pipeline with the given domain sampler and environment values.

PARAMETER	DESCRIPTION
`pipeline`	The pipeline to resolve, which should be a Pipeline object. TYPE: `P`
`domain_sampler`	The DomainSampler to use for sampling from Domain objects. If None, a RandomSampler with no predefined values will be used. TYPE: `DomainSampler \| None` DEFAULT: `None`
`environment_values`	A mapping of environment variable names to their values. If None, an empty mapping will be used. TYPE: `Mapping[str, Any] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`tuple[P, SamplingResolutionContext]`	A tuple containing the resolved pipeline and the SamplingResolutionContext.

RAISES	DESCRIPTION
`ValueError`	If the pipeline is not a Pipeline object or if the domain_sampler is not a DomainSampler or if the environment_values is not a Mapping.

Source code in neps/space/neps_spaces/neps_space.py

def resolve(
    pipeline: P,
    domain_sampler: DomainSampler | None = None,
    environment_values: Mapping[str, Any] | None = None,
) -> tuple[P, SamplingResolutionContext]:
    """Resolve a NePS pipeline with the given domain sampler and environment values.

    Args:
        pipeline: The pipeline to resolve, which should be a Pipeline object.
        domain_sampler: The DomainSampler to use for sampling from Domain objects.
            If None, a RandomSampler with no predefined values will be used.
        environment_values: A mapping of environment variable names to their values.
            If None, an empty mapping will be used.

    Returns:
        A tuple containing the resolved pipeline and the SamplingResolutionContext.

    Raises:
        ValueError: If the pipeline is not a Pipeline object or if the domain_sampler
            is not a DomainSampler or if the environment_values is not a Mapping.
    """
    if domain_sampler is None:
        # By default, use a random sampler with no predefined values.
        domain_sampler = RandomSampler(predefined_samplings={})

    if environment_values is None:
        # By default, have no environment values.
        environment_values = {}
    if isinstance(domain_sampler, IOSampler):
        environment_values = domain_sampler.sample_environment_values(pipeline)

    sampling_resolver = SamplingResolver()
    resolved_pipeline, context = sampling_resolver(
        obj=pipeline,
        domain_sampler=domain_sampler,
        environment_values=environment_values,
    )
    return cast(P, resolved_pipeline), context

Neps space

NepsCompatConverter #

from_neps_config classmethod #

to_neps_config classmethod #

SamplingResolutionContext #

environment_values property #

resolution_root property #

samplings_made property #

add_resolved #

get_resolved #

get_value_from_environment #

resolving #

sample_from #

was_already_resolved #

SamplingResolver #

__call__ #

adjust_evaluation_pipeline_for_neps_space #

check_neps_space_compatibility #

construct_sampling_path #

convert_classic_to_neps_search_space #

convert_neps_to_classic_search_space #

convert_operation_to_callable #

resolve #

from_neps_config `classmethod` #

to_neps_config `classmethod` #

environment_values `property` #

resolution_root `property` #

samplings_made `property` #

call #