Implementation:AUTOMATIC1111 Stable diffusion webui Progress Tracking
| Knowledge Sources | |
|---|---|
| Domains | Task_Management, API |
| Last Updated | 2025-05-15 00:00 GMT |
Overview
Implements a task progress tracking system with queue management, live preview image streaming, and REST API endpoints for monitoring generation progress.
Description
The Progress Tracking module manages the lifecycle of generation tasks through a queue-based system. Tasks are identified by string IDs in the format task(type-RANDOM). The module maintains an OrderedDict of pending tasks, tracks the current active task, and keeps a list of recently finished task IDs (up to 16). It provides two API endpoints: /internal/pending-tasks (GET) returns the count and IDs of pending tasks, and /internal/progress (POST) returns detailed progress information including: active/queued/completed status, progress percentage (computed from job count and sampling steps), ETA estimation, queue position, live preview images (base64-encoded in PNG or other formats), and status text. The restore_progress function allows the UI to recover generation results after a page reload by waiting for a task to complete and returning its recorded results. Pydantic models (ProgressRequest, ProgressResponse, PendingTasksResponse) define the API contract.
Usage
Use this module to track the progress of image generation tasks, display live preview images in the WebUI, show queue position to users, and implement progress polling from the frontend or external API clients.
Code Reference
Source Location
- Repository: AUTOMATIC1111_Stable_diffusion_webui
- File: modules/progress.py
- Lines: 1-152
Signature
def start_task(id_task: str) -> None
def finish_task(id_task: str) -> None
def create_task_id(task_type: str) -> str
def record_results(id_task: str, res) -> None
def add_task_to_queue(id_job: str) -> None
def setup_progress_api(app) -> None
def get_pending_tasks() -> PendingTasksResponse
def progressapi(req: ProgressRequest) -> ProgressResponse
def restore_progress(id_task: str) -> tuple
class ProgressRequest(BaseModel): ...
class ProgressResponse(BaseModel): ...
class PendingTasksResponse(BaseModel): ...
Import
from modules import progress
from modules.progress import setup_progress_api, create_task_id
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| id_task | str | Yes | Task identifier in the format "task(type-RANDOM)" |
| task_type | str | Yes | Type label for task ID generation (used by create_task_id) |
| req | ProgressRequest | Yes | Request object containing id_task, id_live_preview, and live_preview flag |
| app | FastAPI | Yes | The FastAPI application to register routes on (used by setup_progress_api) |
Outputs
| Name | Type | Description |
|---|---|---|
| response | ProgressResponse | Progress data including active, queued, completed, progress (0-1), eta, live_preview (data URI), id_live_preview, and textinfo |
| pending | PendingTasksResponse | Count and list of pending task IDs |
| task_id | str | Generated task identifier string |
Usage Examples
from modules.progress import create_task_id, add_task_to_queue, start_task, finish_task
# Create and queue a task
task_id = create_task_id("txt2img")
add_task_to_queue(task_id)
# Start processing
start_task(task_id)
# ... perform generation ...
finish_task(task_id)
# API client polling for progress
import requests
resp = requests.post("http://localhost:7860/internal/progress", json={
"id_task": task_id,
"id_live_preview": -1,
"live_preview": True
})
print(resp.json()["progress"]) # 0.0 to 1.0