(
    lale_object, hyperparams
)
Hyperparams is a n-tuple of dictionaries of hyper-parameters, each
    dictionary corresponds to an operator in the pipeline
Hyperparams is a n-tuple of dictionaries of hyper-parameters, each
    dictionary corresponds to an operator in the pipeline
Hyperparams is a n - tuple of dictionaries of hyper - parameters each dictionary corresponds to an operator in the pipeline
def create_instance_from_hyperopt_search_space(
    lale_object, hyperparams
) -> "lale.operators.Operator":
    """
    Hyperparams is a n-tuple of dictionaries of hyper-parameters, each
    dictionary corresponds to an operator in the pipeline
    """
    # lale_object can either be an individual operator, a pipeline or an operatorchoice
    # Validate that the number of elements in the n-tuple is the same
    # as the number of steps in the current pipeline

    from lale.operators import (
        BasePipeline,
        OperatorChoice,
        PlannedIndividualOp,
        TrainableOperator,
        TrainablePipeline,
    )

    if isinstance(lale_object, PlannedIndividualOp):
        new_hyperparams: Dict[str, Any] = dict_without(hyperparams, "name")
        hps = lale_object.hyperparams()
        if hps:
            obj_hyperparams = dict(hps)
        else:
            obj_hyperparams = {}

        for k, sub_params in new_hyperparams.items():
            if k in obj_hyperparams:
                sub_op = obj_hyperparams[k]
                updated_params = instantiate_from_hyperopt_search_space(
                    sub_op, sub_params
                )
                if updated_params is not None:
                    new_hyperparams[k] = updated_params

        all_hyperparams = {**obj_hyperparams, **new_hyperparams}
        return lale_object(**all_hyperparams)
    elif isinstance(lale_object, BasePipeline):
        steps = lale_object.steps()
        if len(hyperparams) != len(steps):
            raise ValueError(
                "The number of steps in the hyper-parameter space does not match the number of steps in the pipeline."
            )
        op_instances = []
        edges = lale_object.edges()
        # op_map:Dict[PlannedOpType, TrainableOperator] = {}
        op_map = {}
        for op_index, sub_params in enumerate(hyperparams):
            sub_op = steps[op_index]
            op_instance = create_instance_from_hyperopt_search_space(sub_op, sub_params)
            assert isinstance(op_instance, TrainableOperator)
            assert (
                isinstance(sub_op, OperatorChoice)
                or sub_op.class_name() == op_instance.class_name()
            ), f"sub_op {sub_op.class_name()}, op_instance {op_instance.class_name()}"
            op_instances.append(op_instance)
            op_map[sub_op] = op_instance

        # trainable_edges:List[Tuple[TrainableOperator, TrainableOperator]]
        try:
            trainable_edges = [(op_map[x], op_map[y]) for (x, y) in edges]
        except KeyError as e:
            raise ValueError(
                "An edge was found with an endpoint that is not a step (" + str(e) + ")"
            )

        return TrainablePipeline(op_instances, trainable_edges, ordered=True)  # type: ignore
    elif isinstance(lale_object, OperatorChoice):
        # Hyperopt search space for an OperatorChoice is generated as a dictionary with a single element
        # corresponding to the choice made, the only key is the index of the step and the value is
        # the params corresponding to that step.
        step_index: int
        choices = lale_object.steps()

        if len(choices) == 1:
            step_index = 0
        else:
            step_index_str, hyperparams = list(hyperparams.items())[0]
            step_index = int(step_index_str)
        step_object = choices[step_index]
        return create_instance_from_hyperopt_search_space(step_object, hyperparams)
    else:
        assert False, f"Unknown operator type: {type(lale_object)}"
def create_instance_from_hyperopt_search_space ( lale_object , hyperparams ) -> "lale.operators.Operator" : # lale_object can either be an individual operator, a pipeline or an operatorchoice # Validate that the number of elements in the n-tuple is the same # as the number of steps in the current pipeline from lale . operators import ( BasePipeline , OperatorChoice , PlannedIndividualOp , TrainableOperator , TrainablePipeline , ) if isinstance ( lale_object , PlannedIndividualOp ) : new_hyperparams : Dict [ str , Any ] = dict_without ( hyperparams , "name" ) hps = lale_object . hyperparams ( ) if hps : obj_hyperparams = dict ( hps ) else : obj_hyperparams = { } for k , sub_params in new_hyperparams . items ( ) : if k in obj_hyperparams : sub_op = obj_hyperparams [ k ] updated_params = instantiate_from_hyperopt_search_space ( sub_op , sub_params ) if updated_params is not None : new_hyperparams [ k ] = updated_params all_hyperparams = { * * obj_hyperparams , * * new_hyperparams } return lale_object ( * * all_hyperparams ) elif isinstance ( lale_object , BasePipeline ) : steps = lale_object . steps ( ) if len ( hyperparams ) != len ( steps ) : raise ValueError ( "The number of steps in the hyper-parameter space does not match the number of steps in the pipeline." ) op_instances = [ ] edges = lale_object . edges ( ) # op_map:Dict[PlannedOpType, TrainableOperator] = {} op_map = { } for op_index , sub_params in enumerate ( hyperparams ) : sub_op = steps [ op_index ] op_instance = create_instance_from_hyperopt_search_space ( sub_op , sub_params ) assert isinstance ( op_instance , TrainableOperator ) assert ( isinstance ( sub_op , OperatorChoice ) or sub_op . class_name ( ) == op_instance . class_name ( ) ) , f"sub_op {sub_op.class_name()}, op_instance {op_instance.class_name()}" op_instances . append ( op_instance ) op_map [ sub_op ] = op_instance # trainable_edges:List[Tuple[TrainableOperator, TrainableOperator]] try : trainable_edges = [ ( op_map [ x ] , op_map [ y ] ) for ( x , y ) in edges ] except KeyError as e : raise ValueError ( "An edge was found with an endpoint that is not a step (" + str ( e ) + ")" ) return TrainablePipeline ( op_instances , trainable_edges , ordered = True ) # type: ignore elif isinstance ( lale_object , OperatorChoice ) : # Hyperopt search space for an OperatorChoice is generated as a dictionary with a single element # corresponding to the choice made, the only key is the index of the step and the value is # the params corresponding to that step. step_index : int choices = lale_object . steps ( ) if len ( choices ) == 1 : step_index = 0 else : step_index_str , hyperparams = list ( hyperparams . items ( ) ) [ 0 ] step_index = int ( step_index_str ) step_object = choices [ step_index ] return create_instance_from_hyperopt_search_space ( step_object , hyperparams ) else : assert False , f"Unknown operator type: {type(lale_object)}"