Merge pull request #20 from NREL/pp/batch_background_fix

Batch + monitor + background fix

gaps/batch.py

@@ -136,6 +136,7 @@ def _run_pipelines(self, monitor_background=False):
         """Run the pipeline modules for each batch job."""
 
         for sub_directory in self.sub_dirs:
+            os.chdir(sub_directory)
             pipeline_config = sub_directory / self._pipeline_fp.name
             if not pipeline_config.is_file():
                 raise gapsConfigError(
@@ -216,7 +217,11 @@ def run(self, dry_run=False, monitor_background=False):
         if dry_run:
             return
 
-        self._run_pipelines(monitor_background=monitor_background)
+        cwd = os.getcwd()
+        try:
+            self._run_pipelines(monitor_background=monitor_background)
+        finally:
+            os.chdir(cwd)
 
 
 def _load_batch_config(config_fp):
@@ -358,19 +363,27 @@ def _parse_config(config):
 
         for key, value in args.items():
             if isinstance(value, str):
-                msg = ('Batch arguments should be lists but found '
-                       f'"{key}": "{value}"')
+                msg = (
+                    "Batch arguments should be lists but found "
+                    f"{key!r}: {value!r}"
+                )
                 raise gapsValueError(msg)
 
         sets.add(set_tag)
 
         products = _enumerated_product(args.values())
-        set_str = f' in set "{set_tag}"' if set_tag else ''
-        logger.info(f'Found {len(products)} batch projects{set_str}. '
-                    'Creating jobs...')
-        if len(products) > 1e3:
-            msg = (f'Large number of batch jobs found: {len(products)}! '
-                   'Proceeding, but consider double checking your config.')
+        num_batch_jobs = len(products)
+        set_str = f" in set {set_tag!r}" if set_tag else ""
+        logger.info(
+            "Found %d batch projects%s. Creating jobs...",
+            num_batch_jobs,
+            set_str,
+        )
+        if num_batch_jobs > 1e3:
+            msg = (
+                f"Large number of batch jobs found: {num_batch_jobs:,}! "
+                "Proceeding, but consider double checking your config."
+            )
             warn(msg)
             logger.warning(msg)
 

gaps/cli/pipeline.py

@@ -3,7 +3,6 @@
 GAPs pipeline CLI entry points.
 """
 import os
-import sys
 import logging
 from pathlib import Path
 from warnings import warn
@@ -77,6 +76,7 @@ def pipeline(ctx, config_file, cancel, monitor, background=False):
             raise gapsExecutionError(msg)
         ctx.obj["LOG_STREAM"] = False
         _kickoff_background(config_file)
+        return
 
     project_dir = str(Path(config_file).parent.expanduser().resolve())
     status = Status(project_dir).update_from_all_job_files(purge=False)
@@ -107,7 +107,6 @@ def _kickoff_background(config_file):  # pragma: no cover
         click.echo(
             f"Kicking off pipeline job in the background. Monitor PID: {pid}"
         )
-        sys.exit()
 
 
 def pipeline_command(template_config):

gaps/pipeline.py

@@ -79,9 +79,15 @@ def _cancel_all_jobs(self):
         """Cancel all jobs in this pipeline."""
         status = self.status
         for job_id, hardware in zip(status.job_ids, status.job_hardware):
+            if job_id is None:
+                continue
+
             manager = HardwareOption(hardware).manager
-            if manager is not None:
-                manager.cancel(job_id)
+            if manager is None:
+                continue
+
+            manager.cancel(job_id)
+
         logger.info("Pipeline job %r cancelled.", self.name)
 
     def _main(self):

gaps/version.py

@@ -1,3 +1,3 @@
 """GAPs Version Number. """
 
-__version__ = "0.4.3"
+__version__ = "0.4.4"

tests/cli/test_cli.py

@@ -400,6 +400,8 @@ def preprocess_run_config(config, project_dir, out_dir):
         main, ["pipeline", "-c", pipe_config_fp.as_posix(), "--background"]
     )
 
+    time.sleep(10)  # give job enough time to run a little
+
     status = Status(tmp_cwd).update_from_all_job_files(purge=False)
     assert "monitor_pid" in status
 

tests/test_batch.py

@@ -390,19 +390,24 @@ def test_batch_job_run(typical_batch_config, monkeypatch):
     assert count_0 == 8, "Unknown starting files detected!"
 
     config_cache = []
+    working_dirs = []
 
     def _test_call(config, monitor, *__, **___):
         assert not monitor
         config_cache.append(config)
+        working_dirs.append(os.getcwd())
 
     monkeypatch.setattr(
         gaps.pipeline.Pipeline, "run", _test_call, raising=True
     )
 
+    cwd = os.getcwd()
     BatchJob(typical_batch_config).run()
     assert len(config_cache) == 9
     assert set(fp.name for fp in config_cache) == {"config_pipeline.json"}
     assert len(set(fp.parent for fp in config_cache)) == 9
+    assert len(set(working_dirs)) == 9
+    assert cwd == os.getcwd()
 
     BatchJob(typical_batch_config).delete()
     count_2 = len(list(batch_dir.glob("*")))
@@ -548,16 +553,16 @@ def test_bad_str_arg(typical_batch_config):
     batch_dir = typical_batch_config.parent
 
     config = ConfigType.JSON.load(typical_batch_config)
-    config['sets'][0]['args']['project_points'] = 'bad_str'
-    with open(typical_batch_config, 'w') as f:
+    config["sets"][0]["args"]["project_points"] = "bad_str"
+    with open(typical_batch_config, "w") as f:
         ConfigType.JSON.dump(config, f)
 
     count_0 = len(list(batch_dir.glob("*")))
     assert count_0 == 8, "Unknown starting files detected!"
 
     with pytest.raises(gapsValueError) as exc_info:
         BatchJob(typical_batch_config).run(dry_run=True)
-        assert 'Batch arguments should be lists' in str(exc_info)
+        assert "Batch arguments should be lists" in str(exc_info)
 
 
 if __name__ == "__main__":

tests/test_pipeline.py

@@ -545,6 +545,15 @@ def cache_cancel_calls(__, job_id):
             StatusField.HARDWARE: HardwareOption.EAGLE,
         },
     )
+    Status.mark_job_as_submitted(
+        sample_pipeline_config.parent,
+        "run",
+        "test4",
+        job_attrs={
+            StatusField.JOB_ID: None,
+            StatusField.HARDWARE: HardwareOption.EAGLE,
+        },
+    )
     Pipeline.cancel_all(sample_pipeline_config)
     assert set(cancelled_jobs) == {1, 12}
     assert_message_was_logged("Pipeline job", "INFO")