Merge pull request #525 from Kimoby/updates-for-dashboard

AutoML Results Visualizations Update

Author: guillaume-chevalier

.coveragerc

@@ -0,0 +1,13 @@
+[run]
+parallel = True
+concurrency = multiprocessing, thread
+
+[report]
+exclude_lines =
+    raise NotImplementedError.*
+    if False:
+    if 0:
+    raise AssertionError.*
+    @(abc\.)?abstractmethod
+    if __name__ == __main__:
+

.github/workflows/testpythonpackage.yml

@@ -35,9 +35,9 @@ jobs:
       run: |
         pip install flake8
         # stop the build if there are Python syntax errors or undefined names
-        flake8 neuraxle testing --count --select=E9,F63,F7,F82 --show-source --statistics
+        flake8 neuraxle testing_neuraxle --count --select=E9,F63,F7,F82 --show-source --statistics
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 neuraxle testing --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+        flake8 neuraxle testing_neuraxle --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Test with pytest
       run: |
         python setup.py test

.gitignore

@@ -109,8 +109,8 @@ venv.bak/
 .idea
 .vscode
 .style.yapf
-vandelay-py.js
-appmap.yml
+*-py.js
+*pmap.yml
 tmp
 
 # Other
@@ -121,9 +121,9 @@ todo.txt
 **caching/**
 cache/**
 caching/**
-testing/examples/cache/**
-testing/cache/**
-testing/cache/*
+testing_neuraxle/examples/cache/**
+testing_neuraxle/cache/**
+testing_neuraxle/cache/*
 cov.xml
 profile.sh
 

coverage.sh

@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
 ./flake8.sh
-pytest -n 7 --cov-report html --cov-report xml:cov.xml --cov=neuraxle testing
-# pytest --cov-report html --cov=neuraxle testing; open htmlcov/index.html
+pytest -n 7 --cov-report html --cov-report xml:cov.xml --cov-config=.coveragerc --cov=neuraxle testing_neuraxle
+# pytest --cov-report html --cov=neuraxle testing_neuraxle; open htmlcov/index.html
 

examples/Handler Methods.ipynb

@@ -42,7 +42,7 @@
     "### [handle_transform](https://www.neuraxle.org/stable/api/neuraxle.base.html#neuraxle.base._TransformerStep.handle_transform)\n",
     "\n",
     "1. [\\_will\\_process(data_container, context)](https://www.neuraxle.org/stable/api/neuraxle.base.html#neuraxle.base._TransformerStep._will_process): Apply side effects before any step method\n",
-    "2. [\\_will\\_transform(data_container, context)](https://www.neuraxle.org/stable/api/neuraxle.base.html#neuraxle.base._TransformerStep._will_transform_data_container): Apply side effects before transform.\n",
+    "2. [\\_will\\_transform(data_container, context)](https://www.neuraxle.org/stable/api/neuraxle.base.html#neuraxle.base._TransformerStep._will_transform): Apply side effects before transform.\n",
     "3. [\\_transform\\_data_container(data_container, context)](https://www.neuraxle.org/stable/api/neuraxle.base.html#neuraxle.base._TransformerStep._transform_data_container): Fit transform data container.\n",
     "4. [\\_did\\_transform(data_container, context)](https://www.neuraxle.org/stable/api/neuraxle.base.html#neuraxle.base._TransformerStep._did_transform): Apply side effects after transform. \n",
     "5. [\\_did\\_process(data_container, context)](https://www.neuraxle.org/stable/api/neuraxle.base.html#neuraxle.base._TransformerStep._did_process): Apply side effects after any step method.\n",

examples/Step Saving And Lifecycle.ipynb

@@ -66,20 +66,31 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from neuraxle.base import BaseSaver, BaseStep, ExecutionContext, Identity\n",
-    "from queue import Queue\n",
+    "from multiprocessing import Queue\n",
+    "from neuraxle.base import BaseSaver, BaseTransformer, ExecutionContext, Identity\n",
+    "from neuraxle.base import ExecutionContext as CX\n",
+    "from neuraxle.distributed.streaming import _ProducerConsumerMixin\n",
     "\n",
-    "class ObservableQueueStepSaver(BaseSaver):\n",
-    "    def save_step(self, step: 'BaseStep', context: 'ExecutionContext') -> 'BaseStep':\n",
-    "        step.queue = None\n",
-    "        step.observers = []\n",
+    "class _ProducerConsumerStepSaver(BaseSaver):\n",
+    "    \"\"\"\n",
+    "    Saver for :class:`_ProducerConsumerMixin`.\n",
+    "    This saver class makes sure that the non-picklable queue\n",
+    "    is deleted upon saving for multiprocessing steps.\n",
+    "    \"\"\"\n",
+    "\n",
+    "    def save_step(self, step: BaseTransformer, context: 'CX') -> BaseTransformer:\n",
+    "        step: _ProducerConsumerMixin = step  # typing.\n",
+    "        step._allow_exit_without_queue_flush()\n",
+    "        step.input_queue = None\n",
+    "        step.consumers = []\n",
     "        return step\n",
     "\n",
-    "    def can_load(self, step: 'BaseStep', context: 'ExecutionContext'):\n",
+    "    def can_load(self, step: BaseTransformer, context: 'CX') -> bool:\n",
     "        return True\n",
     "\n",
-    "    def load_step(self, step: 'BaseStep', context: 'ExecutionContext') -> 'BaseStep':\n",
-    "        step.queue = Queue()\n",
+    "    def load_step(self, step: 'BaseTransformer', context: 'CX') -> 'BaseTransformer':\n",
+    "        step: _ProducerConsumerMixin = step  # typing.\n",
+    "        step.input_queue = None\n",
     "        return step"
    ]
   },
@@ -98,7 +109,7 @@
    "source": [
     "class IdentityWithQueue(Identity):\n",
     "    def __init__(self):\n",
-    "        super().__init__(savers=[ObservableQueueStepSaver()])\n",
+    "        super().__init__(savers=[_ProducerConsumerStepSaver()])\n",
     "\n",
     "    def setup(self, context=None):\n",
     "        if not self.is_initialized:\n",

examples/parallel/plot_streaming_pipeline.py

@@ -71,22 +71,22 @@ def main():
     time_vanilla_pipeline, output_classical = eval_run_time(p)
     print(f"Classical 'Pipeline' execution time: {time_vanilla_pipeline} seconds.")
 
-    # Classical minibatch pipeline - minibatch size 25:
+    # Classical minibatch pipeline - minibatch size 5:
     p = MiniBatchSequentialPipeline(preprocessing_and_model_steps,
-                                    batch_size=25)
+                                    batch_size=5)
     time_minibatch_pipeline, output_minibatch = eval_run_time(p)
     print(f"Minibatched 'MiniBatchSequentialPipeline' execution time: {time_minibatch_pipeline} seconds.")
 
-    # Parallel pipeline - minibatch size 25 with 4 parallel workers per step that
+    # Parallel pipeline - minibatch size 5 with 4 parallel workers per step that
     # have a max queue size of 10 batches between preprocessing and the model:
     p = SequentialQueuedPipeline(preprocessing_and_model_steps,
-                                 n_workers_per_step=4, max_queue_size=10, batch_size=25)
+                                 n_workers_per_step=4, max_queued_minibatches=10, batch_size=5)
     time_parallel_pipeline, output_parallel = eval_run_time(p)
     print(f"Parallel 'SequentialQueuedPipeline' execution time: {time_parallel_pipeline} seconds.")
 
-    assert time_parallel_pipeline < time_minibatch_pipeline, str((time_parallel_pipeline, time_vanilla_pipeline))
     assert np.array_equal(output_classical, output_minibatch)
     assert np.array_equal(output_classical, output_parallel)
+    assert time_parallel_pipeline < time_minibatch_pipeline, str((time_parallel_pipeline, time_vanilla_pipeline))
 
 
 if __name__ == '__main__':

flake8.sh

@@ -1,3 +1,3 @@
 #!/usr/bin/env bash
-flake8 neuraxle testing --count --max-line-length=120 --select=E9,F63,F7,F82 --statistics --show-source
+flake8 neuraxle testing_neuraxle --count --max-line-length=120 --select=E9,F63,F7,F82 --statistics --show-source
 

neuraxle/__init__.py

@@ -1 +1 @@
-__version__ = "0.7.1"
+__version__ = "0.7.2"