2010037515

Build # 2010037515

Build Type

github

Committed by GitHub

Commit Message

Merge b6faeac59 into 6122a52f0

Pull Request Pull Request #99: Orm separation

Run Details

1179 of 1816 branches covered (64.92%)

Branch coverage included in aggregate %.

237 of 321 new or added lines in 25 files covered. (73.83%)

2 existing lines in 2 files now uncovered.

2756 of 4610 relevant lines covered (59.78%)

3.3 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

40.11

/simpleml/pipelines/base_pipeline.py

'''
Base Module for Pipelines
'''

__author__ = 'Elisha Yadgaran'

import inspect
import logging
import weakref
from typing import TYPE_CHECKING, Any, Dict, Generator, List, Optional, Union

import pandas as pd
from simpleml.constants import TRAIN_SPLIT
from simpleml.datasets.dataset_splits import Split, SplitContainer
from simpleml.persistables.base_persistable import Persistable
from simpleml.registries import PipelineRegistry
from simpleml.save_patterns.decorators import ExternalArtifactDecorators
from simpleml.utils.errors import PipelineError

from .projected_splits import (IdentityProjectedDatasetSplit,
                               ProjectedDatasetSplit)

if TYPE_CHECKING:
    # Cyclical import hack for type hints
    from simpleml.datasets.base_dataset import Dataset


LOGGER = logging.getLogger(__name__)


@ExternalArtifactDecorators.register_artifact(
    artifact_name='pipeline', save_attribute='external_pipeline', restore_attribute='_external_file')
class Pipeline(Persistable, metaclass=PipelineRegistry):
    '''
    Abstract Base class for all Pipelines objects.

    Relies on mixin classes to define the split_dataset method. Will throw
    an error on use otherwise

    -------
    Schema
    -------
    params: pipeline parameter metadata for easy insight into hyperparameters across trainings
    '''
    object_type: str = 'PIPELINE'

    def __init__(self,
                 has_external_files: bool = True,
                 transformers: Optional[List[Any]] = None,
                 fitted: bool = False,
                 **kwargs):
        # If no save patterns are set, specify a default for disk_pickled
        if 'save_patterns' not in kwargs:
            kwargs['save_patterns'] = {'pipeline': ['disk_pickled']}
        super(Pipeline, self).__init__(
            has_external_files=has_external_files, **kwargs)

        # Instantiate pipeline
        if transformers is None:
            transformers: List[Any] = []
        self._external_file = self._create_external_pipeline(transformers, **kwargs)
        # Initialize fit state -- pass as true to skip fitting transformers
        self.fitted = fitted
        # initialize null dataset reference
        self.dataset_id = None

    '''
    Persistable Management
    '''

    @property
    def fitted(self) -> bool:
        return self.state.get('fitted')

    @fitted.setter
    def fitted(self, value: bool) -> None:
        self.state['fitted'] = value

    @property
    def external_pipeline(self) -> Any:
        '''
        All pipeline objects are going to require some filebase persisted object

        Wrapper around whatever underlying class is desired
        (eg sklearn or native)
        '''
        self.load_if_unloaded('pipeline')
        return self._external_file

    def _create_external_pipeline(self, *args, **kwargs):
        '''
        each subclass should instantiate the respective pipeline library
        '''
        raise NotImplementedError('Must use a specific pipeline implementation')

    def add_dataset(self, dataset: 'Dataset') -> None:
        '''
        Setter method for dataset used
        '''
        if dataset is None:
            return
        self.dataset_id = dataset.id
        self.dataset = dataset

    @property
    def dataset(self):
        '''
        Use a weakref to bind linked dataset so it doesnt bloat usage
        returns dataset if still available or tries to fetch otherwise
        '''
        # still referenced weakref
        if hasattr(self, '_dataset') and self._dataset() is not None:
            return self._dataset()

        # null return if no associated dataset (governed by dataset_id)
        elif self.dataset_id is None:
            return None

        # else regenerate weakref
        LOGGER.info('No referenced object available. Refreshing weakref')
        dataset = self._load_dataset()
        self._dataset = weakref.ref(dataset)
        return dataset

    @dataset.setter
    def dataset(self, dataset: 'Dataset') -> None:
        '''
        Need to be careful not to set as the orm object
        Cannot load if wrong type because of recursive behavior (will
        propagate down the whole dependency chain)
        '''
        self._dataset = weakref.ref(dataset)

    def _load_dataset(self):
        '''
        Helper to fetch the dataset
        '''
        return self.orm_cls.load_dataset(self.dataset_id)

    def assert_dataset(self, msg: str = '') -> None:
        '''
        Helper method to raise an error if dataset isn't present
        '''
        if self.dataset is None:
            raise PipelineError(msg)

    def assert_fitted(self, msg: str = '') -> None:
        '''
        Helper method to raise an error if pipeline isn't fit
        '''
        if not self.fitted:
            raise PipelineError(msg)

    def add_transformer(self, name: str, transformer: Any) -> None:
        '''
        Setter method for new transformer step
        '''
        self.external_pipeline.add_transformer(name, transformer)
        # Need to refit now
        self.fitted = False

    def remove_transformer(self, name: str) -> None:
        '''
        Delete method for transformer step
        '''
        self.external_pipeline.remove_transformer(name)
        # Need to refit now
        self.fitted = False

    def _hash(self) -> str:
        '''
        Hash is the combination of the:
            1) Dataset
            2) Transformers
            3) Transformer Params
            4) Pipeline Config
        '''
        dataset_hash = self.dataset.hash_ or self.dataset._hash()
        transformers = self.get_transformers()
        transformer_params = self.get_params(params_only=True)
        pipeline_config = self.config

        return self.custom_hasher((dataset_hash, transformers, transformer_params, pipeline_config))

    def save(self, **kwargs) -> None:
        '''
        Extend parent function with a few additional save routines

        1) save params
        2) save transformer metadata
        3) features
        '''
        self.assert_dataset('Must set dataset before saving')
        self.assert_fitted('Must fit pipeline before saving')

        # log only attributes - can be refreshed on each save (does not take effect on reloading)
        self.params = self.get_params(params_only=True, **kwargs)
        self.metadata_['transformers'] = self.get_transformers()
        self.metadata_['feature_names'] = self.get_feature_names()

        # Skip file-based persistence if there are no transformers
        if not self.get_transformers():
            self.has_external_files = False

        super().save(**kwargs)

    def __post_restore__(self) -> None:
        '''
        Extend main load routine to load relationship class
        '''
        super().__post_restore__()

        # Create dummy pipeline if one wasnt saved
        if not self.has_external_files:
            self._external_file = self._create_external_pipeline([], **self.params)

    '''
    Data Accessors
    '''

    def split_dataset(self) -> None:
        '''
        Method to create a cached reference to the projected data (cant use dataset
        directly in case of mutation concerns)

        Non-split mixin class. Returns full dataset for any split name
        '''
        default_split = IdentityProjectedDatasetSplit(dataset=self.dataset, split=None)
        # use a single reference to avoid duplicating on different key searches
        self._dataset_splits = SplitContainer(
            default_factory=lambda: default_split
        )

    def get_dataset_split(self, split: Optional[str] = None) -> ProjectedDatasetSplit:
        '''
        Get specific dataset split
        Assumes a ProjectedDatasetSplit object (`simpleml.pipelines.projected_splits.ProjectedDatasetSplit`)
        is returned. Inherit or implement similar expected attributes to replace

        Uses internal `self._dataset_splits` as the split container - assumes
        dictionary like itemgetter
        '''
        if not hasattr(self, '_dataset_splits') or self._dataset_splits is None:
            self.split_dataset()

        return self._dataset_splits[split]

    def get_split_names(self) -> List[str]:
        if not hasattr(self, '_dataset_splits') or self._dataset_splits is None:
            self.split_dataset()
        return list(self._dataset_splits.keys())

    def X(self, split: Optional[str] = None) -> Any:
        '''
        Get X for specific dataset split
        '''
        return self.get_dataset_split(split=split).X

    def y(self, split: Optional[str] = None) -> Any:
        '''
        Get labels for specific dataset split
        '''
        return self.get_dataset_split(split=split).y

    def _filter_fit_params(self, split: ProjectedDatasetSplit) -> Dict[str, Any]:
        '''
        Helper to filter unsupported fit params from dataset splits
        '''
        supported_fit_params = {}

        # Ensure input compatibility with split object
        fit_params = inspect.signature(self.external_pipeline.fit).parameters
        # check if any params are **kwargs (all inputs accepted)
        has_kwarg_params = any([param.kind == param.VAR_KEYWORD for param in fit_params.values()])
        # log ignored args
        if not has_kwarg_params:
            for split_arg, val in split.items():
                if split_arg not in fit_params:
                    LOGGER.warning(f'Unsupported fit param encountered, `{split_arg}`. Dropping...')
                else:
                    supported_fit_params[split_arg] = val
        else:
            supported_fit_params = split

        return supported_fit_params

    def fit(self):
        '''
        Pass through method to external pipeline
        '''
        self.assert_dataset('Must set dataset before fitting')

        if self.fitted:
            LOGGER.warning('Cannot refit pipeline, skipping operation')
            return self

        # Only use default (train) fold to fit
        # No constraint on split -- can be a dataframe, ndarray, or generator
        # but must be encased in a Split object
        split = self.get_dataset_split(split=TRAIN_SPLIT)
        supported_fit_params = self._filter_fit_params(split)
        self.external_pipeline.fit(**supported_fit_params)
        self.fitted = True

        return self

    def transform(self, X: Any, **kwargs) -> Any:
        '''
        Main transform routine - routes to generator or regular method depending
        on the flag

        :param return_generator: boolean, whether to use the transformation method
        that returns a generator object or the regular transformed input
        :param return_sequence: boolean, whether to use method that returns a
        `keras.utils.sequence` object to play nice with keras models
        '''
        self.assert_fitted('Must fit pipeline before transforming')
        return self._transform(X, **kwargs)

    def _transform(self, X: Any, dataset_split: Optional[str] = None) -> Any:
        '''
        Pass through method to external pipeline

        :param X: dataframe/matrix to transform, if None, use internal dataset
        :rtype: Split object if no dataset passed (X is Null). Otherwise matrix
            return of input X
        '''
        if X is None:  # Retrieve dataset split
            split = self.get_dataset_split(split=dataset_split)
            if split.X is None or (isinstance(split.X, pd.DataFrame) and split.X.empty):
                output = None  # Skip transformations on empty dataset
            else:
                output = self.external_pipeline.transform(split.X)

            # Return input with X replaced by output (transformed X)
            # Contains y or other named inputs to propagate downstream
            return Split(X=output, **{k: v for k, v in split.items() if k != 'X'})

        return self.external_pipeline.transform(X)

    def fit_transform(self, **kwargs) -> Any:
        '''
        Wrapper for fit and transform methods
        ASSUMES only applies to default (train) split
        '''
        self.fit()
        return self.transform(X=None, **kwargs)

    '''
    Pass-through methods to external pipeline
    '''

    def get_params(self, **kwargs):
        '''
        Pass through method to external pipeline
        '''
        return self.external_pipeline.get_params(**kwargs)

    def set_params(self, **params):
        '''
        Pass through method to external pipeline
        '''
        return self.external_pipeline.set_params(**params)

    def get_transformers(self):
        '''
        Pass through method to external pipeline
        '''
        return self.external_pipeline.get_transformers()

    def get_feature_names(self) -> List[str]:
        '''
        Pass through method to external pipeline
        Should return a list of the final features generated by this pipeline
        '''
        initial_features = self.dataset.get_feature_names()
        return self.external_pipeline.get_feature_names(feature_names=initial_features)

1	'''
2	Base Module for Pipelines
3	'''
4
5	__author__ = 'Elisha Yadgaran'
6
7	import inspect
8	import logging
9	import weakref
10	from typing import TYPE_CHECKING, Any, Dict, Generator, List, Optional, Union
11
12	import pandas as pd
13	from simpleml.constants import TRAIN_SPLIT
14	from simpleml.datasets.dataset_splits import Split, SplitContainer
15	from simpleml.persistables.base_persistable import Persistable
16	from simpleml.registries import PipelineRegistry
17	from simpleml.save_patterns.decorators import ExternalArtifactDecorators
18	from simpleml.utils.errors import PipelineError
19
20	from .projected_splits import (IdentityProjectedDatasetSplit,
21	ProjectedDatasetSplit)
22
23	if TYPE_CHECKING:	6✔
24	# Cyclical import hack for type hints
25	from simpleml.datasets.base_dataset import Dataset
26
27
28	LOGGER = logging.getLogger(__name__)	6✔
29
30
31	@ExternalArtifactDecorators.register_artifact(	6✔
32	artifact_name='pipeline', save_attribute='external_pipeline', restore_attribute='_external_file')
33	class Pipeline(Persistable, metaclass=PipelineRegistry):	6✔
34	'''
35	Abstract Base class for all Pipelines objects.
36
37	Relies on mixin classes to define the split_dataset method. Will throw
38	an error on use otherwise
39
40	-------
41	Schema
42	-------
43	params: pipeline parameter metadata for easy insight into hyperparameters across trainings
44	'''
45	object_type: str = 'PIPELINE'	6✔
46
47	def __init__(self,	6✔
48	has_external_files: bool = True,
49	transformers: Optional[List[Any]] = None,
50	fitted: bool = False,
51	**kwargs):
52	# If no save patterns are set, specify a default for disk_pickled
53	if 'save_patterns' not in kwargs:	3!
54	kwargs['save_patterns'] = {'pipeline': ['disk_pickled']}	3✔
55	super(Pipeline, self).__init__(	3✔
56	has_external_files=has_external_files, **kwargs)
57
58	# Instantiate pipeline
59	if transformers is None:	3✔
60	transformers: List[Any] = []	3✔
61	self._external_file = self._create_external_pipeline(transformers, **kwargs)	3✔
62	# Initialize fit state -- pass as true to skip fitting transformers
63	self.fitted = fitted	3✔
64	# initialize null dataset reference
65	self.dataset_id = None	3✔
66
67	'''
68	Persistable Management
69	'''
70
71	@property	6✔
72	def fitted(self) -> bool:	6✔
73	return self.state.get('fitted')	×
74
75	@fitted.setter	6✔
76	def fitted(self, value: bool) -> None:	6✔
77	self.state['fitted'] = value	3✔
78
79	@property	6✔
80	def external_pipeline(self) -> Any:	6✔
81	'''
82	All pipeline objects are going to require some filebase persisted object
83
84	Wrapper around whatever underlying class is desired
85	(eg sklearn or native)
86	'''
87	self.load_if_unloaded('pipeline')	×
88	return self._external_file	×
89
90	def _create_external_pipeline(self, args, *kwargs):	6✔
91	'''
92	each subclass should instantiate the respective pipeline library
93	'''
94	raise NotImplementedError('Must use a specific pipeline implementation')
95
96	def add_dataset(self, dataset: 'Dataset') -> None:	6✔
97	'''
98	Setter method for dataset used
99	'''
100	if dataset is None:	3!
NEW 101	return	×
102	self.dataset_id = dataset.id	3✔
103	self.dataset = dataset	3✔
104
105	@property	6✔
106	def dataset(self):	4✔
107	'''
108	Use a weakref to bind linked dataset so it doesnt bloat usage
109	returns dataset if still available or tries to fetch otherwise
110	'''
111	# still referenced weakref
112	if hasattr(self, '_dataset') and self._dataset() is not None:	3!
113	return self._dataset()	3✔
114
115	# null return if no associated dataset (governed by dataset_id)
NEW 116	elif self.dataset_id is None:	×
NEW 117	return None	×
118
119	# else regenerate weakref
NEW 120	LOGGER.info('No referenced object available. Refreshing weakref')	×
NEW 121	dataset = self._load_dataset()	×
NEW 122	self._dataset = weakref.ref(dataset)	×
NEW 123	return dataset	×
124
125	@dataset.setter	6✔
126	def dataset(self, dataset: 'Dataset') -> None:	6✔
127	'''
128	Need to be careful not to set as the orm object
129	Cannot load if wrong type because of recursive behavior (will
130	propagate down the whole dependency chain)
131	'''
132	self._dataset = weakref.ref(dataset)	3✔
133
134	def _load_dataset(self):	6✔
135	'''
136	Helper to fetch the dataset
137	'''
NEW 138	return self.orm_cls.load_dataset(self.dataset_id)	×
139
140	def assert_dataset(self, msg: str = '') -> None:	6✔
141	'''
142	Helper method to raise an error if dataset isn't present
143	'''
144	if self.dataset is None:	×
145	raise PipelineError(msg)	×
146
147	def assert_fitted(self, msg: str = '') -> None:	6✔
148	'''
149	Helper method to raise an error if pipeline isn't fit
150	'''
151	if not self.fitted:	×
152	raise PipelineError(msg)	×
153
154	def add_transformer(self, name: str, transformer: Any) -> None:	6✔
155	'''
156	Setter method for new transformer step
157	'''
158	self.external_pipeline.add_transformer(name, transformer)	×
159	# Need to refit now
160	self.fitted = False	×
161
162	def remove_transformer(self, name: str) -> None:	6✔
163	'''
164	Delete method for transformer step
165	'''
166	self.external_pipeline.remove_transformer(name)	×
167	# Need to refit now
168	self.fitted = False	×
169
170	def _hash(self) -> str:	6✔
171	'''
172	Hash is the combination of the:
173	1) Dataset
174	2) Transformers
175	3) Transformer Params
176	4) Pipeline Config
177	'''
178	dataset_hash = self.dataset.hash_ or self.dataset._hash()	×
179	transformers = self.get_transformers()	×
180	transformer_params = self.get_params(params_only=True)	×
181	pipeline_config = self.config	×
182
183	return self.custom_hasher((dataset_hash, transformers, transformer_params, pipeline_config))	×
184
185	def save(self, **kwargs) -> None:	6✔
186	'''
187	Extend parent function with a few additional save routines
188
189	1) save params
190	2) save transformer metadata
191	3) features
192	'''
193	self.assert_dataset('Must set dataset before saving')	×
194	self.assert_fitted('Must fit pipeline before saving')	×
195
196	# log only attributes - can be refreshed on each save (does not take effect on reloading)
197	self.params = self.get_params(params_only=True, **kwargs)	×
198	self.metadata_['transformers'] = self.get_transformers()	×
199	self.metadata_['feature_names'] = self.get_feature_names()	×
200
201	# Skip file-based persistence if there are no transformers
202	if not self.get_transformers():	×
203	self.has_external_files = False	×
204
NEW 205	super().save(**kwargs)	×
206
207	def __post_restore__(self) -> None:	6✔
208	'''
209	Extend main load routine to load relationship class
210	'''
NEW 211	super().__post_restore__()	×
212
213	# Create dummy pipeline if one wasnt saved
214	if not self.has_external_files:	×
215	self._external_file = self._create_external_pipeline([], **self.params)	×
216
217	'''
218	Data Accessors
219	'''
220
221	def split_dataset(self) -> None:	6✔
222	'''
223	Method to create a cached reference to the projected data (cant use dataset
224	directly in case of mutation concerns)
225
226	Non-split mixin class. Returns full dataset for any split name
227	'''
228	default_split = IdentityProjectedDatasetSplit(dataset=self.dataset, split=None)	3✔
229	# use a single reference to avoid duplicating on different key searches
230	self._dataset_splits = SplitContainer(	3✔
231	default_factory=lambda: default_split
232	)
233
234	def get_dataset_split(self, split: Optional[str] = None) -> ProjectedDatasetSplit:	6✔
235	'''
236	Get specific dataset split
237	Assumes a ProjectedDatasetSplit object (`simpleml.pipelines.projected_splits.ProjectedDatasetSplit`)
238	is returned. Inherit or implement similar expected attributes to replace
239
240	Uses internal `self._dataset_splits` as the split container - assumes
241	dictionary like itemgetter
242	'''
243	if not hasattr(self, '_dataset_splits') or self._dataset_splits is None:	3✔
244	self.split_dataset()	3✔
245
246	return self._dataset_splits[split]	3✔
247
248	def get_split_names(self) -> List[str]:	6✔
249	if not hasattr(self, '_dataset_splits') or self._dataset_splits is None:	×
250	self.split_dataset()	×
251	return list(self._dataset_splits.keys())	×
252
253	def X(self, split: Optional[str] = None) -> Any:	6✔
254	'''
255	Get X for specific dataset split
256	'''
257	return self.get_dataset_split(split=split).X	×
258
259	def y(self, split: Optional[str] = None) -> Any:	6✔
260	'''
261	Get labels for specific dataset split
262	'''
263	return self.get_dataset_split(split=split).y	×
264
265	def _filter_fit_params(self, split: ProjectedDatasetSplit) -> Dict[str, Any]:	6✔
266	'''
267	Helper to filter unsupported fit params from dataset splits
268	'''
269	supported_fit_params = {}	×
270
271	# Ensure input compatibility with split object
272	fit_params = inspect.signature(self.external_pipeline.fit).parameters	×
273	# check if any params are **kwargs (all inputs accepted)
274	has_kwarg_params = any([param.kind == param.VAR_KEYWORD for param in fit_params.values()])	×
275	# log ignored args
276	if not has_kwarg_params:	×
277	for split_arg, val in split.items():	×
278	if split_arg not in fit_params:	×
279	LOGGER.warning(f'Unsupported fit param encountered, `{split_arg}`. Dropping...')	×
280	else:
281	supported_fit_params[split_arg] = val	×
282	else:
283	supported_fit_params = split	×
284
285	return supported_fit_params	×
286
287	def fit(self):	6✔
288	'''
289	Pass through method to external pipeline
290	'''
291	self.assert_dataset('Must set dataset before fitting')	×
292
293	if self.fitted:	×
294	LOGGER.warning('Cannot refit pipeline, skipping operation')	×
295	return self	×
296
297	# Only use default (train) fold to fit
298	# No constraint on split -- can be a dataframe, ndarray, or generator
299	# but must be encased in a Split object
300	split = self.get_dataset_split(split=TRAIN_SPLIT)	×
301	supported_fit_params = self._filter_fit_params(split)	×
302	self.external_pipeline.fit(**supported_fit_params)	×
303	self.fitted = True	×
304
305	return self	×
306
307	def transform(self, X: Any, **kwargs) -> Any:	6✔
308	'''
309	Main transform routine - routes to generator or regular method depending
310	on the flag
311
312	:param return_generator: boolean, whether to use the transformation method
313	that returns a generator object or the regular transformed input
314	:param return_sequence: boolean, whether to use method that returns a
315	`keras.utils.sequence` object to play nice with keras models
316	'''
317	self.assert_fitted('Must fit pipeline before transforming')	×
318	return self._transform(X, **kwargs)	×
319
320	def _transform(self, X: Any, dataset_split: Optional[str] = None) -> Any:	6✔
321	'''
322	Pass through method to external pipeline
323
324	:param X: dataframe/matrix to transform, if None, use internal dataset
325	:rtype: Split object if no dataset passed (X is Null). Otherwise matrix
326	return of input X
327	'''
328	if X is None: # Retrieve dataset split	×
329	split = self.get_dataset_split(split=dataset_split)	×
330	if split.X is None or (isinstance(split.X, pd.DataFrame) and split.X.empty):	×
331	output = None # Skip transformations on empty dataset	×
332	else:
333	output = self.external_pipeline.transform(split.X)	×
334
335	# Return input with X replaced by output (transformed X)
336	# Contains y or other named inputs to propagate downstream
337	return Split(X=output, **{k: v for k, v in split.items() if k != 'X'})	×
338
339	return self.external_pipeline.transform(X)	×
340
341	def fit_transform(self, **kwargs) -> Any:	6✔
342	'''
343	Wrapper for fit and transform methods
344	ASSUMES only applies to default (train) split
345	'''
346	self.fit()	×
347	return self.transform(X=None, **kwargs)	×
348
349	'''
350	Pass-through methods to external pipeline
351	'''
352
353	def get_params(self, **kwargs):	6✔
354	'''
355	Pass through method to external pipeline
356	'''
357	return self.external_pipeline.get_params(**kwargs)	×
358
359	def set_params(self, **params):	6✔
360	'''
361	Pass through method to external pipeline
362	'''
363	return self.external_pipeline.set_params(**params)	×
364
365	def get_transformers(self):	6✔
366	'''
367	Pass through method to external pipeline
368	'''
369	return self.external_pipeline.get_transformers()	×
370
371	def get_feature_names(self) -> List[str]:	6✔
372	'''
373	Pass through method to external pipeline
374	Should return a list of the final features generated by this pipeline
375	'''
376	initial_features = self.dataset.get_feature_names()	×
377	return self.external_pipeline.get_feature_names(feature_names=initial_features)	×

eyadgaran / SimpleML / 2010037515

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous