Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def info(self):
class StatusMock:
status = TaskStatus.RUNNING if self.running else TaskStatus.QUEUED
processing_time = 1
return StatusMock()
task = self.node.create_task(images,
get_submodel_args_dict(),
progress_callback=print_progress,
skip_post_processing=True,
outputs=outputs)
self.remote_task = task
# Cleanup seed file
os.remove(seed_file)
# Keep track of tasks for cleanup
self.params['tasks'].append(task)
# Check status
info = task.info()
if info.status in [TaskStatus.RUNNING, TaskStatus.COMPLETED]:
def monitor():
class nonloc:
status_callback_calls = 0
last_update = 0
def status_callback(info):
# If a task switches from RUNNING to QUEUED, then we need to
# stop the process and re-add the task to the queue.
if info.status == TaskStatus.QUEUED:
log.ODM_WARNING("LRE: %s (%s) turned from RUNNING to QUEUED. Re-adding to back of the queue." % (self, task.uuid))
raise NodeTaskLimitReachedException("Delayed task limit reached")
elif info.status == TaskStatus.RUNNING:
# Print a status message once in a while
nonloc.status_callback_calls += 1
if nonloc.status_callback_calls > 30:
log.ODM_INFO("LRE: %s (%s) is still running" % (self, task.uuid))
from pyodm.types import TaskStatus
QUEUED = TaskStatus.QUEUED.value
RUNNING = TaskStatus.RUNNING.value
FAILED = TaskStatus.FAILED.value
COMPLETED = TaskStatus.COMPLETED.value
CANCELED = TaskStatus.CANCELED.value