Merge pull request #125 from intel/update-branch

352 user story text to image generation microservice with flux1 schnell and flux1 dev (#356)

Author: gooishin

usecases/ai/microservices/text-to-image/Dockerfile

@@ -5,8 +5,8 @@ FROM debian:12-slim
 ARG DEBIAN_FRONTEND=noninteractive
 
 # Define the default allowed models and the default model
-ARG ALLOWED_MODELS="stable-diffusion-v3 stable-diffusion-v3.5 stable-diffusion-xl"
-ARG DEFAULT_MODEL="stable-diffusion-xl"
+ARG ALLOWED_MODELS="stable-diffusion-v3.5 stable-diffusion-xl"
+ARG DEFAULT_MODEL="stable-diffusion-v3.5"
 
 # Set default model as a build argument and runtime environment variable
 ENV DEFAULT_MODEL=${DEFAULT_MODEL}

usecases/ai/microservices/text-to-image/README.md

@@ -8,6 +8,8 @@ It provides endpoints for managing the pipeline, checking the pipeline status, r
 * Stable Diffusion XL
 * Stable Diffusion v3.5 
 * Stable Diffusion v3
+* Flux.1 Schnell
+* Flux.1 Dev
 
 ### Supported Inference Device
 * CPU

usecases/ai/microservices/text-to-image/flux-dev/.gitignore

@@ -0,0 +1,3 @@
+.venv
+flux-dev
+kernel.errors.txt

usecases/ai/microservices/text-to-image/flux-dev/backend/client.py

@@ -0,0 +1,125 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import time
+import requests
+
+
+class PipelineClient:
+    def __init__(self, base_url="http://localhost:8100"):
+        """
+        Initialize the PipelineClient with the base URL.
+        :param base_url: The base URL for the API.
+        """
+        self.base_url = base_url
+
+    def make_request(self, method, endpoint, data=None, timeout=100):
+        """
+        Make a request to the API.
+        :param method: HTTP method (GET, POST, etc.).
+        :param endpoint: API endpoint.
+        :param data: JSON payload for the request.
+        :param timeout: Timeout in seconds.
+        :return: Response object or None on failure.
+        """
+        url = f"{self.base_url}{endpoint}"
+        try:
+            response = requests.request(method, url, json=data, timeout=timeout)
+            response.raise_for_status()
+            return response
+        except requests.RequestException as e:
+            print(f"Error while accessing {url}: {e}")
+            return None
+
+    @staticmethod
+    def save_image(content, filename):
+        """
+        Save the content as an image file.
+        :param content: Binary content of the image.
+        :param filename: Filename to save the image.
+        """
+        try:
+            with open(filename, "wb") as file:
+                file.write(content)
+            print(f"Image saved as {filename}")
+        except IOError as e:
+            print(f"Failed to save image: {e}")
+
+    def wait_for_completion(self, timeout=60, poll_interval=2):
+        """
+        Wait for the pipeline to complete.
+        :param timeout: Maximum time to wait in seconds.
+        :param poll_interval: Time between status checks.
+        :return: True if completed, False otherwise.
+        """
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            response = self.make_request("GET", "/pipeline/status", timeout=10)
+            if response:
+                status = response.json()
+                print(f"Pipeline status: Running={status['running']}, Completed={status['completed']}")
+                if status["completed"]:
+                    return True
+            time.sleep(poll_interval)
+        print("Timeout reached. Pipeline execution did not complete.")
+        return False
+
+    def check_health(self):
+        """
+        Check the health of the API.
+        :return: True if the health check passes, False otherwise.
+        """
+        response = self.make_request("GET", "/health")
+        if response:
+            health_status = response.json()
+            print(f"Health status: {health_status['status']}")
+            return health_status['status'] == "healthy"
+        else:
+            print("Failed to perform health check.")
+            return False
+
+
+def main():
+    client = PipelineClient()
+
+    # Step 1: Health check
+    if not client.check_health():
+        print("Health check failed. Exiting.")
+        return
+
+    # Step 2: Select the device
+    response = client.make_request("POST", "/pipeline/select-device", {"device": "GPU"})
+    if response:
+        print(response.json())
+
+    # Step 3: Trigger the pipeline with additional parameters
+    response = client.make_request(
+        "POST",
+        "/pipeline/run",
+        {
+            "prompt": "A raccoon trapped inside a glass jar full of colorful candies, the background is steamy with vivid colors",
+            "width": 512,  # Additional parameter: width
+            "height": 512,  # Additional parameter: height
+            "num_inference_steps": 5  # Additional parameter: num_inference_steps
+        },
+    )
+    if not response:
+        print("Failed to trigger the pipeline.")
+        return
+    print(response.json())
+
+    # Step 4: Wait for completion
+    if not client.wait_for_completion():
+        return
+
+    # Step 5: Retrieve the generated image
+    response = client.make_request("GET", "/pipeline/image")
+    if response:
+        client.save_image(response.content, "output_image.png")
+    else:
+        print("Failed to retrieve the generated image.")
+
+
+if __name__ == "__main__":
+    main()
+

usecases/ai/microservices/text-to-image/flux-dev/backend/cmd_helper.py

@@ -0,0 +1,43 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import subprocess  # nosec - disable B404:import-subprocess check
+import sys
+from pathlib import Path
+from typing import Dict
+import platform
+
+
+def clone_repo(repo_url: str, revision: str = None, add_to_sys_path: bool = True) -> Path:
+    repo_path = Path(repo_url.split("/")[-1].replace(".git", ""))
+
+    if not repo_path.exists():
+        try:
+            subprocess.run(["git", "clone", repo_url], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        except Exception as exc:
+            print(f"Failed to clone the repository: {exc.stderr}")
+            raise
+
+        if revision:
+            subprocess.Popen(["git", "checkout", revision], cwd=str(repo_path))
+    if add_to_sys_path and str(repo_path.resolve()) not in sys.path:
+        sys.path.insert(0, str(repo_path.resolve()))
+
+    return repo_path
+
+
+def optimum_cli(model_id, output_dir, show_command=True, additional_args: Dict[str, str] = None):
+    export_command = f"optimum-cli export openvino --model {model_id} {output_dir}"
+    if additional_args is not None:
+        for arg, value in additional_args.items():
+            export_command += f" --{arg}"
+            if value:
+                export_command += f" {value}"
+
+    if show_command:
+        from IPython.display import Markdown, display
+
+        display(Markdown("**Export command:**"))
+        display(Markdown(f"`{export_command}`"))
+
+    subprocess.run(export_command.split(" "), shell=(platform.system() == "Windows"), check=True)