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,26 @@ 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
+      - type: bind
+        source: '${EXPERIMENTER_DIR}'
+        target: /d3m-experimenter
+        read_only: true
     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,123 @@ 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, arguments.max_jobs, arguments)
+    #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

@@ -27,7 +27,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 +39,10 @@ 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'))
+

experimenter/databases/d3m_mtl.py

@@ -23,17 +23,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.D3MConfig().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.D3MConfig().d3m_submitter
         # The secret verifying us as the submitter we say we are.
-        self._x_token = config.D3M_DB_TOKEN
+        self._x_token = config.D3MConfig().d3m_token
         if self._is_identifying_as_submitter():
             logger.info(
                 f"Documents will be saved under submitter name: '{self._submitter}'"