update post multi-task server and client

lightx2v/api_multi_servers.py

@@ -82,7 +82,7 @@ def start_server(config: ServerConfig) -> Optional[tuple[subprocess.Popen, str]]
 
         # Wait for server to start, up to 600 seconds
         node_ip = get_node_ip()
-        service_url = f"http://{node_ip}:{config.port}/v1/local/video/generate/service_status"
+        service_url = f"http://{node_ip}:{config.port}/v1/service/status"
 
         # Check once per second, up to 600 times
         for _ in range(600):
@@ -149,14 +149,14 @@ def main():
                 logger.error(f"Error occurred while starting server: {e}")
 
     # Print all server URLs
-    print("\nAll server URLs:")
+    logger.info("\nAll server URLs:")
     for url in urls:
-        print(url)
+        logger.info(url)
 
     # Print node information
     node_ip = get_node_ip()
-    print(f"\nCurrent node IP: {node_ip}")
-    print(f"Number of servers started: {len(urls)}")
+    logger.info(f"\nCurrent node IP: {node_ip}")
+    logger.info(f"Number of servers started: {len(urls)}")
 
     try:
         # Wait for all processes

scripts/server/post_multi_servers.py

@@ -3,132 +3,121 @@ from loguru import logger
 import random
 import string
 import time
+import threading
 from datetime import datetime
+from tqdm import tqdm
 
 
-# same as lightx2v/utils/generate_task_id.py
-# from lightx2v.utils.generate_task_id import generate_task_id
-def generate_task_id():
-    """
-    Generate a random task ID in the format XXXX-XXXX-XXXX-XXXX-XXXX.
-    Features:
-    1. Does not modify the global random state.
-    2. Each X is an uppercase letter or digit (0-9).
-    3. Combines time factors to ensure high randomness.
-    For example: N1PQ-PRM5-N1BN-Z3S1-BGBJ
-    """
-    # Save the current random state (does not affect external randomness)
-    original_state = random.getstate()
-
+def send_and_monitor_task(url, message, task_index, complete_bar, complete_lock):
+    """Send task to server and monitor until completion"""
     try:
-        # Define character set (uppercase letters + digits)
-        characters = string.ascii_uppercase + string.digits
-
-        # Create an independent random instance
-        local_random = random.Random(time.perf_counter_ns())
-
-        # Generate 5 groups of 4-character random strings
-        groups = []
-        for _ in range(5):
-            # Mix new time factor for each group
-            time_mix = int(datetime.now().timestamp())
-            local_random.seed(time_mix + local_random.getstate()[1][0] + time.perf_counter_ns())
-
-            groups.append("".join(local_random.choices(characters, k=4)))
-
-        return "-".join(groups)
-
-    finally:
-        # Restore the original random state
-        random.setstate(original_state)
-
-
-def post_all_tasks(urls, messages):
-    msg_num = len(messages)
-    msg_index = 0
+        # Step 1: Send task and get task_id
+        response = requests.post(f"{url}/v1/tasks/", json=message)
+        response_data = response.json()
+        task_id = response_data.get("task_id")
+
+        if not task_id:
+            logger.error(f"No task_id received from {url}")
+            return False
+
+        # Step 2: Monitor task status until completion
+        while True:
+            try:
+                status_response = requests.get(f"{url}/v1/tasks/{task_id}/status")
+                status_data = status_response.json()
+                task_status = status_data.get("status")
+
+                if task_status == "completed":
+                    # Update completion bar safely
+                    if complete_bar and complete_lock:
+                        with complete_lock:
+                            complete_bar.update(1)
+                    return True
+                elif task_status == "failed":
+                    logger.error(f"Task {task_index + 1} (task_id: {task_id}) failed")
+                    if complete_bar and complete_lock:
+                        with complete_lock:
+                            complete_bar.update(1)  # Still update progress even if failed
+                    return False
+                else:
+                    # Task still running, wait and check again
+                    time.sleep(0.5)
+
+            except Exception as e:
+                logger.error(f"Failed to check status for task_id {task_id}: {e}")
+                time.sleep(0.5)
+
+    except Exception as e:
+        logger.error(f"Failed to send task to {url}: {e}")
+        return False
+
+
+def get_available_urls(urls):
+    """Check which URLs are available and return the list"""
     available_urls = []
     for url in urls:
         try:
             _ = requests.get(f"{url}/v1/service/status").json()
+            available_urls.append(url)
         except Exception as e:
             continue
-        available_urls.append(url)
 
     if not available_urls:
         logger.error("No available urls.")
-        return
+        return None
 
     logger.info(f"available_urls: {available_urls}")
+    return available_urls
+
 
+def find_idle_server(available_urls):
+    """Find an idle server from available URLs"""
     while True:
         for url in available_urls:
-            response = requests.get(f"{url}/v1/service/status").json()
-            if response["service_status"] == "idle":
-                logger.info(f"{url} service is idle, start task...")
-                response = requests.post(f"{url}/v1/tasks/", json=messages[msg_index])
-                logger.info(f"response: {response.json()}")
-                msg_index += 1
-                if msg_index == msg_num:
-                    logger.info("All tasks have been sent.")
-                    return
-        time.sleep(5)
-
-
-if __name__ == "__main__":
-    urls = ["http://localhost:8000", "http://localhost:8001"]
-
-    messages = [
-        {
-            "task_id": generate_task_id(),  # task_id also can be string you like, such as "test_task_001"
-            "task_id_must_unique": True,  # If True, the task_id must be unique, otherwise, it will raise an error. Default is False.
-            "prompt": "A cat walks on the grass, realistic style.",
-            "negative_prompt": "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
-            "image_path": "",
-            "save_video_path": "./output_lightx2v_wan_t2v_t01.mp4",  # It is best to set it to an absolute path.
-        },
-        {
-            "task_id": generate_task_id(),  # task_id also can be string you like, such as "test_task_001"
-            "task_id_must_unique": True,  # If True, the task_id must be unique, otherwise, it will raise an error. Default is False.
-            "prompt": "A person is riding a bike. Realistic, Natural lighting, Casual.",
-            "negative_prompt": "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
-            "image_path": "",
-            "save_video_path": "./output_lightx2v_wan_t2v_t02.mp4",  # It is best to set it to an absolute path.
-        },
-        {
-            "task_id": generate_task_id(),  # task_id also can be string you like, such as "test_task_001"
-            "task_id_must_unique": True,  # If True, the task_id must be unique, otherwise, it will raise an error. Default is False.
-            "prompt": "A car turns a corner. Realistic, Natural lighting, Casual.",
-            "negative_prompt": "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
-            "image_path": "",
-            "save_video_path": "./output_lightx2v_wan_t2v_t03.mp4",  # It is best to set it to an absolute path.
-        },
-        {
-            "task_id": generate_task_id(),  # task_id also can be string you like, such as "test_task_001"
-            "task_id_must_unique": True,  # If True, the task_id must be unique, otherwise, it will raise an error. Default is False.
-            "prompt": "An astronaut is flying in space, Van Gogh style. Dark, Mysterious.",
-            "negative_prompt": "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
-            "image_path": "",
-            "save_video_path": "./output_lightx2v_wan_t2v_t04.mp4",  # It is best to set it to an absolute path.
-        },
-        {
-            "task_id": generate_task_id(),  # task_id also can be string you like, such as "test_task_001"
-            "task_id_must_unique": True,  # If True, the task_id must be unique, otherwise, it will raise an error. Default is False.
-            "prompt": "A beautiful coastal beach in spring, waves gently lapping on the sand, the camera movement is Zoom In. Realistic, Natural lighting, Peaceful.",
-            "negative_prompt": "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
-            "image_path": "",
-            "save_video_path": "./output_lightx2v_wan_t2v_t05.mp4",  # It is best to set it to an absolute path.
-        },
-        {
-            "task_id": generate_task_id(),  # task_id also can be string you like, such as "test_task_001"
-            "task_id_must_unique": True,  # If True, the task_id must be unique, otherwise, it will raise an error. Default is False.
-            "prompt": "Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage.",
-            "negative_prompt": "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
-            "image_path": "",
-            "save_video_path": "./output_lightx2v_wan_t2v_t06.mp4",  # It is best to set it to an absolute path.
-        },
-    ]
-
-    logger.info(f"urls: {urls}")
-    logger.info(f"message: {messages}")
-
-    post_all_tasks(urls, messages)
+            try:
+                response = requests.get(f"{url}/v1/service/status").json()
+                if response["service_status"] == "idle":
+                    return url
+            except Exception as e:
+                continue
+        time.sleep(3)
+
+
+def process_tasks_async(messages, available_urls, show_progress=True):
+    """Process a list of tasks asynchronously across multiple servers"""
+    if not available_urls:
+        logger.error("No available servers to process tasks.")
+        return False
+
+    active_threads = []
+
+    logger.info(f"Sending {len(messages)} tasks to available servers...")
+
+    # Create completion progress bar
+    if show_progress:
+        complete_bar = tqdm(total=len(messages), desc="Completing tasks")
+        complete_lock = threading.Lock()  # Thread-safe updates to completion bar
+
+    for idx, message in enumerate(messages):
+        # Find an idle server
+        server_url = find_idle_server(available_urls)
+
+        # Create and start thread for sending and monitoring task
+        thread = threading.Thread(target=send_and_monitor_task, args=(server_url, message, idx, complete_bar if show_progress else None, complete_lock if show_progress else None))
+        thread.daemon = False
+        thread.start()
+        active_threads.append(thread)
+
+        # Small delay to let thread start
+        time.sleep(0.5)
+
+    # Wait for all threads to complete
+    for thread in active_threads:
+        thread.join()
+
+    # Close completion bar
+    if show_progress:
+        complete_bar.close()
+
+    logger.info("All tasks processing completed!")
+    return True

scripts/server/post_multi_servers_i2v.py

+from loguru import logger
+from post_multi_servers import get_available_urls, process_tasks_async
+
+
+if __name__ == "__main__":
+    urls = [f"http://localhost:{port}" for port in range(8000, 8008)]
+    img_prompts = {
+        "assets/inputs/imgs/img_0.jpg": "Summer beach vacation style, a white cat wearing sunglasses sits on a surfboard. The fluffy-furred feline gazes directly at the camera with a relaxed expression. Blurred beach scenery forms the background featuring crystal-clear waters, distant green hills, and a blue sky dotted with white clouds. The cat assumes a naturally relaxed posture, as if savoring the sea breeze and warm sunlight. A close-up shot highlights the feline's intricate details and the refreshing atmosphere of the seaside.",
+        "assets/inputs/imgs/img_2.jpg": "A close-up cinematic view of a person cooking in a warm,sunlit kitchen, using a wooden spatula to stir-fry a colorful mix of freshvegetables—carrots, broccoli, and bell peppers—in a black frying pan on amodern induction stove. The scene captures the glistening texture of thevegetables, steam gently rising, and subtle reflections on the stove surface.In the background, soft-focus jars, fruits, and a window with natural daylightcreate a cozy atmosphere. The hand motions are smooth and rhythmic, with a realisticsense of motion blur and lighting.",
+    }
+    negative_prompt = "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走"
+
+    messages = []
+    for i, (image_path, prompt) in enumerate(img_prompts.items()):
+        messages.append({"prompt": prompt, "negative_prompt": negative_prompt, "image_path": image_path, "save_video_path": f"./output_lightx2v_wan_i2v_{i + 1}.mp4"})
+
+    logger.info(f"urls: {urls}")
+
+    # Get available servers
+    available_urls = get_available_urls(urls)
+    if not available_urls:
+        exit(1)
+
+    # Process tasks asynchronously
+    success = process_tasks_async(messages, available_urls, show_progress=True)
+
+    if success:
+        logger.info("All tasks completed successfully!")
+    else:
+        logger.error("Some tasks failed.")
+        exit(1)

scripts/server/post_multi_servers_t2v.py

+from loguru import logger
+from post_multi_servers import get_available_urls, process_tasks_async
+
+
+if __name__ == "__main__":
+    urls = ["http://localhost:8000", "http://localhost:8001"]
+    prompts = [
+        "A cat walks on the grass, realistic style.",
+        "A person is riding a bike. Realistic, Natural lighting, Casual.",
+        "A car turns a corner. Realistic, Natural lighting, Casual.",
+        "An astronaut is flying in space, Van Gogh style. Dark, Mysterious.",
+        "A beautiful coastal beach in spring, waves gently lapping on the sand, the camera movement is Zoom In. Realistic, Natural lighting, Peaceful.",
+        "Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage.",
+    ]
+    negative_prompt = "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走"
+
+    messages = []
+    for i, prompt in enumerate(prompts):
+        messages.append({"prompt": prompt, "negative_prompt": negative_prompt, "image_path": "", "save_video_path": f"./output_lightx2v_wan_t2v_{i + 1}.mp4"})
+
+    logger.info(f"urls: {urls}")
+
+    # Get available servers
+    available_urls = get_available_urls(urls)
+    if not available_urls:
+        exit(1)
+
+    # Process tasks asynchronously
+    success = process_tasks_async(messages, available_urls, show_progress=True)
+
+    if success:
+        logger.info("All tasks completed successfully!")
+    else:
+        logger.error("Some tasks failed.")
+        exit(1)

scripts/server/post_vbench_i2v.py

-from tqdm import tqdm
 import argparse
 import glob
 import os
-import requests
-import time
+from loguru import logger
+from post_multi_servers import get_available_urls, process_tasks_async
 
 
-def post_i2v(image_path, output_path):
-    url = "http://localhost:8000"
+def create_i2v_messages(img_files, output_path):
+    """Create messages for image-to-video tasks"""
+    messages = []
+    negative_prompt = "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走"
 
-    file_name = os.path.basename(image_path)
-    prompt = os.path.splitext(file_name)[0]
-    save_video_path = os.path.join(output_path, f"{prompt}.mp4")
+    for img_path in img_files:
+        file_name = os.path.basename(img_path)
+        prompt = os.path.splitext(file_name)[0]
+        save_video_path = os.path.join(output_path, f"{prompt}.mp4")
 
-    message = {
-        "prompt": prompt,
-        "negative_prompt": "镜头晃动，色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
-        "image_path": image_path,
-        "save_video_path": save_video_path,
-    }
+        message = {
+            "prompt": prompt,
+            "negative_prompt": negative_prompt,
+            "image_path": img_path,
+            "save_video_path": save_video_path,
+        }
+        messages.append(message)
 
-    while True:
-        response = requests.get(f"{url}/v1/service/status").json()
-        if response["service_status"] == "idle":
-            response = requests.post(f"{url}/v1/tasks/", json=message)
-            return
-        time.sleep(3)
+    return messages
 
 
 if __name__ == "__main__":
@@ -34,11 +32,35 @@ if __name__ == "__main__":
     parser.add_argument("--output_path", type=str, default="./vbench_i2v", help="output video path.")
     args = parser.parse_args()
 
+    # Create server URLs
+    urls = [f"http://localhost:{port}" for port in range(8000, 8008)]
+
+    # Get available servers
+    available_urls = get_available_urls(urls)
+    if not available_urls:
+        exit(1)
+
+    # Find image files
     if os.path.exists(args.data_path):
         img_files = glob.glob(os.path.join(args.data_path, "*.jpg"))
-        print(f"Found {len(img_files)} image files.")
+        logger.info(f"Found {len(img_files)} image files.")
+
+        if not img_files:
+            logger.error("No image files found.")
+            exit(1)
+
+        # Create messages for all images
+        messages = create_i2v_messages(img_files, args.output_path)
+        logger.info(f"Created {len(messages)} tasks.")
+
+        # Process tasks asynchronously
+        success = process_tasks_async(messages, available_urls, show_progress=True)
 
-        with tqdm(total=len(img_files)) as progress_bar:
-            for idx, img_path in enumerate(img_files):
-                post_i2v(img_path, args.output_path)
-                progress_bar.update()
+        if success:
+            logger.info("All image-to-video tasks completed successfully!")
+        else:
+            logger.error("Some tasks failed.")
+            exit(1)
+    else:
+        logger.error(f"Data path does not exist: {args.data_path}")
+        exit(1)

scripts/server/start_multi_servers.sh

@@ -16,13 +16,13 @@ if [ -z "${num_gpus}" ]; then
 fi
 
 # Check required parameters
-if [ -z "$lightx2v_path" ]; then
-    echo "Error: lightx2v_path not set"
+if [ -z "${lightx2v_path}" ]; then
+    echo "Error: lightx2v_path is not set. Please set this variable first."
     exit 1
 fi
 
-if [ -z "$model_path" ]; then
-    echo "Error: model_path not set"
+if [ -z "${model_path}" ]; then
+    echo "Error: model_path is not set. Please set this variable first."
     exit 1
 fi
 
@@ -37,7 +37,7 @@ export DTYPE=BF16
 python -m lightx2v.api_multi_servers \
     --num_gpus $num_gpus \
     --start_port 8000 \
-    --model_cls wan2.1 \
-    --task t2v \
+    --model_cls wan2.1_distill \
+    --task i2v \
     --model_path $model_path \
-    --config_json ${lightx2v_path}/configs/wan/wan_t2v.json
+    --config_json ${lightx2v_path}/configs/distill/wan_i2v_distill_4step_cfg.json