(
        self, original_op: "Operator", replacement_op: "Operator"
    )
return _replace(self, original_op, replacement_op)
Replaces an original operator with a replacement operator for the given operator.
        Replacement also occurs for all operators within the given operator's steps (i.e. pipelines and
        choices). If a planned operator is given as original_op, all derived operators (including
        trainable and trained versions) will be replaced. Otherwise, only the exact operator
        instance will be replaced.

        Parameters
        ----------
        original_op :
            Operator to replace within given operator. If operator is a planned operator,
            all derived operators (including trainable and trained versions) will be
            replaced. Otherwise, only the exact operator instance will be replaced.

        replacement_op :
            Operator to replace the original with.

        Returns
        -------
        modified_operator :
            Modified operator where original operator is replaced with replacement throughout.
Replaces an original operator with a replacement operator for the given operator.
        Replacement also occurs for all operators within the given operator's steps (i.e. pipelines and
        choices). If a planned operator is given as original_op, all derived operators (including
        trainable and trained versions) will be replaced. Otherwise, only the exact operator
        instance will be replaced.
Replaces an original operator with a replacement operator for the given operator . Replacement also occurs for all operators within the given operator s steps ( i . e . pipelines and choices ) . If a planned operator is given as original_op all derived operators ( including trainable and trained versions ) will be replaced . Otherwise only the exact operator instance will be replaced .
def replace(
        self, original_op: "Operator", replacement_op: "Operator"
    ) -> "Operator":
        """Replaces an original operator with a replacement operator for the given operator.
        Replacement also occurs for all operators within the given operator's steps (i.e. pipelines and
        choices). If a planned operator is given as original_op, all derived operators (including
        trainable and trained versions) will be replaced. Otherwise, only the exact operator
        instance will be replaced.

        Parameters
        ----------
        original_op :
            Operator to replace within given operator. If operator is a planned operator,
            all derived operators (including trainable and trained versions) will be
            replaced. Otherwise, only the exact operator instance will be replaced.

        replacement_op :
            Operator to replace the original with.

        Returns
        -------
        modified_operator :
            Modified operator where original operator is replaced with replacement throughout.
        """

        def _check_match(subject, original_op):
            if (
                not isinstance(original_op, TrainableOperator)
                and isinstance(subject, IndividualOp)
                and isinstance(original_op, IndividualOp)
            ):
                # is planned operator, so replace any matching downstream operator
                if isinstance(subject, original_op):  # type: ignore
                    return True
            else:
                # is trainable or trained operator, only check exact instance match
                if subject == original_op:
                    return True
            return False

        def _replace(subject, original_op, replacement_op):

            # if operator has steps, recursively iterate through steps and recombine
            if hasattr(subject, "steps"):

                # special case if original_op has steps, check if it matches subject first
                if hasattr(original_op, "steps"):
                    if _check_match(subject, original_op):
                        return replacement_op

                new_steps: List[Operator] = []
                if isinstance(subject, BasePipeline):
                    # first convert pipeline edges to index-based representation
                    index_edges = []
                    for edge in subject.edges():
                        index_edges.append(
                            (
                                subject.steps().index(edge[0]),
                                subject.steps().index(edge[1]),
                            )
                        )

                    for step in subject.steps():
                        new_steps.append(_replace(step, original_op, replacement_op))

                    # use previous index-based representation to reconstruct edges
                    new_edges: List[Tuple[Operator, Operator]] = []
                    for index_tuple in index_edges:
                        new_edges.append(
                            (new_steps[index_tuple[0]], new_steps[index_tuple[1]])
                        )

                    return make_pipeline_graph(new_steps, new_edges)

                elif isinstance(subject, OperatorChoice):
                    for step in subject.steps():
                        new_steps.append(_replace(step, original_op, replacement_op))
                    return make_choice(*new_steps)

                else:
                    raise NotImplementedError(
                        "replace() needs to implement recombining this operator with steps"
                    )
            else:
                # base case for recursion: operator with no steps, returns replacement if applicable, original otherwise
                if _check_match(subject, original_op):
                    return replacement_op

                # special case of subject being in a collection
                if isinstance(subject, list):
                    return [_replace(s, original_op, replacement_op) for s in subject]
                elif isinstance(subject, tuple):
                    return tuple(
                        [_replace(s, original_op, replacement_op) for s in subject]
                    )
                elif isinstance(subject, dict):
                    return {
                        k: _replace(v, original_op, replacement_op)
                        for k, v in subject.items()
                    }

                # special case of hyperparams containing operators, usually referring to an estimator
                if hasattr(subject, "hyperparams") and subject.hyperparams():
                    modified_hyperparams = subject.hyperparams().copy()
                    for hyperparam, param_value in modified_hyperparams.items():
                        modified_hyperparams[hyperparam] = _replace(
                            param_value, original_op, replacement_op
                        )

                    return subject(**modified_hyperparams)

            return subject

        return _replace(self, original_op, replacement_op)
def replace ( self , original_op : "Operator" , replacement_op : "Operator" ) -> "Operator" : def _check_match ( subject , original_op ) : if ( not isinstance ( original_op , TrainableOperator ) and isinstance ( subject , IndividualOp ) and isinstance ( original_op , IndividualOp ) ) : # is planned operator, so replace any matching downstream operator if isinstance ( subject , original_op ) : # type: ignore return True else : # is trainable or trained operator, only check exact instance match if subject == original_op : return True return False def _replace ( subject , original_op , replacement_op ) : # if operator has steps, recursively iterate through steps and recombine if hasattr ( subject , "steps" ) : # special case if original_op has steps, check if it matches subject first if hasattr ( original_op , "steps" ) : if _check_match ( subject , original_op ) : return replacement_op new_steps : List [ Operator ] = [ ] if isinstance ( subject , BasePipeline ) : # first convert pipeline edges to index-based representation index_edges = [ ] for edge in subject . edges ( ) : index_edges . append ( ( subject . steps ( ) . index ( edge [ 0 ] ) , subject . steps ( ) . index ( edge [ 1 ] ) , ) ) for step in subject . steps ( ) : new_steps . append ( _replace ( step , original_op , replacement_op ) ) # use previous index-based representation to reconstruct edges new_edges : List [ Tuple [ Operator , Operator ] ] = [ ] for index_tuple in index_edges : new_edges . append ( ( new_steps [ index_tuple [ 0 ] ] , new_steps [ index_tuple [ 1 ] ] ) ) return make_pipeline_graph ( new_steps , new_edges ) elif isinstance ( subject , OperatorChoice ) : for step in subject . steps ( ) : new_steps . append ( _replace ( step , original_op , replacement_op ) ) return make_choice ( * new_steps ) else : raise NotImplementedError ( "replace() needs to implement recombining this operator with steps" ) else : # base case for recursion: operator with no steps, returns replacement if applicable, original otherwise if _check_match ( subject , original_op ) : return replacement_op # special case of subject being in a collection if isinstance ( subject , list ) : return [ _replace ( s , original_op , replacement_op ) for s in subject ] elif isinstance ( subject , tuple ) : return tuple ( [ _replace ( s , original_op , replacement_op ) for s in subject ] ) elif isinstance ( subject , dict ) : return { k : _replace ( v , original_op , replacement_op ) for k , v in subject . items ( ) } # special case of hyperparams containing operators, usually referring to an estimator if hasattr ( subject , "hyperparams" ) and subject . hyperparams ( ) : modified_hyperparams = subject . hyperparams ( ) . copy ( ) for hyperparam , param_value in modified_hyperparams . items ( ) : modified_hyperparams [ hyperparam ] = _replace ( param_value , original_op , replacement_op ) return subject ( * * modified_hyperparams ) return subject return _replace ( self , original_op , replacement_op )