add psij worker

Author: adi611

pydra/engine/workers.py

@@ -891,11 +891,52 @@ def close(self):
         """Finalize the internal pool of tasks."""
         pass
 
+class PsijWorker(Worker):
+    def __init__(self, **kwargs):
+        """Initialize worker."""
+        try:
+            import psij
+        except ImportError:
+            logger.critical("Please install psij.")
+            raise
+        logger.debug("Initialize PsijWorker")
 
+    def run_el(self, interface, rerun=False, **kwargs):
+        """Run a task."""
+        return self.exec_psij(interface, rerun=rerun)
+    
+    def make_spec(self, cmd=None, arg=None, cache_dir=None):
+        spec = self.psij.JobSpec()
+        spec.executable = cmd
+        spec.arguments = arg
+        spec.stdout_path = 'demo.stdout'
+        spec.stderr_path = 'demo.stderr'
+        
+        return spec
+
+    def make_job(self, spec, attributes):
+        job = self.psij.Job()
+        job.spec = spec
+        return job
+
+    async def exec_psij(self, runnable, rerun=False):
+        import psij
+        self.psij = psij
+        jex = psij.JobExecutor.get_instance('local')
+        spec = self.make_spec(runnable.inputs.executable, runnable.inputs.args, runnable.cache_dir)
+        job = self.make_job(spec, None)
+        jex.submit(job)
+        return 
+
+    def close(self):
+        """Finalize the internal pool of tasks."""
+        pass
+    
 WORKERS = {
     "serial": SerialWorker,
     "cf": ConcurrentFuturesWorker,
     "slurm": SlurmWorker,
     "dask": DaskWorker,
     "sge": SGEWorker,
+    "psij": PsijWorker,
 }