Add config filter parameter to simulate_factors()

desmod/config.py

@@ -271,7 +271,7 @@ def factorial_config(base_config, factors, special_key=None):
     for keys, values_list in factors:
         unrolled_factors.append([(keys, values) for values in values_list])
 
-    for keys_values_lists in product(*unrolled_factors):
+    for idx, keys_values_lists in enumerate(product(*unrolled_factors)):
         config = deepcopy(base_config)
         special = []
         if special_key:

desmod/simulation.py

@@ -5,6 +5,12 @@
 from multiprocessing import cpu_count, Process, Queue
 from pprint import pprint
 from threading import Thread
+try:
+    from future_builtins import filter
+except ImportError:
+    # Assume Python 3, which already has filter built in
+    pass
+
 import json
 import os
 import random
@@ -21,6 +27,7 @@
                              consume_multi_progress)
 from desmod.timescale import parse_time, scale_time
 from desmod.tracer import TraceManager
+from desmod.workspacesync.s3 import S3Sync
 
 
 class SimEnvironment(simpy.Environment):
@@ -124,11 +131,12 @@ def schedule(self, delay=0):
 class _Workspace(object):
     """Context manager for workspace directory management."""
     def __init__(self, config):
+        self.config = config
         self.workspace = config.setdefault('meta.sim.workspace',
                                            config.setdefault('sim.workspace',
                                                              os.curdir))
         self.overwrite = config.setdefault('sim.workspace.overwrite', False)
-        self.prev_dir = os.getcwd()
+        self.orig_dir = os.getcwd()
 
     def __enter__(self):
         if os.path.relpath(self.workspace) != os.curdir:
@@ -140,7 +148,20 @@ def __enter__(self):
             os.chdir(self.workspace)
 
     def __exit__(self, *exc):
-        os.chdir(self.prev_dir)
+        os.chdir(self.orig_dir)
+        try:
+            os.chdir(self.config['sim.workspace'])
+            self.sync()
+        finally:
+            os.chdir(self.orig_dir)
+
+    def sync(self):
+        if self.config.setdefault('sim.sync.s3.enable', False):
+            artifacts = []
+            for root, _, files in os.walk(os.curdir, topdown=False):
+                for filename in files:
+                    artifacts.append(os.path.join(root, filename))
+            S3Sync(self.config, artifacts).sync()
 
 
 def simulate(config, top_type, env_type=SimEnvironment, reraise=True,
@@ -205,7 +226,7 @@ def simulate(config, top_type, env_type=SimEnvironment, reraise=True,
 
 
 def simulate_factors(base_config, factors, top_type,
-                     env_type=SimEnvironment, jobs=None):
+                     env_type=SimEnvironment, jobs=None, config_filter=None):
     """Run multi-factor simulations in separate processes.
 
     The `factors` are used to compose specialized config dictionaries for the
@@ -220,15 +241,20 @@ def simulate_factors(base_config, factors, top_type,
     :param top_type: The model's top-level Component subclass.
     :param env_type: :class:`SimEnvironment` subclass.
     :param int jobs: User specified number of concurent processes.
+    :param function config_filter:
+        A function which will be passed a config and returns a bool to filter.
     :returns: Sequence of result dictionaries for each simulation.
 
     """
     configs = list(factorial_config(base_config, factors, 'meta.sim.special'))
     ws = base_config.setdefault('sim.workspace', os.curdir)
     overwrite = base_config.setdefault('sim.workspace.overwrite', False)
+
     for index, config in enumerate(configs):
         config['meta.sim.index'] = index
         config['meta.sim.workspace'] = os.path.join(ws, str(index))
+    if config_filter is not None:
+        configs[:] = filter(config_filter, configs)
     if overwrite and os.path.relpath(ws) != os.curdir and os.path.isdir(ws):
         shutil.rmtree(ws)
     return simulate_many(configs, top_type, env_type, jobs)

desmod/workspacesync/s3.py

@@ -0,0 +1,50 @@
+"""Synchronization of workspace artifacts to Amazon S3 cloud storage."""
+import os
+
+
+class S3Sync(object):
+    """S3 workspace synchronization.
+
+    The :class:`S3Sync` class implements a sync method that synchronizes the
+    workspace artifacts to an S3 bucket with a configured key prefix.  The
+    s3 destination is of the format: /{prefix}/{workspace name}/{artifact path}
+
+    :param dict config: A fully-initialized configuration dictionary.
+    :param list artifacts: A list of artifact paths relative to the workspace
+        directory. E.g., ["./0/results.yaml"].
+
+    """
+
+    MAX_THREADS = 10
+
+    def __init__(self, config, artifacts):
+        self.client = None
+        self.config = config
+        self.workspace = config['sim.workspace']
+        self.artifacts = artifacts
+
+    def _upload_artifact(self, artifact):
+        dest = os.path.join(
+            self.config['sim.sync.s3.prefix'],
+            os.path.split(self.workspace)[1],
+            (artifact[2:]))
+        self.client.upload_file(
+            artifact, self.config['sim.sync.s3.bucket'], dest)
+
+    def sync(self):
+        """Concurrently upload the artifacts to s3."""
+        from concurrent.futures import ThreadPoolExecutor
+        import boto3
+
+        self.config.setdefault('sim.sync.s3.prefix', '')
+        self.client = boto3.client('s3')
+
+        if len(self.artifacts) == 0:
+            return
+
+        futures = []
+        with ThreadPoolExecutor(max_workers=self.MAX_THREADS) as executor:
+            for artifact in self.artifacts:
+                futures.append(
+                    executor.submit(self._upload_artifact, artifact))
+        [future.result() for future in futures]

requirements.txt

@@ -1,3 +1,5 @@
+boto3 >= 1.4.2
+futures >= 3.1.1; python_version < '3.0'
 pyvcd >= 0.1.1
 PyYAML >= 3.11
 simpy >= 3.0.8

tests/test_simulation.py

@@ -38,6 +38,7 @@ def config():
         'sim.result.file': 'result.yaml',
         'sim.workspace': 'workspace',
         'sim.workspace.overwrite': False,
+        'sim.sync.s3.enable': False,
         'sim.timescale': '1 us',
         'sim.seed': 1234,
         'sim.duration': '1 us',
@@ -196,6 +197,22 @@ def test_simulate_factors(config):
                          result['config']['sim.result.file']))
 
 
+def test_simulate_factors_only_factor(config):
+    FACTOR_NUM = 2
+    cfg_filter_fn = lambda cfg: cfg['meta.sim.index'] == FACTOR_NUM
+    factors = [(['sim.seed'], [[1], [2], [3]])]
+    results = simulate_factors(
+        config, factors, TopTest, config_filter=cfg_filter_fn)
+    assert len(results) == 1
+    for result in results:
+        assert result['sim.exception'] is None
+        assert result['config']['meta.sim.workspace'] == os.path.join(
+            config['sim.workspace'], str(FACTOR_NUM))
+        assert os.path.exists(
+            os.path.join(result['config']['meta.sim.workspace'],
+                         result['config']['sim.result.file']))
+
+
 def test_simulate_factors_progress(config, capfd):
     config['sim.progress.enable'] = True
     config['sim.duration'] = '10 us'

tests/test_tracer.py

@@ -37,6 +37,7 @@ def config():
         'sim.vcd.start_time': '',
         'sim.vcd.stop_time': '',
         'sim.workspace': 'workspace',
+        'sim.sync.s3.enable': False,
         'test.raise': False,
     }