Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
..
Thanks to Umaneo Technologies Inc. for their contributions to this Machine Learning
project, visit https://www.umaneo.com/ for more information on Umaneo Technologies Inc.
"""
from neuraxle.base import MetaStepMixin, BaseStep
class MetaSKLearnWrapper(MetaStepMixin, BaseStep):
def __init__(self, wrapped):
"""
Wrap a scikit-learn MetaEstimatorMixin for usage in Neuraxle.
This class is similar to the SKLearnWrapper class of Neuraxle that can wrap a scikit-learn BaseEstimator.
:param wrapped: a scikit-learn object of type "MetaEstimatorMixin".
"""
MetaStepMixin.__init__(self)
BaseStep.__init__(self)
self.wrapped_sklearn_metaestimator = wrapped # TODO: use self.set_step of the MetaStepMixin instead?
# sklearn.model_selection.RandomizedSearchCV
def fit(self, data_inputs, expected_outputs=None) -> 'BaseStep':
self.wrapped_sklearn_metaestimator = self.wrapped_sklearn_metaestimator.fit(data_inputs, expected_outputs)
return self
np.expand_dims(np.array(data_inputs)[:, i], axis=-1)
for i in self.column_selection
]
return np.concatenate(columns, axis=-1)
if self.column_selection is None:
return data_inputs
raise ValueError(
'column selection type not supported : {0}\nSupported types'.format(
self.column_selection,
repr(ColumnSelectionType)
))
class ColumnsSelectorND(MetaStepMixin, BaseStep):
"""
ColumnSelectorND wraps a ColumnSelector2D by as many ForEachDataInput step
as needed to select the last dimension.
"""
def __init__(self, columns_selection, n_dimension=3):
BaseStep.__init__(self)
col_selector = ColumnSelector2D(columns_selection=columns_selection)
for _ in range(min(0, n_dimension - 2)):
col_selector = ForEachDataInput(col_selector)
MetaStepMixin.__init__(self, col_selector)
self.n_dimension = n_dimension
def fit(self, data_inputs, expected_outputs=None) -> 'BaseStep':
def __init__(self, scoring_function=r2_score, joiner=NumpyConcatenateOuterBatch()):
MetaStepMixin.__init__(self)
BaseStep.__init__(self)
self.scoring_function = scoring_function
self.joiner = joiner
def __init__(self, wrapped: BaseStep):
ForceHandleMixin.__init__(self)
MetaStepMixin.__init__(self, wrapped=wrapped)
BaseStep.__init__(self)
from neuraxle.base import MetaStepMixin, BaseStep, ExecutionContext, DataContainer, NonTransformableMixin, \
ExecutionMode, NonFittableMixin, ForceHandleMixin
class TransformOnlyWrapper(
NonTransformableMixin,
NonFittableMixin,
MetaStepMixin,
BaseStep
):
"""
A wrapper step that makes its wrapped step only executes in the transform execution mode.
.. seealso:: :class:`ExecutionMode`,
:class:`neuraxle.base.DataContainer`,
:class:`neuraxle.base.NonTransformableMixin`,
:class:`neuraxle.base.NonFittableMixin`,
:class:`neuraxle.base.MetaStepMixin`,
:class:`neuraxle.base.BaseStep`
"""
def __init__(self, wrapped: BaseStep):
NonTransformableMixin.__init__(self)
NonFittableMixin.__init__(self)
def __init__(
self,
wrapped = None,
n_iter: int = 10,
higher_score_is_better: bool = True,
validation_technique: BaseCrossValidation = KFoldCrossValidation(),
refit=True,
):
if wrapped is not None:
MetaStepMixin.__init__(self, wrapped)
BaseStep.__init__(self)
self.n_iter = n_iter
self.higher_score_is_better = higher_score_is_better
self.validation_technique: BaseCrossValidation = validation_technique
self.refit = refit
def set_hyperparams(self, hyperparams: HyperparameterSamples) -> BaseStep:
MetaStepMixin.set_hyperparams(self, hyperparams)
self.steps = [s.set_hyperparams(self.wrapped.get_hyperparams()) for s in self.steps]
return self
:param data_inputs: (data inputs, expected outputs) tuple to shuffle
:return:
"""
if self.increment_seed_after_each_fit:
self.seed += 1
di, eo = data_inputs
data = list(zip(di, eo))
random.Random(self.seed).shuffle(data)
data_inputs_shuffled, expected_outputs_shuffled = list(zip(*data))
return list(data_inputs_shuffled), list(expected_outputs_shuffled)
class EpochRepeater(MetaStepMixin, BaseStep):
"""
Repeat wrapped step fit, or transform for the number of epochs passed in the constructor.
.. code-block:: python
p = Pipeline([
TrainOnlyWrapper(DataShuffler(seed=42, increment_seed_after_each_fit=True, increment_seed_after_each_fit=False)),
EpochRepeater(ForecastingPipeline(), epochs=EPOCHS, repeat_in_test_mode=False)
])
.. seealso::
:class:`DataShuffler`,
:class:`MetaStepMixin`,
:class:`TrainOnlyWrapper`,
:class:`TestOnlyWrapper`,
:class:`BaseStep`
def __init__(self, columns_selection, n_dimension=3):
BaseStep.__init__(self)
col_selector = ColumnSelector2D(columns_selection=columns_selection)
for _ in range(min(0, n_dimension - 2)):
col_selector = ForEachDataInput(col_selector)
MetaStepMixin.__init__(self, col_selector)
self.n_dimension = n_dimension