GSoC 2023: Add PSI/J Worker

[adi611]

@djarecka @ghisvail - I am currently working on mapping the ShellCommandTask workflow to PSI/J.

1. I am using pydra's runnable.inputs.executable and runnable.inputs.args to pass to psij's spec.executable and spec.arguments respectively which I then use to create a psij Job and submit the Job to the Job Executor. I am successfully getting the result on a file as defined by psij's spec.stdout_path. Now how can I pass that output to ShellCommandTask's output stdout?
2. I am also getting an error - RuntimeError: Could not find results of 'shelly1' node in a sub-directory named 'ShellCommandTask_2b3decbead6264e36eed881610112561' in any of the cache locations: /tmp/tmpc0axxwur.

[adi611]

@djarecka @ghisvail - I am currently working on mapping the ShellCommandTask workflow to PSI/J.

1. I am using pydra's runnable.inputs.executable and runnable.inputs.args to pass to psij's spec.executable and spec.arguments respectively which I then use to create a psij Job and submit the Job to the Job Executor. I am successfully getting the result on a file as defined by psij's spec.stdout_path. Now how can I pass that output to ShellCommandTask's output stdout?
2. I am also getting an error - RuntimeError: Could not find results of 'shelly1' node in a sub-directory named 'ShellCommandTask_2b3decbead6264e36eed881610112561' in any of the cache locations: /tmp/tmpc0axxwur.

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):
        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)
        job = self.make_job(spec, None)
        jex.submit(job)
        return

    def close(self):
        """Finalize the internal pool of tasks."""
        pass

[djarecka]

could you create a draft PR so I can try the worker?

let's first try to fix the error you're getting. Do you actually have the directory /tmp/tmpc0axxwur? What is inside? Is this the output directory for the task you are running?
Can you point me which test you're running?

[adi611]

could you create a draft PR so I can try the worker?

let's first try to fix the error you're getting. Do you actually have the directory /tmp/tmpc0axxwur? What is inside? Is this the output directory for the task you are running? Can you point me which test you're running?

Yes I am working on creating the draft PR. I made some changes and I am not getting an error now.

I am running a simple ShellCommandTask code which is not part of the existing tests:

import nest_asyncio
nest_asyncio.apply()

import pydra

cmd1 = 'ls'
args = ['-a']

shelly1 = pydra.ShellCommandTask(name='shelly1', executable=cmd1, args=args)
print('cmndline = ', shelly1.cmdline)

with pydra.Submitter(plugin='psij') as sub:
    sub(shelly1)
print(shelly1.result())

[adi611]

@djarecka @ghisvail I have opened a draft PR for the same - #694 .

[adi611]

@djarecka @ghisvail I want to run all the tests in test_shelltask.py using the psij plugin. How can I achieve this?

[djarecka]

@adi611 - you could edit the conftest file: https://github.com/nipype/pydra/blob/master/pydra/conftest.py to change plugin that is a pytest's fixture (or add another fixture as it is done for dask)

and for just a temporarily testing you can overwrite the variable in the specific file