(
        self, X, y=None, num_epochs_batching=None
    )
return result
This method handles the case where all but the last node of a pipeline are frozen_trained and
        only the last node needs to be fitted batch-wise.
This method handles the case where all but the last node of a pipeline are frozen_trained and
        only the last node needs to be fitted batch-wise.
This method handles the case where all but the last node of a pipeline are frozen_trained and only the last node needs to be fitted batch - wise .
def _fit_last_node_with_batches(
        self, X, y=None, num_epochs_batching=None
    ) -> "TrainedPipeline[TrainedIndividualOp]":
        """
        This method handles the case where all but the last node of a pipeline are frozen_trained and
        only the last node needs to be fitted batch-wise.
        """
        sink_nodes = self._find_sink_nodes()
        if len(sink_nodes) > 1:
            raise ValueError(
                "A pipeline with more than one sink node is not supported."
            )
        sink_node = sink_nodes[0]
        new_sink_node: Union[TrainableIndividualOp, TrainedIndividualOp] = sink_node

        try:
            num_epochs = sink_node._impl_instance().num_epochs
        except AttributeError:
            if num_epochs_batching is None:
                warnings.warn(
                    "Operator {} does not have num_epochs and none given to Batching operator, using 1 as a default".format(
                        sink_node.name()
                    )
                )
                num_epochs = 1
            else:
                num_epochs = num_epochs_batching
        assert num_epochs >= 0
        outputs: Dict[Operator, Any] = {}
        edges: List[Tuple[TrainableOpType, TrainableOpType]] = self.edges()
        trained_map: Dict[
            Union[TrainedIndividualOp, TrainableIndividualOp], TrainedIndividualOp
        ] = {}
        trained_steps: List[IndividualOp] = []
        for operator in self._steps[
            :-1
        ]:  # Again a strict assumption on the structure of the pipeline
            trained_map[operator] = operator  # type:ignore
            trained_steps.append(operator)

        for epoch in range(num_epochs):
            for _, batch_data in enumerate(
                X  # With Batching, X is a dataloader which contains y too if given
            ):  # batching_transformer will output only one obj
                if isinstance(batch_data, tuple):
                    batch_X, batch_y = batch_data
                elif isinstance(batch_data, list):
                    batch_X = batch_data[0]
                    batch_y = batch_data[1]
                else:
                    batch_X = batch_data
                    batch_y = None
                batch_output = None
                for operator in self._steps[
                    :-1
                ]:  # Again a strict assumption on the structure of the pipeline
                    if not operator.is_frozen_trained():
                        raise ValueError(
                            "All the nodes but the last one should be frozen trained to use this mode of batch-wise training."
                        )
                    preds = self._preds[operator]
                    if len(preds) == 0:
                        batch_input = [batch_X]
                    else:
                        batch_input = [
                            outputs[pred][0]
                            if isinstance(outputs[pred], tuple)
                            else outputs[pred]
                            for pred in preds
                        ]
                    if len(batch_input) == 1:
                        batch_input = batch_input[0]

                    if operator.is_transformer():
                        batch_output = operator.transform(batch_input, batch_y)
                    else:
                        # This is ok because trainable pipelines steps
                        # must only be individual operators
                        if operator.has_method("predict_proba"):  # type: ignore
                            batch_output = operator.predict_proba(X=batch_input)
                        elif operator.has_method("decision_function"):  # type: ignore
                            batch_output = operator.decision_function(X=batch_input)
                        else:
                            batch_output = operator._predict(X=batch_input)
                    outputs[operator] = batch_output

                if sink_node.is_supervised():
                    try:
                        new_sink_node = new_sink_node.partial_fit(
                            batch_output, batch_y, classes=y
                        )
                    except TypeError:
                        new_sink_node = new_sink_node.partial_fit(batch_output, batch_y)
                else:
                    new_sink_node = new_sink_node.partial_fit(batch_output)
        trained = TrainedIndividualOp(
            new_sink_node.name(),
            new_sink_node._impl,
            new_sink_node._schemas,
            None,
            _lale_trained=True,
        )
        trained_steps.append(trained)
        trained_map[sink_node] = trained
        trained_edges = [(trained_map[x], trained_map[y]) for (x, y) in edges]

        trained_steps2: Any = list(trained_steps)
        result: TrainedPipeline[TrainedIndividualOp] = TrainedPipeline(
            trained_steps2, trained_edges, ordered=True, _lale_trained=True
        )
        self._trained = result
        return result
def _fit_last_node_with_batches ( self , X , y = None , num_epochs_batching = None ) -> "TrainedPipeline[TrainedIndividualOp]" : sink_nodes = self . _find_sink_nodes ( ) if len ( sink_nodes ) > 1 : raise ValueError ( "A pipeline with more than one sink node is not supported." ) sink_node = sink_nodes [ 0 ] new_sink_node : Union [ TrainableIndividualOp , TrainedIndividualOp ] = sink_node try : num_epochs = sink_node . _impl_instance ( ) . num_epochs except AttributeError : if num_epochs_batching is None : warnings . warn ( "Operator {} does not have num_epochs and none given to Batching operator, using 1 as a default" . format ( sink_node . name ( ) ) ) num_epochs = 1 else : num_epochs = num_epochs_batching assert num_epochs >= 0 outputs : Dict [ Operator , Any ] = { } edges : List [ Tuple [ TrainableOpType , TrainableOpType ] ] = self . edges ( ) trained_map : Dict [ Union [ TrainedIndividualOp , TrainableIndividualOp ] , TrainedIndividualOp ] = { } trained_steps : List [ IndividualOp ] = [ ] for operator in self . _steps [ : - 1 ] : # Again a strict assumption on the structure of the pipeline trained_map [ operator ] = operator # type:ignore trained_steps . append ( operator ) for epoch in range ( num_epochs ) : for _ , batch_data in enumerate ( X # With Batching, X is a dataloader which contains y too if given ) : # batching_transformer will output only one obj if isinstance ( batch_data , tuple ) : batch_X , batch_y = batch_data elif isinstance ( batch_data , list ) : batch_X = batch_data [ 0 ] batch_y = batch_data [ 1 ] else : batch_X = batch_data batch_y = None batch_output = None for operator in self . _steps [ : - 1 ] : # Again a strict assumption on the structure of the pipeline if not operator . is_frozen_trained ( ) : raise ValueError ( "All the nodes but the last one should be frozen trained to use this mode of batch-wise training." ) preds = self . _preds [ operator ] if len ( preds ) == 0 : batch_input = [ batch_X ] else : batch_input = [ outputs [ pred ] [ 0 ] if isinstance ( outputs [ pred ] , tuple ) else outputs [ pred ] for pred in preds ] if len ( batch_input ) == 1 : batch_input = batch_input [ 0 ] if operator . is_transformer ( ) : batch_output = operator . transform ( batch_input , batch_y ) else : # This is ok because trainable pipelines steps # must only be individual operators if operator . has_method ( "predict_proba" ) : # type: ignore batch_output = operator . predict_proba ( X = batch_input ) elif operator . has_method ( "decision_function" ) : # type: ignore batch_output = operator . decision_function ( X = batch_input ) else : batch_output = operator . _predict ( X = batch_input ) outputs [ operator ] = batch_output if sink_node . is_supervised ( ) : try : new_sink_node = new_sink_node . partial_fit ( batch_output , batch_y , classes = y ) except TypeError : new_sink_node = new_sink_node . partial_fit ( batch_output , batch_y ) else : new_sink_node = new_sink_node . partial_fit ( batch_output ) trained = TrainedIndividualOp ( new_sink_node . name ( ) , new_sink_node . _impl , new_sink_node . _schemas , None , _lale_trained = True , ) trained_steps . append ( trained ) trained_map [ sink_node ] = trained trained_edges = [ ( trained_map [ x ] , trained_map [ y ] ) for ( x , y ) in edges ] trained_steps2 : Any = list ( trained_steps ) result : TrainedPipeline [ TrainedIndividualOp ] = TrainedPipeline ( trained_steps2 , trained_edges , ordered = True , _lale_trained = True ) self . _trained = result return result