Skip to content

Splitter

framework3.base.base_splitter

__all__ = ['BaseSplitter'] module-attribute

BaseSplitter

Bases: BaseFilter

Base class for splitter components in the framework.

This abstract class extends BaseFilter and defines the interface for splitter operations. It provides a structure for implementing data splitting strategies for pipelines.

Key Features
  • Abstract methods for starting splitting process, logging metrics, and splitting pipelines
  • Support for verbose output control
  • Integration with optimizer components
Usage

To create a new splitter type, inherit from this class and implement the required methods. For example:

class MyCustomSplitter(BaseSplitter):
    def __init__(self, splitting_params):
        super().__init__()
        self.splitting_params = splitting_params

    def start(self, x: XYData, y: Optional[XYData], X_: Optional[XYData]) -> Optional[XYData]:
        # Implement splitting start logic
        pass

    def split(self, pipeline: BaseFilter) -> None:
        # Implement pipeline splitting logic
        pass

    def log_metrics(self) -> None:
        # Implement metric logging logic
        pass

    def evaluate(self, x_data: XYData, y_true: XYData | None, y_pred: XYData) -> Dict[str, Any]:
        # Implement evaluation logic
        pass

Attributes:

Name Type Description
pipeline BaseFilter

The pipeline to be split.

Methods:

Name Description
start

XYData, y: Optional[XYData], X_: Optional[XYData]) -> Optional[XYData]: Abstract method to start the splitting process.
log_metrics

Abstract method to log the metrics of the pipeline.
split

BaseFilter) -> None: Abstract method to split the given pipeline.
verbose

bool) -> None: Sets the verbosity level for the splitter and its pipeline.
evaluate

XYData, y_true: XYData | None, y_pred: XYData) -> Dict[str, Any]: Abstract method to evaluate the metric based on the provided data.
optimizer

BaseOptimizer) -> BaseOptimizer: Applies an optimizer to the splitter.
unwrap

Returns the underlying pipeline.
Note

This is an abstract base class. Concrete implementations should override the abstract methods to provide specific splitting functionality.

Source code in framework3/base/base_splitter.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class BaseSplitter(BaseFilter):
    """
    Base class for splitter components in the framework.

    This abstract class extends BaseFilter and defines the interface for splitter operations.
    It provides a structure for implementing data splitting strategies for pipelines.

    Key Features:
        - Abstract methods for starting splitting process, logging metrics, and splitting pipelines
        - Support for verbose output control
        - Integration with optimizer components

    Usage:
        To create a new splitter type, inherit from this class and implement
        the required methods. For example:

        ```python
        class MyCustomSplitter(BaseSplitter):
            def __init__(self, splitting_params):
                super().__init__()
                self.splitting_params = splitting_params

            def start(self, x: XYData, y: Optional[XYData], X_: Optional[XYData]) -> Optional[XYData]:
                # Implement splitting start logic
                pass

            def split(self, pipeline: BaseFilter) -> None:
                # Implement pipeline splitting logic
                pass

            def log_metrics(self) -> None:
                # Implement metric logging logic
                pass

            def evaluate(self, x_data: XYData, y_true: XYData | None, y_pred: XYData) -> Dict[str, Any]:
                # Implement evaluation logic
                pass
        ```

    Attributes:
        pipeline (BaseFilter): The pipeline to be split.

    Methods:
        start(x: XYData, y: Optional[XYData], X_: Optional[XYData]) -> Optional[XYData]:
            Abstract method to start the splitting process.

        log_metrics() -> None:
            Abstract method to log the metrics of the pipeline.

        split(pipeline: BaseFilter) -> None:
            Abstract method to split the given pipeline.

        verbose(value: bool) -> None:
            Sets the verbosity level for the splitter and its pipeline.

        evaluate(x_data: XYData, y_true: XYData | None, y_pred: XYData) -> Dict[str, Any]:
            Abstract method to evaluate the metric based on the provided data.

        optimizer(optimizer: BaseOptimizer) -> BaseOptimizer:
            Applies an optimizer to the splitter.

        unwrap() -> BaseFilter:
            Returns the underlying pipeline.

    Note:
        This is an abstract base class. Concrete implementations should override
        the abstract methods to provide specific splitting functionality.
    """

    @abstractmethod
    def start(
        self, x: XYData, y: Optional[XYData], X_: Optional[XYData]
    ) -> Optional[XYData]:
        """
        Start the splitting process.

        This abstract method should be implemented by subclasses to define
        the specific logic for initiating the splitting process.

        Args:
            x (XYData): The primary input data.
            y (Optional[XYData]): Optional target data.
            X_ (Optional[XYData]): Optional additional input data.

        Returns:
            Optional[XYData]: The processed data, if any.

        Raises:
            NotImplementedError: If the subclass does not implement this method.
        """
        ...

    @abstractmethod
    def split(self, pipeline: BaseFilter) -> None:
        """
        Split the given pipeline.

        This abstract method should be implemented by subclasses to define
        the specific logic for splitting a pipeline.

        Args:
            pipeline (BaseFilter): The pipeline to be split.

        Raises:
            NotImplementedError: If the subclass does not implement this method.
        """
        ...

    def verbose(self, value: bool) -> None:
        """
        Set the verbosity of the splitter and its pipeline.

        This method controls the verbosity of both the splitter itself and its associated pipeline.

        Args:
            value (bool): If True, enables verbose output; if False, disables it.
        """
        self._verbose = value
        self.pipeline.verbose(value)

    @abstractmethod
    def evaluate(
        self, x_data: XYData, y_true: XYData | None, y_pred: XYData
    ) -> Dict[str, Any]:
        """
        Evaluate the metric based on the provided data.

        This abstract method should be implemented by subclasses to calculate the specific metric.

        Args:
            x_data (XYData): The input data used for the prediction.
            y_true (XYData | None): The ground truth or actual values.
            y_pred (XYData): The predicted values.

        Returns:
            Dict[str, Any]: A dictionary containing the calculated metric values.

        Raises:
            NotImplementedError: If the subclass does not implement this method.
        """
        ...

    def _pre_fit_wrapp(self, x: XYData, y: Optional[XYData]) -> float | None:
        """
        Wrapper method for pre-fitting.

        This method calls the original fit method of the pipeline.

        Args:
            x (XYData): The input data for fitting.
            y (Optional[XYData]): The target data for fitting, if applicable.

        Returns:
            float | None: The result of the original fit method.
        """
        return self._original_fit(x, y)

    def _pre_predict_wrapp(self, x: XYData) -> XYData:
        """
        Wrapper method for pre-prediction.

        This method calls the original predict method of the pipeline.

        Args:
            x (XYData): The input data for prediction.

        Returns:
            XYData: The result of the original predict method.
        """
        return self._original_predict(x)

    def optimizer(self, optimizer: BaseOptimizer) -> BaseOptimizer:
        """
        Apply an optimizer to the splitter.

        This method allows an optimizer to be applied to the entire splitter.

        Args:
            optimizer (BaseOptimizer): The optimizer to apply to the splitter.

        Returns:
            BaseOptimizer: The optimizer after optimization.
        """
        optimizer.optimize(self)
        return optimizer

    def unwrap(self) -> BaseFilter:
        """
        Unwrap the splitter to get the underlying pipeline.

        This method returns the pipeline that the splitter is operating on.

        Returns:
            BaseFilter: The underlying pipeline.
        """
        return self.pipeline
evaluate(x_data, y_true, y_pred) abstractmethod

Evaluate the metric based on the provided data.

This abstract method should be implemented by subclasses to calculate the specific metric.

Parameters:

Name Type Description Default
x_data XYData

The input data used for the prediction.

 required
y_true XYData | None

The ground truth or actual values.

 required
y_pred XYData

The predicted values.

 required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: A dictionary containing the calculated metric values.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Source code in framework3/base/base_splitter.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@abstractmethod
def evaluate(
    self, x_data: XYData, y_true: XYData | None, y_pred: XYData
) -> Dict[str, Any]:
    """
    Evaluate the metric based on the provided data.

    This abstract method should be implemented by subclasses to calculate the specific metric.

    Args:
        x_data (XYData): The input data used for the prediction.
        y_true (XYData | None): The ground truth or actual values.
        y_pred (XYData): The predicted values.

    Returns:
        Dict[str, Any]: A dictionary containing the calculated metric values.

    Raises:
        NotImplementedError: If the subclass does not implement this method.
    """
    ...
optimizer(optimizer)

Apply an optimizer to the splitter.

This method allows an optimizer to be applied to the entire splitter.

Parameters:

Name Type Description Default
optimizer BaseOptimizer

The optimizer to apply to the splitter.

 required

Returns:

Name Type Description
BaseOptimizer BaseOptimizer

The optimizer after optimization.
Source code in framework3/base/base_splitter.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def optimizer(self, optimizer: BaseOptimizer) -> BaseOptimizer:
    """
    Apply an optimizer to the splitter.

    This method allows an optimizer to be applied to the entire splitter.

    Args:
        optimizer (BaseOptimizer): The optimizer to apply to the splitter.

    Returns:
        BaseOptimizer: The optimizer after optimization.
    """
    optimizer.optimize(self)
    return optimizer
split(pipeline) abstractmethod

Split the given pipeline.

This abstract method should be implemented by subclasses to define the specific logic for splitting a pipeline.

Parameters:

Name Type Description Default
pipeline BaseFilter

The pipeline to be split.

 required

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Source code in framework3/base/base_splitter.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@abstractmethod
def split(self, pipeline: BaseFilter) -> None:
    """
    Split the given pipeline.

    This abstract method should be implemented by subclasses to define
    the specific logic for splitting a pipeline.

    Args:
        pipeline (BaseFilter): The pipeline to be split.

    Raises:
        NotImplementedError: If the subclass does not implement this method.
    """
    ...
start(x, y, X_) abstractmethod

Start the splitting process.

This abstract method should be implemented by subclasses to define the specific logic for initiating the splitting process.

Parameters:

Name Type Description Default
x XYData

The primary input data.

 required
y Optional[XYData]

Optional target data.

 required
X_ Optional[XYData]

Optional additional input data.

 required

Returns:

Type Description
Optional[XYData]

Optional[XYData]: The processed data, if any.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Source code in framework3/base/base_splitter.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@abstractmethod
def start(
    self, x: XYData, y: Optional[XYData], X_: Optional[XYData]
) -> Optional[XYData]:
    """
    Start the splitting process.

    This abstract method should be implemented by subclasses to define
    the specific logic for initiating the splitting process.

    Args:
        x (XYData): The primary input data.
        y (Optional[XYData]): Optional target data.
        X_ (Optional[XYData]): Optional additional input data.

    Returns:
        Optional[XYData]: The processed data, if any.

    Raises:
        NotImplementedError: If the subclass does not implement this method.
    """
    ...
unwrap()

Unwrap the splitter to get the underlying pipeline.

This method returns the pipeline that the splitter is operating on.

Returns:

Name Type Description
BaseFilter BaseFilter

The underlying pipeline.
Source code in framework3/base/base_splitter.py 
197
198
199
200
201
202
203
204
205
206
def unwrap(self) -> BaseFilter:
    """
    Unwrap the splitter to get the underlying pipeline.

    This method returns the pipeline that the splitter is operating on.

    Returns:
        BaseFilter: The underlying pipeline.
    """
    return self.pipeline
verbose(value)

Set the verbosity of the splitter and its pipeline.

This method controls the verbosity of both the splitter itself and its associated pipeline.

Parameters:

Name Type Description Default
value bool

If True, enables verbose output; if False, disables it.

 required
Source code in framework3/base/base_splitter.py 
119
120
121
122
123
124
125
126
127
128
129
def verbose(self, value: bool) -> None:
    """
    Set the verbosity of the splitter and its pipeline.

    This method controls the verbosity of both the splitter itself and its associated pipeline.

    Args:
        value (bool): If True, enables verbose output; if False, disables it.
    """
    self._verbose = value
    self.pipeline.verbose(value)