Refactor modifier

.env-example

@@ -13,3 +13,4 @@ RQ_QUEUES="default"
 
 D3M_DB_SUBMITTER=submitter_name
 D3M_DB_TOKEN=token
+SAVE_TO_D3M=false

docker-compose.yml

@@ -7,14 +7,22 @@ services:
     volumes:
       - type: bind
         source: '${DATA_DIR}/${REDIS_DATA_DIR}'
-        target: '/data'
+        target: /data
     networks:
       - default
 
   rq_worker:
     image: 'd3m-experimenter:latest'
     env_file:
       - ./.env
+    volumes:
+      - type: bind
+        source: '${DATASETS_DIR}'
+        target: /datasets
+        read_only: true
+      - type: bind
+        source: '${DATA_DIR}'
+        target: /data
     command: 'rq worker --url redis://${REDIS_HOST} ${RQ_QUEUES}'
     networks:
       - default

experimenter/cli.py

@@ -1,7 +1,9 @@
 import argparse
 import typing
 
-from experimenter import exceptions, queue
+
+from experimenter.modify_generator import ModifyGenerator
+from experimenter import config, exceptions, queue
 
 
 def main(argv: typing.Sequence) -> None:
@@ -21,13 +23,21 @@ def configure_parser(parser: argparse.ArgumentParser) -> None:
     )
     configure_queue_parser(queue_parser)
 
+    generator_parser = subparsers.add_parser(
+        'generator',
+        description='generates new pipelines and queues them to run on available datasets',
+    )
+    configure_generator_parser(generator_parser) 
+
 
 def handler(arguments: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
     experimenter_command = arguments.experimenter_command
     subparser = parser._subparsers._group_actions[0].choices[experimenter_command]  # type: ignore
 
     if experimenter_command == 'queue':
         queue_handler(arguments, subparser)
+    elif experimenter_command == 'generator':
+        generator_handler(arguments, subparser)
     else:
         raise exceptions.InvalidStateError('Unknown experimenter command: {}'.format(experimenter_command))
 
@@ -40,6 +50,12 @@ def configure_queue_parser(parser: argparse.ArgumentParser) -> None:
 
     empty_parser = subparsers.add_parser('empty', help='remove all jobs from a queue')
     empty_parser.add_argument('-q', '--queue-name', help='the name of the queue to empty')
+    empty_parser.add_argument('-f', '--failed', help='remove the failed queue', action='store_true')
+
+    #save a failed traceback parser
+    save_failed_parser = subparsers.add_parser('save-failed', help='save failed job error output')
+    save_failed_parser.add_argument('-q', '--queue-name', help='the name of the queue to empty')
+    save_failed_parser.add_argument('-j', '--job-num', type=int, default=0, help='the failed job number')
 
 
 def queue_handler(arguments: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
@@ -48,6 +64,127 @@ def queue_handler(arguments: argparse.Namespace, parser: argparse.ArgumentParser
     if queue_command == 'status':
         queue.status()
     elif queue_command == 'empty':
-        queue.empty(arguments.queue_name)
+        queue.empty(arguments.queue_name, arguments.failed)
+    elif queue_command == 'save-failed':
+        queue.save_failed_job(arguments.queue_name, arguments.job_num)
     else:
         raise exceptions.InvalidStateError('Unknown queue command: {}'.format(queue_command))
+
+
+def configure_generator_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument('-j', '--max-jobs', type=int, default=None, action='store', help='maximum number of jobs generated')
+    parser.add_argument('-t', '--job-timeout', type=int, default=None, action='store', help='maximum runtime for a single job in seconds')
+
+    subparsers = parser.add_subparsers(dest='generator_command')
+    subparsers.required = True  # type: ignore
+
+    search_subparser = subparsers.add_parser(
+        'search',
+        help='searches for new pipelines not found in the metalearning database',
+    )
+    configure_search_parser(search_subparser)
+
+    modify_subparser = subparsers.add_parser(
+        'modify',
+        help='modifies existing pipelines in the metalearning database',
+    )
+    configure_modify_parser(modify_subparser)
+
+    update_subparser = subparsers.add_parser(
+        'update',
+        help='updates existing pipeline runs in the metalearning database to use the current versions of datasets and primitives',
+    )
+    configure_update_parser(update_subparser)
+
+
+def generator_handler(arguments: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+    generator_command = arguments.generator_command
+    subparser = parser._subparsers._group_actions[0].choices[generator_command]  # type: ignore
+
+    if generator_command == 'search':
+        search_handler(arguments, subparser)
+    elif generator_command == 'modify':
+        modify_handler(arguments, subparser)
+    elif generator_command == 'update':
+        update_handler(arguments, subparser)
+    else:
+        raise exceptions.InvalidStateError('Unknown queue command: {}'.format(generator_command))
+
+
+def configure_search_parser(parser: argparse.ArgumentParser) -> None:
+    pass
+
+
+def search_handler(arguments: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+    raise exceptions.NotImplementedError()
+
+
+def configure_modify_parser(parser: argparse.ArgumentParser) -> None:
+    #create the subparsers for the different types of modifications
+
+    #seed swapper functionality
+    subparsers = parser.add_subparsers(dest='modify_type')
+    subparsers.required = True
+    swap_seed_subparser = subparsers.add_parser(
+         'random-seed',
+         description='Uses database data to search pipelines and run functional pipelines on different random seeds',
+     )
+    #subparser arguments
+    swap_seed_subparser.add_argument(
+         '--pipeline_id',
+         help='The pipeline id to search for in the query, if none, searches all pipelines',
+         default=None,
+         type=str)
+    swap_seed_subparser.add_argument(
+         '--submitter',
+         help='The pipeline submitter to add to the query',
+         default=None,
+         type=str)
+    swap_seed_subparser.add_argument(
+         '--seed-limit',
+         help='The amount of random seeds that each ran pipeline will have at the end of the test',
+         default=2,
+         type=int)
+    swap_seed_subparser.add_argument(
+         '--test',
+         help='run the test instead of random pipeline generation',
+         action='store_true')
+
+    #Primitive swapper functionality
+    primitive_swap_subparser = subparsers.add_parser(
+        'primitive-swap',
+        description='Searches database for pipeline runs containing a primitive and swaps out primitive for a different given primitive')
+    #subparser arguments
+    primitive_swap_subparser.add_argument(
+         '--primitive_id',
+         help='The id of the primitive to swap out',
+         default=None,
+         type=str)
+    primitive_swap_subparser.add_argument(
+         '--limit_indeces',
+         help='Details for primitive swapping',
+         default=None)
+    primitive_swap_subparser.add_argument(
+         '--swap_primitive_id',
+         help='The id of the primitve to swap in',
+         default=None,
+         type=str)
+
+
+def modify_handler(arguments: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+    modify_type = arguments.modify_type
+    modify_generator = ModifyGenerator(modify_type = modify_type, 
+                                       max_jobs = arguments.max_jobs, 
+                                       seed_limit = arguments.seed_limit, 
+                                       submitter = arguments.submitter,
+                                       pipeline_id = arguments.pipeline_id)
+    #now run the enqueuer part
+    queue.enqueue_jobs(jobs=modify_generator, job_timeout=arguments.job_timeout)
+
+
+def configure_update_parser(parser: argparse.ArgumentParser) -> None:
+    pass
+
+
+def update_handler(arguments: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+    raise exceptions.NotImplementedError()

experimenter/config.py

@@ -10,6 +10,7 @@
 _ERROR_MESSAGE = 'environment variable not set: {}'
 
 
+#parse the .env file
 datasets_dir: str = os.environ.get('DATASETS_DIR', None)
 def validate_datasets_dir():
     if datasets_dir is None:
@@ -27,7 +28,7 @@ def validate_data_dir():
 def validate_redis_host():
     if redis_host is None:
         raise exceptions.ConfigError(_ERROR_MESSAGE.format('REDIS_HOST'))
-
+        
 
 d3m_db_submitter: str = os.environ.get('D3M_DB_SUBMITTER', None)
 def validate_d3m_db_submitter():
@@ -39,3 +40,46 @@ def validate_d3m_db_submitter():
 def validate_d3m_db_token():
     if d3m_db_token is None:
         raise exceptions.ConfigError(_ERROR_MESSAGE.format('D3M_DB_TOKEN'))
+
+
+save_to_d3m: bool = os.environ.get('SAVE_TO_D3M', None) == 'true'
+def validate_save():
+    if save_to_d3m is None:
+        raise exceptions.ConfigError(_ERROR_MESSAGE.format('SAVE_TO_D3M'))
+
+
+query_host: str = os.environ.get('QUERY_HOST', 'https://metalearning.datadrivendiscovery.org/es')
+def validate_query_host():
+    if query_host is None:
+        raise exceptions.ConfigError(_ERROR_MESSAGE.format('QUERY_HOST'))
+
+
+query_timeout: int = int(os.environ.get('QUERY_TIMEOUT', '500'))
+def validate_query_timeout():
+    if query_timeout is None:
+        raise exceptions.ConfigError(_ERROR_MESSAGE.format('QUERY_TIMEOUT'))
+
+
+#get the save paths for the experimenter from the point of view of the docker container
+output_run_path: str = os.path.abspath(os.path.join('/data', 'pipeline_runs'))
+if (not os.path.exists(output_run_path)):
+    #create the directory
+    os.makedirs(output_run_path, exist_ok=True)
+
+
+pipelines_path: str = os.path.abspath(os.path.join('/data', 'pipelines'))
+if (not os.path.exists(pipelines_path)):
+    #create the directory
+    os.makedirs(pipelines_path, exist_ok=True)
+
+
+data_prep_pipelines_path: str = os.path.abspath(os.path.join('/data', 'data-preparation-pipelines'))
+if (not os.path.exists(data_prep_pipelines_path)):
+    #create the directory
+    os.makedirs(data_prep_pipelines_path, exist_ok=True)
+
+
+scoring_pipelines_path: str = os.path.abspath(os.path.join('/data', 'scoring-pipelines'))
+if (not os.path.exists(scoring_pipelines_path)):
+    #create the directory
+    os.makedirs(scoring_pipelines_path, exist_ok=True)

experimenter/databases/d3m_mtl.py

@@ -1,5 +1,6 @@
 import logging
 import json
+import yaml
 
 import requests
 from d3m.primitive_interfaces.base import PrimitiveBase
@@ -23,17 +24,17 @@ def __init__(self) -> None:
         self._post_url = D3M_MTL_DB_POST_URL
         # This env var allows code calling this class to be run during
         # unit tests without actually saving to the production DB.
-        self.should_save = config.SAVE_TO_D3M
+        self.should_save = config.save_to_d3m
         # A reference to a low-level elasticsearch client. This can be
         # used to query the D3M DB, or this classe's `search` method
         # can be used, and is preferred, since its API is more straightforward.
         # This low-level client is the only way to accomplish
         # certain things though.
         self.es = Elasticsearch(hosts=[D3M_MTL_DB_GET_URL], timeout=30)
         # Our submitter name.
-        self._submitter = config.D3M_DB_SUBMITTER
+        self._submitter = config.d3m_db_submitter
         # The secret verifying us as the submitter we say we are.
-        self._x_token = config.D3M_DB_TOKEN
+        self._x_token = config.d3m_db_token
         if self._is_identifying_as_submitter():
             logger.info(
                 f"Documents will be saved under submitter name: '{self._submitter}'"
@@ -105,7 +106,15 @@ def does_pipeline_exist_in_db(self, pipeline: Pipeline) -> bool:
             .count()
         )
         return num_pipeline_matches > 0
-
+
+    def save_pipeline_runs_from_path(self, pipeline_run_path: str) -> requests.Response:
+        responses = list()
+        with open(pipeline_run_path, 'r') as pipeline_data:
+            pipeline_runs = yaml.safe_load_all(pipeline_data)
+            for pipeline_run in pipeline_runs:
+                responses.append(self.save_pipeline_run(pipeline_run).content)
+        return responses
+
     def save_pipeline_run(self, pipeline_run: dict) -> requests.Response:
         return self._save(pipeline_run, "pipeline-run")
 
@@ -156,7 +165,7 @@ def _create_no_save_response(self) -> requests.Response:
         response.status_code = 200
         response._content = (
             b'{ "result" : "No request was made to the D3M DB API to save a record, '
-            b'since the SAVE_TO_D3M environment variable is not set." }'
+            b'since the SAVE_TO_D3M environment variable is not set to true." }'
         )
         return response