Fixes

Author: calvinp0

arc/job/adapters/gaussian.py

@@ -155,7 +155,7 @@ def __init__(self,
         self.incore_capacity = 1
         self.job_adapter = 'gaussian'
         self.execution_type = execution_type or 'queue'
-        self.command = ['g03', 'g09', 'g16']
+        self.command = ['g16', 'g09', 'g03']
         self.url = 'https://gaussian.com/'
 
         if species is None:
@@ -500,13 +500,15 @@ def execute_incore(self):
         """
         Execute a job incore.
         """
-        which(self.command,
-              return_bool=True,
-              raise_error=True,
-              raise_msg=f'Please install {self.job_adapter}, see {self.url} for more information.',
-              )
+        binary = which(self.command,
+                       return_bool=False,
+                       raise_error=True,
+                       raise_msg=f'Please install {self.job_adapter}, see {self.url} for more information.',
+                       )
+        binary_name = os.path.basename(binary)
         self._log_job_execution()
-        execute_command(incore_commands[self.job_adapter])
+        commands = [cmd.replace('g16', binary_name) for cmd in incore_commands[self.job_adapter]]
+        execute_command(commands)
 
     def execute_queue(self):
         """

arc/job/pipe/pipe_run.py

@@ -211,13 +211,19 @@ def write_submit_script(self) -> str:
                 f'No pipe submit template for cluster software: {self.cluster_software}. '
                 f'Available templates: {list(pipe_submit.keys())}')
         cpus, memory_mb, array_size = self._submission_resources()
+        server = servers_dict.get('local', {})
+        queue, _ = next(iter(server.get('queues', {}).items()), ('', None))
+        engine = self.tasks[0].engine if self.tasks else ''
+        env_setup = pipe_settings.get('env_setup', {}).get(engine, '')
         content = pipe_submit[template_key].format(
             name=f'pipe_{self.run_id}',
             max_task_num=array_size,
             pipe_root=self.pipe_root,
             python_exe=sys.executable,
             cpus=cpus,
             memory=memory_mb,
+            queue=queue,
+            env_setup=env_setup,
         )
         filename = 'submit.sub' if self.cluster_software == 'htcondor' else 'submit.sh'
         submit_path = os.path.join(self.pipe_root, filename)

arc/job/pipe/pipe_run_test.py

@@ -159,8 +159,8 @@ def test_pbs_content(self):
         path = run.write_submit_script()
         with open(path) as f:
             content = f.read()
-        self.assertIn('#PBS -t 1-8', content)
-        self.assertIn('WORKER_ID=$PBS_ARRAYID', content)
+        self.assertIn('#PBS -J 1-8', content)
+        self.assertIn('WORKER_ID="$PBS_ARRAY_INDEX"', content)
 
     def test_htcondor_content(self):
         run = self._make_run('htcondor', max_workers=12, n_tasks=12)

arc/scripts/pipe_worker.py

@@ -262,6 +262,10 @@ def _run_adapter(spec: TaskSpec, scratch_dir: str, job_type: str, **extra_kwargs
         **extra_kwargs,
     )
     job.execute()
+    output_file = getattr(job, 'local_path_to_output_file', None)
+    if output_file and not os.path.isfile(output_file):
+        raise RuntimeError(f'{spec.engine} produced no output file at {output_file}. '
+                           f'The engine may not be installed or configured on this node.')
 
 
 # ---------------------------------------------------------------------------

arc/settings/settings.py

@@ -313,6 +313,9 @@
     'max_workers': 100,        # Upper bound on array worker slots per PipeRun.
     'max_attempts': 3,         # Retry budget per task before terminal failure.
     'lease_duration_s': 86400, # Worker lease duration in seconds (default 24h).
+    'env_setup': {},           # Engine-specific shell setup commands, e.g.,
+                               # {'gaussian': 'source /usr/local/g09/setup.sh',
+                               #  'orca': 'source /usr/local/orca/setup.sh'}
 }
 
 # Criteria for identification of imaginary frequencies for transition states.

arc/settings/submit.py

@@ -50,37 +50,41 @@
 pipe_submit = {
     'slurm': """#!/bin/bash -l
 #SBATCH -J {name}
+#SBATCH -p {queue}
 #SBATCH -N 1
 #SBATCH -n {cpus}
 #SBATCH --mem={memory}
 #SBATCH --array=1-{max_task_num}
 #SBATCH -o {pipe_root}/out_%a.txt
 #SBATCH -e {pipe_root}/err_%a.txt
 
+{env_setup}
 WORKER_ID=$SLURM_ARRAY_TASK_ID
 
 {python_exe} -m arc.scripts.pipe_worker --pipe_root {pipe_root} --worker_id $WORKER_ID
 """,
     'pbs': """#!/bin/bash -l
 #PBS -N {name}
+#PBS -q {queue}
 #PBS -l ncpus={cpus}
 #PBS -l mem={memory}mb
-#PBS -t 1-{max_task_num}
-#PBS -o {pipe_root}/out_$PBS_ARRAYID.txt
-#PBS -e {pipe_root}/err_$PBS_ARRAYID.txt
+#PBS -J 1-{max_task_num}
 
-WORKER_ID=$PBS_ARRAYID
+{env_setup}
+WORKER_ID="$PBS_ARRAY_INDEX"
 
-{python_exe} -m arc.scripts.pipe_worker --pipe_root {pipe_root} --worker_id $WORKER_ID
+{python_exe} -m arc.scripts.pipe_worker --pipe_root {pipe_root} --worker_id "$WORKER_ID"
 """,
     'sge': """#!/bin/bash -l
 #$ -N {name}
+#$ -q {queue}
 #$ -pe smp {cpus}
 #$ -l h_vmem={memory}M
 #$ -t 1-{max_task_num}
 #$ -o {pipe_root}/out_$SGE_TASK_ID.txt
 #$ -e {pipe_root}/err_$SGE_TASK_ID.txt
 
+{env_setup}
 WORKER_ID=$SGE_TASK_ID
 
 {python_exe} -m arc.scripts.pipe_worker --pipe_root {pipe_root} --worker_id $WORKER_ID