Method f

scripts/task_scheduler.py:196–234

(request: gr.Request, *args)

    def wrap_register_ui_task(self):
        def f(request: gr.Request, *args):
            if len(args) == 0:
                raise Exception("Invalid call")

            checkpoint: str = args[0]
            task_id = args[1]
            args = args[1:]
            task_name = None

            if task_id == queue_with_every_checkpoints:
                task_id = str(uuid4())
                checkpoint = list_checkpoint_tiles()
            else:
                if not task_id.startswith("task("):
                    task_name = task_id
                    task_id = str(uuid4())

                if checkpoint is None or checkpoint == "" or checkpoint == checkpoint_current:
                    checkpoint = [shared.sd_model.sd_checkpoint_info.title]
                elif checkpoint == checkpoint_runtime:
                    checkpoint = [None]
                elif checkpoint.endswith(" checkpoints)"):
                    checkpoint_dir = " ".join(checkpoint.split(" ")[0:-2])
                    checkpoint = list(filter(lambda c: c.startswith(checkpoint_dir), list_checkpoint_tiles()))
                else:
                    checkpoint = [checkpoint]

            for i, c in enumerate(checkpoint):
                t_id = task_id if i == 0 else f"{task_id}.{i}"
                task_runner.register_ui_task(
                    t_id,
                    self.is_img2img,
                    *args,
                    checkpoint=c,
                    task_name=task_name,
                    request=request,
                )

            task_runner.execute_pending_tasks_threading()

        return f