Components

amltk.pipeline.components

The provided subclasses of a Node that can be used can be assembled into a pipeline.

Choice dataclass

Choice(
    *nodes: Node | NodeLike,
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: (
        Callable[[Config, Any], Config] | None
    ) = None,
    meta: Mapping[str, Any] | None = None
)

Bases: Node[Item, Space]

A Choice between different subcomponents.

This indicates that a choice should be made between the different children in .nodes, usually done when you configure() with some config from a search_space().

from amltk.pipeline import Choice, Component
from sklearn.ensemble import RandomForestClassifier
from sklearn.neural_network import MLPClassifier

rf = Component(RandomForestClassifier, space={"n_estimators": (10, 100)})
mlp = Component(MLPClassifier, space={"activation": ["logistic", "relu", "tanh"]})

estimator_choice = Choice(rf, mlp, name="estimator")

╭─ Choice(estimator) ──────────────────────────────────────────────────────────╮
│ ╭─ Component(MLPClassifier) ─────╮ ╭─ Component(RandomForestClassifier)─╮    │
│ │ item  class MLPClassifier(...) │ │ item  class                        │    │
│ │ space {                        │ │       RandomForestClassifier(...)  │    │
│ │           'activation': [      │ │ space {'n_estimators': (10, 100)}  │    │
│ │               'logistic',      │ ╰────────────────────────────────────╯    │
│ │               'relu',          │                                           │
│ │               'tanh'           │                                           │
│ │           ]                    │                                           │
│ │       }                        │                                           │
│ ╰────────────────────────────────╯                                           │
╰──────────────────────────────────────────────────────────────────────────────╯

Order of nodes

The given nodes of a choice are always ordered according to their name, so indexing choice.nodes may not be reliable if modifying the choice dynamically.

Please use choice["name"] to access the nodes instead.

See Also

• Node

PARAMETER	DESCRIPTION
nodes	The nodes that should be chosen between for this node. TYPE: Node | NodeLike DEFAULT: ()
item	The item attached to this node (if any). TYPE: Item | Callable[[Item], Item] | None DEFAULT: None
name	The name of the node. If not specified, the name will be randomly generated. TYPE: str | None DEFAULT: None
config	The configuration for this node. TYPE: Config | None DEFAULT: None
space	The search space for this node. This will be used when search_space() is called. TYPE: Space | None DEFAULT: None
fidelities	The fidelities for this node. TYPE: Mapping[str, Any] | None DEFAULT: None
config_transform	A function that transforms the config= parameter during configure(config) before return the new configured node. Useful for times where you need to combine multiple parameters into one. TYPE: Callable[[Config, Any], Config] | None DEFAULT: None
meta	Any meta information about this node. TYPE: Mapping[str, Any] | None DEFAULT: None

Source code in src/amltk/pipeline/components.py

def __init__(
    self,
    *nodes: Node | NodeLike,
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: Callable[[Config, Any], Config] | None = None,
    meta: Mapping[str, Any] | None = None,
):
    """Initialize a choice node.

    Args:
        nodes: The nodes that should be chosen between for this node.
        item: The item attached to this node (if any).
        name: The name of the node. If not specified, the name will be
            randomly generated.
        config: The configuration for this node.
        space: The search space for this node. This will be used when
            [`search_space()`][amltk.pipeline.node.Node.search_space] is called.
        fidelities: The fidelities for this node.
        config_transform: A function that transforms the `config=` parameter
            during [`configure(config)`][amltk.pipeline.node.Node.configure]
            before return the new configured node. Useful for times where
            you need to combine multiple parameters into one.
        meta: Any meta information about this node.
    """
    _nodes: tuple[Node, ...] = tuple(
        sorted((as_node(n) for n in nodes), key=lambda n: n.name),
    )
    if not all_unique(_nodes, key=lambda node: node.name):
        raise ValueError(
            f"Can't handle nodes as we can not generate a __choice__ for {nodes=}."
            "\nAll nodes must have a unique name. Please provide a `name=` to them",
        )

    if name is None:
        name = f"Choice-{randuid(8)}"

    super().__init__(
        *_nodes,
        name=name,
        item=item,
        config=config,
        space=space,
        fidelities=fidelities,
        config_transform=config_transform,
        meta=meta,
    )

config class-attribute instance-attribute

config: Config | None = field(hash=False)

The configuration for this node

config_transform class-attribute instance-attribute

config_transform: Callable[[Config, Any], Config] | None = (
    field(hash=False)
)

A function that transforms the configuration of this node

fidelities class-attribute instance-attribute

fidelities: Mapping[str, Any] | None = field(hash=False)

The fidelities for this node

item class-attribute instance-attribute

item: Callable[..., Item] | Item | None = field(hash=False)

The item attached to this node

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = field(hash=False)

Any meta information about this node

name class-attribute instance-attribute

name: str = field(hash=True)

Name of the node

nodes instance-attribute

nodes: tuple[Node, ...]

The choice of possible nodes that this choice could take.

space class-attribute instance-attribute

space: Space | None = field(hash=False)

The search space for this node

__getitem__

__getitem__(key: str) -> Node

Get the first from .nodes with key.

Source code in src/amltk/pipeline/node.py

def __getitem__(self, key: str) -> Node:
    """Get the first from [`.nodes`][amltk.pipeline.node.Node.nodes] with `key`."""
    found = first_true(
        self.nodes,
        None,
        lambda node: node.name == key,
    )
    if found is None:
        raise KeyError(
            f"Could not find node with name {key} in '{self.name}'."
            f" Available nodes are: {', '.join(node.name for node in self.nodes)}",
        )

    return found

build

build(
    builder: (
        Callable[Concatenate[Node, P], BuilderOutput]
        | Literal["sklearn"]
    ),
    *builder_args: args,
    **builder_kwargs: kwargs
) -> BuilderOutput | Pipeline

Build a concrete object out of this node.

PARAMETER	DESCRIPTION
builder	The builder to use. This can be a function that takes in the node and returns the object or a string that is one of: "sklearn": Build a sklearn.pipeline.Pipeline out of this node. TYPE: Callable[Concatenate[Node, P], BuilderOutput] | Literal['sklearn']
builder_args	The positional arguments to pass to the builder TYPE: args DEFAULT: ()
builder_kwargs	The keyword arguments to pass to the builder TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BuilderOutput | Pipeline	The built object

Source code in src/amltk/pipeline/node.py

def build(
    self,
    builder: Callable[Concatenate[Node, P], BuilderOutput] | Literal["sklearn"],
    *builder_args: P.args,
    **builder_kwargs: P.kwargs,
) -> BuilderOutput | SklearnPipeline:
    """Build a concrete object out of this node.

    Args:
        builder: The builder to use. This can be a function that takes in
            the node and returns the object or a string that is one of:

            * `#!python "sklearn"`: Build a
                [`sklearn.pipeline.Pipeline`][sklearn.pipeline.Pipeline]
                out of this node.

        builder_args: The positional arguments to pass to the builder
        builder_kwargs: The keyword arguments to pass to the builder

    Returns:
        The built object
    """
    match builder:
        case "sklearn":
            from amltk.pipeline.builders.sklearn import build as _build

            return _build(self, *builder_args, **builder_kwargs)  # type: ignore
        case _:
            return builder(self, *builder_args, **builder_kwargs)

chosen

chosen() -> Node

The chosen branch.

RETURNS	DESCRIPTION
Node	The chosen branch

Source code in src/amltk/pipeline/components.py

def chosen(self) -> Node:
    """The chosen branch.

    Returns:
        The chosen branch
    """
    match self.config:
        case {"__choice__": choice}:
            chosen = first_true(
                self.nodes,
                pred=lambda node: node.name == choice,
                default=None,
            )
            if chosen is None:
                raise NodeNotFoundError(choice, self.name)

            return chosen
        case _:
            raise NoChoiceMadeError(self.name)

configure

configure(
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None
) -> Self

Configure this node and anything following it with the given config.

Configuring a choice

For a Choice, if the config has a __choice__ key, then only the node chosen will be configured. The others will not be configured at all and their config will be discarded.

PARAMETER	DESCRIPTION
config	The configuration to apply TYPE: Config
prefixed_name	Whether items in the config are prefixed by the names of the nodes. * If None, the default, then prefixed_name will be assumed to be True if this node has a next node or if the config has keys that begin with this nodes name. * If True, then the config will be searched for items prefixed by the name of the node (and subsequent chained nodes). * If False, then the config will be searched for items without the prefix, i.e. the config keys are exactly those matching this nodes search space. TYPE: bool | None DEFAULT: None
transform_context	Any context to give to config_transform= of individual nodes. TYPE: Any | None DEFAULT: None
params	The params to match any requests when configuring this node. These will match against any ParamRequests in the config and will be used to fill in any missing values. TYPE: Mapping[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Self	The configured node

Source code in src/amltk/pipeline/components.py

@override
def configure(
    self,
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None,
) -> Self:
    """Configure this node and anything following it with the given config.

    !!! note "Configuring a choice"

        For a Choice, if the config has a `__choice__` key, then only the node
        chosen will be configured. The others will not be configured at all and
        their config will be discarded.

    Args:
        config: The configuration to apply
        prefixed_name: Whether items in the config are prefixed by the names
            of the nodes.
            * If `None`, the default, then `prefixed_name` will be assumed to
                be `True` if this node has a next node or if the config has
                keys that begin with this nodes name.
            * If `True`, then the config will be searched for items prefixed
                by the name of the node (and subsequent chained nodes).
            * If `False`, then the config will be searched for items without
                the prefix, i.e. the config keys are exactly those matching
                this nodes search space.
        transform_context: Any context to give to `config_transform=` of individual
            nodes.
        params: The params to match any requests when configuring this node.
            These will match against any ParamRequests in the config and will
            be used to fill in any missing values.

    Returns:
        The configured node
    """
    # Get the config for this node
    match prefixed_name:
        case True:
            config = mapping_select(config, f"{self.name}:")
        case False:
            pass
        case None if any(k.startswith(f"{self.name}:") for k in config):
            config = mapping_select(config, f"{self.name}:")
        case None:
            pass

    _kwargs: dict[str, Any] = {}

    # Configure all the branches if exists
    # This part is what differs for a Choice
    if len(self.nodes) > 0:
        choice_made = config.get("__choice__", None)
        if choice_made is not None:
            matching_child = first_true(
                self.nodes,
                pred=lambda node: node.name == choice_made,
                default=None,
            )
            if matching_child is None:
                raise ValueError(
                    f"Can not find matching child for choice {self.name} with child"
                    f" {choice_made}."
                    "\nPlease check the config and ensure that the choice is one of"
                    f" {[n.name for n in self.nodes]}."
                    f"\nThe config recieved at this choice node was {config=}.",
                )

            # We still iterate over all of them just to ensure correct ordering
            nodes = tuple(
                node.copy()
                if node.name != choice_made
                else matching_child.configure(
                    config,
                    prefixed_name=True,
                    transform_context=transform_context,
                    params=params,
                )
                for node in self.nodes
            )
            _kwargs["nodes"] = nodes
        else:
            nodes = tuple(
                node.configure(
                    config,
                    prefixed_name=True,
                    transform_context=transform_context,
                    params=params,
                )
                for node in self.nodes
            )
            _kwargs["nodes"] = nodes

    this_config = {
        hp: v
        for hp, v in config.items()
        if (
            ":" not in hp
            and not any(hp.startswith(f"{node.name}") for node in self.nodes)
        )
    }
    if self.config is not None:
        this_config = {**self.config, **this_config}

    this_config = dict(self._fufill_param_requests(this_config, params=params))

    if self.config_transform is not None:
        this_config = dict(self.config_transform(this_config, transform_context))

    if len(this_config) > 0:
        _kwargs["config"] = dict(this_config)

    return self.mutate(**_kwargs)

copy

copy() -> Self

Copy this node, removing all links in the process.

Source code in src/amltk/pipeline/node.py

def copy(self) -> Self:
    """Copy this node, removing all links in the process."""
    return self.mutate()

display

display(*, full: bool = False) -> RenderableType

Display this node.

PARAMETER	DESCRIPTION
full	Whether to display the full node or just a summary TYPE: bool DEFAULT: False

Source code in src/amltk/pipeline/node.py

def display(self, *, full: bool = False) -> RenderableType:
    """Display this node.

    Args:
        full: Whether to display the full node or just a summary
    """
    if not full:
        return self.__rich__()

    from rich.console import Group as RichGroup

    return RichGroup(*self._rich_iter())

factorize

factorize(
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None
) -> Iterator[Self]

Please see factorize().

Source code in src/amltk/pipeline/node.py

def factorize(
    self,
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None,
) -> Iterator[Self]:
    """Please see [`factorize()`][amltk.pipeline.ops.factorize]."""  # noqa: D402
    from amltk.pipeline.ops import factorize

    yield from factorize(
        self,
        min_depth=min_depth,
        max_depth=max_depth,
        current_depth=current_depth,
        factor_by=factor_by,
        assign_child=assign_child,
    )

fidelity_space

fidelity_space() -> dict[str, Any]

Get the fidelities for this node and any connected nodes.

Source code in src/amltk/pipeline/node.py

def fidelity_space(self) -> dict[str, Any]:
    """Get the fidelities for this node and any connected nodes."""
    fids = {}
    for node in self.nodes:
        fids.update(prefix_keys(node.fidelity_space(), f"{self.name}:"))

    return fids

find

find(
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None

Find a node in that's nested deeper from this node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]
default	The value to return if the node is not found. Defaults to None TYPE: T | None DEFAULT: None

RETURNS	DESCRIPTION
Node | T | None	The node if found, otherwise the default value. Defaults to None

Source code in src/amltk/pipeline/node.py

def find(
    self,
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None:
    """Find a node in that's nested deeper from this node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node
        default: The value to return if the node is not found. Defaults to None

    Returns:
        The node if found, otherwise the default value. Defaults to None
    """
    itr = self.iter()
    match key:
        case Node():
            return first_true(itr, default, lambda node: node == key)
        case str():
            return first_true(itr, default, lambda node: node.name == key)
        case _:
            return first_true(itr, default, key)  # type: ignore

iter

iter() -> Iterator[Node]

Iterate the the nodes, including this node.

YIELDS	DESCRIPTION
Node	The nodes connected to this node

Source code in src/amltk/pipeline/node.py

def iter(self) -> Iterator[Node]:
    """Iterate the the nodes, including this node.

    Yields:
        The nodes connected to this node
    """
    yield self
    for node in self.nodes:
        yield from node.iter()

linearized_fidelity

linearized_fidelity(
    value: float,
) -> dict[str, int | float | Any]

Get the liniearized fidelities for this node and any connected nodes.

PARAMETER	DESCRIPTION
value	The value to linearize. Must be between [0, 1] TYPE: float

Return

dictionary from key to it's linearized fidelity.

Source code in src/amltk/pipeline/node.py

def linearized_fidelity(self, value: float) -> dict[str, int | float | Any]:
    """Get the liniearized fidelities for this node and any connected nodes.

    Args:
        value: The value to linearize. Must be between [0, 1]

    Return:
        dictionary from key to it's linearized fidelity.
    """
    assert 1.0 <= value <= 100.0, f"{value=} not in [1.0, 100.0]"  # noqa: PLR2004
    d = {}
    for node in self.nodes:
        node_fids = prefix_keys(
            node.linearized_fidelity(value),
            f"{self.name}:",
        )
        d.update(node_fids)

    if self.fidelities is None:
        return d

    for f_name, f_range in self.fidelities.items():
        match f_range:
            case (int() | float(), int() | float()):
                low, high = f_range
                fid = low + (high - low) * value
                fid = low + (high - low) * (value - 1) / 100
                fid = fid if isinstance(low, float) else round(fid)
                d[f_name] = fid
            case _:
                raise ValueError(
                    f"Invalid fidelities to linearize {f_range} for {f_name}"
                    f" in {self}. Only supports ranges of the form (low, high)",
                )

    return prefix_keys(d, f"{self.name}:")

mutate

mutate(**kwargs: Any) -> Self

Mutate the node with the given keyword arguments.

PARAMETER	DESCRIPTION
**kwargs	The keyword arguments to mutate TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	Self The mutated node

Source code in src/amltk/pipeline/node.py

def mutate(self, **kwargs: Any) -> Self:
    """Mutate the node with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to mutate

    Returns:
        Self
            The mutated node
    """
    _args = ()
    _kwargs = {**self.__dict__, **kwargs}

    # If there's nodes in kwargs, we have to check if it's
    # a positional or keyword argument and handle accordingly.
    if (nodes := _kwargs.pop("nodes", None)) is not None:
        match self._NODES_INIT:
            case "args":
                _args = nodes
            case "kwargs":
                _kwargs["nodes"] = nodes
            case None if len(nodes) == 0:
                pass  # Just ignore it, it's popped out
            case None:
                raise ValueError(
                    "Cannot mutate nodes when __init__ does not accept nodes",
                )

    # If there's a config in kwargs, we have to check if it's actually got values
    config = _kwargs.pop("config", None)
    if config is not None and len(config) > 0:
        _kwargs["config"] = config

    # Lastly, we remove anything that can't be passed to kwargs of the
    # subclasses __init__
    _available_kwargs = inspect.signature(self.__init__).parameters.keys()  # type: ignore
    for k in list(_kwargs.keys()):
        if k not in _available_kwargs:
            _kwargs.pop(k)

    return self.__class__(*_args, **_kwargs)

path_to

path_to(
    key: str | Node | Callable[[Node], bool]
) -> list[Node] | None

Find a path to the given node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]

RETURNS	DESCRIPTION
list[Node] | None	The path to the node if found, else None

Source code in src/amltk/pipeline/node.py

def path_to(self, key: str | Node | Callable[[Node], bool]) -> list[Node] | None:
    """Find a path to the given node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node

    Returns:
        The path to the node if found, else None
    """
    # We found our target, just return now

    match key:
        case Node():
            pred = lambda node: node == key
        case str():
            pred = lambda node: node.name == key
        case _:
            pred = key

    for path, node in self.walk():
        if pred(node):
            return path

    return None

search_space

search_space(
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: args,
    **parser_kwargs: kwargs
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace

Get the search space for this node.

PARAMETER	DESCRIPTION
parser	The parser to use. This can be a function that takes in the node and returns the search space or a string that is one of: "configspace": Build a ConfigSpace.ConfigurationSpace out of this node. "optuna": Build a dict of hyperparameters that Optuna can use in its ask and tell methods TYPE: Callable[Concatenate[Node, P], ParserOutput] | Literal['configspace', 'optuna']
parser_args	The positional arguments to pass to the parser TYPE: args DEFAULT: ()
parser_kwargs	The keyword arguments to pass to the parser TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
ParserOutput | ConfigurationSpace | OptunaSearchSpace	The search space

Source code in src/amltk/pipeline/node.py

def search_space(
    self,
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: P.args,
    **parser_kwargs: P.kwargs,
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace:
    """Get the search space for this node.

    Args:
        parser: The parser to use. This can be a function that takes in
            the node and returns the search space or a string that is one of:

            * `#!python "configspace"`: Build a
                [`ConfigSpace.ConfigurationSpace`](https://automl.github.io/ConfigSpace/master/)
                out of this node.
            * `#!python "optuna"`: Build a dict of hyperparameters that Optuna can
                use in its [ask and tell methods](https://optuna.readthedocs.io/en/stable/tutorial/20_recipes/009_ask_and_tell.html#define-and-run)

        parser_args: The positional arguments to pass to the parser
        parser_kwargs: The keyword arguments to pass to the parser

    Returns:
        The search space
    """
    match parser:
        case "configspace":
            from amltk.pipeline.parsers.configspace import parser as cs_parser

            return cs_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case "optuna":
            from amltk.pipeline.parsers.optuna import parser as optuna_parser

            return optuna_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case str():  # type: ignore
            raise ValueError(
                f"Invalid str for parser {parser}. "
                "Please use 'configspace' or 'optuna' or pass in your own"
                " parser function",
            )
        case _:
            return parser(self, *parser_args, **parser_kwargs)

walk

walk(
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]

Walk the nodes in this chain.

PARAMETER	DESCRIPTION
path	The current path to this node TYPE: Sequence[Node] | None DEFAULT: None

YIELDS	DESCRIPTION
list[Node]	The parents of the node and the node itself

Source code in src/amltk/pipeline/node.py

def walk(
    self,
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]:
    """Walk the nodes in this chain.

    Args:
        path: The current path to this node

    Yields:
        The parents of the node and the node itself
    """
    path = list(path) if path is not None else []
    yield path, self

    for node in self.nodes:
        yield from node.walk(path=[*path, self])

Component dataclass

Component(
    item: Callable[..., Item],
    *,
    name: str | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: (
        Callable[[Config, Any], Config] | None
    ) = None,
    meta: Mapping[str, Any] | None = None
)

Bases: Node[Item, Space]

A Component of the pipeline with a possible item and no children.

This is the basic building block of most pipelines, it accepts as it's item= some function that will be called with build_item() to build that one part of the pipeline.

When build_item() is called, whatever the config of the component is at that time, will be used to construct the item.

A common pattern is to use a Component to wrap a constructor, specifying the space= and config= to be used when building the item.

from amltk.pipeline import Component
from sklearn.ensemble import RandomForestClassifier

rf = Component(
    RandomForestClassifier,
    config={"max_depth": 3},
    space={"n_estimators": (10, 100)}
)

config = {"n_estimators": 50}  # Sample from some space or something
configured_rf = rf.configure(config)

estimator = configured_rf.build_item()

╭─ Component(RandomForestClassifier) ──────╮
│ item   class RandomForestClassifier(...) │
│ config {'max_depth': 3}                  │
│ space  {'n_estimators': (10, 100)}       │
╰──────────────────────────────────────────╯

RandomForestClassifier(max_depth=3, n_estimators=50)

In a Jupyter environment, please rerun this cell to show the HTML representation or trust the notebook.
On GitHub, the HTML representation is unable to render, please try loading this page with nbviewer.org.

See Also

• Node

PARAMETER	DESCRIPTION
item	The item attached to this node. TYPE: Callable[..., Item]
name	The name of the node. If not specified, the name will be generated from the item. TYPE: str | None DEFAULT: None
config	The configuration for this node. TYPE: Config | None DEFAULT: None
space	The search space for this node. This will be used when search_space() is called. TYPE: Space | None DEFAULT: None
fidelities	The fidelities for this node. TYPE: Mapping[str, Any] | None DEFAULT: None
config_transform	A function that transforms the config= parameter during configure(config) before return the new configured node. Useful for times where you need to combine multiple parameters into one. TYPE: Callable[[Config, Any], Config] | None DEFAULT: None
meta	Any meta information about this node. TYPE: Mapping[str, Any] | None DEFAULT: None

Source code in src/amltk/pipeline/components.py

def __init__(
    self,
    item: Callable[..., Item],
    *,
    name: str | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: Callable[[Config, Any], Config] | None = None,
    meta: Mapping[str, Any] | None = None,
):
    """Initialize a component.

    Args:
        item: The item attached to this node.
        name: The name of the node. If not specified, the name will be
            generated from the item.
        config: The configuration for this node.
        space: The search space for this node. This will be used when
            [`search_space()`][amltk.pipeline.node.Node.search_space] is called.
        fidelities: The fidelities for this node.
        config_transform: A function that transforms the `config=` parameter
            during [`configure(config)`][amltk.pipeline.node.Node.configure]
            before return the new configured node. Useful for times where
            you need to combine multiple parameters into one.
        meta: Any meta information about this node.
    """
    super().__init__(
        name=name if name is not None else entity_name(item),
        item=item,
        config=config,
        space=space,
        fidelities=fidelities,
        config_transform=config_transform,
        meta=meta,
    )

config class-attribute instance-attribute

config: Config | None = field(hash=False)

The configuration for this node

config_transform class-attribute instance-attribute

config_transform: Callable[[Config, Any], Config] | None = (
    field(hash=False)
)

A function that transforms the configuration of this node

fidelities class-attribute instance-attribute

fidelities: Mapping[str, Any] | None = field(hash=False)

The fidelities for this node

item instance-attribute

item: Callable[..., Item]

A node which constructs an item in the pipeline.

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = field(hash=False)

Any meta information about this node

name class-attribute instance-attribute

name: str = field(hash=True)

Name of the node

nodes instance-attribute

nodes: tuple[]

A component has no children.

space class-attribute instance-attribute

space: Space | None = field(hash=False)

The search space for this node

__getitem__

__getitem__(key: str) -> Node

Get the first from .nodes with key.

Source code in src/amltk/pipeline/node.py

def __getitem__(self, key: str) -> Node:
    """Get the first from [`.nodes`][amltk.pipeline.node.Node.nodes] with `key`."""
    found = first_true(
        self.nodes,
        None,
        lambda node: node.name == key,
    )
    if found is None:
        raise KeyError(
            f"Could not find node with name {key} in '{self.name}'."
            f" Available nodes are: {', '.join(node.name for node in self.nodes)}",
        )

    return found

build

build(
    builder: (
        Callable[Concatenate[Node, P], BuilderOutput]
        | Literal["sklearn"]
    ),
    *builder_args: args,
    **builder_kwargs: kwargs
) -> BuilderOutput | Pipeline

Build a concrete object out of this node.

PARAMETER	DESCRIPTION
builder	The builder to use. This can be a function that takes in the node and returns the object or a string that is one of: "sklearn": Build a sklearn.pipeline.Pipeline out of this node. TYPE: Callable[Concatenate[Node, P], BuilderOutput] | Literal['sklearn']
builder_args	The positional arguments to pass to the builder TYPE: args DEFAULT: ()
builder_kwargs	The keyword arguments to pass to the builder TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BuilderOutput | Pipeline	The built object

Source code in src/amltk/pipeline/node.py

def build(
    self,
    builder: Callable[Concatenate[Node, P], BuilderOutput] | Literal["sklearn"],
    *builder_args: P.args,
    **builder_kwargs: P.kwargs,
) -> BuilderOutput | SklearnPipeline:
    """Build a concrete object out of this node.

    Args:
        builder: The builder to use. This can be a function that takes in
            the node and returns the object or a string that is one of:

            * `#!python "sklearn"`: Build a
                [`sklearn.pipeline.Pipeline`][sklearn.pipeline.Pipeline]
                out of this node.

        builder_args: The positional arguments to pass to the builder
        builder_kwargs: The keyword arguments to pass to the builder

    Returns:
        The built object
    """
    match builder:
        case "sklearn":
            from amltk.pipeline.builders.sklearn import build as _build

            return _build(self, *builder_args, **builder_kwargs)  # type: ignore
        case _:
            return builder(self, *builder_args, **builder_kwargs)

build_item

build_item(**kwargs: Any) -> Item

Build the item attached to this component.

PARAMETER	DESCRIPTION
**kwargs	Any additional arguments to pass to the item TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Item	Item The built item

Source code in src/amltk/pipeline/components.py

def build_item(self, **kwargs: Any) -> Item:
    """Build the item attached to this component.

    Args:
        **kwargs: Any additional arguments to pass to the item

    Returns:
        Item
            The built item
    """
    config = self.config or {}
    try:
        return self.item(**{**config, **kwargs})
    except TypeError as e:
        new_msg = f"Failed to build `{self.item=}` with `{self.config=}`.\n"
        if any(kwargs):
            new_msg += f"Extra {kwargs=} were also provided.\n"
        new_msg += (
            "If the item failed to initialize, a common reason can be forgetting"
            " to call `configure()` on the `Component` or the pipeline it is in or"
            " not calling `build()`/`build_item()` on the **returned** value of"
            " `configure()`.\n"
            "Reasons may also include not having fully specified the `config`"
            " initially, it having not being configured fully from `configure()`"
            " or from misspecfying parameters in the `space`."
        )
        raise ComponentBuildError(new_msg) from e

configure

configure(
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None
) -> Self

Configure this node and anything following it with the given config.

PARAMETER	DESCRIPTION
config	The configuration to apply TYPE: Config
prefixed_name	Whether items in the config are prefixed by the names of the nodes. * If None, the default, then prefixed_name will be assumed to be True if this node has a next node or if the config has keys that begin with this nodes name. * If True, then the config will be searched for items prefixed by the name of the node (and subsequent chained nodes). * If False, then the config will be searched for items without the prefix, i.e. the config keys are exactly those matching this nodes search space. TYPE: bool | None DEFAULT: None
transform_context	Any context to give to config_transform= of individual nodes. TYPE: Any | None DEFAULT: None
params	The params to match any requests when configuring this node. These will match against any ParamRequests in the config and will be used to fill in any missing values. TYPE: Mapping[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Self	The configured node

Source code in src/amltk/pipeline/node.py

def configure(
    self,
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None,
) -> Self:
    """Configure this node and anything following it with the given config.

    Args:
        config: The configuration to apply
        prefixed_name: Whether items in the config are prefixed by the names
            of the nodes.
            * If `None`, the default, then `prefixed_name` will be assumed to
                be `True` if this node has a next node or if the config has
                keys that begin with this nodes name.
            * If `True`, then the config will be searched for items prefixed
                by the name of the node (and subsequent chained nodes).
            * If `False`, then the config will be searched for items without
                the prefix, i.e. the config keys are exactly those matching
                this nodes search space.
        transform_context: Any context to give to `config_transform=` of individual
            nodes.
        params: The params to match any requests when configuring this node.
            These will match against any ParamRequests in the config and will
            be used to fill in any missing values.

    Returns:
        The configured node
    """
    # Get the config for this node
    match prefixed_name:
        case True:
            config = mapping_select(config, f"{self.name}:")
        case False:
            pass
        case None if any(k.startswith(f"{self.name}:") for k in config):
            config = mapping_select(config, f"{self.name}:")
        case None:
            pass

    _kwargs: dict[str, Any] = {}

    # Configure all the branches if exists
    if len(self.nodes) > 0:
        nodes = tuple(
            node.configure(
                config,
                prefixed_name=True,
                transform_context=transform_context,
                params=params,
            )
            for node in self.nodes
        )
        _kwargs["nodes"] = nodes

    this_config = {
        hp: v
        for hp, v in config.items()
        if (
            ":" not in hp
            and not any(hp.startswith(f"{node.name}") for node in self.nodes)
        )
    }
    if self.config is not None:
        this_config = {**self.config, **this_config}

    this_config = dict(self._fufill_param_requests(this_config, params=params))

    if self.config_transform is not None:
        this_config = dict(self.config_transform(this_config, transform_context))

    if len(this_config) > 0:
        _kwargs["config"] = dict(this_config)

    return self.mutate(**_kwargs)

copy

copy() -> Self

Copy this node, removing all links in the process.

Source code in src/amltk/pipeline/node.py

def copy(self) -> Self:
    """Copy this node, removing all links in the process."""
    return self.mutate()

display

display(*, full: bool = False) -> RenderableType

Display this node.

PARAMETER	DESCRIPTION
full	Whether to display the full node or just a summary TYPE: bool DEFAULT: False

Source code in src/amltk/pipeline/node.py

def display(self, *, full: bool = False) -> RenderableType:
    """Display this node.

    Args:
        full: Whether to display the full node or just a summary
    """
    if not full:
        return self.__rich__()

    from rich.console import Group as RichGroup

    return RichGroup(*self._rich_iter())

factorize

factorize(
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None
) -> Iterator[Self]

Please see factorize().

Source code in src/amltk/pipeline/node.py

def factorize(
    self,
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None,
) -> Iterator[Self]:
    """Please see [`factorize()`][amltk.pipeline.ops.factorize]."""  # noqa: D402
    from amltk.pipeline.ops import factorize

    yield from factorize(
        self,
        min_depth=min_depth,
        max_depth=max_depth,
        current_depth=current_depth,
        factor_by=factor_by,
        assign_child=assign_child,
    )

fidelity_space

fidelity_space() -> dict[str, Any]

Get the fidelities for this node and any connected nodes.

Source code in src/amltk/pipeline/node.py

def fidelity_space(self) -> dict[str, Any]:
    """Get the fidelities for this node and any connected nodes."""
    fids = {}
    for node in self.nodes:
        fids.update(prefix_keys(node.fidelity_space(), f"{self.name}:"))

    return fids

find

find(
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None

Find a node in that's nested deeper from this node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]
default	The value to return if the node is not found. Defaults to None TYPE: T | None DEFAULT: None

RETURNS	DESCRIPTION
Node | T | None	The node if found, otherwise the default value. Defaults to None

Source code in src/amltk/pipeline/node.py

def find(
    self,
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None:
    """Find a node in that's nested deeper from this node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node
        default: The value to return if the node is not found. Defaults to None

    Returns:
        The node if found, otherwise the default value. Defaults to None
    """
    itr = self.iter()
    match key:
        case Node():
            return first_true(itr, default, lambda node: node == key)
        case str():
            return first_true(itr, default, lambda node: node.name == key)
        case _:
            return first_true(itr, default, key)  # type: ignore

iter

iter() -> Iterator[Node]

Iterate the the nodes, including this node.

YIELDS	DESCRIPTION
Node	The nodes connected to this node

Source code in src/amltk/pipeline/node.py

def iter(self) -> Iterator[Node]:
    """Iterate the the nodes, including this node.

    Yields:
        The nodes connected to this node
    """
    yield self
    for node in self.nodes:
        yield from node.iter()

linearized_fidelity

linearized_fidelity(
    value: float,
) -> dict[str, int | float | Any]

Get the liniearized fidelities for this node and any connected nodes.

PARAMETER	DESCRIPTION
value	The value to linearize. Must be between [0, 1] TYPE: float

Return

dictionary from key to it's linearized fidelity.

Source code in src/amltk/pipeline/node.py

def linearized_fidelity(self, value: float) -> dict[str, int | float | Any]:
    """Get the liniearized fidelities for this node and any connected nodes.

    Args:
        value: The value to linearize. Must be between [0, 1]

    Return:
        dictionary from key to it's linearized fidelity.
    """
    assert 1.0 <= value <= 100.0, f"{value=} not in [1.0, 100.0]"  # noqa: PLR2004
    d = {}
    for node in self.nodes:
        node_fids = prefix_keys(
            node.linearized_fidelity(value),
            f"{self.name}:",
        )
        d.update(node_fids)

    if self.fidelities is None:
        return d

    for f_name, f_range in self.fidelities.items():
        match f_range:
            case (int() | float(), int() | float()):
                low, high = f_range
                fid = low + (high - low) * value
                fid = low + (high - low) * (value - 1) / 100
                fid = fid if isinstance(low, float) else round(fid)
                d[f_name] = fid
            case _:
                raise ValueError(
                    f"Invalid fidelities to linearize {f_range} for {f_name}"
                    f" in {self}. Only supports ranges of the form (low, high)",
                )

    return prefix_keys(d, f"{self.name}:")

mutate

mutate(**kwargs: Any) -> Self

Mutate the node with the given keyword arguments.

PARAMETER	DESCRIPTION
**kwargs	The keyword arguments to mutate TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	Self The mutated node

Source code in src/amltk/pipeline/node.py

def mutate(self, **kwargs: Any) -> Self:
    """Mutate the node with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to mutate

    Returns:
        Self
            The mutated node
    """
    _args = ()
    _kwargs = {**self.__dict__, **kwargs}

    # If there's nodes in kwargs, we have to check if it's
    # a positional or keyword argument and handle accordingly.
    if (nodes := _kwargs.pop("nodes", None)) is not None:
        match self._NODES_INIT:
            case "args":
                _args = nodes
            case "kwargs":
                _kwargs["nodes"] = nodes
            case None if len(nodes) == 0:
                pass  # Just ignore it, it's popped out
            case None:
                raise ValueError(
                    "Cannot mutate nodes when __init__ does not accept nodes",
                )

    # If there's a config in kwargs, we have to check if it's actually got values
    config = _kwargs.pop("config", None)
    if config is not None and len(config) > 0:
        _kwargs["config"] = config

    # Lastly, we remove anything that can't be passed to kwargs of the
    # subclasses __init__
    _available_kwargs = inspect.signature(self.__init__).parameters.keys()  # type: ignore
    for k in list(_kwargs.keys()):
        if k not in _available_kwargs:
            _kwargs.pop(k)

    return self.__class__(*_args, **_kwargs)

path_to

path_to(
    key: str | Node | Callable[[Node], bool]
) -> list[Node] | None

Find a path to the given node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]

RETURNS	DESCRIPTION
list[Node] | None	The path to the node if found, else None

Source code in src/amltk/pipeline/node.py

def path_to(self, key: str | Node | Callable[[Node], bool]) -> list[Node] | None:
    """Find a path to the given node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node

    Returns:
        The path to the node if found, else None
    """
    # We found our target, just return now

    match key:
        case Node():
            pred = lambda node: node == key
        case str():
            pred = lambda node: node.name == key
        case _:
            pred = key

    for path, node in self.walk():
        if pred(node):
            return path

    return None

search_space

search_space(
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: args,
    **parser_kwargs: kwargs
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace

Get the search space for this node.

PARAMETER	DESCRIPTION
parser	The parser to use. This can be a function that takes in the node and returns the search space or a string that is one of: "configspace": Build a ConfigSpace.ConfigurationSpace out of this node. "optuna": Build a dict of hyperparameters that Optuna can use in its ask and tell methods TYPE: Callable[Concatenate[Node, P], ParserOutput] | Literal['configspace', 'optuna']
parser_args	The positional arguments to pass to the parser TYPE: args DEFAULT: ()
parser_kwargs	The keyword arguments to pass to the parser TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
ParserOutput | ConfigurationSpace | OptunaSearchSpace	The search space

Source code in src/amltk/pipeline/node.py

def search_space(
    self,
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: P.args,
    **parser_kwargs: P.kwargs,
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace:
    """Get the search space for this node.

    Args:
        parser: The parser to use. This can be a function that takes in
            the node and returns the search space or a string that is one of:

            * `#!python "configspace"`: Build a
                [`ConfigSpace.ConfigurationSpace`](https://automl.github.io/ConfigSpace/master/)
                out of this node.
            * `#!python "optuna"`: Build a dict of hyperparameters that Optuna can
                use in its [ask and tell methods](https://optuna.readthedocs.io/en/stable/tutorial/20_recipes/009_ask_and_tell.html#define-and-run)

        parser_args: The positional arguments to pass to the parser
        parser_kwargs: The keyword arguments to pass to the parser

    Returns:
        The search space
    """
    match parser:
        case "configspace":
            from amltk.pipeline.parsers.configspace import parser as cs_parser

            return cs_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case "optuna":
            from amltk.pipeline.parsers.optuna import parser as optuna_parser

            return optuna_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case str():  # type: ignore
            raise ValueError(
                f"Invalid str for parser {parser}. "
                "Please use 'configspace' or 'optuna' or pass in your own"
                " parser function",
            )
        case _:
            return parser(self, *parser_args, **parser_kwargs)

walk

walk(
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]

Walk the nodes in this chain.

PARAMETER	DESCRIPTION
path	The current path to this node TYPE: Sequence[Node] | None DEFAULT: None

YIELDS	DESCRIPTION
list[Node]	The parents of the node and the node itself

Source code in src/amltk/pipeline/node.py

def walk(
    self,
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]:
    """Walk the nodes in this chain.

    Args:
        path: The current path to this node

    Yields:
        The parents of the node and the node itself
    """
    path = list(path) if path is not None else []
    yield path, self

    for node in self.nodes:
        yield from node.walk(path=[*path, self])

Fixed dataclass

Fixed(
    item: Item,
    *,
    name: str | None = None,
    config: None = None,
    space: None = None,
    fidelities: None = None,
    config_transform: None = None,
    meta: Mapping[str, Any] | None = None
)

Bases: Node[Item, None]

A Fixed part of the pipeline that represents something that can not be configured and used directly as is.

It consists of an .item that is fixed, non-configurable and non-searchable. It also has no children.

This is useful for representing parts of the pipeline that are fixed, for example if you have a pipeline that is a Sequential of nodes, but you want to fix the first component to be a PCA with n_components=3, you can use a Fixed to represent that.

from amltk.pipeline import Component, Fixed, Sequential
from sklearn.ensemble import RandomForestClassifier
from sklearn.decomposition import PCA

rf = Component(RandomForestClassifier, space={"n_estimators": (10, 100)})
pca = Fixed(PCA(n_components=3))

pipeline = Sequential(pca, rf, name="my_pipeline")

╭─ Sequential(my_pipeline) ───────────────────╮
│ ╭─ Fixed(PCA) ─────────────╮                │
│ │ item PCA(n_components=3) │                │
│ ╰──────────────────────────╯                │
│                      ↓                      │
│ ╭─ Component(RandomForestClassifier) ─────╮ │
│ │ item  class RandomForestClassifier(...) │ │
│ │ space {'n_estimators': (10, 100)}       │ │
│ ╰─────────────────────────────────────────╯ │
╰─────────────────────────────────────────────╯

See Also

• Node

PARAMETER	DESCRIPTION
item	The item attached to this node. Will be fixed and can not be configured. TYPE: Item
name	The name of the node. If not specified, the name will be generated from the item. TYPE: str | None DEFAULT: None
meta	Any meta information about this node. TYPE: Mapping[str, Any] | None DEFAULT: None

Source code in src/amltk/pipeline/components.py

def __init__(  # noqa: D417
    self,
    item: Item,
    *,
    name: str | None = None,
    config: None = None,
    space: None = None,
    fidelities: None = None,
    config_transform: None = None,
    meta: Mapping[str, Any] | None = None,
):
    """Initialize a fixed node.

    Args:
        item: The item attached to this node. Will be fixed and can not
            be configured.
        name: The name of the node. If not specified, the name will be
            generated from the item.
        meta: Any meta information about this node.
    """
    super().__init__(
        name=name if name is not None else entity_name(item),
        item=item,
        config=config,
        space=space,
        fidelities=fidelities,
        config_transform=config_transform,
        meta=meta,
    )

config class-attribute instance-attribute

config: None = None

A fixed node has no config.

config_transform class-attribute instance-attribute

config_transform: None = None

A fixed node has no config so no transform.

fidelities class-attribute instance-attribute

fidelities: None = None

A fixed node has no search space.

item instance-attribute

item: Item

The fixed item that this node represents.

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = field(hash=False)

Any meta information about this node

name class-attribute instance-attribute

name: str = field(hash=True)

Name of the node

nodes class-attribute instance-attribute

nodes: tuple[] = ()

A fixed node has no children.

space class-attribute instance-attribute

space: None = None

A fixed node has no search space.

__getitem__

__getitem__(key: str) -> Node

Get the first from .nodes with key.

Source code in src/amltk/pipeline/node.py

def __getitem__(self, key: str) -> Node:
    """Get the first from [`.nodes`][amltk.pipeline.node.Node.nodes] with `key`."""
    found = first_true(
        self.nodes,
        None,
        lambda node: node.name == key,
    )
    if found is None:
        raise KeyError(
            f"Could not find node with name {key} in '{self.name}'."
            f" Available nodes are: {', '.join(node.name for node in self.nodes)}",
        )

    return found

build

build(
    builder: (
        Callable[Concatenate[Node, P], BuilderOutput]
        | Literal["sklearn"]
    ),
    *builder_args: args,
    **builder_kwargs: kwargs
) -> BuilderOutput | Pipeline

Build a concrete object out of this node.

PARAMETER	DESCRIPTION
builder	The builder to use. This can be a function that takes in the node and returns the object or a string that is one of: "sklearn": Build a sklearn.pipeline.Pipeline out of this node. TYPE: Callable[Concatenate[Node, P], BuilderOutput] | Literal['sklearn']
builder_args	The positional arguments to pass to the builder TYPE: args DEFAULT: ()
builder_kwargs	The keyword arguments to pass to the builder TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BuilderOutput | Pipeline	The built object

Source code in src/amltk/pipeline/node.py

def build(
    self,
    builder: Callable[Concatenate[Node, P], BuilderOutput] | Literal["sklearn"],
    *builder_args: P.args,
    **builder_kwargs: P.kwargs,
) -> BuilderOutput | SklearnPipeline:
    """Build a concrete object out of this node.

    Args:
        builder: The builder to use. This can be a function that takes in
            the node and returns the object or a string that is one of:

            * `#!python "sklearn"`: Build a
                [`sklearn.pipeline.Pipeline`][sklearn.pipeline.Pipeline]
                out of this node.

        builder_args: The positional arguments to pass to the builder
        builder_kwargs: The keyword arguments to pass to the builder

    Returns:
        The built object
    """
    match builder:
        case "sklearn":
            from amltk.pipeline.builders.sklearn import build as _build

            return _build(self, *builder_args, **builder_kwargs)  # type: ignore
        case _:
            return builder(self, *builder_args, **builder_kwargs)

configure

configure(
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None
) -> Self

Configure this node and anything following it with the given config.

PARAMETER	DESCRIPTION
config	The configuration to apply TYPE: Config
prefixed_name	Whether items in the config are prefixed by the names of the nodes. * If None, the default, then prefixed_name will be assumed to be True if this node has a next node or if the config has keys that begin with this nodes name. * If True, then the config will be searched for items prefixed by the name of the node (and subsequent chained nodes). * If False, then the config will be searched for items without the prefix, i.e. the config keys are exactly those matching this nodes search space. TYPE: bool | None DEFAULT: None
transform_context	Any context to give to config_transform= of individual nodes. TYPE: Any | None DEFAULT: None
params	The params to match any requests when configuring this node. These will match against any ParamRequests in the config and will be used to fill in any missing values. TYPE: Mapping[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Self	The configured node

Source code in src/amltk/pipeline/node.py

def configure(
    self,
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None,
) -> Self:
    """Configure this node and anything following it with the given config.

    Args:
        config: The configuration to apply
        prefixed_name: Whether items in the config are prefixed by the names
            of the nodes.
            * If `None`, the default, then `prefixed_name` will be assumed to
                be `True` if this node has a next node or if the config has
                keys that begin with this nodes name.
            * If `True`, then the config will be searched for items prefixed
                by the name of the node (and subsequent chained nodes).
            * If `False`, then the config will be searched for items without
                the prefix, i.e. the config keys are exactly those matching
                this nodes search space.
        transform_context: Any context to give to `config_transform=` of individual
            nodes.
        params: The params to match any requests when configuring this node.
            These will match against any ParamRequests in the config and will
            be used to fill in any missing values.

    Returns:
        The configured node
    """
    # Get the config for this node
    match prefixed_name:
        case True:
            config = mapping_select(config, f"{self.name}:")
        case False:
            pass
        case None if any(k.startswith(f"{self.name}:") for k in config):
            config = mapping_select(config, f"{self.name}:")
        case None:
            pass

    _kwargs: dict[str, Any] = {}

    # Configure all the branches if exists
    if len(self.nodes) > 0:
        nodes = tuple(
            node.configure(
                config,
                prefixed_name=True,
                transform_context=transform_context,
                params=params,
            )
            for node in self.nodes
        )
        _kwargs["nodes"] = nodes

    this_config = {
        hp: v
        for hp, v in config.items()
        if (
            ":" not in hp
            and not any(hp.startswith(f"{node.name}") for node in self.nodes)
        )
    }
    if self.config is not None:
        this_config = {**self.config, **this_config}

    this_config = dict(self._fufill_param_requests(this_config, params=params))

    if self.config_transform is not None:
        this_config = dict(self.config_transform(this_config, transform_context))

    if len(this_config) > 0:
        _kwargs["config"] = dict(this_config)

    return self.mutate(**_kwargs)

copy

copy() -> Self

Copy this node, removing all links in the process.

Source code in src/amltk/pipeline/node.py

def copy(self) -> Self:
    """Copy this node, removing all links in the process."""
    return self.mutate()

display

display(*, full: bool = False) -> RenderableType

Display this node.

PARAMETER	DESCRIPTION
full	Whether to display the full node or just a summary TYPE: bool DEFAULT: False

Source code in src/amltk/pipeline/node.py

def display(self, *, full: bool = False) -> RenderableType:
    """Display this node.

    Args:
        full: Whether to display the full node or just a summary
    """
    if not full:
        return self.__rich__()

    from rich.console import Group as RichGroup

    return RichGroup(*self._rich_iter())

factorize

factorize(
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None
) -> Iterator[Self]

Please see factorize().

Source code in src/amltk/pipeline/node.py

def factorize(
    self,
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None,
) -> Iterator[Self]:
    """Please see [`factorize()`][amltk.pipeline.ops.factorize]."""  # noqa: D402
    from amltk.pipeline.ops import factorize

    yield from factorize(
        self,
        min_depth=min_depth,
        max_depth=max_depth,
        current_depth=current_depth,
        factor_by=factor_by,
        assign_child=assign_child,
    )

fidelity_space

fidelity_space() -> dict[str, Any]

Get the fidelities for this node and any connected nodes.

Source code in src/amltk/pipeline/node.py

def fidelity_space(self) -> dict[str, Any]:
    """Get the fidelities for this node and any connected nodes."""
    fids = {}
    for node in self.nodes:
        fids.update(prefix_keys(node.fidelity_space(), f"{self.name}:"))

    return fids

find

find(
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None

Find a node in that's nested deeper from this node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]
default	The value to return if the node is not found. Defaults to None TYPE: T | None DEFAULT: None

RETURNS	DESCRIPTION
Node | T | None	The node if found, otherwise the default value. Defaults to None

Source code in src/amltk/pipeline/node.py

def find(
    self,
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None:
    """Find a node in that's nested deeper from this node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node
        default: The value to return if the node is not found. Defaults to None

    Returns:
        The node if found, otherwise the default value. Defaults to None
    """
    itr = self.iter()
    match key:
        case Node():
            return first_true(itr, default, lambda node: node == key)
        case str():
            return first_true(itr, default, lambda node: node.name == key)
        case _:
            return first_true(itr, default, key)  # type: ignore

iter

iter() -> Iterator[Node]

Iterate the the nodes, including this node.

YIELDS	DESCRIPTION
Node	The nodes connected to this node

Source code in src/amltk/pipeline/node.py

def iter(self) -> Iterator[Node]:
    """Iterate the the nodes, including this node.

    Yields:
        The nodes connected to this node
    """
    yield self
    for node in self.nodes:
        yield from node.iter()

linearized_fidelity

linearized_fidelity(
    value: float,
) -> dict[str, int | float | Any]

Get the liniearized fidelities for this node and any connected nodes.

PARAMETER	DESCRIPTION
value	The value to linearize. Must be between [0, 1] TYPE: float

Return

dictionary from key to it's linearized fidelity.

Source code in src/amltk/pipeline/node.py

def linearized_fidelity(self, value: float) -> dict[str, int | float | Any]:
    """Get the liniearized fidelities for this node and any connected nodes.

    Args:
        value: The value to linearize. Must be between [0, 1]

    Return:
        dictionary from key to it's linearized fidelity.
    """
    assert 1.0 <= value <= 100.0, f"{value=} not in [1.0, 100.0]"  # noqa: PLR2004
    d = {}
    for node in self.nodes:
        node_fids = prefix_keys(
            node.linearized_fidelity(value),
            f"{self.name}:",
        )
        d.update(node_fids)

    if self.fidelities is None:
        return d

    for f_name, f_range in self.fidelities.items():
        match f_range:
            case (int() | float(), int() | float()):
                low, high = f_range
                fid = low + (high - low) * value
                fid = low + (high - low) * (value - 1) / 100
                fid = fid if isinstance(low, float) else round(fid)
                d[f_name] = fid
            case _:
                raise ValueError(
                    f"Invalid fidelities to linearize {f_range} for {f_name}"
                    f" in {self}. Only supports ranges of the form (low, high)",
                )

    return prefix_keys(d, f"{self.name}:")

mutate

mutate(**kwargs: Any) -> Self

Mutate the node with the given keyword arguments.

PARAMETER	DESCRIPTION
**kwargs	The keyword arguments to mutate TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	Self The mutated node

Source code in src/amltk/pipeline/node.py

def mutate(self, **kwargs: Any) -> Self:
    """Mutate the node with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to mutate

    Returns:
        Self
            The mutated node
    """
    _args = ()
    _kwargs = {**self.__dict__, **kwargs}

    # If there's nodes in kwargs, we have to check if it's
    # a positional or keyword argument and handle accordingly.
    if (nodes := _kwargs.pop("nodes", None)) is not None:
        match self._NODES_INIT:
            case "args":
                _args = nodes
            case "kwargs":
                _kwargs["nodes"] = nodes
            case None if len(nodes) == 0:
                pass  # Just ignore it, it's popped out
            case None:
                raise ValueError(
                    "Cannot mutate nodes when __init__ does not accept nodes",
                )

    # If there's a config in kwargs, we have to check if it's actually got values
    config = _kwargs.pop("config", None)
    if config is not None and len(config) > 0:
        _kwargs["config"] = config

    # Lastly, we remove anything that can't be passed to kwargs of the
    # subclasses __init__
    _available_kwargs = inspect.signature(self.__init__).parameters.keys()  # type: ignore
    for k in list(_kwargs.keys()):
        if k not in _available_kwargs:
            _kwargs.pop(k)

    return self.__class__(*_args, **_kwargs)

path_to

path_to(
    key: str | Node | Callable[[Node], bool]
) -> list[Node] | None

Find a path to the given node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]

RETURNS	DESCRIPTION
list[Node] | None	The path to the node if found, else None

Source code in src/amltk/pipeline/node.py

def path_to(self, key: str | Node | Callable[[Node], bool]) -> list[Node] | None:
    """Find a path to the given node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node

    Returns:
        The path to the node if found, else None
    """
    # We found our target, just return now

    match key:
        case Node():
            pred = lambda node: node == key
        case str():
            pred = lambda node: node.name == key
        case _:
            pred = key

    for path, node in self.walk():
        if pred(node):
            return path

    return None

search_space

search_space(
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: args,
    **parser_kwargs: kwargs
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace

Get the search space for this node.

PARAMETER	DESCRIPTION
parser	The parser to use. This can be a function that takes in the node and returns the search space or a string that is one of: "configspace": Build a ConfigSpace.ConfigurationSpace out of this node. "optuna": Build a dict of hyperparameters that Optuna can use in its ask and tell methods TYPE: Callable[Concatenate[Node, P], ParserOutput] | Literal['configspace', 'optuna']
parser_args	The positional arguments to pass to the parser TYPE: args DEFAULT: ()
parser_kwargs	The keyword arguments to pass to the parser TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
ParserOutput | ConfigurationSpace | OptunaSearchSpace	The search space

Source code in src/amltk/pipeline/node.py

def search_space(
    self,
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: P.args,
    **parser_kwargs: P.kwargs,
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace:
    """Get the search space for this node.

    Args:
        parser: The parser to use. This can be a function that takes in
            the node and returns the search space or a string that is one of:

            * `#!python "configspace"`: Build a
                [`ConfigSpace.ConfigurationSpace`](https://automl.github.io/ConfigSpace/master/)
                out of this node.
            * `#!python "optuna"`: Build a dict of hyperparameters that Optuna can
                use in its [ask and tell methods](https://optuna.readthedocs.io/en/stable/tutorial/20_recipes/009_ask_and_tell.html#define-and-run)

        parser_args: The positional arguments to pass to the parser
        parser_kwargs: The keyword arguments to pass to the parser

    Returns:
        The search space
    """
    match parser:
        case "configspace":
            from amltk.pipeline.parsers.configspace import parser as cs_parser

            return cs_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case "optuna":
            from amltk.pipeline.parsers.optuna import parser as optuna_parser

            return optuna_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case str():  # type: ignore
            raise ValueError(
                f"Invalid str for parser {parser}. "
                "Please use 'configspace' or 'optuna' or pass in your own"
                " parser function",
            )
        case _:
            return parser(self, *parser_args, **parser_kwargs)

walk

walk(
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]

Walk the nodes in this chain.

PARAMETER	DESCRIPTION
path	The current path to this node TYPE: Sequence[Node] | None DEFAULT: None

YIELDS	DESCRIPTION
list[Node]	The parents of the node and the node itself

Source code in src/amltk/pipeline/node.py

def walk(
    self,
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]:
    """Walk the nodes in this chain.

    Args:
        path: The current path to this node

    Yields:
        The parents of the node and the node itself
    """
    path = list(path) if path is not None else []
    yield path, self

    for node in self.nodes:
        yield from node.walk(path=[*path, self])

Join dataclass

Join(
    *nodes: Node | NodeLike,
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: (
        Callable[[Config, Any], Config] | None
    ) = None,
    meta: Mapping[str, Any] | None = None
)

Bases: Node[Item, Space]

Join together different parts of the pipeline.

This indicates the different children in .nodes should act in tandem with one another, for example, concatenating the outputs of the various members of the Join.

from amltk.pipeline import Join, Component
from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest

pca = Component(PCA, space={"n_components": (1, 3)})
kbest = Component(SelectKBest, space={"k": (1, 3)})

join = Join(pca, kbest, name="my_feature_union")

╭─ Join(my_feature_union) ────────────────────────────────────────────╮
│ ╭─ Component(PCA) ───────────────╮ ╭─ Component(SelectKBest) ─────╮ │
│ │ item  class PCA(...)           │ │ item  class SelectKBest(...) │ │
│ │ space {'n_components': (1, 3)} │ │ space {'k': (1, 3)}          │ │
│ ╰────────────────────────────────╯ ╰──────────────────────────────╯ │
╰─────────────────────────────────────────────────────────────────────╯

See Also

• Node

PARAMETER	DESCRIPTION
nodes	The nodes that should be joined together in parallel. TYPE: Node | NodeLike DEFAULT: ()
item	The item attached to this node (if any). TYPE: Item | Callable[[Item], Item] | None DEFAULT: None
name	The name of the node. If not specified, the name will be randomly generated. TYPE: str | None DEFAULT: None
config	The configuration for this node. TYPE: Config | None DEFAULT: None
space	The search space for this node. This will be used when search_space() is called. TYPE: Space | None DEFAULT: None
fidelities	The fidelities for this node. TYPE: Mapping[str, Any] | None DEFAULT: None
config_transform	A function that transforms the config= parameter during configure(config) before return the new configured node. Useful for times where you need to combine multiple parameters into one. TYPE: Callable[[Config, Any], Config] | None DEFAULT: None
meta	Any meta information about this node. TYPE: Mapping[str, Any] | None DEFAULT: None

Source code in src/amltk/pipeline/components.py

def __init__(
    self,
    *nodes: Node | NodeLike,
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: Callable[[Config, Any], Config] | None = None,
    meta: Mapping[str, Any] | None = None,
):
    """Initialize a join node.

    Args:
        nodes: The nodes that should be joined together in parallel.
        item: The item attached to this node (if any).
        name: The name of the node. If not specified, the name will be
            randomly generated.
        config: The configuration for this node.
        space: The search space for this node. This will be used when
            [`search_space()`][amltk.pipeline.node.Node.search_space] is called.
        fidelities: The fidelities for this node.
        config_transform: A function that transforms the `config=` parameter
            during [`configure(config)`][amltk.pipeline.node.Node.configure]
            before return the new configured node. Useful for times where
            you need to combine multiple parameters into one.
        meta: Any meta information about this node.
    """
    _nodes = tuple(as_node(n) for n in nodes)
    if not all_unique(_nodes, key=lambda node: node.name):
        raise ValueError(
            f"Can't handle nodes they do not all contain unique names, {nodes=}."
            "\nAll nodes must have a unique name. Please provide a `name=` to them",
        )

    if name is None:
        name = f"Join-{randuid(8)}"

    super().__init__(
        *_nodes,
        name=name,
        item=item,
        config=config,
        space=space,
        fidelities=fidelities,
        config_transform=config_transform,
        meta=meta,
    )

config class-attribute instance-attribute

config: Config | None = field(hash=False)

The configuration for this node

config_transform class-attribute instance-attribute

config_transform: Callable[[Config, Any], Config] | None = (
    field(hash=False)
)

A function that transforms the configuration of this node

fidelities class-attribute instance-attribute

fidelities: Mapping[str, Any] | None = field(hash=False)

The fidelities for this node

item class-attribute instance-attribute

item: Callable[..., Item] | Item | None = field(hash=False)

The item attached to this node

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = field(hash=False)

Any meta information about this node

name class-attribute instance-attribute

name: str = field(hash=True)

Name of the node

nodes instance-attribute

nodes: tuple[Node, ...]

The nodes that should be joined together in parallel.

space class-attribute instance-attribute

space: Space | None = field(hash=False)

The search space for this node

__getitem__

__getitem__(key: str) -> Node

Get the first from .nodes with key.

Source code in src/amltk/pipeline/node.py

def __getitem__(self, key: str) -> Node:
    """Get the first from [`.nodes`][amltk.pipeline.node.Node.nodes] with `key`."""
    found = first_true(
        self.nodes,
        None,
        lambda node: node.name == key,
    )
    if found is None:
        raise KeyError(
            f"Could not find node with name {key} in '{self.name}'."
            f" Available nodes are: {', '.join(node.name for node in self.nodes)}",
        )

    return found

build

build(
    builder: (
        Callable[Concatenate[Node, P], BuilderOutput]
        | Literal["sklearn"]
    ),
    *builder_args: args,
    **builder_kwargs: kwargs
) -> BuilderOutput | Pipeline

Build a concrete object out of this node.

PARAMETER	DESCRIPTION
builder	The builder to use. This can be a function that takes in the node and returns the object or a string that is one of: "sklearn": Build a sklearn.pipeline.Pipeline out of this node. TYPE: Callable[Concatenate[Node, P], BuilderOutput] | Literal['sklearn']
builder_args	The positional arguments to pass to the builder TYPE: args DEFAULT: ()
builder_kwargs	The keyword arguments to pass to the builder TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BuilderOutput | Pipeline	The built object

Source code in src/amltk/pipeline/node.py

def build(
    self,
    builder: Callable[Concatenate[Node, P], BuilderOutput] | Literal["sklearn"],
    *builder_args: P.args,
    **builder_kwargs: P.kwargs,
) -> BuilderOutput | SklearnPipeline:
    """Build a concrete object out of this node.

    Args:
        builder: The builder to use. This can be a function that takes in
            the node and returns the object or a string that is one of:

            * `#!python "sklearn"`: Build a
                [`sklearn.pipeline.Pipeline`][sklearn.pipeline.Pipeline]
                out of this node.

        builder_args: The positional arguments to pass to the builder
        builder_kwargs: The keyword arguments to pass to the builder

    Returns:
        The built object
    """
    match builder:
        case "sklearn":
            from amltk.pipeline.builders.sklearn import build as _build

            return _build(self, *builder_args, **builder_kwargs)  # type: ignore
        case _:
            return builder(self, *builder_args, **builder_kwargs)

configure

configure(
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None
) -> Self

Configure this node and anything following it with the given config.

PARAMETER	DESCRIPTION
config	The configuration to apply TYPE: Config
prefixed_name	Whether items in the config are prefixed by the names of the nodes. * If None, the default, then prefixed_name will be assumed to be True if this node has a next node or if the config has keys that begin with this nodes name. * If True, then the config will be searched for items prefixed by the name of the node (and subsequent chained nodes). * If False, then the config will be searched for items without the prefix, i.e. the config keys are exactly those matching this nodes search space. TYPE: bool | None DEFAULT: None
transform_context	Any context to give to config_transform= of individual nodes. TYPE: Any | None DEFAULT: None
params	The params to match any requests when configuring this node. These will match against any ParamRequests in the config and will be used to fill in any missing values. TYPE: Mapping[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Self	The configured node

Source code in src/amltk/pipeline/node.py

def configure(
    self,
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None,
) -> Self:
    """Configure this node and anything following it with the given config.

    Args:
        config: The configuration to apply
        prefixed_name: Whether items in the config are prefixed by the names
            of the nodes.
            * If `None`, the default, then `prefixed_name` will be assumed to
                be `True` if this node has a next node or if the config has
                keys that begin with this nodes name.
            * If `True`, then the config will be searched for items prefixed
                by the name of the node (and subsequent chained nodes).
            * If `False`, then the config will be searched for items without
                the prefix, i.e. the config keys are exactly those matching
                this nodes search space.
        transform_context: Any context to give to `config_transform=` of individual
            nodes.
        params: The params to match any requests when configuring this node.
            These will match against any ParamRequests in the config and will
            be used to fill in any missing values.

    Returns:
        The configured node
    """
    # Get the config for this node
    match prefixed_name:
        case True:
            config = mapping_select(config, f"{self.name}:")
        case False:
            pass
        case None if any(k.startswith(f"{self.name}:") for k in config):
            config = mapping_select(config, f"{self.name}:")
        case None:
            pass

    _kwargs: dict[str, Any] = {}

    # Configure all the branches if exists
    if len(self.nodes) > 0:
        nodes = tuple(
            node.configure(
                config,
                prefixed_name=True,
                transform_context=transform_context,
                params=params,
            )
            for node in self.nodes
        )
        _kwargs["nodes"] = nodes

    this_config = {
        hp: v
        for hp, v in config.items()
        if (
            ":" not in hp
            and not any(hp.startswith(f"{node.name}") for node in self.nodes)
        )
    }
    if self.config is not None:
        this_config = {**self.config, **this_config}

    this_config = dict(self._fufill_param_requests(this_config, params=params))

    if self.config_transform is not None:
        this_config = dict(self.config_transform(this_config, transform_context))

    if len(this_config) > 0:
        _kwargs["config"] = dict(this_config)

    return self.mutate(**_kwargs)

copy

copy() -> Self

Copy this node, removing all links in the process.

Source code in src/amltk/pipeline/node.py

def copy(self) -> Self:
    """Copy this node, removing all links in the process."""
    return self.mutate()

display

display(*, full: bool = False) -> RenderableType

Display this node.

PARAMETER	DESCRIPTION
full	Whether to display the full node or just a summary TYPE: bool DEFAULT: False

Source code in src/amltk/pipeline/node.py

def display(self, *, full: bool = False) -> RenderableType:
    """Display this node.

    Args:
        full: Whether to display the full node or just a summary
    """
    if not full:
        return self.__rich__()

    from rich.console import Group as RichGroup

    return RichGroup(*self._rich_iter())

factorize

factorize(
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None
) -> Iterator[Self]

Please see factorize().

Source code in src/amltk/pipeline/node.py

def factorize(
    self,
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None,
) -> Iterator[Self]:
    """Please see [`factorize()`][amltk.pipeline.ops.factorize]."""  # noqa: D402
    from amltk.pipeline.ops import factorize

    yield from factorize(
        self,
        min_depth=min_depth,
        max_depth=max_depth,
        current_depth=current_depth,
        factor_by=factor_by,
        assign_child=assign_child,
    )

fidelity_space

fidelity_space() -> dict[str, Any]

Get the fidelities for this node and any connected nodes.

Source code in src/amltk/pipeline/node.py

def fidelity_space(self) -> dict[str, Any]:
    """Get the fidelities for this node and any connected nodes."""
    fids = {}
    for node in self.nodes:
        fids.update(prefix_keys(node.fidelity_space(), f"{self.name}:"))

    return fids

find

find(
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None

Find a node in that's nested deeper from this node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]
default	The value to return if the node is not found. Defaults to None TYPE: T | None DEFAULT: None

RETURNS	DESCRIPTION
Node | T | None	The node if found, otherwise the default value. Defaults to None

Source code in src/amltk/pipeline/node.py

def find(
    self,
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None:
    """Find a node in that's nested deeper from this node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node
        default: The value to return if the node is not found. Defaults to None

    Returns:
        The node if found, otherwise the default value. Defaults to None
    """
    itr = self.iter()
    match key:
        case Node():
            return first_true(itr, default, lambda node: node == key)
        case str():
            return first_true(itr, default, lambda node: node.name == key)
        case _:
            return first_true(itr, default, key)  # type: ignore

iter

iter() -> Iterator[Node]

Iterate the the nodes, including this node.

YIELDS	DESCRIPTION
Node	The nodes connected to this node

Source code in src/amltk/pipeline/node.py

def iter(self) -> Iterator[Node]:
    """Iterate the the nodes, including this node.

    Yields:
        The nodes connected to this node
    """
    yield self
    for node in self.nodes:
        yield from node.iter()

linearized_fidelity

linearized_fidelity(
    value: float,
) -> dict[str, int | float | Any]

Get the liniearized fidelities for this node and any connected nodes.

PARAMETER	DESCRIPTION
value	The value to linearize. Must be between [0, 1] TYPE: float

Return

dictionary from key to it's linearized fidelity.

Source code in src/amltk/pipeline/node.py

def linearized_fidelity(self, value: float) -> dict[str, int | float | Any]:
    """Get the liniearized fidelities for this node and any connected nodes.

    Args:
        value: The value to linearize. Must be between [0, 1]

    Return:
        dictionary from key to it's linearized fidelity.
    """
    assert 1.0 <= value <= 100.0, f"{value=} not in [1.0, 100.0]"  # noqa: PLR2004
    d = {}
    for node in self.nodes:
        node_fids = prefix_keys(
            node.linearized_fidelity(value),
            f"{self.name}:",
        )
        d.update(node_fids)

    if self.fidelities is None:
        return d

    for f_name, f_range in self.fidelities.items():
        match f_range:
            case (int() | float(), int() | float()):
                low, high = f_range
                fid = low + (high - low) * value
                fid = low + (high - low) * (value - 1) / 100
                fid = fid if isinstance(low, float) else round(fid)
                d[f_name] = fid
            case _:
                raise ValueError(
                    f"Invalid fidelities to linearize {f_range} for {f_name}"
                    f" in {self}. Only supports ranges of the form (low, high)",
                )

    return prefix_keys(d, f"{self.name}:")

mutate

mutate(**kwargs: Any) -> Self

Mutate the node with the given keyword arguments.

PARAMETER	DESCRIPTION
**kwargs	The keyword arguments to mutate TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	Self The mutated node

Source code in src/amltk/pipeline/node.py

def mutate(self, **kwargs: Any) -> Self:
    """Mutate the node with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to mutate

    Returns:
        Self
            The mutated node
    """
    _args = ()
    _kwargs = {**self.__dict__, **kwargs}

    # If there's nodes in kwargs, we have to check if it's
    # a positional or keyword argument and handle accordingly.
    if (nodes := _kwargs.pop("nodes", None)) is not None:
        match self._NODES_INIT:
            case "args":
                _args = nodes
            case "kwargs":
                _kwargs["nodes"] = nodes
            case None if len(nodes) == 0:
                pass  # Just ignore it, it's popped out
            case None:
                raise ValueError(
                    "Cannot mutate nodes when __init__ does not accept nodes",
                )

    # If there's a config in kwargs, we have to check if it's actually got values
    config = _kwargs.pop("config", None)
    if config is not None and len(config) > 0:
        _kwargs["config"] = config

    # Lastly, we remove anything that can't be passed to kwargs of the
    # subclasses __init__
    _available_kwargs = inspect.signature(self.__init__).parameters.keys()  # type: ignore
    for k in list(_kwargs.keys()):
        if k not in _available_kwargs:
            _kwargs.pop(k)

    return self.__class__(*_args, **_kwargs)

path_to

path_to(
    key: str | Node | Callable[[Node], bool]
) -> list[Node] | None

Find a path to the given node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]

RETURNS	DESCRIPTION
list[Node] | None	The path to the node if found, else None

Source code in src/amltk/pipeline/node.py

def path_to(self, key: str | Node | Callable[[Node], bool]) -> list[Node] | None:
    """Find a path to the given node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node

    Returns:
        The path to the node if found, else None
    """
    # We found our target, just return now

    match key:
        case Node():
            pred = lambda node: node == key
        case str():
            pred = lambda node: node.name == key
        case _:
            pred = key

    for path, node in self.walk():
        if pred(node):
            return path

    return None

search_space

search_space(
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: args,
    **parser_kwargs: kwargs
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace

Get the search space for this node.

PARAMETER	DESCRIPTION
parser	The parser to use. This can be a function that takes in the node and returns the search space or a string that is one of: "configspace": Build a ConfigSpace.ConfigurationSpace out of this node. "optuna": Build a dict of hyperparameters that Optuna can use in its ask and tell methods TYPE: Callable[Concatenate[Node, P], ParserOutput] | Literal['configspace', 'optuna']
parser_args	The positional arguments to pass to the parser TYPE: args DEFAULT: ()
parser_kwargs	The keyword arguments to pass to the parser TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
ParserOutput | ConfigurationSpace | OptunaSearchSpace	The search space

Source code in src/amltk/pipeline/node.py

def search_space(
    self,
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: P.args,
    **parser_kwargs: P.kwargs,
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace:
    """Get the search space for this node.

    Args:
        parser: The parser to use. This can be a function that takes in
            the node and returns the search space or a string that is one of:

            * `#!python "configspace"`: Build a
                [`ConfigSpace.ConfigurationSpace`](https://automl.github.io/ConfigSpace/master/)
                out of this node.
            * `#!python "optuna"`: Build a dict of hyperparameters that Optuna can
                use in its [ask and tell methods](https://optuna.readthedocs.io/en/stable/tutorial/20_recipes/009_ask_and_tell.html#define-and-run)

        parser_args: The positional arguments to pass to the parser
        parser_kwargs: The keyword arguments to pass to the parser

    Returns:
        The search space
    """
    match parser:
        case "configspace":
            from amltk.pipeline.parsers.configspace import parser as cs_parser

            return cs_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case "optuna":
            from amltk.pipeline.parsers.optuna import parser as optuna_parser

            return optuna_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case str():  # type: ignore
            raise ValueError(
                f"Invalid str for parser {parser}. "
                "Please use 'configspace' or 'optuna' or pass in your own"
                " parser function",
            )
        case _:
            return parser(self, *parser_args, **parser_kwargs)

walk

walk(
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]

Walk the nodes in this chain.

PARAMETER	DESCRIPTION
path	The current path to this node TYPE: Sequence[Node] | None DEFAULT: None

YIELDS	DESCRIPTION
list[Node]	The parents of the node and the node itself

Source code in src/amltk/pipeline/node.py

def walk(
    self,
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]:
    """Walk the nodes in this chain.

    Args:
        path: The current path to this node

    Yields:
        The parents of the node and the node itself
    """
    path = list(path) if path is not None else []
    yield path, self

    for node in self.nodes:
        yield from node.walk(path=[*path, self])

Searchable dataclass

Searchable(
    space: Space | None = None,
    *,
    name: str | None = None,
    config: Config | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: (
        Callable[[Config, Any], Config] | None
    ) = None,
    meta: Mapping[str, Any] | None = None
)

Bases: Node[None, Space]

A Searchable node of the pipeline which just represents a search space, no item attached.

While not usually applicable to pipelines you want to build, this node is useful for creating a search space, especially if the real pipeline you want to optimize can not be built directly. For example, if you are optimize a script, you may wish to use a Searchable to represent the search space of that script.

from amltk.pipeline import Searchable

script_space = Searchable({"mode": ["orange", "blue", "red"], "n": (10, 100)})

╭─ Searchable(Searchable-JMC3PIeV) ─────────────────────────╮
│ space {'mode': ['orange', 'blue', 'red'], 'n': (10, 100)} │
╰───────────────────────────────────────────────────────────╯

See Also

• Node

PARAMETER	DESCRIPTION
space	The search space for this node. This will be used when search_space() is called. TYPE: Space | None DEFAULT: None
name	The name of the node. If not specified, a random one will be generated. TYPE: str | None DEFAULT: None
config	The configuration for this node. Useful for setting some default values. TYPE: Config | None DEFAULT: None
fidelities	The fidelities for this node. TYPE: Mapping[str, Any] | None DEFAULT: None
config_transform	A function that transforms the config= parameter during configure(config) before return the new configured node. Useful for times where you need to combine multiple parameters into one. TYPE: Callable[[Config, Any], Config] | None DEFAULT: None
meta	Any meta information about this node. TYPE: Mapping[str, Any] | None DEFAULT: None

Source code in src/amltk/pipeline/components.py

def __init__(
    self,
    space: Space | None = None,
    *,
    name: str | None = None,
    config: Config | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: Callable[[Config, Any], Config] | None = None,
    meta: Mapping[str, Any] | None = None,
):
    """Initialize a choice.

    Args:
        space: The search space for this node. This will be used when
            [`search_space()`][amltk.pipeline.node.Node.search_space] is called.
        name: The name of the node. If not specified, a random one will
            be generated.
        config: The configuration for this node. Useful for setting some
            default values.
        fidelities: The fidelities for this node.
        config_transform: A function that transforms the `config=` parameter
            during [`configure(config)`][amltk.pipeline.node.Node.configure]
            before return the new configured node. Useful for times where
            you need to combine multiple parameters into one.
        meta: Any meta information about this node.
    """
    if name is None:
        name = f"Searchable-{randuid(8)}"

    super().__init__(
        name=name,
        config=config,
        space=space,
        fidelities=fidelities,
        config_transform=config_transform,
        meta=meta,
    )

config class-attribute instance-attribute

config: Config | None = field(hash=False)

The configuration for this node

config_transform class-attribute instance-attribute

config_transform: Callable[[Config, Any], Config] | None = (
    field(hash=False)
)

A function that transforms the configuration of this node

fidelities class-attribute instance-attribute

fidelities: Mapping[str, Any] | None = field(hash=False)

The fidelities for this node

item class-attribute instance-attribute

item: None = None

A searchable has no item.

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = field(hash=False)

Any meta information about this node

name class-attribute instance-attribute

name: str = field(hash=True)

Name of the node

nodes class-attribute instance-attribute

nodes: tuple[] = ()

A searchable has no children.

space class-attribute instance-attribute

space: Space | None = field(hash=False)

The search space for this node

__getitem__

__getitem__(key: str) -> Node

Get the first from .nodes with key.

Source code in src/amltk/pipeline/node.py

def __getitem__(self, key: str) -> Node:
    """Get the first from [`.nodes`][amltk.pipeline.node.Node.nodes] with `key`."""
    found = first_true(
        self.nodes,
        None,
        lambda node: node.name == key,
    )
    if found is None:
        raise KeyError(
            f"Could not find node with name {key} in '{self.name}'."
            f" Available nodes are: {', '.join(node.name for node in self.nodes)}",
        )

    return found

build

build(
    builder: (
        Callable[Concatenate[Node, P], BuilderOutput]
        | Literal["sklearn"]
    ),
    *builder_args: args,
    **builder_kwargs: kwargs
) -> BuilderOutput | Pipeline

Build a concrete object out of this node.

PARAMETER	DESCRIPTION
builder	The builder to use. This can be a function that takes in the node and returns the object or a string that is one of: "sklearn": Build a sklearn.pipeline.Pipeline out of this node. TYPE: Callable[Concatenate[Node, P], BuilderOutput] | Literal['sklearn']
builder_args	The positional arguments to pass to the builder TYPE: args DEFAULT: ()
builder_kwargs	The keyword arguments to pass to the builder TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BuilderOutput | Pipeline	The built object

Source code in src/amltk/pipeline/node.py

def build(
    self,
    builder: Callable[Concatenate[Node, P], BuilderOutput] | Literal["sklearn"],
    *builder_args: P.args,
    **builder_kwargs: P.kwargs,
) -> BuilderOutput | SklearnPipeline:
    """Build a concrete object out of this node.

    Args:
        builder: The builder to use. This can be a function that takes in
            the node and returns the object or a string that is one of:

            * `#!python "sklearn"`: Build a
                [`sklearn.pipeline.Pipeline`][sklearn.pipeline.Pipeline]
                out of this node.

        builder_args: The positional arguments to pass to the builder
        builder_kwargs: The keyword arguments to pass to the builder

    Returns:
        The built object
    """
    match builder:
        case "sklearn":
            from amltk.pipeline.builders.sklearn import build as _build

            return _build(self, *builder_args, **builder_kwargs)  # type: ignore
        case _:
            return builder(self, *builder_args, **builder_kwargs)

configure

configure(
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None
) -> Self

Configure this node and anything following it with the given config.

PARAMETER	DESCRIPTION
config	The configuration to apply TYPE: Config
prefixed_name	Whether items in the config are prefixed by the names of the nodes. * If None, the default, then prefixed_name will be assumed to be True if this node has a next node or if the config has keys that begin with this nodes name. * If True, then the config will be searched for items prefixed by the name of the node (and subsequent chained nodes). * If False, then the config will be searched for items without the prefix, i.e. the config keys are exactly those matching this nodes search space. TYPE: bool | None DEFAULT: None
transform_context	Any context to give to config_transform= of individual nodes. TYPE: Any | None DEFAULT: None
params	The params to match any requests when configuring this node. These will match against any ParamRequests in the config and will be used to fill in any missing values. TYPE: Mapping[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Self	The configured node

Source code in src/amltk/pipeline/node.py

def configure(
    self,
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None,
) -> Self:
    """Configure this node and anything following it with the given config.

    Args:
        config: The configuration to apply
        prefixed_name: Whether items in the config are prefixed by the names
            of the nodes.
            * If `None`, the default, then `prefixed_name` will be assumed to
                be `True` if this node has a next node or if the config has
                keys that begin with this nodes name.
            * If `True`, then the config will be searched for items prefixed
                by the name of the node (and subsequent chained nodes).
            * If `False`, then the config will be searched for items without
                the prefix, i.e. the config keys are exactly those matching
                this nodes search space.
        transform_context: Any context to give to `config_transform=` of individual
            nodes.
        params: The params to match any requests when configuring this node.
            These will match against any ParamRequests in the config and will
            be used to fill in any missing values.

    Returns:
        The configured node
    """
    # Get the config for this node
    match prefixed_name:
        case True:
            config = mapping_select(config, f"{self.name}:")
        case False:
            pass
        case None if any(k.startswith(f"{self.name}:") for k in config):
            config = mapping_select(config, f"{self.name}:")
        case None:
            pass

    _kwargs: dict[str, Any] = {}

    # Configure all the branches if exists
    if len(self.nodes) > 0:
        nodes = tuple(
            node.configure(
                config,
                prefixed_name=True,
                transform_context=transform_context,
                params=params,
            )
            for node in self.nodes
        )
        _kwargs["nodes"] = nodes

    this_config = {
        hp: v
        for hp, v in config.items()
        if (
            ":" not in hp
            and not any(hp.startswith(f"{node.name}") for node in self.nodes)
        )
    }
    if self.config is not None:
        this_config = {**self.config, **this_config}

    this_config = dict(self._fufill_param_requests(this_config, params=params))

    if self.config_transform is not None:
        this_config = dict(self.config_transform(this_config, transform_context))

    if len(this_config) > 0:
        _kwargs["config"] = dict(this_config)

    return self.mutate(**_kwargs)

copy

copy() -> Self

Copy this node, removing all links in the process.

Source code in src/amltk/pipeline/node.py

def copy(self) -> Self:
    """Copy this node, removing all links in the process."""
    return self.mutate()

display

display(*, full: bool = False) -> RenderableType

Display this node.

PARAMETER	DESCRIPTION
full	Whether to display the full node or just a summary TYPE: bool DEFAULT: False

Source code in src/amltk/pipeline/node.py

def display(self, *, full: bool = False) -> RenderableType:
    """Display this node.

    Args:
        full: Whether to display the full node or just a summary
    """
    if not full:
        return self.__rich__()

    from rich.console import Group as RichGroup

    return RichGroup(*self._rich_iter())

factorize

factorize(
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None
) -> Iterator[Self]

Please see factorize().

Source code in src/amltk/pipeline/node.py

def factorize(
    self,
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None,
) -> Iterator[Self]:
    """Please see [`factorize()`][amltk.pipeline.ops.factorize]."""  # noqa: D402
    from amltk.pipeline.ops import factorize

    yield from factorize(
        self,
        min_depth=min_depth,
        max_depth=max_depth,
        current_depth=current_depth,
        factor_by=factor_by,
        assign_child=assign_child,
    )

fidelity_space

fidelity_space() -> dict[str, Any]

Get the fidelities for this node and any connected nodes.

Source code in src/amltk/pipeline/node.py

def fidelity_space(self) -> dict[str, Any]:
    """Get the fidelities for this node and any connected nodes."""
    fids = {}
    for node in self.nodes:
        fids.update(prefix_keys(node.fidelity_space(), f"{self.name}:"))

    return fids

find

find(
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None

Find a node in that's nested deeper from this node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]
default	The value to return if the node is not found. Defaults to None TYPE: T | None DEFAULT: None

RETURNS	DESCRIPTION
Node | T | None	The node if found, otherwise the default value. Defaults to None

Source code in src/amltk/pipeline/node.py

def find(
    self,
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None:
    """Find a node in that's nested deeper from this node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node
        default: The value to return if the node is not found. Defaults to None

    Returns:
        The node if found, otherwise the default value. Defaults to None
    """
    itr = self.iter()
    match key:
        case Node():
            return first_true(itr, default, lambda node: node == key)
        case str():
            return first_true(itr, default, lambda node: node.name == key)
        case _:
            return first_true(itr, default, key)  # type: ignore

iter

iter() -> Iterator[Node]

Iterate the the nodes, including this node.

YIELDS	DESCRIPTION
Node	The nodes connected to this node

Source code in src/amltk/pipeline/node.py

def iter(self) -> Iterator[Node]:
    """Iterate the the nodes, including this node.

    Yields:
        The nodes connected to this node
    """
    yield self
    for node in self.nodes:
        yield from node.iter()

linearized_fidelity

linearized_fidelity(
    value: float,
) -> dict[str, int | float | Any]

Get the liniearized fidelities for this node and any connected nodes.

PARAMETER	DESCRIPTION
value	The value to linearize. Must be between [0, 1] TYPE: float

Return

dictionary from key to it's linearized fidelity.

Source code in src/amltk/pipeline/node.py

def linearized_fidelity(self, value: float) -> dict[str, int | float | Any]:
    """Get the liniearized fidelities for this node and any connected nodes.

    Args:
        value: The value to linearize. Must be between [0, 1]

    Return:
        dictionary from key to it's linearized fidelity.
    """
    assert 1.0 <= value <= 100.0, f"{value=} not in [1.0, 100.0]"  # noqa: PLR2004
    d = {}
    for node in self.nodes:
        node_fids = prefix_keys(
            node.linearized_fidelity(value),
            f"{self.name}:",
        )
        d.update(node_fids)

    if self.fidelities is None:
        return d

    for f_name, f_range in self.fidelities.items():
        match f_range:
            case (int() | float(), int() | float()):
                low, high = f_range
                fid = low + (high - low) * value
                fid = low + (high - low) * (value - 1) / 100
                fid = fid if isinstance(low, float) else round(fid)
                d[f_name] = fid
            case _:
                raise ValueError(
                    f"Invalid fidelities to linearize {f_range} for {f_name}"
                    f" in {self}. Only supports ranges of the form (low, high)",
                )

    return prefix_keys(d, f"{self.name}:")

mutate

mutate(**kwargs: Any) -> Self

Mutate the node with the given keyword arguments.

PARAMETER	DESCRIPTION
**kwargs	The keyword arguments to mutate TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	Self The mutated node

Source code in src/amltk/pipeline/node.py

def mutate(self, **kwargs: Any) -> Self:
    """Mutate the node with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to mutate

    Returns:
        Self
            The mutated node
    """
    _args = ()
    _kwargs = {**self.__dict__, **kwargs}

    # If there's nodes in kwargs, we have to check if it's
    # a positional or keyword argument and handle accordingly.
    if (nodes := _kwargs.pop("nodes", None)) is not None:
        match self._NODES_INIT:
            case "args":
                _args = nodes
            case "kwargs":
                _kwargs["nodes"] = nodes
            case None if len(nodes) == 0:
                pass  # Just ignore it, it's popped out
            case None:
                raise ValueError(
                    "Cannot mutate nodes when __init__ does not accept nodes",
                )

    # If there's a config in kwargs, we have to check if it's actually got values
    config = _kwargs.pop("config", None)
    if config is not None and len(config) > 0:
        _kwargs["config"] = config

    # Lastly, we remove anything that can't be passed to kwargs of the
    # subclasses __init__
    _available_kwargs = inspect.signature(self.__init__).parameters.keys()  # type: ignore
    for k in list(_kwargs.keys()):
        if k not in _available_kwargs:
            _kwargs.pop(k)

    return self.__class__(*_args, **_kwargs)

path_to

path_to(
    key: str | Node | Callable[[Node], bool]
) -> list[Node] | None

Find a path to the given node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]

RETURNS	DESCRIPTION
list[Node] | None	The path to the node if found, else None

Source code in src/amltk/pipeline/node.py

def path_to(self, key: str | Node | Callable[[Node], bool]) -> list[Node] | None:
    """Find a path to the given node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node

    Returns:
        The path to the node if found, else None
    """
    # We found our target, just return now

    match key:
        case Node():
            pred = lambda node: node == key
        case str():
            pred = lambda node: node.name == key
        case _:
            pred = key

    for path, node in self.walk():
        if pred(node):
            return path

    return None

search_space

search_space(
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: args,
    **parser_kwargs: kwargs
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace

Get the search space for this node.

PARAMETER	DESCRIPTION
parser	The parser to use. This can be a function that takes in the node and returns the search space or a string that is one of: "configspace": Build a ConfigSpace.ConfigurationSpace out of this node. "optuna": Build a dict of hyperparameters that Optuna can use in its ask and tell methods TYPE: Callable[Concatenate[Node, P], ParserOutput] | Literal['configspace', 'optuna']
parser_args	The positional arguments to pass to the parser TYPE: args DEFAULT: ()
parser_kwargs	The keyword arguments to pass to the parser TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
ParserOutput | ConfigurationSpace | OptunaSearchSpace	The search space

Source code in src/amltk/pipeline/node.py

def search_space(
    self,
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: P.args,
    **parser_kwargs: P.kwargs,
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace:
    """Get the search space for this node.

    Args:
        parser: The parser to use. This can be a function that takes in
            the node and returns the search space or a string that is one of:

            * `#!python "configspace"`: Build a
                [`ConfigSpace.ConfigurationSpace`](https://automl.github.io/ConfigSpace/master/)
                out of this node.
            * `#!python "optuna"`: Build a dict of hyperparameters that Optuna can
                use in its [ask and tell methods](https://optuna.readthedocs.io/en/stable/tutorial/20_recipes/009_ask_and_tell.html#define-and-run)

        parser_args: The positional arguments to pass to the parser
        parser_kwargs: The keyword arguments to pass to the parser

    Returns:
        The search space
    """
    match parser:
        case "configspace":
            from amltk.pipeline.parsers.configspace import parser as cs_parser

            return cs_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case "optuna":
            from amltk.pipeline.parsers.optuna import parser as optuna_parser

            return optuna_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case str():  # type: ignore
            raise ValueError(
                f"Invalid str for parser {parser}. "
                "Please use 'configspace' or 'optuna' or pass in your own"
                " parser function",
            )
        case _:
            return parser(self, *parser_args, **parser_kwargs)

walk

walk(
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]

Walk the nodes in this chain.

PARAMETER	DESCRIPTION
path	The current path to this node TYPE: Sequence[Node] | None DEFAULT: None

YIELDS	DESCRIPTION
list[Node]	The parents of the node and the node itself

Source code in src/amltk/pipeline/node.py

def walk(
    self,
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]:
    """Walk the nodes in this chain.

    Args:
        path: The current path to this node

    Yields:
        The parents of the node and the node itself
    """
    path = list(path) if path is not None else []
    yield path, self

    for node in self.nodes:
        yield from node.walk(path=[*path, self])

Sequential dataclass

Sequential(
    *nodes: Node | NodeLike,
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: (
        Callable[[Config, Any], Config] | None
    ) = None,
    meta: Mapping[str, Any] | None = None
)

Bases: Node[Item, Space]

A Sequential set of operations in a pipeline.

This indicates the different children in .nodes should act one after another, feeding the output of one into the next.

from amltk.pipeline import Component, Sequential
from sklearn.decomposition import PCA
from sklearn.ensemble import RandomForestClassifier

pipeline = Sequential(
    PCA(n_components=3),
    Component(RandomForestClassifier, space={"n_estimators": (10, 100)}),
    name="my_pipeline"
)

╭─ Sequential(my_pipeline) ───────────────────╮
│ ╭─ Fixed(PCA) ─────────────╮                │
│ │ item PCA(n_components=3) │                │
│ ╰──────────────────────────╯                │
│                      ↓                      │
│ ╭─ Component(RandomForestClassifier) ─────╮ │
│ │ item  class RandomForestClassifier(...) │ │
│ │ space {'n_estimators': (10, 100)}       │ │
│ ╰─────────────────────────────────────────╯ │
╰─────────────────────────────────────────────╯

See Also

• Node

PARAMETER	DESCRIPTION
nodes	The nodes that this node leads to. In the case of a Sequential, the order here matters and it signifies that data should first be passed through the first node, then the second, etc. TYPE: Node | NodeLike DEFAULT: ()
item	The item attached to this node (if any). TYPE: Item | Callable[[Item], Item] | None DEFAULT: None
name	The name of the node. If not specified, the name will be randomly generated. TYPE: str | None DEFAULT: None
config	The configuration for this node. TYPE: Config | None DEFAULT: None
space	The search space for this node. This will be used when search_space() is called. TYPE: Space | None DEFAULT: None
fidelities	The fidelities for this node. TYPE: Mapping[str, Any] | None DEFAULT: None
config_transform	A function that transforms the config= parameter during configure(config) before return the new configured node. Useful for times where you need to combine multiple parameters into one. TYPE: Callable[[Config, Any], Config] | None DEFAULT: None
meta	Any meta information about this node. TYPE: Mapping[str, Any] | None DEFAULT: None

Source code in src/amltk/pipeline/components.py

def __init__(
    self,
    *nodes: Node | NodeLike,
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: Callable[[Config, Any], Config] | None = None,
    meta: Mapping[str, Any] | None = None,
):
    """Initialize a sequential node.

    Args:
        nodes: The nodes that this node leads to. In the case of a `Sequential`,
            the order here matters and it signifies that data should first
            be passed through the first node, then the second, etc.
        item: The item attached to this node (if any).
        name: The name of the node. If not specified, the name will be
            randomly generated.
        config: The configuration for this node.
        space: The search space for this node. This will be used when
            [`search_space()`][amltk.pipeline.node.Node.search_space] is called.
        fidelities: The fidelities for this node.
        config_transform: A function that transforms the `config=` parameter
            during [`configure(config)`][amltk.pipeline.node.Node.configure]
            before return the new configured node. Useful for times where
            you need to combine multiple parameters into one.
        meta: Any meta information about this node.
    """
    _nodes = tuple(as_node(n) for n in nodes)

    # Perhaps we need to do a deeper check on this...
    if not all_unique(_nodes, key=lambda node: node.name):
        raise DuplicateNamesError(self)

    if name is None:
        name = f"Seq-{randuid(8)}"

    super().__init__(
        *_nodes,
        name=name,
        item=item,
        config=config,
        space=space,
        fidelities=fidelities,
        config_transform=config_transform,
        meta=meta,
    )

config class-attribute instance-attribute

config: Config | None = field(hash=False)

The configuration for this node

config_transform class-attribute instance-attribute

config_transform: Callable[[Config, Any], Config] | None = (
    field(hash=False)
)

A function that transforms the configuration of this node

fidelities class-attribute instance-attribute

fidelities: Mapping[str, Any] | None = field(hash=False)

The fidelities for this node

item class-attribute instance-attribute

item: Callable[..., Item] | Item | None = field(hash=False)

The item attached to this node

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = field(hash=False)

Any meta information about this node

name class-attribute instance-attribute

name: str = field(hash=True)

Name of the node

nodes instance-attribute

nodes: tuple[Node, ...]

The nodes ordered in series.

space class-attribute instance-attribute

space: Space | None = field(hash=False)

The search space for this node

tail property

tail: Node

The last step in the pipeline.

__getitem__

__getitem__(key: str) -> Node

Get the first from .nodes with key.

Source code in src/amltk/pipeline/node.py

def __getitem__(self, key: str) -> Node:
    """Get the first from [`.nodes`][amltk.pipeline.node.Node.nodes] with `key`."""
    found = first_true(
        self.nodes,
        None,
        lambda node: node.name == key,
    )
    if found is None:
        raise KeyError(
            f"Could not find node with name {key} in '{self.name}'."
            f" Available nodes are: {', '.join(node.name for node in self.nodes)}",
        )

    return found

__len__

__len__() -> int

Get the number of nodes in the pipeline.

Source code in src/amltk/pipeline/components.py

def __len__(self) -> int:
    """Get the number of nodes in the pipeline."""
    return len(self.nodes)

build

build(
    builder: (
        Callable[Concatenate[Node, P], BuilderOutput]
        | Literal["sklearn"]
    ),
    *builder_args: args,
    **builder_kwargs: kwargs
) -> BuilderOutput | Pipeline

Build a concrete object out of this node.

PARAMETER	DESCRIPTION
builder	The builder to use. This can be a function that takes in the node and returns the object or a string that is one of: "sklearn": Build a sklearn.pipeline.Pipeline out of this node. TYPE: Callable[Concatenate[Node, P], BuilderOutput] | Literal['sklearn']
builder_args	The positional arguments to pass to the builder TYPE: args DEFAULT: ()
builder_kwargs	The keyword arguments to pass to the builder TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BuilderOutput | Pipeline	The built object

Source code in src/amltk/pipeline/node.py

def build(
    self,
    builder: Callable[Concatenate[Node, P], BuilderOutput] | Literal["sklearn"],
    *builder_args: P.args,
    **builder_kwargs: P.kwargs,
) -> BuilderOutput | SklearnPipeline:
    """Build a concrete object out of this node.

    Args:
        builder: The builder to use. This can be a function that takes in
            the node and returns the object or a string that is one of:

            * `#!python "sklearn"`: Build a
                [`sklearn.pipeline.Pipeline`][sklearn.pipeline.Pipeline]
                out of this node.

        builder_args: The positional arguments to pass to the builder
        builder_kwargs: The keyword arguments to pass to the builder

    Returns:
        The built object
    """
    match builder:
        case "sklearn":
            from amltk.pipeline.builders.sklearn import build as _build

            return _build(self, *builder_args, **builder_kwargs)  # type: ignore
        case _:
            return builder(self, *builder_args, **builder_kwargs)

configure

configure(
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None
) -> Self

Configure this node and anything following it with the given config.

PARAMETER	DESCRIPTION
config	The configuration to apply TYPE: Config
prefixed_name	Whether items in the config are prefixed by the names of the nodes. * If None, the default, then prefixed_name will be assumed to be True if this node has a next node or if the config has keys that begin with this nodes name. * If True, then the config will be searched for items prefixed by the name of the node (and subsequent chained nodes). * If False, then the config will be searched for items without the prefix, i.e. the config keys are exactly those matching this nodes search space. TYPE: bool | None DEFAULT: None
transform_context	Any context to give to config_transform= of individual nodes. TYPE: Any | None DEFAULT: None
params	The params to match any requests when configuring this node. These will match against any ParamRequests in the config and will be used to fill in any missing values. TYPE: Mapping[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Self	The configured node

Source code in src/amltk/pipeline/node.py

def configure(
    self,
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None,
) -> Self:
    """Configure this node and anything following it with the given config.

    Args:
        config: The configuration to apply
        prefixed_name: Whether items in the config are prefixed by the names
            of the nodes.
            * If `None`, the default, then `prefixed_name` will be assumed to
                be `True` if this node has a next node or if the config has
                keys that begin with this nodes name.
            * If `True`, then the config will be searched for items prefixed
                by the name of the node (and subsequent chained nodes).
            * If `False`, then the config will be searched for items without
                the prefix, i.e. the config keys are exactly those matching
                this nodes search space.
        transform_context: Any context to give to `config_transform=` of individual
            nodes.
        params: The params to match any requests when configuring this node.
            These will match against any ParamRequests in the config and will
            be used to fill in any missing values.

    Returns:
        The configured node
    """
    # Get the config for this node
    match prefixed_name:
        case True:
            config = mapping_select(config, f"{self.name}:")
        case False:
            pass
        case None if any(k.startswith(f"{self.name}:") for k in config):
            config = mapping_select(config, f"{self.name}:")
        case None:
            pass

    _kwargs: dict[str, Any] = {}

    # Configure all the branches if exists
    if len(self.nodes) > 0:
        nodes = tuple(
            node.configure(
                config,
                prefixed_name=True,
                transform_context=transform_context,
                params=params,
            )
            for node in self.nodes
        )
        _kwargs["nodes"] = nodes

    this_config = {
        hp: v
        for hp, v in config.items()
        if (
            ":" not in hp
            and not any(hp.startswith(f"{node.name}") for node in self.nodes)
        )
    }
    if self.config is not None:
        this_config = {**self.config, **this_config}

    this_config = dict(self._fufill_param_requests(this_config, params=params))

    if self.config_transform is not None:
        this_config = dict(self.config_transform(this_config, transform_context))

    if len(this_config) > 0:
        _kwargs["config"] = dict(this_config)

    return self.mutate(**_kwargs)

copy

copy() -> Self

Copy this node, removing all links in the process.

Source code in src/amltk/pipeline/node.py

def copy(self) -> Self:
    """Copy this node, removing all links in the process."""
    return self.mutate()

display

display(*, full: bool = False) -> RenderableType

Display this node.

PARAMETER	DESCRIPTION
full	Whether to display the full node or just a summary TYPE: bool DEFAULT: False

Source code in src/amltk/pipeline/node.py

def display(self, *, full: bool = False) -> RenderableType:
    """Display this node.

    Args:
        full: Whether to display the full node or just a summary
    """
    if not full:
        return self.__rich__()

    from rich.console import Group as RichGroup

    return RichGroup(*self._rich_iter())

factorize

factorize(
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None
) -> Iterator[Self]

Please see factorize().

Source code in src/amltk/pipeline/node.py

def factorize(
    self,
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None,
) -> Iterator[Self]:
    """Please see [`factorize()`][amltk.pipeline.ops.factorize]."""  # noqa: D402
    from amltk.pipeline.ops import factorize

    yield from factorize(
        self,
        min_depth=min_depth,
        max_depth=max_depth,
        current_depth=current_depth,
        factor_by=factor_by,
        assign_child=assign_child,
    )

fidelity_space

fidelity_space() -> dict[str, Any]

Get the fidelities for this node and any connected nodes.

Source code in src/amltk/pipeline/node.py

def fidelity_space(self) -> dict[str, Any]:
    """Get the fidelities for this node and any connected nodes."""
    fids = {}
    for node in self.nodes:
        fids.update(prefix_keys(node.fidelity_space(), f"{self.name}:"))

    return fids

find

find(
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None

Find a node in that's nested deeper from this node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]
default	The value to return if the node is not found. Defaults to None TYPE: T | None DEFAULT: None

RETURNS	DESCRIPTION
Node | T | None	The node if found, otherwise the default value. Defaults to None

Source code in src/amltk/pipeline/node.py

def find(
    self,
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None:
    """Find a node in that's nested deeper from this node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node
        default: The value to return if the node is not found. Defaults to None

    Returns:
        The node if found, otherwise the default value. Defaults to None
    """
    itr = self.iter()
    match key:
        case Node():
            return first_true(itr, default, lambda node: node == key)
        case str():
            return first_true(itr, default, lambda node: node.name == key)
        case _:
            return first_true(itr, default, key)  # type: ignore

iter

iter() -> Iterator[Node]

Iterate the the nodes, including this node.

YIELDS	DESCRIPTION
Node	The nodes connected to this node

Source code in src/amltk/pipeline/node.py

def iter(self) -> Iterator[Node]:
    """Iterate the the nodes, including this node.

    Yields:
        The nodes connected to this node
    """
    yield self
    for node in self.nodes:
        yield from node.iter()

linearized_fidelity

linearized_fidelity(
    value: float,
) -> dict[str, int | float | Any]

Get the liniearized fidelities for this node and any connected nodes.

PARAMETER	DESCRIPTION
value	The value to linearize. Must be between [0, 1] TYPE: float

Return

dictionary from key to it's linearized fidelity.

Source code in src/amltk/pipeline/node.py

def linearized_fidelity(self, value: float) -> dict[str, int | float | Any]:
    """Get the liniearized fidelities for this node and any connected nodes.

    Args:
        value: The value to linearize. Must be between [0, 1]

    Return:
        dictionary from key to it's linearized fidelity.
    """
    assert 1.0 <= value <= 100.0, f"{value=} not in [1.0, 100.0]"  # noqa: PLR2004
    d = {}
    for node in self.nodes:
        node_fids = prefix_keys(
            node.linearized_fidelity(value),
            f"{self.name}:",
        )
        d.update(node_fids)

    if self.fidelities is None:
        return d

    for f_name, f_range in self.fidelities.items():
        match f_range:
            case (int() | float(), int() | float()):
                low, high = f_range
                fid = low + (high - low) * value
                fid = low + (high - low) * (value - 1) / 100
                fid = fid if isinstance(low, float) else round(fid)
                d[f_name] = fid
            case _:
                raise ValueError(
                    f"Invalid fidelities to linearize {f_range} for {f_name}"
                    f" in {self}. Only supports ranges of the form (low, high)",
                )

    return prefix_keys(d, f"{self.name}:")

mutate

mutate(**kwargs: Any) -> Self

Mutate the node with the given keyword arguments.

PARAMETER	DESCRIPTION
**kwargs	The keyword arguments to mutate TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	Self The mutated node

Source code in src/amltk/pipeline/node.py

def mutate(self, **kwargs: Any) -> Self:
    """Mutate the node with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to mutate

    Returns:
        Self
            The mutated node
    """
    _args = ()
    _kwargs = {**self.__dict__, **kwargs}

    # If there's nodes in kwargs, we have to check if it's
    # a positional or keyword argument and handle accordingly.
    if (nodes := _kwargs.pop("nodes", None)) is not None:
        match self._NODES_INIT:
            case "args":
                _args = nodes
            case "kwargs":
                _kwargs["nodes"] = nodes
            case None if len(nodes) == 0:
                pass  # Just ignore it, it's popped out
            case None:
                raise ValueError(
                    "Cannot mutate nodes when __init__ does not accept nodes",
                )

    # If there's a config in kwargs, we have to check if it's actually got values
    config = _kwargs.pop("config", None)
    if config is not None and len(config) > 0:
        _kwargs["config"] = config

    # Lastly, we remove anything that can't be passed to kwargs of the
    # subclasses __init__
    _available_kwargs = inspect.signature(self.__init__).parameters.keys()  # type: ignore
    for k in list(_kwargs.keys()):
        if k not in _available_kwargs:
            _kwargs.pop(k)

    return self.__class__(*_args, **_kwargs)

path_to

path_to(
    key: str | Node | Callable[[Node], bool]
) -> list[Node] | None

Find a path to the given node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]

RETURNS	DESCRIPTION
list[Node] | None	The path to the node if found, else None

Source code in src/amltk/pipeline/node.py

def path_to(self, key: str | Node | Callable[[Node], bool]) -> list[Node] | None:
    """Find a path to the given node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node

    Returns:
        The path to the node if found, else None
    """
    # We found our target, just return now

    match key:
        case Node():
            pred = lambda node: node == key
        case str():
            pred = lambda node: node.name == key
        case _:
            pred = key

    for path, node in self.walk():
        if pred(node):
            return path

    return None

search_space

search_space(
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: args,
    **parser_kwargs: kwargs
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace

Get the search space for this node.

PARAMETER	DESCRIPTION
parser	The parser to use. This can be a function that takes in the node and returns the search space or a string that is one of: "configspace": Build a ConfigSpace.ConfigurationSpace out of this node. "optuna": Build a dict of hyperparameters that Optuna can use in its ask and tell methods TYPE: Callable[Concatenate[Node, P], ParserOutput] | Literal['configspace', 'optuna']
parser_args	The positional arguments to pass to the parser TYPE: args DEFAULT: ()
parser_kwargs	The keyword arguments to pass to the parser TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
ParserOutput | ConfigurationSpace | OptunaSearchSpace	The search space

Source code in src/amltk/pipeline/node.py

def search_space(
    self,
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: P.args,
    **parser_kwargs: P.kwargs,
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace:
    """Get the search space for this node.

    Args:
        parser: The parser to use. This can be a function that takes in
            the node and returns the search space or a string that is one of:

            * `#!python "configspace"`: Build a
                [`ConfigSpace.ConfigurationSpace`](https://automl.github.io/ConfigSpace/master/)
                out of this node.
            * `#!python "optuna"`: Build a dict of hyperparameters that Optuna can
                use in its [ask and tell methods](https://optuna.readthedocs.io/en/stable/tutorial/20_recipes/009_ask_and_tell.html#define-and-run)

        parser_args: The positional arguments to pass to the parser
        parser_kwargs: The keyword arguments to pass to the parser

    Returns:
        The search space
    """
    match parser:
        case "configspace":
            from amltk.pipeline.parsers.configspace import parser as cs_parser

            return cs_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case "optuna":
            from amltk.pipeline.parsers.optuna import parser as optuna_parser

            return optuna_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case str():  # type: ignore
            raise ValueError(
                f"Invalid str for parser {parser}. "
                "Please use 'configspace' or 'optuna' or pass in your own"
                " parser function",
            )
        case _:
            return parser(self, *parser_args, **parser_kwargs)

walk

walk(
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]

Walk the nodes in this chain.

PARAMETER	DESCRIPTION
path	The current path to this node TYPE: Sequence[Node] | None DEFAULT: None

YIELDS	DESCRIPTION
list[Node]	The parents of the node and the node itself

Source code in src/amltk/pipeline/components.py

@override
def walk(
    self,
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]:
    """Walk the nodes in this chain.

    Args:
        path: The current path to this node

    Yields:
        The parents of the node and the node itself
    """
    path = list(path) if path is not None else []
    yield path, self

    path = [*path, self]
    for node in self.nodes:
        yield from node.walk(path=path)

        # Append the previous node so that the next node in the sequence is
        # lead to from the previous node
        path = [*path, node]

Split dataclass

Split(
    *nodes: Node | NodeLike | dict[str, Node | NodeLike],
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: (
        Callable[[Config, Any], Config] | None
    ) = None,
    meta: Mapping[str, Any] | None = None
)

Bases: Node[Item, Space]

A Split of data in a pipeline.

This indicates the different children in .nodes should act in parallel but on different subsets of data.

from amltk.pipeline import Component, Split
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import OneHotEncoder
from sklearn.compose import make_column_selector

categorical_pipeline = [
    SimpleImputer(strategy="constant", fill_value="missing"),
    OneHotEncoder(drop="first"),
]
numerical_pipeline = Component(SimpleImputer, space={"strategy": ["mean", "median"]})

preprocessor = Split(
    {
        "categories": categorical_pipeline,
        "numerical": numerical_pipeline,
    },
    config={
        "categories": make_column_selector(dtype_include="category"),
        "numerical": make_column_selector(dtype_exclude="category"),
    },
    name="my_split"
)

╭─ Split(my_split) ────────────────────────────────────────────────────────────╮
│ config {                                                                     │
│            'categories':                                                     │
│        <sklearn.compose._column_transformer.make_column_selector object at   │
│        0x7f95fa3631f0>,                                                      │
│            'numerical':                                                      │
│        <sklearn.compose._column_transformer.make_column_selector object at   │
│        0x7f95fa360730>                                                       │
│        }                                                                     │
│ ╭─ Sequential(categories) ──────────╮ ╭─ Sequential(numerical) ────────────╮ │
│ │ ╭─ Fixed(SimpleImputer) ────────╮ │ │ ╭─ Component(SimpleImputer) ─────╮ │ │
│ │ │ item SimpleImputer(fill_valu… │ │ │ │ item  class SimpleImputer(...) │ │ │
│ │ │      strategy='constant')     │ │ │ │ space {                        │ │ │
│ │ ╰───────────────────────────────╯ │ │ │           'strategy': [        │ │ │
│ │                 ↓                 │ │ │               'mean',          │ │ │
│ │ ╭─ Fixed(OneHotEncoder) ────────╮ │ │ │               'median'         │ │ │
│ │ │ item OneHotEncoder(drop='fir… │ │ │ │           ]                    │ │ │
│ │ ╰───────────────────────────────╯ │ │ │       }                        │ │ │
│ ╰───────────────────────────────────╯ │ ╰────────────────────────────────╯ │ │
│                                       ╰────────────────────────────────────╯ │
╰──────────────────────────────────────────────────────────────────────────────╯

See Also

• Node

PARAMETER	DESCRIPTION
nodes	The nodes that this node leads to. You may also provide a dictionary where the keys are the names of the nodes and the values are the nodes or list of nodes themselves. TYPE: Node | NodeLike | dict[str, Node | NodeLike] DEFAULT: ()
item	The item attached to this node. The object created by item should be capable of figuring out how to deal with its child nodes. TYPE: Item | Callable[[Item], Item] | None DEFAULT: None
name	The name of the node. If not specified, the name will be generated from the item. TYPE: str | None DEFAULT: None
config	The configuration for this split. TYPE: Config | None DEFAULT: None
space	The search space for this node. This will be used when search_space() is called. TYPE: Space | None DEFAULT: None
fidelities	The fidelities for this node. TYPE: Mapping[str, Any] | None DEFAULT: None
config_transform	A function that transforms the config= parameter during configure(config) before return the new configured node. Useful for times where you need to combine multiple parameters into one. TYPE: Callable[[Config, Any], Config] | None DEFAULT: None
meta	Any meta information about this node. TYPE: Mapping[str, Any] | None DEFAULT: None

Source code in src/amltk/pipeline/components.py

def __init__(
    self,
    *nodes: Node | NodeLike | dict[str, Node | NodeLike],
    name: str | None = None,
    item: Item | Callable[[Item], Item] | None = None,
    config: Config | None = None,
    space: Space | None = None,
    fidelities: Mapping[str, Any] | None = None,
    config_transform: Callable[[Config, Any], Config] | None = None,
    meta: Mapping[str, Any] | None = None,
):
    """Initialize a split node.

    Args:
        nodes: The nodes that this node leads to. You may also provide
            a dictionary where the keys are the names of the nodes and
            the values are the nodes or list of nodes themselves.
        item: The item attached to this node. The object created by `item`
            should be capable of figuring out how to deal with its child nodes.
        name: The name of the node. If not specified, the name will be
            generated from the item.
        config: The configuration for this split.
        space: The search space for this node. This will be used when
            [`search_space()`][amltk.pipeline.node.Node.search_space] is called.
        fidelities: The fidelities for this node.
        config_transform: A function that transforms the `config=` parameter
            during [`configure(config)`][amltk.pipeline.node.Node.configure]
            before return the new configured node. Useful for times where
            you need to combine multiple parameters into one.
        meta: Any meta information about this node.
    """
    if any(isinstance(n, dict) for n in nodes):
        if len(nodes) > 1:
            raise ValueError(
                "Can't handle multiple nodes with a dictionary as a node.\n"
                f"{nodes=}",
            )
        _node = nodes[0]
        assert isinstance(_node, dict)

        def _construct(key: str, value: Node | NodeLike) -> Node:
            match value:
                case list():
                    return Sequential(*value, name=key)
                case set() | tuple():
                    return as_node(value, name=key)
                case _:
                    return Sequential(value, name=key)

        _nodes = tuple(_construct(key, value) for key, value in _node.items())
    else:
        _nodes = tuple(as_node(n) for n in nodes)

    if not all_unique(_nodes, key=lambda node: node.name):
        raise ValueError(
            f"Can't handle nodes they do not all contain unique names, {nodes=}."
            "\nAll nodes must have a unique name. Please provide a `name=` to them",
        )

    if name is None:
        name = f"Split-{randuid(8)}"

    super().__init__(
        *_nodes,
        name=name,
        item=item,
        config=config,
        space=space,
        fidelities=fidelities,
        config_transform=config_transform,
        meta=meta,
    )

config class-attribute instance-attribute

config: Config | None = field(hash=False)

The configuration for this node

config_transform class-attribute instance-attribute

config_transform: Callable[[Config, Any], Config] | None = (
    field(hash=False)
)

A function that transforms the configuration of this node

fidelities class-attribute instance-attribute

fidelities: Mapping[str, Any] | None = field(hash=False)

The fidelities for this node

item class-attribute instance-attribute

item: Callable[..., Item] | Item | None = field(hash=False)

The item attached to this node

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = field(hash=False)

Any meta information about this node

name class-attribute instance-attribute

name: str = field(hash=True)

Name of the node

nodes class-attribute instance-attribute

nodes: tuple[Node, ...] = field(hash=True)

The nodes that this node leads to.

space class-attribute instance-attribute

space: Space | None = field(hash=False)

The search space for this node

__getitem__

__getitem__(key: str) -> Node

Get the first from .nodes with key.

Source code in src/amltk/pipeline/node.py

def __getitem__(self, key: str) -> Node:
    """Get the first from [`.nodes`][amltk.pipeline.node.Node.nodes] with `key`."""
    found = first_true(
        self.nodes,
        None,
        lambda node: node.name == key,
    )
    if found is None:
        raise KeyError(
            f"Could not find node with name {key} in '{self.name}'."
            f" Available nodes are: {', '.join(node.name for node in self.nodes)}",
        )

    return found

build

build(
    builder: (
        Callable[Concatenate[Node, P], BuilderOutput]
        | Literal["sklearn"]
    ),
    *builder_args: args,
    **builder_kwargs: kwargs
) -> BuilderOutput | Pipeline

Build a concrete object out of this node.

PARAMETER	DESCRIPTION
builder	The builder to use. This can be a function that takes in the node and returns the object or a string that is one of: "sklearn": Build a sklearn.pipeline.Pipeline out of this node. TYPE: Callable[Concatenate[Node, P], BuilderOutput] | Literal['sklearn']
builder_args	The positional arguments to pass to the builder TYPE: args DEFAULT: ()
builder_kwargs	The keyword arguments to pass to the builder TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BuilderOutput | Pipeline	The built object

Source code in src/amltk/pipeline/node.py

def build(
    self,
    builder: Callable[Concatenate[Node, P], BuilderOutput] | Literal["sklearn"],
    *builder_args: P.args,
    **builder_kwargs: P.kwargs,
) -> BuilderOutput | SklearnPipeline:
    """Build a concrete object out of this node.

    Args:
        builder: The builder to use. This can be a function that takes in
            the node and returns the object or a string that is one of:

            * `#!python "sklearn"`: Build a
                [`sklearn.pipeline.Pipeline`][sklearn.pipeline.Pipeline]
                out of this node.

        builder_args: The positional arguments to pass to the builder
        builder_kwargs: The keyword arguments to pass to the builder

    Returns:
        The built object
    """
    match builder:
        case "sklearn":
            from amltk.pipeline.builders.sklearn import build as _build

            return _build(self, *builder_args, **builder_kwargs)  # type: ignore
        case _:
            return builder(self, *builder_args, **builder_kwargs)

configure

configure(
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None
) -> Self

Configure this node and anything following it with the given config.

PARAMETER	DESCRIPTION
config	The configuration to apply TYPE: Config
prefixed_name	Whether items in the config are prefixed by the names of the nodes. * If None, the default, then prefixed_name will be assumed to be True if this node has a next node or if the config has keys that begin with this nodes name. * If True, then the config will be searched for items prefixed by the name of the node (and subsequent chained nodes). * If False, then the config will be searched for items without the prefix, i.e. the config keys are exactly those matching this nodes search space. TYPE: bool | None DEFAULT: None
transform_context	Any context to give to config_transform= of individual nodes. TYPE: Any | None DEFAULT: None
params	The params to match any requests when configuring this node. These will match against any ParamRequests in the config and will be used to fill in any missing values. TYPE: Mapping[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
Self	The configured node

Source code in src/amltk/pipeline/node.py

def configure(
    self,
    config: Config,
    *,
    prefixed_name: bool | None = None,
    transform_context: Any | None = None,
    params: Mapping[str, Any] | None = None,
) -> Self:
    """Configure this node and anything following it with the given config.

    Args:
        config: The configuration to apply
        prefixed_name: Whether items in the config are prefixed by the names
            of the nodes.
            * If `None`, the default, then `prefixed_name` will be assumed to
                be `True` if this node has a next node or if the config has
                keys that begin with this nodes name.
            * If `True`, then the config will be searched for items prefixed
                by the name of the node (and subsequent chained nodes).
            * If `False`, then the config will be searched for items without
                the prefix, i.e. the config keys are exactly those matching
                this nodes search space.
        transform_context: Any context to give to `config_transform=` of individual
            nodes.
        params: The params to match any requests when configuring this node.
            These will match against any ParamRequests in the config and will
            be used to fill in any missing values.

    Returns:
        The configured node
    """
    # Get the config for this node
    match prefixed_name:
        case True:
            config = mapping_select(config, f"{self.name}:")
        case False:
            pass
        case None if any(k.startswith(f"{self.name}:") for k in config):
            config = mapping_select(config, f"{self.name}:")
        case None:
            pass

    _kwargs: dict[str, Any] = {}

    # Configure all the branches if exists
    if len(self.nodes) > 0:
        nodes = tuple(
            node.configure(
                config,
                prefixed_name=True,
                transform_context=transform_context,
                params=params,
            )
            for node in self.nodes
        )
        _kwargs["nodes"] = nodes

    this_config = {
        hp: v
        for hp, v in config.items()
        if (
            ":" not in hp
            and not any(hp.startswith(f"{node.name}") for node in self.nodes)
        )
    }
    if self.config is not None:
        this_config = {**self.config, **this_config}

    this_config = dict(self._fufill_param_requests(this_config, params=params))

    if self.config_transform is not None:
        this_config = dict(self.config_transform(this_config, transform_context))

    if len(this_config) > 0:
        _kwargs["config"] = dict(this_config)

    return self.mutate(**_kwargs)

copy

copy() -> Self

Copy this node, removing all links in the process.

Source code in src/amltk/pipeline/node.py

def copy(self) -> Self:
    """Copy this node, removing all links in the process."""
    return self.mutate()

display

display(*, full: bool = False) -> RenderableType

Display this node.

PARAMETER	DESCRIPTION
full	Whether to display the full node or just a summary TYPE: bool DEFAULT: False

Source code in src/amltk/pipeline/node.py

def display(self, *, full: bool = False) -> RenderableType:
    """Display this node.

    Args:
        full: Whether to display the full node or just a summary
    """
    if not full:
        return self.__rich__()

    from rich.console import Group as RichGroup

    return RichGroup(*self._rich_iter())

factorize

factorize(
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None
) -> Iterator[Self]

Please see factorize().

Source code in src/amltk/pipeline/node.py

def factorize(
    self,
    *,
    min_depth: int = 0,
    max_depth: int | None = None,
    current_depth: int = 0,
    factor_by: Callable[[Node], bool] | None = None,
    assign_child: Callable[[Node, Node], Node] | None = None,
) -> Iterator[Self]:
    """Please see [`factorize()`][amltk.pipeline.ops.factorize]."""  # noqa: D402
    from amltk.pipeline.ops import factorize

    yield from factorize(
        self,
        min_depth=min_depth,
        max_depth=max_depth,
        current_depth=current_depth,
        factor_by=factor_by,
        assign_child=assign_child,
    )

fidelity_space

fidelity_space() -> dict[str, Any]

Get the fidelities for this node and any connected nodes.

Source code in src/amltk/pipeline/node.py

def fidelity_space(self) -> dict[str, Any]:
    """Get the fidelities for this node and any connected nodes."""
    fids = {}
    for node in self.nodes:
        fids.update(prefix_keys(node.fidelity_space(), f"{self.name}:"))

    return fids

find

find(
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None

Find a node in that's nested deeper from this node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]
default	The value to return if the node is not found. Defaults to None TYPE: T | None DEFAULT: None

RETURNS	DESCRIPTION
Node | T | None	The node if found, otherwise the default value. Defaults to None

Source code in src/amltk/pipeline/node.py

def find(
    self,
    key: str | Node | Callable[[Node], bool],
    default: T | None = None,
) -> Node | T | None:
    """Find a node in that's nested deeper from this node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node
        default: The value to return if the node is not found. Defaults to None

    Returns:
        The node if found, otherwise the default value. Defaults to None
    """
    itr = self.iter()
    match key:
        case Node():
            return first_true(itr, default, lambda node: node == key)
        case str():
            return first_true(itr, default, lambda node: node.name == key)
        case _:
            return first_true(itr, default, key)  # type: ignore

iter

iter() -> Iterator[Node]

Iterate the the nodes, including this node.

YIELDS	DESCRIPTION
Node	The nodes connected to this node

Source code in src/amltk/pipeline/node.py

def iter(self) -> Iterator[Node]:
    """Iterate the the nodes, including this node.

    Yields:
        The nodes connected to this node
    """
    yield self
    for node in self.nodes:
        yield from node.iter()

linearized_fidelity

linearized_fidelity(
    value: float,
) -> dict[str, int | float | Any]

Get the liniearized fidelities for this node and any connected nodes.

PARAMETER	DESCRIPTION
value	The value to linearize. Must be between [0, 1] TYPE: float

Return

dictionary from key to it's linearized fidelity.

Source code in src/amltk/pipeline/node.py

def linearized_fidelity(self, value: float) -> dict[str, int | float | Any]:
    """Get the liniearized fidelities for this node and any connected nodes.

    Args:
        value: The value to linearize. Must be between [0, 1]

    Return:
        dictionary from key to it's linearized fidelity.
    """
    assert 1.0 <= value <= 100.0, f"{value=} not in [1.0, 100.0]"  # noqa: PLR2004
    d = {}
    for node in self.nodes:
        node_fids = prefix_keys(
            node.linearized_fidelity(value),
            f"{self.name}:",
        )
        d.update(node_fids)

    if self.fidelities is None:
        return d

    for f_name, f_range in self.fidelities.items():
        match f_range:
            case (int() | float(), int() | float()):
                low, high = f_range
                fid = low + (high - low) * value
                fid = low + (high - low) * (value - 1) / 100
                fid = fid if isinstance(low, float) else round(fid)
                d[f_name] = fid
            case _:
                raise ValueError(
                    f"Invalid fidelities to linearize {f_range} for {f_name}"
                    f" in {self}. Only supports ranges of the form (low, high)",
                )

    return prefix_keys(d, f"{self.name}:")

mutate

mutate(**kwargs: Any) -> Self

Mutate the node with the given keyword arguments.

PARAMETER	DESCRIPTION
**kwargs	The keyword arguments to mutate TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	Self The mutated node

Source code in src/amltk/pipeline/node.py

def mutate(self, **kwargs: Any) -> Self:
    """Mutate the node with the given keyword arguments.

    Args:
        **kwargs: The keyword arguments to mutate

    Returns:
        Self
            The mutated node
    """
    _args = ()
    _kwargs = {**self.__dict__, **kwargs}

    # If there's nodes in kwargs, we have to check if it's
    # a positional or keyword argument and handle accordingly.
    if (nodes := _kwargs.pop("nodes", None)) is not None:
        match self._NODES_INIT:
            case "args":
                _args = nodes
            case "kwargs":
                _kwargs["nodes"] = nodes
            case None if len(nodes) == 0:
                pass  # Just ignore it, it's popped out
            case None:
                raise ValueError(
                    "Cannot mutate nodes when __init__ does not accept nodes",
                )

    # If there's a config in kwargs, we have to check if it's actually got values
    config = _kwargs.pop("config", None)
    if config is not None and len(config) > 0:
        _kwargs["config"] = config

    # Lastly, we remove anything that can't be passed to kwargs of the
    # subclasses __init__
    _available_kwargs = inspect.signature(self.__init__).parameters.keys()  # type: ignore
    for k in list(_kwargs.keys()):
        if k not in _available_kwargs:
            _kwargs.pop(k)

    return self.__class__(*_args, **_kwargs)

path_to

path_to(
    key: str | Node | Callable[[Node], bool]
) -> list[Node] | None

Find a path to the given node.

PARAMETER	DESCRIPTION
key	The key to search for or a function that returns True if the node is the desired node TYPE: str | Node | Callable[[Node], bool]

RETURNS	DESCRIPTION
list[Node] | None	The path to the node if found, else None

Source code in src/amltk/pipeline/node.py

def path_to(self, key: str | Node | Callable[[Node], bool]) -> list[Node] | None:
    """Find a path to the given node.

    Args:
        key: The key to search for or a function that returns True if the node
            is the desired node

    Returns:
        The path to the node if found, else None
    """
    # We found our target, just return now

    match key:
        case Node():
            pred = lambda node: node == key
        case str():
            pred = lambda node: node.name == key
        case _:
            pred = key

    for path, node in self.walk():
        if pred(node):
            return path

    return None

search_space

search_space(
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: args,
    **parser_kwargs: kwargs
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace

Get the search space for this node.

PARAMETER	DESCRIPTION
parser	The parser to use. This can be a function that takes in the node and returns the search space or a string that is one of: "configspace": Build a ConfigSpace.ConfigurationSpace out of this node. "optuna": Build a dict of hyperparameters that Optuna can use in its ask and tell methods TYPE: Callable[Concatenate[Node, P], ParserOutput] | Literal['configspace', 'optuna']
parser_args	The positional arguments to pass to the parser TYPE: args DEFAULT: ()
parser_kwargs	The keyword arguments to pass to the parser TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
ParserOutput | ConfigurationSpace | OptunaSearchSpace	The search space

Source code in src/amltk/pipeline/node.py

def search_space(
    self,
    parser: (
        Callable[Concatenate[Node, P], ParserOutput]
        | Literal["configspace", "optuna"]
    ),
    *parser_args: P.args,
    **parser_kwargs: P.kwargs,
) -> ParserOutput | ConfigurationSpace | OptunaSearchSpace:
    """Get the search space for this node.

    Args:
        parser: The parser to use. This can be a function that takes in
            the node and returns the search space or a string that is one of:

            * `#!python "configspace"`: Build a
                [`ConfigSpace.ConfigurationSpace`](https://automl.github.io/ConfigSpace/master/)
                out of this node.
            * `#!python "optuna"`: Build a dict of hyperparameters that Optuna can
                use in its [ask and tell methods](https://optuna.readthedocs.io/en/stable/tutorial/20_recipes/009_ask_and_tell.html#define-and-run)

        parser_args: The positional arguments to pass to the parser
        parser_kwargs: The keyword arguments to pass to the parser

    Returns:
        The search space
    """
    match parser:
        case "configspace":
            from amltk.pipeline.parsers.configspace import parser as cs_parser

            return cs_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case "optuna":
            from amltk.pipeline.parsers.optuna import parser as optuna_parser

            return optuna_parser(self, *parser_args, **parser_kwargs)  # type: ignore
        case str():  # type: ignore
            raise ValueError(
                f"Invalid str for parser {parser}. "
                "Please use 'configspace' or 'optuna' or pass in your own"
                " parser function",
            )
        case _:
            return parser(self, *parser_args, **parser_kwargs)

walk

walk(
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]

Walk the nodes in this chain.

PARAMETER	DESCRIPTION
path	The current path to this node TYPE: Sequence[Node] | None DEFAULT: None

YIELDS	DESCRIPTION
list[Node]	The parents of the node and the node itself

Source code in src/amltk/pipeline/node.py

def walk(
    self,
    path: Sequence[Node] | None = None,
) -> Iterator[tuple[list[Node], Node]]:
    """Walk the nodes in this chain.

    Args:
        path: The current path to this node

    Yields:
        The parents of the node and the node itself
    """
    path = list(path) if path is not None else []
    yield path, self

    for node in self.nodes:
        yield from node.walk(path=[*path, self])

as_node

as_node(
    thing: Node | NodeLike[Item], name: str | None = None
) -> Node | Choice | Join | Sequential | Fixed[Item]

Convert a node, pipeline, set or tuple into a component, copying anything in the process and removing all linking to other nodes.

PARAMETER	DESCRIPTION
thing	The thing to convert TYPE: Node | NodeLike[Item]
name	The name of the node. If it already a node, it will be renamed to that one. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
Node | Choice | Join | Sequential | Fixed[Item]	The component

Source code in src/amltk/pipeline/components.py

def as_node(  # noqa: PLR0911
    thing: Node | NodeLike[Item],
    name: str | None = None,
) -> Node | Choice | Join | Sequential | Fixed[Item]:
    """Convert a node, pipeline, set or tuple into a component, copying anything
    in the process and removing all linking to other nodes.

    Args:
        thing: The thing to convert
        name: The name of the node. If it already a node, it will be renamed to that
            one.

    Returns:
        The component
    """
    match thing:
        case set():
            return Choice(*thing, name=name)
        case tuple():
            return Join(*thing, name=name)
        case list():
            return Sequential(*thing, name=name)
        case Node():
            name = thing.name if name is None else name
            return thing.mutate(name=name)
        case type():
            return Component(thing, name=name)
        case thing if (inspect.isfunction(thing) or inspect.ismethod(thing)):
            return Component(thing, name=name)
        case _:
            return Fixed(thing, name=name)