Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Unfulfilled dependency at run time with is_tmp=True #68

Open
oshimikam opened this issue Nov 1, 2023 · 0 comments
Open

Unfulfilled dependency at run time with is_tmp=True #68

oshimikam opened this issue Nov 1, 2023 · 0 comments

Comments

@oshimikam
Copy link

Is there a way to automatically remove intermediate files after they are not needed anymore?

I thought to use the is_tmp argument from TargetInfo, but then I get an Unfulfilled dependency error.
For example, if I add is_tmp=True to the example from the tutorial:

import sciluigi as sl

class MyFooWriter(sl.Task):
    # We have no inputs here
    # Define outputs:
    def out_foo(self):
        return sl.TargetInfo(self, 'foo.txt', is_tmp=True)
    def run(self):
        with self.out_foo().open('w') as foofile:
            foofile.write('foo\n')


class MyFooReplacer(sl.Task):
    replacement = sl.Parameter() # Here, we take as a parameter
                                  # what to replace foo with.
    # Here we have one input, a "foo file":
    in_foo = None
    # ... and an output, a "bar file":
    def out_replaced(self):
        # As the path to the returned target(info), we
        # use the path of the foo file:
        return sl.TargetInfo(self, self.in_foo().path + '.bar.txt')
    def run(self):
        with self.in_foo().open() as in_f:
            with self.out_replaced().open('w') as out_f:
                # Here we see that we use the parameter self.replacement:
                out_f.write(in_f.read().replace('foo', self.replacement))


class MyWorkflow(sl.WorkflowTask):
    def workflow(self):
        foowriter = self.new_task('foowriter', MyFooWriter)
        fooreplacer = self.new_task('fooreplacer', MyFooReplacer,
            replacement='bar')

        # Here we do the *magic*: Connecting outputs to inputs:
        fooreplacer.in_foo = foowriter.out_foo

        # Return the last task(s) in the workflow chain.
        return fooreplacer
                
if __name__ == '__main__':
    sl.run_local(main_task_cls=MyWorkflow)

I get:

raise RuntimeError('Unfulfilled %s at run time: %s' % (deps, ', '.join(missing)))
RuntimeError: Unfulfilled dependency at run time: MyFooWriter_foowriter_MyWorkflow_insta_1c49888331
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant