docker file added

.gitignore

@@ -0,0 +1,8 @@
+env
+env_old
+*.bak
+*.pyc
+__pycache__
+results
+# excluding user settings for the GUI frontend
+configs/settings.yaml

Dockerfile

@@ -0,0 +1,22 @@
+FROM ghcr.io/astral-sh/uv:python3.11-trixie-slim
+
+RUN apt-get update && apt-get install -y \
+    git curl build-essential libgl1 libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+ENV DEVICE=cpu
+
+WORKDIR /app
+COPY . .
+
+RUN uv pip install --system --no-cache-dir \
+    torch==2.8.0 torchvision==0.23.0 \
+    --index-url https://download.pytorch.org/whl/cpu
+
+RUN uv pip install --system --no-cache-dir -r requirements.txt
+
+EXPOSE 7860
+
+ENV GRADIO_SERVER_NAME="0.0.0.0"
+
+CMD ["python", "src/frontend/webui/hf_demo.py"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Rupesh Sreeraman
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -3,11 +3,8 @@ title: Fastsdcpu
 emoji: 📉
 colorFrom: blue
 colorTo: blue
-sdk: gradio
-sdk_version: 5.6.0
-python_version: "3.11.9"
-app_file: app.py
-pinned: false
+sdk: docker
+app_port: 7860
 license: mit
 ---
 

THIRD-PARTY-LICENSES

@@ -0,0 +1,143 @@
+stablediffusion.cpp - MIT
+
+OpenVINO stablediffusion engine - Apache 2
+
+SD Turbo -  STABILITY AI NON-COMMERCIAL RESEARCH COMMUNITY LICENSE AGREEMENT
+
+MIT License
+
+Copyright (c) 2023 leejet
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
+ERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+Definitions.
+
+"License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document.
+
+"Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License.
+
+"Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity.
+
+"You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License.
+
+"Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files.
+
+"Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types.
+
+"Work" shall mean the work of authorship, whether in Source or Object form, made available under the License, as indicated by a copyright notice that is included in or attached to the work (an example is provided in the Appendix below).
+
+"Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. For the purposes of this License, Derivative Works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work and Derivative Works thereof.
+
+"Contribution" shall mean any work of authorship, including the original version of the Work and any modifications or additions to that Work or Derivative Works thereof, that is intentionally submitted to Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submitted" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or otherwise designated in writing by the copyright owner as "Not a Contribution."
+
+"Contributor" shall mean Licensor and any individual or Legal Entity on behalf of whom a Contribution has been received by Licensor and subsequently incorporated within the Work.
+
+Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form.
+
+Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work, where such license applies only to those patent claims licensable by such Contributor that are necessarily infringed by their Contribution(s) alone or by combination of their Contribution(s) with the Work to which such Contribution(s) was submitted. If You institute patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Work or a Contribution incorporated within the Work constitutes direct or contributory patent infringement, then any patent licenses granted to You under this License for that Work shall terminate as of the date such litigation is filed.
+
+Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions:
+
+(a) You must give any other recipients of the Work or Derivative Works a copy of this License; and
+
+(b) You must cause any modified files to carry prominent notices stating that You changed the files; and
+
+(c) You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work, excluding those notices that do not pertain to any part of the Derivative Works; and
+
+(d) If the Work includes a "NOTICE" text file as part of its distribution, then any Derivative Works that You distribute must include a readable copy of the attribution notices contained within such NOTICE file, excluding those notices that do not pertain to any part of the Derivative Works, in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. The contents of the NOTICE file are for informational purposes only and do not modify the License. You may add Your own attribution notices within Derivative Works that You distribute, alongside or as an addendum to the NOTICE text from the Work, provided that such additional attribution notices cannot be construed as modifying the License.
+
+You may add Your own copyright statement to Your modifications and may provide additional or different license terms and conditions for use, reproduction, or distribution of Your modifications, or for any such Derivative Works as a whole, provided Your use, reproduction, and distribution of the Work otherwise complies with the conditions stated in this License.
+
+Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. Notwithstanding the above, nothing herein shall supersede or modify the terms of any separate license agreement you may have executed with Licensor regarding such Contributions.
+
+Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work and reproducing the content of the NOTICE file.
+
+Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied, including, without limitation, any warranties or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are solely responsible for determining the appropriateness of using or redistributing the Work and assume any risks associated with Your exercise of permissions under this License.
+
+Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or consequential damages of any character arising as a result of this License or out of the use or inability to use the Work (including but not limited to damages for loss of goodwill, work stoppage, computer failure or malfunction, or any and all other commercial damages or losses), even if such Contributor has been advised of the possibility of such damages.
+
+Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. However, in accepting such obligations, You may act only on Your own behalf and on Your sole responsibility, not on behalf of any other Contributor, and only if You agree to indemnify, defend, and hold each Contributor harmless for any liability incurred by, or claims asserted against, such Contributor by reason of your accepting any such warranty or additional liability.
+
+END OF TERMS AND CONDITIONS
+
+APPENDIX: How to apply the Apache License to your work.
+
+  To apply the Apache License to your work, attach the following
+  boilerplate notice, with the fields enclosed by brackets "[]"
+  replaced with your own identifying information. (Don't include
+  the brackets!)  The text should be enclosed in the appropriate
+  comment syntax for the file format. We also recommend that a
+  file or class name and description of purpose be included on the
+  same "printed page" as the copyright notice for easier
+  identification within third-party archives.
+Copyright [yyyy] [name of copyright owner]
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
+
+   <http://www.apache.org/licenses/LICENSE-2.0>
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
+
+STABILITY AI NON-COMMERCIAL RESEARCH COMMUNITY LICENSE AGREEMENT
+Dated: November 28, 2023
+
+By using or distributing any portion or element of the Models, Software, Software Products or Derivative Works, you agree to be bound by this Agreement.
+
+"Agreement" means this Stable Non-Commercial Research Community License Agreement.
+
+“AUP” means the Stability AI Acceptable Use Policy available at <https://stability.ai/use-policy>, as may be updated from time to time.
+
+"Derivative Work(s)” means (a) any derivative work of the Software Products as recognized by U.S. copyright laws and (b) any modifications to a Model, and any other model created which is based on or derived from the Model or the Model’s output. For clarity, Derivative Works do not include the output of any Model.
+
+“Documentation” means any specifications, manuals, documentation, and other written information provided by Stability AI related to the Software.
+
+"Licensee" or "you" means you, or your employer or any other person or entity (if you are entering into this Agreement on such person or entity's behalf), of the age required under applicable laws, rules or regulations to provide legal consent and that has legal authority to bind your employer or such other person or entity if you are entering in this Agreement on their behalf.
+
+“Model(s)" means, collectively, Stability AI’s proprietary models and algorithms, including machine-learning models, trained model weights and other elements of the foregoing, made available under this Agreement.
+
+“Non-Commercial Uses” means exercising any of the rights granted herein for the purpose of research or non-commercial purposes. Non-Commercial Uses does not include any production use of the Software Products or any Derivative Works.
+
+"Stability AI" or "we" means Stability AI Ltd. and its affiliates.
+
+"Software" means Stability AI’s proprietary software made available under this Agreement.
+
+“Software Products” means the Models, Software and Documentation, individually or in any combination.
+
+1.     License Rights and Redistribution. 
+
+a.  Subject to your compliance with this Agreement, the AUP (which is hereby incorporated herein by reference), and the Documentation, Stability AI grants you a non-exclusive, worldwide, non-transferable, non-sublicensable, revocable, royalty free and limited license under Stability AI’s intellectual property or other rights owned or controlled by Stability AI embodied in the Software Products to use, reproduce, distribute, and create Derivative Works of, the Software Products, in each case for Non-Commercial Uses only.
+
+b.  You may not use the Software Products or Derivative Works to enable third parties to use the Software Products or Derivative Works as part of your hosted service or via your APIs, whether you are adding substantial additional functionality thereto or not. Merely distributing the Software Products or Derivative Works for download online without offering any related service (ex. by distributing the Models on HuggingFace) is not a violation of this subsection. If you wish to use the Software Products or any Derivative Works for commercial or production use or you wish to make the Software Products or any Derivative Works available to third parties via your hosted service or your APIs, contact Stability AI at <https://stability.ai/contact>.
+
+c.  If you distribute or make the Software Products, or any Derivative Works thereof, available to a third party, the Software Products, Derivative Works, or any portion thereof, respectively, will remain subject to this Agreement and you must (i) provide a copy of this Agreement to such third party, and (ii) retain the following attribution notice within a "Notice" text file distributed as a part of such copies: "This Stability AI Model is licensed under the Stability AI Non-Commercial Research Community License, Copyright (c) Stability AI Ltd. All Rights Reserved.” If you create a Derivative Work of a Software Product, you may add your own attribution notices to the Notice file included with the Software Product, provided that you clearly indicate which attributions apply to the Software Product and you must state in the NOTICE file that you changed the Software Product and how it was modified.
+
+2.     Disclaimer of Warranty. UNLESS REQUIRED BY APPLICABLE LAW, THE SOFTWARE PRODUCTS  AND ANY OUTPUT AND RESULTS THERE FROM ARE PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING, WITHOUT LIMITATION, ANY WARRANTIES OF TITLE, NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. YOU ARE SOLELY RESPONSIBLE FOR DETERMINING THE APPROPRIATENESS OF USING OR REDISTRIBUTING THE SOFTWARE PRODUCTS, DERIVATIVE WORKS OR ANY OUTPUT OR RESULTS AND ASSUME ANY RISKS ASSOCIATED WITH YOUR USE OF THE SOFTWARE PRODUCTS, DERIVATIVE WORKS AND ANY OUTPUT AND RESULTS. 
+
+3.     Limitation of Liability. IN NO EVENT WILL STABILITY AI OR ITS AFFILIATES BE LIABLE UNDER ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, TORT, NEGLIGENCE, PRODUCTS LIABILITY, OR OTHERWISE, ARISING OUT OF THIS AGREEMENT, FOR ANY LOST PROFITS OR ANY DIRECT, INDIRECT, SPECIAL, CONSEQUENTIAL, INCIDENTAL, EXEMPLARY OR PUNITIVE DAMAGES, EVEN IF STABILITY AI OR ITS AFFILIATES HAVE BEEN ADVISED OF THE POSSIBILITY OF ANY OF THE FOREGOING. 
+
+4.     Intellectual Property.
+
+a.  No trademark licenses are granted under this Agreement, and in connection with the Software Products or Derivative Works, neither Stability AI nor Licensee may use any name or mark owned by or associated with the other or any of its affiliates, except as required for reasonable and customary use in describing and redistributing the Software Products or Derivative Works.
+
+b.  Subject to Stability AI’s ownership of the Software Products and Derivative Works made by or for Stability AI, with respect to any Derivative Works that are made by you, as between you and Stability AI, you are and will be the owner of such Derivative Works
+
+c.  If you institute litigation or other proceedings against Stability AI (including a cross-claim or counterclaim in a lawsuit) alleging that the Software Products, Derivative Works or associated outputs or results, or any portion of any of the foregoing, constitutes infringement of intellectual property or other rights owned or licensable by you, then any licenses granted to you under this Agreement shall terminate as of the date such litigation or claim is filed or instituted. You will indemnify and hold harmless Stability AI from and against any claim by any third party arising out of or related to your use or distribution of the Software Products or Derivative Works in violation of this Agreement.
+
+5.      Term and Termination. The term of this Agreement will commence upon your acceptance of this Agreement or access to the Software Products and will continue in full force and effect until terminated in accordance with the terms and conditions herein. Stability AI may terminate this Agreement if you are in breach of any term or condition of this Agreement. Upon termination of this Agreement, you shall delete and cease use of any Software Products or Derivative Works. Sections 2-4 shall survive the termination of this Agreement.

benchmark-openvino.bat

@@ -0,0 +1,23 @@
+@echo off
+setlocal
+
+set "PYTHON_COMMAND=python"
+
+call python --version > nul 2>&1
+if %errorlevel% equ 0 (
+    echo Python command check :OK
+) else (
+    echo "Error: Python command not found, please install Python (Recommended : Python 3.10 or Python 3.11) and try again"
+    pause
+    exit /b 1
+    
+)
+
+:check_python_version
+for /f "tokens=2" %%I in ('%PYTHON_COMMAND% --version 2^>^&1') do (
+    set "python_version=%%I"
+)
+
+echo Python version: %python_version%
+
+call "%~dp0env\Scripts\activate.bat"  && %PYTHON_COMMAND%  src/app.py -b --use_openvino  --openvino_lcm_model_id "rupeshs/sd-turbo-openvino"

benchmark.bat

@@ -0,0 +1,23 @@
+@echo off
+setlocal
+
+set "PYTHON_COMMAND=python"
+
+call python --version > nul 2>&1
+if %errorlevel% equ 0 (
+    echo Python command check :OK
+) else (
+    echo "Error: Python command not found, please install Python (Recommended : Python 3.10 or Python 3.11) and try again"
+    pause
+    exit /b 1
+    
+)
+
+:check_python_version
+for /f "tokens=2" %%I in ('%PYTHON_COMMAND% --version 2^>^&1') do (
+    set "python_version=%%I"
+)
+
+echo Python version: %python_version%
+
+call "%~dp0env\Scripts\activate.bat"  && %PYTHON_COMMAND%  src/app.py -b

configs/lcm-lora-models.txt

@@ -0,0 +1,4 @@
+latent-consistency/lcm-lora-sdv1-5
+latent-consistency/lcm-lora-sdxl
+latent-consistency/lcm-lora-ssd-1b
+rupeshs/hypersd-sd1-5-1-step-lora

configs/lcm-models.txt

@@ -0,0 +1,6 @@
+stabilityai/sd-turbo
+rupeshs/sdxs-512-0.9-orig-vae
+rupeshs/hyper-sd-sdxl-1-step
+rupeshs/SDXL-Lightning-2steps
+stabilityai/sdxl-turbo
+SimianLuo/LCM_Dreamshaper_v7

configs/openvino-lcm-models.txt

@@ -0,0 +1,10 @@
+rupeshs/sd-turbo-openvino
+rupeshs/sdxs-512-0.9-openvino
+rupeshs/hyper-sd-sdxl-1-step-openvino-int8
+rupeshs/SDXL-Lightning-2steps-openvino-int8
+rupeshs/sdxl-turbo-openvino-int8
+rupeshs/LCM-dreamshaper-v7-openvino
+Disty0/LCM_SoteMix
+rupeshs/sd15-lcm-square-openvino-int8
+OpenVINO/FLUX.1-schnell-int4-ov
+rupeshs/sana-sprint-0.6b-openvino-int4

configs/stable-diffusion-models.txt

@@ -0,0 +1,7 @@
+Lykon/dreamshaper-8
+Fictiverse/Stable_Diffusion_PaperCut_Model
+stabilityai/stable-diffusion-xl-base-1.0
+runwayml/stable-diffusion-v1-5
+segmind/SSD-1B
+stablediffusionapi/anything-v5
+prompthero/openjourney-v4

controlnet_models/Readme.txt

@@ -0,0 +1,3 @@
+Place your ControlNet models in this folder.
+You can download controlnet model (.safetensors) from  https://huggingface.co/comfyanonymous/ControlNet-v1-1_fp16_safetensors/tree/main
+E.g: https://huggingface.co/comfyanonymous/ControlNet-v1-1_fp16_safetensors/blob/main/control_v11p_sd15_canny_fp16.safetensors

install-mac.sh

@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+echo Starting FastSD CPU env installation...
+set -e
+PYTHON_COMMAND="python3"
+
+if ! command -v python3 &>/dev/null; then
+    if ! command -v python &>/dev/null; then
+        echo "Error: Python not found, please install python 3.8 or higher and try again"
+        exit 1
+    fi
+fi
+
+if command -v python &>/dev/null; then
+   PYTHON_COMMAND="python"
+fi
+
+echo "Found $PYTHON_COMMAND command"
+
+python_version=$($PYTHON_COMMAND --version 2>&1 | awk '{print $2}')  
+echo "Python version : $python_version"
+
+if ! command -v uv &>/dev/null; then
+    echo "Error: uv command not found,please install https://docs.astral.sh/uv/getting-started/installation/#__tabbed_1_1 and try again."
+    exit 1
+fi
+
+BASEDIR=$(pwd)
+
+uv venv --python 3.11.6 "$BASEDIR/env"
+# shellcheck disable=SC1091
+source "$BASEDIR/env/bin/activate"
+uv pip install torch==2.8.0
+uv pip install -r "$BASEDIR/requirements.txt"
+chmod +x "start.sh"
+chmod +x "start-webui.sh"
+read -n1 -r -p "FastSD CPU installation completed,press any key to continue..." key

install.bat

@@ -0,0 +1,38 @@
+
+@echo off
+setlocal
+echo Starting FastSD CPU env installation...
+
+set "PYTHON_COMMAND=python"
+
+call python --version > nul 2>&1
+if %errorlevel% equ 0 (
+    echo Python command check :OK
+) else (
+    echo "Error: Python command not found,please install Python(Recommended : Python 3.10 or Python 3.11) and try again."
+    pause
+    exit /b 1
+    
+)
+
+call uv --version > nul 2>&1
+if %errorlevel% equ 0 (
+    echo uv command check :OK
+) else (
+    echo "Error: uv command not found,please install https://docs.astral.sh/uv/getting-started/installation/#__tabbed_1_2 and try again."
+    pause
+    exit /b 1
+    
+)
+:check_python_version
+for /f "tokens=2" %%I in ('%PYTHON_COMMAND% --version 2^>^&1') do (
+    set "python_version=%%I"
+)
+
+echo Python version: %python_version%
+
+uv venv --python 3.11.6 "%~dp0env" 
+call "%~dp0env\Scripts\activate.bat" && uv pip install torch==2.8.0 --index-url https://download.pytorch.org/whl/cpu 
+call "%~dp0env\Scripts\activate.bat" && uv pip install -r "%~dp0requirements.txt"
+echo FastSD CPU env installation completed.
+pause

install.sh

@@ -0,0 +1,44 @@
+#!/usr/bin/env bash
+echo Starting FastSD CPU env installation...
+set -e
+PYTHON_COMMAND="python3"
+
+if ! command -v python3 &>/dev/null; then
+    if ! command -v python &>/dev/null; then
+        echo "Error: Python not found, please install python 3.8 or higher and try again"
+        exit 1
+    fi
+fi
+
+if command -v python &>/dev/null; then
+   PYTHON_COMMAND="python"
+fi
+
+echo "Found $PYTHON_COMMAND command"
+
+python_version=$($PYTHON_COMMAND --version 2>&1 | awk '{print $2}')  
+echo "Python version : $python_version"
+
+if ! command -v uv &>/dev/null; then
+    echo "Error: uv command not found,please install https://docs.astral.sh/uv/getting-started/installation/#__tabbed_1_1 and try again."
+    exit 1
+fi
+
+BASEDIR=$(pwd)
+
+uv venv --python 3.11.6 "$BASEDIR/env"
+# shellcheck disable=SC1091
+source "$BASEDIR/env/bin/activate"
+uv pip install torch==2.8.0 torchvision==0.23.0 --index-url https://download.pytorch.org/whl/cpu
+if [[ "$1" == "--disable-gui" ]]; then
+    #! For termux , we don't need Qt based GUI
+    packages="$(grep -v "^ *#\|^PyQt5" requirements.txt | grep .)" 
+    # shellcheck disable=SC2086
+    uv pip install $packages
+else
+    uv pip install -r "$BASEDIR/requirements.txt"
+fi
+
+chmod +x "start.sh"
+chmod +x "start-webui.sh"
+read -n1 -r -p "FastSD CPU installation completed,press any key to continue..." key

lora_models/Readme.txt

@@ -0,0 +1,3 @@
+Place your lora models in this folder.
+You can download lora model (.safetensors/Safetensor) from Civitai (https://civitai.com/) or Hugging Face(https://huggingface.co/)
+E.g: https://civitai.com/models/207984/cutecartoonredmond-15v-cute-cartoon-lora-for-liberteredmond-sd-15?modelVersionId=234192

models/gguf/clip/readme.txt

@@ -0,0 +1 @@
+Place CLIP model files here"

models/gguf/diffusion/readme.txt

@@ -0,0 +1 @@
+Place your diffusion gguf model files here

models/gguf/t5xxl/readme.txt

@@ -0,0 +1 @@
+Place T5-XXL model files here

models/gguf/vae/readme.txt

@@ -0,0 +1 @@
+Place VAE model files here 

requirements.txt

@@ -0,0 +1,23 @@
+accelerate==1.6.0
+diffusers==0.33.0
+transformers==4.48.0
+PyQt5
+Pillow==9.4.0
+openvino==2025.1.0
+optimum-intel==1.23.0
+onnx==1.16.0
+numpy==1.26.4
+onnxruntime==1.17.3
+pydantic
+typing-extensions==4.10.0
+pyyaml==6.0.1
+gradio==5.6.0
+peft==0.6.1
+opencv-python==4.8.1.78
+omegaconf==2.3.0
+controlnet-aux==0.0.7
+mediapipe==0.10.18
+tomesd==0.1.3
+mcp==1.6.0
+fastapi-mcp==0.3.0
+hf_xet

src/app.py

@@ -0,0 +1,554 @@
+import json
+from argparse import ArgumentParser
+
+from PIL import Image
+
+import constants
+from backend.controlnet import controlnet_settings_from_dict
+from backend.device import get_device_name
+from backend.models.gen_images import ImageFormat
+from backend.models.lcmdiffusion_setting import DiffusionTask
+from backend.upscale.tiled_upscale import generate_upscaled_image
+from constants import APP_VERSION, DEVICE
+from frontend.webui.image_variations_ui import generate_image_variations
+from models.interface_types import InterfaceType
+from paths import FastStableDiffusionPaths, ensure_path
+from state import get_context, get_settings
+from utils import show_system_info
+
+parser = ArgumentParser(description=f"FAST SD CPU {constants.APP_VERSION}")
+parser.add_argument(
+    "-s",
+    "--share",
+    action="store_true",
+    help="Create sharable link(Web UI)",
+    required=False,
+)
+group = parser.add_mutually_exclusive_group(required=False)
+group.add_argument(
+    "-g",
+    "--gui",
+    action="store_true",
+    help="Start desktop GUI",
+)
+group.add_argument(
+    "-w",
+    "--webui",
+    action="store_true",
+    help="Start Web UI",
+)
+group.add_argument(
+    "-a",
+    "--api",
+    action="store_true",
+    help="Start Web API server",
+)
+group.add_argument(
+    "-m",
+    "--mcp",
+    action="store_true",
+    help="Start MCP(Model Context Protocol) server",
+)
+group.add_argument(
+    "-r",
+    "--realtime",
+    action="store_true",
+    help="Start realtime inference UI(experimental)",
+)
+group.add_argument(
+    "-v",
+    "--version",
+    action="store_true",
+    help="Version",
+)
+
+parser.add_argument(
+    "-b",
+    "--benchmark",
+    action="store_true",
+    help="Run inference benchmark on the selected device",
+)
+parser.add_argument(
+    "--lcm_model_id",
+    type=str,
+    help="Model ID or path,Default stabilityai/sd-turbo",
+    default="stabilityai/sd-turbo",
+)
+parser.add_argument(
+    "--openvino_lcm_model_id",
+    type=str,
+    help="OpenVINO Model ID or path,Default rupeshs/sd-turbo-openvino",
+    default="rupeshs/sd-turbo-openvino",
+)
+parser.add_argument(
+    "--prompt",
+    type=str,
+    help="Describe the image you want to generate",
+    default="",
+)
+parser.add_argument(
+    "--negative_prompt",
+    type=str,
+    help="Describe what you want to exclude from the generation",
+    default="",
+)
+parser.add_argument(
+    "--image_height",
+    type=int,
+    help="Height of the image",
+    default=512,
+)
+parser.add_argument(
+    "--image_width",
+    type=int,
+    help="Width of the image",
+    default=512,
+)
+parser.add_argument(
+    "--inference_steps",
+    type=int,
+    help="Number of steps,default : 1",
+    default=1,
+)
+parser.add_argument(
+    "--guidance_scale",
+    type=float,
+    help="Guidance scale,default : 1.0",
+    default=1.0,
+)
+
+parser.add_argument(
+    "--number_of_images",
+    type=int,
+    help="Number of images to generate ,default : 1",
+    default=1,
+)
+parser.add_argument(
+    "--seed",
+    type=int,
+    help="Seed,default : -1 (disabled) ",
+    default=-1,
+)
+parser.add_argument(
+    "--use_openvino",
+    action="store_true",
+    help="Use OpenVINO model",
+)
+
+parser.add_argument(
+    "--use_offline_model",
+    action="store_true",
+    help="Use offline model",
+)
+parser.add_argument(
+    "--clip_skip",
+    type=int,
+    help="CLIP Skip (1-12), default : 1 (disabled) ",
+    default=1,
+)
+parser.add_argument(
+    "--token_merging",
+    type=float,
+    help="Token merging scale, 0.0 - 1.0, default : 0.0",
+    default=0.0,
+)
+
+parser.add_argument(
+    "--use_safety_checker",
+    action="store_true",
+    help="Use safety checker",
+)
+parser.add_argument(
+    "--use_lcm_lora",
+    action="store_true",
+    help="Use LCM-LoRA",
+)
+parser.add_argument(
+    "--base_model_id",
+    type=str,
+    help="LCM LoRA base model ID,Default Lykon/dreamshaper-8",
+    default="Lykon/dreamshaper-8",
+)
+parser.add_argument(
+    "--lcm_lora_id",
+    type=str,
+    help="LCM LoRA model ID,Default latent-consistency/lcm-lora-sdv1-5",
+    default="latent-consistency/lcm-lora-sdv1-5",
+)
+parser.add_argument(
+    "-i",
+    "--interactive",
+    action="store_true",
+    help="Interactive CLI mode",
+)
+parser.add_argument(
+    "-t",
+    "--use_tiny_auto_encoder",
+    action="store_true",
+    help="Use Tiny AutoEncoder for TAESD/TAESDXL/TAEF1",
+)
+parser.add_argument(
+    "-f",
+    "--file",
+    type=str,
+    help="Input image for img2img mode",
+    default="",
+)
+parser.add_argument(
+    "--img2img",
+    action="store_true",
+    help="img2img mode; requires input file via -f argument",
+)
+parser.add_argument(
+    "--batch_count",
+    type=int,
+    help="Number of sequential generations",
+    default=1,
+)
+parser.add_argument(
+    "--strength",
+    type=float,
+    help="Denoising strength for img2img and Image variations",
+    default=0.3,
+)
+parser.add_argument(
+    "--sdupscale",
+    action="store_true",
+    help="Tiled SD upscale,works only for the resolution 512x512,(2x upscale)",
+)
+parser.add_argument(
+    "--upscale",
+    action="store_true",
+    help="EDSR SD upscale ",
+)
+parser.add_argument(
+    "--custom_settings",
+    type=str,
+    help="JSON file containing custom generation settings",
+    default=None,
+)
+parser.add_argument(
+    "--usejpeg",
+    action="store_true",
+    help="Images will be saved as JPEG format",
+)
+parser.add_argument(
+    "--noimagesave",
+    action="store_true",
+    help="Disable image saving",
+)
+parser.add_argument(
+    "--imagequality", type=int, help="Output image quality [0 to 100]", default=90
+)
+parser.add_argument(
+    "--lora",
+    type=str,
+    help="LoRA model full path e.g D:\lora_models\CuteCartoon15V-LiberteRedmodModel-Cartoon-CuteCartoonAF.safetensors",
+    default=None,
+)
+parser.add_argument(
+    "--lora_weight",
+    type=float,
+    help="LoRA adapter weight [0 to 1.0]",
+    default=0.5,
+)
+parser.add_argument(
+    "--port",
+    type=int,
+    help="Web server port",
+    default=8000,
+)
+
+args = parser.parse_args()
+
+if args.version:
+    print(APP_VERSION)
+    exit()
+
+# parser.print_help()
+print("FastSD CPU - ", APP_VERSION)
+show_system_info()
+print(f"Using device : {constants.DEVICE}")
+
+
+if args.webui:
+    app_settings = get_settings()
+else:
+    app_settings = get_settings()
+
+print(f"Output path : {app_settings.settings.generated_images.path}")
+ensure_path(app_settings.settings.generated_images.path)
+
+print(f"Found {len(app_settings.lcm_models)} LCM models in config/lcm-models.txt")
+print(
+    f"Found {len(app_settings.stable_diffsuion_models)} stable diffusion models in config/stable-diffusion-models.txt"
+)
+print(
+    f"Found {len(app_settings.lcm_lora_models)} LCM-LoRA models in config/lcm-lora-models.txt"
+)
+print(
+    f"Found {len(app_settings.openvino_lcm_models)} OpenVINO LCM models in config/openvino-lcm-models.txt"
+)
+
+if args.noimagesave:
+    app_settings.settings.generated_images.save_image = False
+else:
+    app_settings.settings.generated_images.save_image = True
+
+app_settings.settings.generated_images.save_image_quality = args.imagequality
+
+if not args.realtime:
+    # To minimize realtime mode dependencies
+    from backend.upscale.upscaler import upscale_image
+    from frontend.cli_interactive import interactive_mode
+
+if args.gui:
+    from frontend.gui.ui import start_gui
+
+    print("Starting desktop GUI mode(Qt)")
+    start_gui(
+        [],
+        app_settings,
+    )
+elif args.webui:
+    from frontend.webui.ui import start_webui
+
+    print("Starting web UI mode")
+    start_webui(
+        args.share,
+    )
+elif args.realtime:
+    from frontend.webui.realtime_ui import start_realtime_text_to_image
+
+    print("Starting realtime text to image(EXPERIMENTAL)")
+    start_realtime_text_to_image(args.share)
+elif args.api:
+    from backend.api.web import start_web_server
+
+    start_web_server(args.port)
+elif args.mcp:
+    from backend.api.mcp_server import start_mcp_server
+
+    start_mcp_server(args.port)
+else:
+    context = get_context(InterfaceType.CLI)
+    config = app_settings.settings
+
+    if args.use_openvino:
+        config.lcm_diffusion_setting.openvino_lcm_model_id = args.openvino_lcm_model_id
+    else:
+        config.lcm_diffusion_setting.lcm_model_id = args.lcm_model_id
+
+    config.lcm_diffusion_setting.prompt = args.prompt
+    config.lcm_diffusion_setting.negative_prompt = args.negative_prompt
+    config.lcm_diffusion_setting.image_height = args.image_height
+    config.lcm_diffusion_setting.image_width = args.image_width
+    config.lcm_diffusion_setting.guidance_scale = args.guidance_scale
+    config.lcm_diffusion_setting.number_of_images = args.number_of_images
+    config.lcm_diffusion_setting.inference_steps = args.inference_steps
+    config.lcm_diffusion_setting.strength = args.strength
+    config.lcm_diffusion_setting.seed = args.seed
+    config.lcm_diffusion_setting.use_openvino = args.use_openvino
+    config.lcm_diffusion_setting.use_tiny_auto_encoder = args.use_tiny_auto_encoder
+    config.lcm_diffusion_setting.use_lcm_lora = args.use_lcm_lora
+    config.lcm_diffusion_setting.lcm_lora.base_model_id = args.base_model_id
+    config.lcm_diffusion_setting.lcm_lora.lcm_lora_id = args.lcm_lora_id
+    config.lcm_diffusion_setting.diffusion_task = DiffusionTask.text_to_image.value
+    config.lcm_diffusion_setting.lora.enabled = False
+    config.lcm_diffusion_setting.lora.path = args.lora
+    config.lcm_diffusion_setting.lora.weight = args.lora_weight
+    config.lcm_diffusion_setting.lora.fuse = False
+    if config.lcm_diffusion_setting.lora.path:
+        config.lcm_diffusion_setting.lora.enabled = True
+    if args.usejpeg:
+        config.generated_images.format = ImageFormat.JPEG.value.upper()
+    if args.seed > -1:
+        config.lcm_diffusion_setting.use_seed = True
+    else:
+        config.lcm_diffusion_setting.use_seed = False
+    config.lcm_diffusion_setting.use_offline_model = args.use_offline_model
+    config.lcm_diffusion_setting.clip_skip = args.clip_skip
+    config.lcm_diffusion_setting.token_merging = args.token_merging
+    config.lcm_diffusion_setting.use_safety_checker = args.use_safety_checker
+
+    # Read custom settings from JSON file
+    custom_settings = {}
+    if args.custom_settings:
+        with open(args.custom_settings) as f:
+            custom_settings = json.load(f)
+
+    # Basic ControlNet settings; if ControlNet is enabled, an image is
+    # required even in txt2img mode
+    config.lcm_diffusion_setting.controlnet = None
+    controlnet_settings_from_dict(
+        config.lcm_diffusion_setting,
+        custom_settings,
+    )
+
+    # Interactive mode
+    if args.interactive:
+        # wrapper(interactive_mode, config, context)
+        config.lcm_diffusion_setting.lora.fuse = False
+        interactive_mode(config, context)
+
+    # Start of non-interactive CLI image generation
+    if args.img2img and args.file != "":
+        config.lcm_diffusion_setting.init_image = Image.open(args.file)
+        config.lcm_diffusion_setting.diffusion_task = DiffusionTask.image_to_image.value
+    elif args.img2img and args.file == "":
+        print("Error : You need to specify a file in img2img mode")
+        exit()
+    elif args.upscale and args.file == "" and args.custom_settings == None:
+        print("Error : You need to specify a file in SD upscale mode")
+        exit()
+    elif (
+        args.prompt == ""
+        and args.file == ""
+        and args.custom_settings == None
+        and not args.benchmark
+    ):
+        print("Error : You need to provide a prompt")
+        exit()
+
+    if args.upscale:
+        # image = Image.open(args.file)
+        output_path = FastStableDiffusionPaths.get_upscale_filepath(
+            args.file,
+            2,
+            config.generated_images.format,
+        )
+        result = upscale_image(
+            context,
+            args.file,
+            output_path,
+            2,
+        )
+    # Perform Tiled SD upscale (EXPERIMENTAL)
+    elif args.sdupscale:
+        if args.use_openvino:
+            config.lcm_diffusion_setting.strength = 0.3
+        upscale_settings = None
+        if custom_settings != {}:
+            upscale_settings = custom_settings
+        filepath = args.file
+        output_format = config.generated_images.format
+        if upscale_settings:
+            filepath = upscale_settings["source_file"]
+            output_format = upscale_settings["output_format"].upper()
+        output_path = FastStableDiffusionPaths.get_upscale_filepath(
+            filepath,
+            2,
+            output_format,
+        )
+
+        generate_upscaled_image(
+            config,
+            filepath,
+            config.lcm_diffusion_setting.strength,
+            upscale_settings=upscale_settings,
+            context=context,
+            tile_overlap=32 if config.lcm_diffusion_setting.use_openvino else 16,
+            output_path=output_path,
+            image_format=output_format,
+        )
+        exit()
+    # If img2img argument is set and prompt is empty, use image variations mode
+    elif args.img2img and args.prompt == "":
+        for i in range(0, args.batch_count):
+            generate_image_variations(
+                config.lcm_diffusion_setting.init_image, args.strength
+            )
+    else:
+        if args.benchmark:
+            print("Initializing benchmark...")
+            bench_lcm_setting = config.lcm_diffusion_setting
+            bench_lcm_setting.prompt = "a cat"
+            bench_lcm_setting.use_tiny_auto_encoder = False
+            context.generate_text_to_image(
+                settings=config,
+                device=DEVICE,
+            )
+
+            latencies = []
+
+            print("Starting benchmark please wait...")
+            for _ in range(3):
+                context.generate_text_to_image(
+                    settings=config,
+                    device=DEVICE,
+                )
+                latencies.append(context.latency)
+
+            avg_latency = sum(latencies) / 3
+
+            bench_lcm_setting.use_tiny_auto_encoder = True
+
+            context.generate_text_to_image(
+                settings=config,
+                device=DEVICE,
+            )
+            latencies = []
+            for _ in range(3):
+                context.generate_text_to_image(
+                    settings=config,
+                    device=DEVICE,
+                )
+                latencies.append(context.latency)
+
+            avg_latency_taesd = sum(latencies) / 3
+
+            benchmark_name = ""
+
+            if config.lcm_diffusion_setting.use_openvino:
+                benchmark_name = "OpenVINO"
+            else:
+                benchmark_name = "PyTorch"
+
+            bench_model_id = ""
+            if bench_lcm_setting.use_openvino:
+                bench_model_id = bench_lcm_setting.openvino_lcm_model_id
+            elif bench_lcm_setting.use_lcm_lora:
+                bench_model_id = bench_lcm_setting.lcm_lora.base_model_id
+            else:
+                bench_model_id = bench_lcm_setting.lcm_model_id
+
+            benchmark_result = [
+                ["Device", f"{DEVICE.upper()},{get_device_name()}"],
+                ["Stable Diffusion Model", bench_model_id],
+                [
+                    "Image Size ",
+                    f"{bench_lcm_setting.image_width}x{bench_lcm_setting.image_height}",
+                ],
+                [
+                    "Inference Steps",
+                    f"{bench_lcm_setting.inference_steps}",
+                ],
+                [
+                    "Benchmark Passes",
+                    3,
+                ],
+                [
+                    "Average Latency",
+                    f"{round(avg_latency, 3)} sec",
+                ],
+                [
+                    "Average Latency(TAESD* enabled)",
+                    f"{round(avg_latency_taesd, 3)} sec",
+                ],
+            ]
+            print()
+            print(
+                f"                          FastSD Benchmark - {benchmark_name:8}                         "
+            )
+            print(f"-" * 80)
+            for benchmark in benchmark_result:
+                print(f"{benchmark[0]:35} - {benchmark[1]}")
+            print(f"-" * 80)
+            print("*TAESD - Tiny AutoEncoder for Stable Diffusion")
+
+        else:
+            for i in range(0, args.batch_count):
+                context.generate_text_to_image(
+                    settings=config,
+                    device=DEVICE,
+                )

src/app_settings.py

@@ -0,0 +1,124 @@
+from copy import deepcopy
+from os import makedirs, path
+
+import yaml
+from constants import (
+    LCM_LORA_MODELS_FILE,
+    LCM_MODELS_FILE,
+    OPENVINO_LCM_MODELS_FILE,
+    SD_MODELS_FILE,
+)
+from paths import FastStableDiffusionPaths, join_paths
+from utils import get_files_in_dir, get_models_from_text_file
+
+from models.settings import Settings
+
+
+class AppSettings:
+    def __init__(self):
+        self.config_path = FastStableDiffusionPaths().get_app_settings_path()
+        self._stable_diffsuion_models = get_models_from_text_file(
+            FastStableDiffusionPaths().get_models_config_path(SD_MODELS_FILE)
+        )
+        self._lcm_lora_models = get_models_from_text_file(
+            FastStableDiffusionPaths().get_models_config_path(LCM_LORA_MODELS_FILE)
+        )
+        self._openvino_lcm_models = get_models_from_text_file(
+            FastStableDiffusionPaths().get_models_config_path(OPENVINO_LCM_MODELS_FILE)
+        )
+        self._lcm_models = get_models_from_text_file(
+            FastStableDiffusionPaths().get_models_config_path(LCM_MODELS_FILE)
+        )
+        self._gguf_diffusion_models = get_files_in_dir(
+            join_paths(FastStableDiffusionPaths().get_gguf_models_path(), "diffusion")
+        )
+        self._gguf_clip_models = get_files_in_dir(
+            join_paths(FastStableDiffusionPaths().get_gguf_models_path(), "clip")
+        )
+        self._gguf_vae_models = get_files_in_dir(
+            join_paths(FastStableDiffusionPaths().get_gguf_models_path(), "vae")
+        )
+        self._gguf_t5xxl_models = get_files_in_dir(
+            join_paths(FastStableDiffusionPaths().get_gguf_models_path(), "t5xxl")
+        )
+        self._config = None
+
+    @property
+    def settings(self):
+        return self._config
+
+    @property
+    def stable_diffsuion_models(self):
+        return self._stable_diffsuion_models
+
+    @property
+    def openvino_lcm_models(self):
+        return self._openvino_lcm_models
+
+    @property
+    def lcm_models(self):
+        return self._lcm_models
+
+    @property
+    def lcm_lora_models(self):
+        return self._lcm_lora_models
+
+    @property
+    def gguf_diffusion_models(self):
+        return self._gguf_diffusion_models
+
+    @property
+    def gguf_clip_models(self):
+        return self._gguf_clip_models
+
+    @property
+    def gguf_vae_models(self):
+        return self._gguf_vae_models
+
+    @property
+    def gguf_t5xxl_models(self):
+        return self._gguf_t5xxl_models
+
+    def load(self, skip_file=False):
+        if skip_file:
+            print("Skipping config file")
+            settings_dict = self._load_default()
+            self._config = Settings.model_validate(settings_dict)
+        else:
+            if not path.exists(self.config_path):
+                base_dir = path.dirname(self.config_path)
+                if not path.exists(base_dir):
+                    makedirs(base_dir)
+                try:
+                    print("Settings not found creating default settings")
+                    with open(self.config_path, "w") as file:
+                        yaml.dump(
+                            self._load_default(),
+                            file,
+                        )
+                except Exception as ex:
+                    print(f"Error in creating settings : {ex}")
+                    exit()
+            try:
+                with open(self.config_path) as file:
+                    settings_dict = yaml.safe_load(file)
+                    self._config = Settings.model_validate(settings_dict)
+            except Exception as ex:
+                print(f"Error in loading settings : {ex}")
+
+    def save(self):
+        try:
+            with open(self.config_path, "w") as file:
+                tmp_cfg = deepcopy(self._config)
+                tmp_cfg.lcm_diffusion_setting.init_image = None
+                configurations = tmp_cfg.model_dump(
+                    exclude=["init_image"],
+                )
+                if configurations:
+                    yaml.dump(configurations, file)
+        except Exception as ex:
+            print(f"Error in saving settings : {ex}")
+
+    def _load_default(self) -> dict:
+        default_config = Settings()
+        return default_config.model_dump()

src/backend/annotators/canny_control.py

@@ -0,0 +1,15 @@
+import numpy as np
+from backend.annotators.control_interface import ControlInterface
+from cv2 import Canny
+from PIL import Image
+
+
+class CannyControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        low_threshold = 100
+        high_threshold = 200
+        image = np.array(image)
+        image = Canny(image, low_threshold, high_threshold)
+        image = image[:, :, None]
+        image = np.concatenate([image, image, image], axis=2)
+        return Image.fromarray(image)

src/backend/annotators/control_interface.py

@@ -0,0 +1,12 @@
+from abc import ABC, abstractmethod
+
+from PIL import Image
+
+
+class ControlInterface(ABC):
+    @abstractmethod
+    def get_control_image(
+        self,
+        image: Image,
+    ) -> Image:
+        pass

src/backend/annotators/depth_control.py

@@ -0,0 +1,15 @@
+import numpy as np
+from backend.annotators.control_interface import ControlInterface
+from PIL import Image
+from transformers import pipeline
+
+
+class DepthControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        depth_estimator = pipeline("depth-estimation")
+        image = depth_estimator(image)["depth"]
+        image = np.array(image)
+        image = image[:, :, None]
+        image = np.concatenate([image, image, image], axis=2)
+        image = Image.fromarray(image)
+        return image

src/backend/annotators/image_control_factory.py

@@ -0,0 +1,31 @@
+from backend.annotators.canny_control import CannyControl
+from backend.annotators.depth_control import DepthControl
+from backend.annotators.lineart_control import LineArtControl
+from backend.annotators.mlsd_control import MlsdControl
+from backend.annotators.normal_control import NormalControl
+from backend.annotators.pose_control import PoseControl
+from backend.annotators.shuffle_control import ShuffleControl
+from backend.annotators.softedge_control import SoftEdgeControl
+
+
+class ImageControlFactory:
+    def create_control(self, controlnet_type: str):
+        if controlnet_type == "Canny":
+            return CannyControl()
+        elif controlnet_type == "Pose":
+            return PoseControl()
+        elif controlnet_type == "MLSD":
+            return MlsdControl()
+        elif controlnet_type == "Depth":
+            return DepthControl()
+        elif controlnet_type == "LineArt":
+            return LineArtControl()
+        elif controlnet_type == "Shuffle":
+            return ShuffleControl()
+        elif controlnet_type == "NormalBAE":
+            return NormalControl()
+        elif controlnet_type == "SoftEdge":
+            return SoftEdgeControl()
+        else:
+            print("Error: Control type not implemented!")
+            raise Exception("Error: Control type not implemented!")

src/backend/annotators/lineart_control.py

@@ -0,0 +1,11 @@
+import numpy as np
+from backend.annotators.control_interface import ControlInterface
+from controlnet_aux import LineartDetector
+from PIL import Image
+
+
+class LineArtControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        processor = LineartDetector.from_pretrained("lllyasviel/Annotators")
+        control_image = processor(image)
+        return control_image

src/backend/annotators/mlsd_control.py

@@ -0,0 +1,10 @@
+from backend.annotators.control_interface import ControlInterface
+from controlnet_aux import MLSDdetector
+from PIL import Image
+
+
+class MlsdControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        mlsd = MLSDdetector.from_pretrained("lllyasviel/ControlNet")
+        image = mlsd(image)
+        return image

src/backend/annotators/normal_control.py

@@ -0,0 +1,10 @@
+from backend.annotators.control_interface import ControlInterface
+from controlnet_aux import NormalBaeDetector
+from PIL import Image
+
+
+class NormalControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        processor = NormalBaeDetector.from_pretrained("lllyasviel/Annotators")
+        control_image = processor(image)
+        return control_image

src/backend/annotators/pose_control.py

@@ -0,0 +1,10 @@
+from backend.annotators.control_interface import ControlInterface
+from controlnet_aux import OpenposeDetector
+from PIL import Image
+
+
+class PoseControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        openpose = OpenposeDetector.from_pretrained("lllyasviel/ControlNet")
+        image = openpose(image)
+        return image

src/backend/annotators/shuffle_control.py

@@ -0,0 +1,10 @@
+from backend.annotators.control_interface import ControlInterface
+from controlnet_aux import ContentShuffleDetector
+from PIL import Image
+
+
+class ShuffleControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        shuffle_processor = ContentShuffleDetector()
+        image = shuffle_processor(image)
+        return image

src/backend/annotators/softedge_control.py

@@ -0,0 +1,10 @@
+from backend.annotators.control_interface import ControlInterface
+from controlnet_aux import PidiNetDetector
+from PIL import Image
+
+
+class SoftEdgeControl(ControlInterface):
+    def get_control_image(self, image: Image) -> Image:
+        processor = PidiNetDetector.from_pretrained("lllyasviel/Annotators")
+        control_image = processor(image)
+        return control_image

src/backend/api/mcp_server.py

@@ -0,0 +1,100 @@
+import platform
+
+import uvicorn
+from backend.device import get_device_name
+from backend.models.device import DeviceInfo
+from constants import APP_VERSION, DEVICE
+from context import Context
+from fastapi import FastAPI, Request
+from fastapi_mcp import FastApiMCP
+from state import get_settings
+from fastapi.middleware.cors import CORSMiddleware
+from models.interface_types import InterfaceType
+from fastapi.staticfiles import StaticFiles
+
+SERVER_PORT = 8000
+
+app_settings = get_settings()
+app = FastAPI(
+    title="FastSD CPU",
+    description="Fast stable diffusion on CPU",
+    version=APP_VERSION,
+    license_info={
+        "name": "MIT",
+        "identifier": "MIT",
+    },
+    describe_all_responses=True,
+    describe_full_response_schema=True,
+)
+origins = ["*"]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+context = Context(InterfaceType.API_SERVER)
+app.mount("/results", StaticFiles(directory="results"), name="results")
+
+
+@app.get(
+    "/info",
+    description="Get system information",
+    summary="Get system information",
+    operation_id="get_system_info",
+)
+async def info() -> dict:
+    device_info = DeviceInfo(
+        device_type=DEVICE,
+        device_name=get_device_name(),
+        os=platform.system(),
+        platform=platform.platform(),
+        processor=platform.processor(),
+    )
+    return device_info.model_dump()
+
+
+@app.post(
+    "/generate",
+    description="Generate image from text prompt",
+    summary="Text to image generation",
+    operation_id="generate",
+)
+async def generate(
+    prompt: str,
+    request: Request,
+) -> str:
+    """
+    Returns URL of the generated image for text prompt
+    """
+    app_settings.settings.lcm_diffusion_setting.prompt = prompt
+    images = context.generate_text_to_image(app_settings.settings)
+    image_names = context.save_images(
+        images,
+        app_settings.settings,
+    )
+    # url = request.url_for("results", path=image_names[0]) - Claude Desktop returns api_server
+    url = f"http://localhost:{SERVER_PORT}/results/{image_names[0]}"
+    image_url = f"The generated image available at the URL {url}"
+    return image_url
+
+
+def start_mcp_server(port: int = 8000):
+    global SERVER_PORT
+    SERVER_PORT = port
+    print(f"Starting MCP server on port {port}...")
+    mcp = FastApiMCP(
+        app,
+        name="FastSDCPU MCP",
+        description="MCP server for FastSD CPU API",
+    )
+
+    mcp.mount()
+    uvicorn.run(
+        app,
+        host="0.0.0.0",
+        port=port,
+    )

src/backend/api/models/response.py

@@ -0,0 +1,18 @@
+from typing import List
+
+from pydantic import BaseModel
+
+
+class StableDiffusionResponse(BaseModel):
+    """
+    Stable diffusion response model
+
+    Attributes:
+        images (List[str]): List of JPEG image as base64 encoded
+        latency (float): Latency in seconds
+        error (str): Error message if any
+    """
+
+    images: List[str]
+    latency: float
+    error: str = ""

src/backend/api/web.py

@@ -0,0 +1,116 @@
+import platform
+
+import uvicorn
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from backend.api.models.response import StableDiffusionResponse
+from backend.base64_image import base64_image_to_pil, pil_image_to_base64_str
+from backend.device import get_device_name
+from backend.models.device import DeviceInfo
+from backend.models.lcmdiffusion_setting import DiffusionTask, LCMDiffusionSetting
+from constants import APP_VERSION, DEVICE
+from context import Context
+from models.interface_types import InterfaceType
+from state import get_settings
+
+app_settings = get_settings()
+app = FastAPI(
+    title="FastSD CPU",
+    description="Fast stable diffusion on CPU",
+    version=APP_VERSION,
+    license_info={
+        "name": "MIT",
+        "identifier": "MIT",
+    },
+    docs_url="/api/docs",
+    redoc_url="/api/redoc",
+    openapi_url="/api/openapi.json",
+)
+print(app_settings.settings.lcm_diffusion_setting)
+origins = ["*"]
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+context = Context(InterfaceType.API_SERVER)
+
+
+@app.get("/api/")
+async def root():
+    return {"message": "Welcome to FastSD CPU API"}
+
+
+@app.get(
+    "/api/info",
+    description="Get system information",
+    summary="Get system information",
+)
+async def info():
+    device_info = DeviceInfo(
+        device_type=DEVICE,
+        device_name=get_device_name(),
+        os=platform.system(),
+        platform=platform.platform(),
+        processor=platform.processor(),
+    )
+    return device_info.model_dump()
+
+
+@app.get(
+    "/api/config",
+    description="Get current configuration",
+    summary="Get configurations",
+)
+async def config():
+    return app_settings.settings
+
+
+@app.get(
+    "/api/models",
+    description="Get available models",
+    summary="Get available models",
+)
+async def models():
+    return {
+        "lcm_lora_models": app_settings.lcm_lora_models,
+        "stable_diffusion": app_settings.stable_diffsuion_models,
+        "openvino_models": app_settings.openvino_lcm_models,
+        "lcm_models": app_settings.lcm_models,
+    }
+
+
+@app.post(
+    "/api/generate",
+    description="Generate image(Text to image,Image to Image)",
+    summary="Generate image(Text to image,Image to Image)",
+)
+async def generate(diffusion_config: LCMDiffusionSetting) -> StableDiffusionResponse:
+    app_settings.settings.lcm_diffusion_setting = diffusion_config
+    if diffusion_config.diffusion_task == DiffusionTask.image_to_image:
+        app_settings.settings.lcm_diffusion_setting.init_image = base64_image_to_pil(
+            diffusion_config.init_image
+        )
+
+    images = context.generate_text_to_image(app_settings.settings)
+
+    if images:
+        images_base64 = [pil_image_to_base64_str(img) for img in images]
+    else:
+        images_base64 = []
+    return StableDiffusionResponse(
+        latency=round(context.latency, 2),
+        images=images_base64,
+        error=context.error,
+    )
+
+
+def start_web_server(port: int = 8000):
+    uvicorn.run(
+        app,
+        host="0.0.0.0",
+        port=port,
+    )

src/backend/base64_image.py

@@ -0,0 +1,21 @@
+from io import BytesIO
+from base64 import b64encode, b64decode
+from PIL import Image
+
+
+def pil_image_to_base64_str(
+    image: Image,
+    format: str = "JPEG",
+) -> str:
+    buffer = BytesIO()
+    image.save(buffer, format=format)
+    buffer.seek(0)
+    img_base64 = b64encode(buffer.getvalue()).decode("utf-8")
+    return img_base64
+
+
+def base64_image_to_pil(base64_str) -> Image:
+    image_data = b64decode(base64_str)
+    image_buffer = BytesIO(image_data)
+    image = Image.open(image_buffer)
+    return image

src/backend/controlnet.py

@@ -0,0 +1,156 @@
+import logging
+from PIL import Image
+from typing import Any
+from diffusers import (
+    ControlNetModel,
+    AutoPipelineForText2Image,
+    LCMScheduler,
+    AutoencoderKL,
+    StableDiffusionPipeline,
+    StableDiffusionXLPipeline,
+    StableDiffusionControlNetPipeline,
+    StableDiffusionXLControlNetPipeline,
+    StableDiffusionControlNetImg2ImgPipeline,
+    StableDiffusionXLControlNetImg2ImgPipeline,
+)
+from backend.models.lcmdiffusion_setting import (
+    DiffusionTask,
+    ControlNetSetting,
+)
+
+
+# Prepares ControlNet adapters for use with FastSD CPU
+#
+# This function loads the ControlNet adapters defined by the
+# _lcm_diffusion_setting.controlnet_ object and returns a dictionary
+# with the pipeline arguments required to use the loaded adapters
+def load_controlnet_adapters(lcm_diffusion_setting) -> dict:
+    controlnet_args = {}
+    if (
+        lcm_diffusion_setting.controlnet is None
+        or not lcm_diffusion_setting.controlnet.enabled
+    ):
+        return controlnet_args
+
+    logging.info("Loading ControlNet adapter")
+    controlnet_adapter = ControlNetModel.from_single_file(
+        lcm_diffusion_setting.controlnet.adapter_path,
+        # local_files_only=True,
+        use_safetensors=True,
+    )
+    controlnet_args["controlnet"] = controlnet_adapter
+    return controlnet_args
+
+
+# Updates the ControlNet pipeline arguments to use for image generation
+#
+# This function uses the contents of the _lcm_diffusion_setting.controlnet_
+# object to generate a dictionary with the corresponding pipeline arguments
+# to be used for image generation; in particular, it sets the ControlNet control
+# image and conditioning scale
+def update_controlnet_arguments(lcm_diffusion_setting) -> dict:
+    controlnet_args = {}
+    if (
+        lcm_diffusion_setting.controlnet is None
+        or not lcm_diffusion_setting.controlnet.enabled
+    ):
+        return controlnet_args
+
+    controlnet_args["controlnet_conditioning_scale"] = (
+        lcm_diffusion_setting.controlnet.conditioning_scale
+    )
+    if lcm_diffusion_setting.diffusion_task == DiffusionTask.text_to_image.value:
+        controlnet_args["image"] = lcm_diffusion_setting.controlnet._control_image
+    elif lcm_diffusion_setting.diffusion_task == DiffusionTask.image_to_image.value:
+        controlnet_args["control_image"] = (
+            lcm_diffusion_setting.controlnet._control_image
+        )
+    return controlnet_args
+
+
+# Helper function to adjust ControlNet settings from a dictionary
+def controlnet_settings_from_dict(
+    lcm_diffusion_setting,
+    dictionary,
+) -> None:
+    if lcm_diffusion_setting is None or dictionary is None:
+        logging.error("Invalid arguments!")
+        return
+    if (
+        "controlnet" not in dictionary
+        or dictionary["controlnet"] is None
+        or len(dictionary["controlnet"]) == 0
+    ):
+        logging.warning("ControlNet settings not found, ControlNet will be disabled")
+        lcm_diffusion_setting.controlnet = None
+        return
+
+    controlnet = ControlNetSetting()
+    controlnet.enabled = dictionary["controlnet"][0]["enabled"]
+    controlnet.conditioning_scale = dictionary["controlnet"][0]["conditioning_scale"]
+    controlnet.adapter_path = dictionary["controlnet"][0]["adapter_path"]
+    controlnet._control_image = None
+    image_path = dictionary["controlnet"][0]["control_image"]
+    if controlnet.enabled:
+        try:
+            controlnet._control_image = Image.open(image_path)
+        except (AttributeError, FileNotFoundError) as err:
+            print(err)
+        if controlnet._control_image is None:
+            logging.error("Wrong ControlNet control image! Disabling ControlNet")
+            controlnet.enabled = False
+    lcm_diffusion_setting.controlnet = controlnet
+
+
+def get_controlnet_pipeline(
+    pipeline: Any, lcm_diffusion_setting, diffusion_task: DiffusionTask
+) -> Any:
+    """Creates a ControlNet pipeline from the base txt2img _pipeline_"""
+    if (
+        lcm_diffusion_setting.controlnet is None
+        or not lcm_diffusion_setting.controlnet.enabled
+    ):
+        return None
+    components = pipeline.components
+    pipeline_class = pipeline.__class__.__name__
+    controlnet_args = load_controlnet_adapters(lcm_diffusion_setting)
+    if diffusion_task == DiffusionTask.text_to_image.value:
+        if (
+            pipeline_class == "LatentConsistencyModelPipeline"
+            or pipeline_class == "StableDiffusionPipeline"
+        ):
+            controlnet_pipeline = StableDiffusionControlNetPipeline.from_pipe(
+                pipeline,
+                vae=None,
+                **controlnet_args,
+            )
+            controlnet_pipeline.vae = pipeline.vae
+            return controlnet_pipeline
+        elif pipeline_class == "StableDiffusionXLPipeline":
+            controlnet_pipeline = StableDiffusionXLControlNetPipeline.from_pipe(
+                pipeline,
+                vae=None,
+                **controlnet_args,
+            )
+            controlnet_pipeline.vae = pipeline.vae
+            return controlnet_pipeline
+    elif diffusion_task == DiffusionTask.image_to_image.value:
+        if (
+            pipeline_class == "LatentConsistencyModelPipeline"
+            or pipeline_class == "StableDiffusionPipeline"
+        ):
+            controlnet_pipeline = StableDiffusionControlNetImg2ImgPipeline.from_pipe(
+                pipeline,
+                vae=None,
+                **controlnet_args,
+            )
+            controlnet_pipeline.vae = pipeline.vae
+            return controlnet_pipeline
+        elif pipeline_class == "StableDiffusionXLPipeline":
+            controlnet_pipeline = StableDiffusionXLControlNetImg2ImgPipeline.from_pipe(
+                pipeline,
+                vae=None,
+                **controlnet_args,
+            )
+            controlnet_pipeline.vae = pipeline.vae
+            return controlnet_pipeline

src/backend/device.py

@@ -0,0 +1,23 @@
+import platform
+from constants import DEVICE
+import torch
+import openvino as ov
+
+core = ov.Core()
+
+
+def is_openvino_device() -> bool:
+    if DEVICE.lower() == "cpu" or DEVICE.lower()[0] == "g" or DEVICE.lower()[0] == "n":
+        return True
+    else:
+        return False
+
+
+def get_device_name() -> str:
+    if DEVICE == "cuda" or DEVICE == "mps":
+        default_gpu_index = torch.cuda.current_device()
+        return torch.cuda.get_device_name(default_gpu_index)
+    elif platform.system().lower() == "darwin":
+        return platform.processor()
+    elif is_openvino_device():
+        return core.get_property(DEVICE.upper(), "FULL_DEVICE_NAME")

src/backend/gguf/gguf_diffusion.py

@@ -0,0 +1,319 @@
+"""
+Wrapper class to call the stablediffusion.cpp shared library for GGUF support
+"""
+
+import ctypes
+import platform
+from ctypes import (
+    POINTER,
+    c_bool,
+    c_char_p,
+    c_float,
+    c_int,
+    c_int64,
+    c_void_p,
+)
+from dataclasses import dataclass
+from os import path
+from typing import List, Any
+
+import numpy as np
+from PIL import Image
+
+from backend.gguf.sdcpp_types import (
+    RngType,
+    SampleMethod,
+    Schedule,
+    SDCPPLogLevel,
+    SDImage,
+    SdType,
+)
+
+
+@dataclass
+class ModelConfig:
+    model_path: str = ""
+    clip_l_path: str = ""
+    t5xxl_path: str = ""
+    diffusion_model_path: str = ""
+    vae_path: str = ""
+    taesd_path: str = ""
+    control_net_path: str = ""
+    lora_model_dir: str = ""
+    embed_dir: str = ""
+    stacked_id_embed_dir: str = ""
+    vae_decode_only: bool = True
+    vae_tiling: bool = False
+    free_params_immediately: bool = False
+    n_threads: int = 4
+    wtype: SdType = SdType.SD_TYPE_Q4_0
+    rng_type: RngType = RngType.CUDA_RNG
+    schedule: Schedule = Schedule.DEFAULT
+    keep_clip_on_cpu: bool = False
+    keep_control_net_cpu: bool = False
+    keep_vae_on_cpu: bool = False
+
+
+@dataclass
+class Txt2ImgConfig:
+    prompt: str = "a man wearing sun glasses, highly detailed"
+    negative_prompt: str = ""
+    clip_skip: int = -1
+    cfg_scale: float = 2.0
+    guidance: float = 3.5
+    width: int = 512
+    height: int = 512
+    sample_method: SampleMethod = SampleMethod.EULER_A
+    sample_steps: int = 1
+    seed: int = -1
+    batch_count: int = 2
+    control_cond: Image = None
+    control_strength: float = 0.90
+    style_strength: float = 0.5
+    normalize_input: bool = False
+    input_id_images_path: bytes = b""
+
+
+class GGUFDiffusion:
+    """GGUF Diffusion
+    To support GGUF diffusion model based on stablediffusion.cpp
+    https://github.com/ggerganov/ggml/blob/master/docs/gguf.md
+    Implmented based on stablediffusion.h
+    """
+
+    def __init__(
+        self,
+        libpath: str,
+        config: ModelConfig,
+        logging_enabled: bool = False,
+    ):
+        sdcpp_shared_lib_path = self._get_sdcpp_shared_lib_path(libpath)
+        try:
+            self.libsdcpp = ctypes.CDLL(sdcpp_shared_lib_path)
+        except OSError as e:
+            print(f"Failed to load library {sdcpp_shared_lib_path}")
+            raise ValueError(f"Error: {e}")
+
+        if not config.clip_l_path or not path.exists(config.clip_l_path):
+            raise ValueError(
+                "CLIP model file not found,please check readme.md for GGUF model usage"
+            )
+
+        if not config.t5xxl_path or not path.exists(config.t5xxl_path):
+            raise ValueError(
+                "T5XXL model file not found,please check readme.md for GGUF model usage"
+            )
+
+        if not config.diffusion_model_path or not path.exists(
+            config.diffusion_model_path
+        ):
+            raise ValueError(
+                "Diffusion model file not found,please check readme.md for GGUF model usage"
+            )
+
+        if not config.vae_path or not path.exists(config.vae_path):
+            raise ValueError(
+                "VAE model file not found,please check readme.md for GGUF model usage"
+            )
+
+        self.model_config = config
+
+        self.libsdcpp.new_sd_ctx.argtypes = [
+            c_char_p,  # const char* model_path
+            c_char_p,  # const char* clip_l_path
+            c_char_p,  # const char* t5xxl_path
+            c_char_p,  # const char* diffusion_model_path
+            c_char_p,  # const char* vae_path
+            c_char_p,  # const char* taesd_path
+            c_char_p,  # const char* control_net_path_c_str
+            c_char_p,  # const char* lora_model_dir
+            c_char_p,  # const char* embed_dir_c_str
+            c_char_p,  # const char* stacked_id_embed_dir_c_str
+            c_bool,  # bool vae_decode_only
+            c_bool,  # bool vae_tiling
+            c_bool,  # bool free_params_immediately
+            c_int,  # int n_threads
+            SdType,  # enum sd_type_t wtype
+            RngType,  # enum rng_type_t rng_type
+            Schedule,  # enum schedule_t s
+            c_bool,  # bool keep_clip_on_cpu
+            c_bool,  # bool keep_control_net_cpu
+            c_bool,  # bool keep_vae_on_cpu
+        ]
+
+        self.libsdcpp.new_sd_ctx.restype = POINTER(c_void_p)
+
+        self.sd_ctx = self.libsdcpp.new_sd_ctx(
+            self._str_to_bytes(self.model_config.model_path),
+            self._str_to_bytes(self.model_config.clip_l_path),
+            self._str_to_bytes(self.model_config.t5xxl_path),
+            self._str_to_bytes(self.model_config.diffusion_model_path),
+            self._str_to_bytes(self.model_config.vae_path),
+            self._str_to_bytes(self.model_config.taesd_path),
+            self._str_to_bytes(self.model_config.control_net_path),
+            self._str_to_bytes(self.model_config.lora_model_dir),
+            self._str_to_bytes(self.model_config.embed_dir),
+            self._str_to_bytes(self.model_config.stacked_id_embed_dir),
+            self.model_config.vae_decode_only,
+            self.model_config.vae_tiling,
+            self.model_config.free_params_immediately,
+            self.model_config.n_threads,
+            self.model_config.wtype,
+            self.model_config.rng_type,
+            self.model_config.schedule,
+            self.model_config.keep_clip_on_cpu,
+            self.model_config.keep_control_net_cpu,
+            self.model_config.keep_vae_on_cpu,
+        )
+
+        if logging_enabled:
+            self._set_logcallback()
+
+    def _set_logcallback(self):
+        print("Setting logging callback")
+        # Define function callback
+        SdLogCallbackType = ctypes.CFUNCTYPE(
+            None,
+            SDCPPLogLevel,
+            ctypes.c_char_p,
+            ctypes.c_void_p,
+        )
+
+        self.libsdcpp.sd_set_log_callback.argtypes = [
+            SdLogCallbackType,
+            ctypes.c_void_p,
+        ]
+        self.libsdcpp.sd_set_log_callback.restype = None
+        # Convert the Python callback to a C func pointer
+        self.c_log_callback = SdLogCallbackType(
+            self.log_callback
+        )  # prevent GC,keep callback as member variable
+        self.libsdcpp.sd_set_log_callback(self.c_log_callback, None)
+
+    def _get_sdcpp_shared_lib_path(
+        self,
+        root_path: str,
+    ) -> str:
+        system_name = platform.system()
+        print(f"GGUF Diffusion on {system_name}")
+        lib_name = "stable-diffusion.dll"
+        sdcpp_lib_path = ""
+
+        if system_name == "Windows":
+            sdcpp_lib_path = path.join(root_path, lib_name)
+        elif system_name == "Linux":
+            lib_name = "libstable-diffusion.so"
+            sdcpp_lib_path = path.join(root_path, lib_name)
+        elif system_name == "Darwin":
+            lib_name = "libstable-diffusion.dylib"
+            sdcpp_lib_path = path.join(root_path, lib_name)
+        else:
+            print("Unknown platform.")
+
+        return sdcpp_lib_path
+
+    @staticmethod
+    def log_callback(
+        level,
+        text,
+        data,
+    ):
+        print(f"{text.decode('utf-8')}", end="")
+
+    def _str_to_bytes(self, in_str: str, encoding: str = "utf-8") -> bytes:
+        if in_str:
+            return in_str.encode(encoding)
+        else:
+            return b""
+
+    def generate_text2mg(self, txt2img_cfg: Txt2ImgConfig) -> List[Any]:
+        self.libsdcpp.txt2img.restype = POINTER(SDImage)
+        self.libsdcpp.txt2img.argtypes = [
+            c_void_p,  # sd_ctx_t* sd_ctx (pointer to context object)
+            c_char_p,  # const char* prompt
+            c_char_p,  # const char* negative_prompt
+            c_int,  # int clip_skip
+            c_float,  # float cfg_scale
+            c_float,  # float guidance
+            c_int,  # int width
+            c_int,  # int height
+            SampleMethod,  # enum sample_method_t sample_method
+            c_int,  # int sample_steps
+            c_int64,  # int64_t seed
+            c_int,  # int batch_count
+            POINTER(SDImage),  # const sd_image_t* control_cond (pointer to SDImage)
+            c_float,  # float control_strength
+            c_float,  # float style_strength
+            c_bool,  # bool normalize_input
+            c_char_p,  # const char* input_id_images_path
+        ]
+
+        image_buffer = self.libsdcpp.txt2img(
+            self.sd_ctx,
+            self._str_to_bytes(txt2img_cfg.prompt),
+            self._str_to_bytes(txt2img_cfg.negative_prompt),
+            txt2img_cfg.clip_skip,
+            txt2img_cfg.cfg_scale,
+            txt2img_cfg.guidance,
+            txt2img_cfg.width,
+            txt2img_cfg.height,
+            txt2img_cfg.sample_method,
+            txt2img_cfg.sample_steps,
+            txt2img_cfg.seed,
+            txt2img_cfg.batch_count,
+            txt2img_cfg.control_cond,
+            txt2img_cfg.control_strength,
+            txt2img_cfg.style_strength,
+            txt2img_cfg.normalize_input,
+            txt2img_cfg.input_id_images_path,
+        )
+
+        images = self._get_sd_images_from_buffer(
+            image_buffer,
+            txt2img_cfg.batch_count,
+        )
+
+        return images
+
+    def _get_sd_images_from_buffer(
+        self,
+        image_buffer: Any,
+        batch_count: int,
+    ) -> List[Any]:
+        images = []
+        if image_buffer:
+            for i in range(batch_count):
+                image = image_buffer[i]
+                print(
+                    f"Generated image: {image.width}x{image.height} with {image.channel} channels"
+                )
+
+                width = image.width
+                height = image.height
+                channels = image.channel
+                pixel_data = np.ctypeslib.as_array(
+                    image.data, shape=(height, width, channels)
+                )
+
+                if channels == 1:
+                    pil_image = Image.fromarray(pixel_data.squeeze(), mode="L")
+                elif channels == 3:
+                    pil_image = Image.fromarray(pixel_data, mode="RGB")
+                elif channels == 4:
+                    pil_image = Image.fromarray(pixel_data, mode="RGBA")
+                else:
+                    raise ValueError(f"Unsupported number of channels: {channels}")
+
+                images.append(pil_image)
+        return images
+
+    def terminate(self):
+        if self.libsdcpp:
+            if self.sd_ctx:
+                self.libsdcpp.free_sd_ctx.argtypes = [c_void_p]
+                self.libsdcpp.free_sd_ctx.restype = None
+                self.libsdcpp.free_sd_ctx(self.sd_ctx)
+                del self.sd_ctx
+                self.sd_ctx = None
+                del self.libsdcpp
+                self.libsdcpp = None

src/backend/gguf/sdcpp_types.py

@@ -0,0 +1,104 @@
+"""
+Ctypes for stablediffusion.cpp shared library
+This is as per the stablediffusion.h  file
+"""
+
+from enum import IntEnum
+from ctypes import (
+    c_int,
+    c_uint32,
+    c_uint8,
+    POINTER,
+    Structure,
+)
+
+
+class CtypesEnum(IntEnum):
+    """A ctypes-compatible IntEnum superclass."""
+
+    @classmethod
+    def from_param(cls, obj):
+        return int(obj)
+
+
+class RngType(CtypesEnum):
+    STD_DEFAULT_RNG = 0
+    CUDA_RNG = 1
+
+
+class SampleMethod(CtypesEnum):
+    EULER_A = 0
+    EULER = 1
+    HEUN = 2
+    DPM2 = 3
+    DPMPP2S_A = 4
+    DPMPP2M = 5
+    DPMPP2Mv2 = 6
+    IPNDM = 7
+    IPNDM_V = 7
+    LCM = 8
+    N_SAMPLE_METHODS = 9
+
+
+class Schedule(CtypesEnum):
+    DEFAULT = 0
+    DISCRETE = 1
+    KARRAS = 2
+    EXPONENTIAL = 3
+    AYS = 4
+    GITS = 5
+    N_SCHEDULES = 5
+
+
+class SdType(CtypesEnum):
+    SD_TYPE_F32 = 0
+    SD_TYPE_F16 = 1
+    SD_TYPE_Q4_0 = 2
+    SD_TYPE_Q4_1 = 3
+    # SD_TYPE_Q4_2 = 4, support has been removed
+    # SD_TYPE_Q4_3 = 5, support has been removed
+    SD_TYPE_Q5_0 = 6
+    SD_TYPE_Q5_1 = 7
+    SD_TYPE_Q8_0 = 8
+    SD_TYPE_Q8_1 = 9
+    SD_TYPE_Q2_K = 10
+    SD_TYPE_Q3_K = 11
+    SD_TYPE_Q4_K = 12
+    SD_TYPE_Q5_K = 13
+    SD_TYPE_Q6_K = 14
+    SD_TYPE_Q8_K = 15
+    SD_TYPE_IQ2_XXS = 16
+    SD_TYPE_IQ2_XS = 17
+    SD_TYPE_IQ3_XXS = 18
+    SD_TYPE_IQ1_S = 19
+    SD_TYPE_IQ4_NL = 20
+    SD_TYPE_IQ3_S = 21
+    SD_TYPE_IQ2_S = 22
+    SD_TYPE_IQ4_XS = 23
+    SD_TYPE_I8 = 24
+    SD_TYPE_I16 = 25
+    SD_TYPE_I32 = 26
+    SD_TYPE_I64 = 27
+    SD_TYPE_F64 = 28
+    SD_TYPE_IQ1_M = 29
+    SD_TYPE_BF16 = 30
+    SD_TYPE_Q4_0_4_4 = 31
+    SD_TYPE_Q4_0_4_8 = 32
+    SD_TYPE_Q4_0_8_8 = 33
+    SD_TYPE_COUNT = 34
+
+
+class SDImage(Structure):
+    _fields_ = [
+        ("width", c_uint32),
+        ("height", c_uint32),
+        ("channel", c_uint32),
+        ("data", POINTER(c_uint8)),
+    ]
+
+
+class SDCPPLogLevel(c_int):
+    SD_LOG_LEVEL_DEBUG = 0
+    SD_LOG_LEVEL_INFO = 1
+    SD_LOG_LEVEL_WARNING = 2
+    SD_LOG_LEVEL_ERROR = 3

src/backend/image_saver.py

@@ -0,0 +1,75 @@
+import json
+from os import path, mkdir
+from typing import Any
+from uuid import uuid4
+from backend.models.lcmdiffusion_setting import LCMDiffusionSetting
+from utils import get_image_file_extension
+
+
+def get_exclude_keys():
+    exclude_keys = {
+        "init_image": True,
+        "generated_images": True,
+        "lora": {
+            "models_dir": True,
+            "path": True,
+        },
+        "dirs": True,
+        "controlnet": {
+            "adapter_path": True,
+        },
+    }
+    return exclude_keys
+
+
+class ImageSaver:
+    @staticmethod
+    def save_images(
+        output_path: str,
+        images: Any,
+        folder_name: str = "",
+        format: str = "PNG",
+        jpeg_quality: int = 90,
+        lcm_diffusion_setting: LCMDiffusionSetting = None,
+    ) -> list[str]:
+        gen_id = uuid4()
+        image_ids = []
+
+        if images:
+            image_seeds = []
+
+            for index, image in enumerate(images):
+
+                image_seed = image.info.get('image_seed')
+                if image_seed is not None:
+                    image_seeds.append(image_seed)
+
+                if not path.exists(output_path):
+                    mkdir(output_path)
+
+                if folder_name:
+                    out_path = path.join(
+                        output_path,
+                        folder_name,
+                    )
+                else:
+                    out_path = output_path
+
+                if not path.exists(out_path):
+                    mkdir(out_path)
+                image_extension = get_image_file_extension(format)
+                image_file_name = f"{gen_id}-{index+1}{image_extension}"
+                image_ids.append(image_file_name)
+                image.save(path.join(out_path, image_file_name), quality = jpeg_quality)
+            if lcm_diffusion_setting:
+                data = lcm_diffusion_setting.model_dump(exclude=get_exclude_keys())
+                if image_seeds:
+                    data['image_seeds'] = image_seeds
+                with open(path.join(out_path, f"{gen_id}.json"), "w") as json_file:
+                    json.dump(
+                        data,
+                        json_file,
+                        indent=4,
+                    )
+        return image_ids
+            

src/backend/lcm_text_to_image.py

@@ -0,0 +1,686 @@
+import gc
+from math import ceil
+from typing import Any, List
+import random
+
+import numpy as np
+import torch
+from backend.device import is_openvino_device
+from backend.lora import reset_active_lora_weights
+from backend.controlnet import (
+    load_controlnet_adapters,
+    update_controlnet_arguments,
+    get_controlnet_pipeline,
+)
+from backend.models.lcmdiffusion_setting import (
+    DiffusionTask,
+    LCMDiffusionSetting,
+    LCMLora,
+)
+from backend.openvino.pipelines import (
+    get_ov_image_to_image_pipeline,
+    get_ov_text_to_image_pipeline,
+    ov_load_tiny_autoencoder,
+    get_ov_diffusion_pipeline,
+)
+from backend.pipelines.lcm import (
+    get_image_to_image_pipeline,
+    get_lcm_model_pipeline,
+    load_taesd,
+)
+from backend.pipelines.lcm_lora import get_lcm_lora_pipeline
+from constants import DEVICE, GGUF_THREADS
+from diffusers import LCMScheduler
+from image_ops import resize_pil_image
+from backend.openvino.ov_hc_stablediffusion_pipeline import OvHcLatentConsistency
+from backend.gguf.gguf_diffusion import (
+    GGUFDiffusion,
+    ModelConfig,
+    Txt2ImgConfig,
+    SampleMethod,
+)
+from paths import get_app_path
+from pprint import pprint
+
+try:
+    # support for token merging; keeping it optional for now
+    import tomesd
+except ImportError:
+    print("tomesd library unavailable; disabling token merging support")
+    tomesd = None
+
+
+class LCMTextToImage:
+    def __init__(
+        self,
+        device: str = "cpu",
+    ) -> None:
+        self.pipeline = None
+        self.txt2img_pipeline = None
+        self.img2img_pipeline = None
+        self.controlnet_pipeline = None
+        self.controlnet_img2img_pipeline = None
+        self.use_openvino = False
+        self.device = ""
+        self.previous_model_id = None
+        self.previous_use_tae_sd = False
+        self.previous_use_lcm_lora = False
+        self.previous_ov_model_id = ""
+        self.previous_token_merging = 0.0
+        self.previous_safety_checker = False
+        self.previous_use_openvino = False
+        self.img_to_img_pipeline = None
+        self.is_openvino_init = False
+        self.previous_lora = None
+        self.task_type = DiffusionTask.text_to_image
+        self.previous_use_gguf_model = False
+        self.previous_gguf_model = None
+        self.torch_data_type = (
+            torch.float32 if is_openvino_device() or DEVICE == "mps" else torch.float16
+        )
+        self.ov_model_id = None
+        print(f"Torch datatype : {self.torch_data_type}")
+
+    def _pipeline_to_device(self):
+        print(f"Pipeline device : {DEVICE}")
+        print(f"Pipeline dtype : {self.torch_data_type}")
+        self.pipeline.to(
+            torch_device=DEVICE,
+            torch_dtype=self.torch_data_type,
+        )
+
+    def _add_freeu(self):
+        pipeline_class = self.pipeline.__class__.__name__
+        if isinstance(self.pipeline.scheduler, LCMScheduler):
+            if pipeline_class == "StableDiffusionPipeline":
+                print("Add FreeU - SD")
+                self.pipeline.enable_freeu(
+                    s1=0.9,
+                    s2=0.2,
+                    b1=1.2,
+                    b2=1.4,
+                )
+            elif pipeline_class == "StableDiffusionXLPipeline":
+                print("Add FreeU - SDXL")
+                self.pipeline.enable_freeu(
+                    s1=0.6,
+                    s2=0.4,
+                    b1=1.1,
+                    b2=1.2,
+                )
+
+    def _enable_vae_tiling(self):
+        self.pipeline.vae.enable_tiling()
+
+    def _update_lcm_scheduler_params(self):
+        if isinstance(self.pipeline.scheduler, LCMScheduler):
+            self.pipeline.scheduler = LCMScheduler.from_config(
+                self.pipeline.scheduler.config,
+                beta_start=0.001,
+                beta_end=0.01,
+            )
+
+    def _is_hetero_pipeline(self) -> bool:
+        return "square" in self.ov_model_id.lower()
+
+    def _load_ov_hetero_pipeline(self):
+        print("Loading Heterogeneous Compute pipeline")
+        if DEVICE.upper() == "NPU":
+            device = ["NPU", "NPU", "NPU"]
+            self.pipeline = OvHcLatentConsistency(self.ov_model_id, device)
+        else:
+            self.pipeline = OvHcLatentConsistency(self.ov_model_id)
+
+    def _generate_images_hetero_compute(
+        self,
+        lcm_diffusion_setting: LCMDiffusionSetting,
+    ):
+        print("Using OpenVINO ")
+        if lcm_diffusion_setting.diffusion_task == DiffusionTask.text_to_image.value:
+            return [
+                self.pipeline.generate(
+                    prompt=lcm_diffusion_setting.prompt,
+                    neg_prompt=lcm_diffusion_setting.negative_prompt,
+                    init_image=None,
+                    strength=1.0,
+                    num_inference_steps=lcm_diffusion_setting.inference_steps,
+                )
+            ]
+        else:
+            return [
+                self.pipeline.generate(
+                    prompt=lcm_diffusion_setting.prompt,
+                    neg_prompt=lcm_diffusion_setting.negative_prompt,
+                    init_image=lcm_diffusion_setting.init_image,
+                    strength=lcm_diffusion_setting.strength,
+                    num_inference_steps=lcm_diffusion_setting.inference_steps,
+                )
+            ]
+
+    def _is_valid_mode(
+        self,
+        modes: List,
+    ) -> bool:
+        return modes.count(True) == 1 or modes.count(False) == 3
+
+    def _validate_mode(
+        self,
+        modes: List,
+    ) -> None:
+        if not self._is_valid_mode(modes):
+            raise ValueError("Invalid mode,delete configs/settings.yaml and retry!")
+
+    def _is_sana_model(self) -> bool:
+        return "sana" in self.ov_model_id.lower()
+
+    def init(
+        self,
+        device: str = "cpu",
+        lcm_diffusion_setting: LCMDiffusionSetting = LCMDiffusionSetting(),
+    ) -> None:
+        # Mode validation either LCM LoRA or OpenVINO or GGUF
+
+        modes = [
+            lcm_diffusion_setting.use_gguf_model,
+            lcm_diffusion_setting.use_openvino,
+            lcm_diffusion_setting.use_lcm_lora,
+        ]
+        self._validate_mode(modes)
+        self.device = device
+        self.use_openvino = lcm_diffusion_setting.use_openvino
+        model_id = lcm_diffusion_setting.lcm_model_id
+        use_local_model = lcm_diffusion_setting.use_offline_model
+        use_tiny_auto_encoder = lcm_diffusion_setting.use_tiny_auto_encoder
+        use_lora = lcm_diffusion_setting.use_lcm_lora
+        lcm_lora: LCMLora = lcm_diffusion_setting.lcm_lora
+        token_merging = lcm_diffusion_setting.token_merging
+        self.ov_model_id = lcm_diffusion_setting.openvino_lcm_model_id
+
+        if lcm_diffusion_setting.diffusion_task == DiffusionTask.image_to_image.value:
+            lcm_diffusion_setting.init_image = resize_pil_image(
+                lcm_diffusion_setting.init_image,
+                lcm_diffusion_setting.image_width,
+                lcm_diffusion_setting.image_height,
+            )
+
+        # Reset current pipeline references to allow the garbage
+        # collector to correctly free the pipeline when necessary
+        if self.txt2img_pipeline:  # In LCM or LCM-LoRA modes
+            self.pipeline = self.txt2img_pipeline
+            self.img_to_img_pipeline = self.img2img_pipeline
+            # Regenerate the ControlNet pipelines if the variable
+            # lcm_diffusion_setting.rebuild_controlnet_pipeline is set,
+            # this is done here because rebuilding the ControlNet pipelines
+            # doesn't necessarily implies a full pipeline rebuild.
+            if lcm_diffusion_setting.rebuild_controlnet_pipeline:
+                if self.controlnet_pipeline:
+                    del self.controlnet_pipeline
+                    self.controlnet_pipeline = None
+                if self.controlnet_img2img_pipeline:
+                    del self.controlnet_img2img_pipeline
+                    self.controlnet_img2img_pipeline = None
+                gc.collect()
+                self.controlnet_pipeline = get_controlnet_pipeline(
+                    self.txt2img_pipeline,
+                    lcm_diffusion_setting,
+                    DiffusionTask.text_to_image,
+                )
+                self.controlnet_img2img_pipeline = get_controlnet_pipeline(
+                    self.txt2img_pipeline,
+                    lcm_diffusion_setting,
+                    DiffusionTask.image_to_image,
+                )
+                lcm_diffusion_setting.rebuild_controlnet_pipeline = False
+
+        if (
+            self.pipeline is None
+            or self.previous_model_id != model_id
+            or self.previous_use_tae_sd != use_tiny_auto_encoder
+            or self.previous_lcm_lora_base_id != lcm_lora.base_model_id
+            or self.previous_lcm_lora_id != lcm_lora.lcm_lora_id
+            or self.previous_use_lcm_lora != use_lora
+            or self.previous_ov_model_id != self.ov_model_id
+            or self.previous_token_merging != token_merging
+            or self.previous_safety_checker != lcm_diffusion_setting.use_safety_checker
+            or self.previous_use_openvino != lcm_diffusion_setting.use_openvino
+            or self.previous_use_gguf_model != lcm_diffusion_setting.use_gguf_model
+            or self.previous_gguf_model != lcm_diffusion_setting.gguf_model
+            or (
+                self.use_openvino
+                and (
+                    self.previous_task_type != lcm_diffusion_setting.diffusion_task
+                    or self.previous_lora != lcm_diffusion_setting.lora
+                )
+            )
+            or lcm_diffusion_setting.rebuild_pipeline
+        ):
+            if self.use_openvino and is_openvino_device():
+                if self.pipeline:
+                    del self.pipeline
+                    self.pipeline = None
+                    gc.collect()
+                self.is_openvino_init = True
+                if (
+                    lcm_diffusion_setting.diffusion_task
+                    == DiffusionTask.text_to_image.value
+                ):
+                    print(
+                        f"***** Init Text to image (OpenVINO) - {self.ov_model_id} *****"
+                    )
+                    if "flux" in self.ov_model_id.lower() or self._is_sana_model():
+                        if self._is_sana_model():
+                            print("Loading OpenVINO SANA Sprint pipeline")
+                        else:
+                            print("Loading OpenVINO Flux pipeline")
+                        self.pipeline = get_ov_diffusion_pipeline(self.ov_model_id)
+                    elif self._is_hetero_pipeline():
+                        self._load_ov_hetero_pipeline()
+                    else:
+                        self.pipeline = get_ov_text_to_image_pipeline(
+                            self.ov_model_id,
+                            use_local_model,
+                        )
+                elif (
+                    lcm_diffusion_setting.diffusion_task
+                    == DiffusionTask.image_to_image.value
+                ):
+                    if not self.pipeline and self._is_hetero_pipeline():
+                        self._load_ov_hetero_pipeline()
+                    else:
+                        print(
+                            f"***** Image to image (OpenVINO) - {self.ov_model_id} *****"
+                        )
+                        self.pipeline = get_ov_image_to_image_pipeline(
+                            self.ov_model_id,
+                            use_local_model,
+                        )
+            elif lcm_diffusion_setting.use_gguf_model:
+                model = lcm_diffusion_setting.gguf_model.diffusion_path
+                print(f"***** Init Text to image (GGUF) - {model} *****")
+                # if self.pipeline:
+                #     self.pipeline.terminate()
+                #     del self.pipeline
+                #     self.pipeline = None
+                self._init_gguf_diffusion(lcm_diffusion_setting)
+            else:
+                # Code for pipeline rebuild in LCM or LCM-LoRA modes
+                reset_active_lora_weights()
+                if self.txt2img_pipeline:  # In LCM or LCM-LoRA modes
+                    self.pipeline = self.txt2img_pipeline
+                    if self.txt2img_pipeline:
+                        del self.txt2img_pipeline
+                        self.txt2img_pipeline = None
+                    if self.img2img_pipeline:
+                        del self.img2img_pipeline
+                        self.img2img_pipeline = None
+                    if self.img_to_img_pipeline:
+                        del self.img_to_img_pipeline
+                        self.img_to_img_pipeline = None
+                    if self.controlnet_pipeline:
+                        del self.controlnet_pipeline
+                        self.controlnet_pipeline = None
+                    if self.controlnet_img2img_pipeline:
+                        del self.controlnet_img2img_pipeline
+                        self.controlnet_img2img_pipeline = None
+                    del self.pipeline
+                    self.pipeline = None
+                    gc.collect()
+
+                if use_lora:
+                    print(
+                        f"***** Init LCM-LoRA pipeline - {lcm_lora.base_model_id} *****"
+                    )
+                    self.pipeline = get_lcm_lora_pipeline(
+                        lcm_lora.base_model_id,
+                        lcm_lora.lcm_lora_id,
+                        use_local_model,
+                        torch_data_type=self.torch_data_type,
+                    )
+
+                else:
+                    print(f"***** Init LCM Model pipeline - {model_id} *****")
+                    extra_args = {}
+                    self.pipeline = get_lcm_model_pipeline(
+                        model_id,
+                        use_local_model,
+                        extra_args,
+                    )
+
+                # Prepare alternative generation pipelines using the newly
+                # created pipeline from which all extra pipelines are derived
+                self.txt2img_pipeline = self.pipeline
+                self.img2img_pipeline = get_image_to_image_pipeline(self.pipeline)
+                self.controlnet_pipeline = get_controlnet_pipeline(
+                    self.pipeline,
+                    lcm_diffusion_setting,
+                    DiffusionTask.text_to_image,
+                )
+                self.controlnet_img2img_pipeline = get_controlnet_pipeline(
+                    self.pipeline,
+                    lcm_diffusion_setting,
+                    DiffusionTask.image_to_image,
+                )
+                self.img_to_img_pipeline = self.img2img_pipeline
+
+                if tomesd and token_merging > 0.001:
+                    print(f"***** Token Merging: {token_merging} *****")
+                    tomesd.apply_patch(self.pipeline, ratio=token_merging)
+                    tomesd.apply_patch(self.img_to_img_pipeline, ratio=token_merging)
+
+            if use_tiny_auto_encoder:
+                if self.use_openvino and is_openvino_device():
+                    if not self._is_sana_model():
+                        print("Using Tiny AutoEncoder (OpenVINO)")
+                        ov_load_tiny_autoencoder(
+                            self.pipeline,
+                            use_local_model,
+                        )
+                else:
+                    print("Using Tiny Auto Encoder")
+                    load_taesd(
+                        self.pipeline,
+                        use_local_model,
+                        self.torch_data_type,
+                    )
+                    load_taesd(
+                        self.img_to_img_pipeline,
+                        use_local_model,
+                        self.torch_data_type,
+                    )
+
+            if not self.use_openvino and not is_openvino_device():
+                self._pipeline_to_device()
+
+            if not self._is_hetero_pipeline():
+                if (
+                    lcm_diffusion_setting.diffusion_task
+                    == DiffusionTask.image_to_image.value
+                    and lcm_diffusion_setting.use_openvino
+                ):
+                    self.pipeline.scheduler = LCMScheduler.from_config(
+                        self.pipeline.scheduler.config,
+                    )
+                else:
+                    if not lcm_diffusion_setting.use_gguf_model:
+                        self._update_lcm_scheduler_params()
+
+            if use_lora:
+                self._add_freeu()
+
+            self.previous_model_id = model_id
+            self.previous_ov_model_id = self.ov_model_id
+            self.previous_use_tae_sd = use_tiny_auto_encoder
+            self.previous_lcm_lora_base_id = lcm_lora.base_model_id
+            self.previous_lcm_lora_id = lcm_lora.lcm_lora_id
+            self.previous_use_lcm_lora = use_lora
+            self.previous_token_merging = lcm_diffusion_setting.token_merging
+            self.previous_safety_checker = lcm_diffusion_setting.use_safety_checker
+            self.previous_use_openvino = lcm_diffusion_setting.use_openvino
+            self.previous_task_type = lcm_diffusion_setting.diffusion_task
+            self.previous_lora = lcm_diffusion_setting.lora.model_copy(deep=True)
+            self.previous_use_gguf_model = lcm_diffusion_setting.use_gguf_model
+            self.previous_gguf_model = lcm_diffusion_setting.gguf_model.model_copy(
+                deep=True
+            )
+            lcm_diffusion_setting.rebuild_pipeline = False
+            if (
+                lcm_diffusion_setting.diffusion_task
+                == DiffusionTask.text_to_image.value
+            ):
+                print(f"Pipeline : {self.pipeline}")
+            elif (
+                lcm_diffusion_setting.diffusion_task
+                == DiffusionTask.image_to_image.value
+            ):
+                if self.use_openvino and is_openvino_device():
+                    print(f"Pipeline : {self.pipeline}")
+                else:
+                    print(f"Pipeline : {self.img_to_img_pipeline}")
+            if self.use_openvino:
+                if lcm_diffusion_setting.lora.enabled:
+                    print("Warning: Lora models not supported on OpenVINO mode")
+            elif not lcm_diffusion_setting.use_gguf_model:
+                adapters = self.pipeline.get_active_adapters()
+                print(f"Active adapters : {adapters}")
+
+    def _get_timesteps(self):
+        time_steps = self.pipeline.scheduler.config.get("timesteps")
+        time_steps_value = [int(time_steps)] if time_steps else None
+        return time_steps_value
+
+    def _compile_ov_pipeline(
+        self,
+        lcm_diffusion_setting,
+    ):
+        self.pipeline.reshape(
+            batch_size=-1,
+            height=lcm_diffusion_setting.image_height,
+            width=lcm_diffusion_setting.image_width,
+            num_images_per_prompt=lcm_diffusion_setting.number_of_images,
+        )
+        self.pipeline.compile()
+
+    def generate(
+        self,
+        lcm_diffusion_setting: LCMDiffusionSetting,
+        reshape: bool = False,
+    ) -> Any:
+        guidance_scale = lcm_diffusion_setting.guidance_scale
+        img_to_img_inference_steps = lcm_diffusion_setting.inference_steps
+        check_step_value = int(
+            lcm_diffusion_setting.inference_steps * lcm_diffusion_setting.strength
+        )
+        if (
+            lcm_diffusion_setting.diffusion_task == DiffusionTask.image_to_image.value
+            and check_step_value < 1
+        ):
+            img_to_img_inference_steps = ceil(1 / lcm_diffusion_setting.strength)
+            print(
+                f"Strength: {lcm_diffusion_setting.strength},{img_to_img_inference_steps}"
+            )
+
+        # In LCM and LCM-LoRA modes, set the pipeline for the current
+        # generation by setting self.pipeline and self.img_to_img_pipeline
+        # to either the normal generation pipelines or the ControlNet
+        # generation pipelines, depending on the user settings. Note that,
+        # after generation, the pipelines are reset for compatibility with
+        # other generation modes.
+        if self.txt2img_pipeline:  # In LCM or LCM-LoRA modes
+            self.pipeline = self.txt2img_pipeline
+            self.img_to_img_pipeline = self.img2img_pipeline
+            if (
+                lcm_diffusion_setting.controlnet
+                and lcm_diffusion_setting.controlnet.enabled
+            ):
+                if self.controlnet_pipeline != None:
+                    self.pipeline = self.controlnet_pipeline
+                if self.controlnet_img2img_pipeline != None:
+                    self.img_to_img_pipeline = self.controlnet_img2img_pipeline
+        pipeline_extra_args = {}
+
+        if lcm_diffusion_setting.use_seed:
+            cur_seed = lcm_diffusion_setting.seed
+            # for multiple images with a fixed seed, use sequential seeds
+            seeds = [
+                (cur_seed + i) for i in range(lcm_diffusion_setting.number_of_images)
+            ]
+        else:
+            seeds = [
+                random.randint(0, 999999999)
+                for i in range(lcm_diffusion_setting.number_of_images)
+            ]
+
+        if self.use_openvino:
+            # no support for generators; try at least to ensure reproducible results for single images
+            torch.manual_seed(seeds[0])
+            if self._is_hetero_pipeline():
+                torch.manual_seed(seeds[0])
+                lcm_diffusion_setting.seed = seeds[0]
+        else:
+            pipeline_extra_args["generator"] = [
+                torch.Generator(device=self.device).manual_seed(s) for s in seeds
+            ]
+
+        is_openvino_pipe = lcm_diffusion_setting.use_openvino and is_openvino_device()
+        if is_openvino_pipe and not self._is_hetero_pipeline():
+            print("Using OpenVINO")
+            if self.is_openvino_init and self._is_sana_model():
+                self._compile_ov_pipeline(lcm_diffusion_setting)
+
+            if reshape and not self.is_openvino_init:
+                print("Reshape and compile")
+                self._compile_ov_pipeline(lcm_diffusion_setting)
+
+            if self.is_openvino_init:
+                self.is_openvino_init = False
+
+        if is_openvino_pipe and self._is_hetero_pipeline():
+            return self._generate_images_hetero_compute(lcm_diffusion_setting)
+        elif lcm_diffusion_setting.use_gguf_model:
+            return self._generate_images_gguf(lcm_diffusion_setting)
+
+        if lcm_diffusion_setting.clip_skip > 1:
+            # We follow the convention that "CLIP Skip == 2" means "skip
+            # the last layer", so "CLIP Skip == 1" means "no skipping"
+            pipeline_extra_args["clip_skip"] = lcm_diffusion_setting.clip_skip - 1
+
+        self.pipeline.safety_checker = None
+        if (
+            lcm_diffusion_setting.diffusion_task == DiffusionTask.image_to_image.value
+            and not is_openvino_pipe
+        ):
+            self.img_to_img_pipeline.safety_checker = None
+
+        if (
+            not lcm_diffusion_setting.use_lcm_lora
+            and not lcm_diffusion_setting.use_openvino
+            and lcm_diffusion_setting.guidance_scale != 1.0
+        ):
+            print("Not using LCM-LoRA so setting guidance_scale 1.0")
+            guidance_scale = 1.0
+
+        controlnet_args = update_controlnet_arguments(lcm_diffusion_setting)
+        if lcm_diffusion_setting.use_openvino:
+            if (
+                lcm_diffusion_setting.diffusion_task
+                == DiffusionTask.text_to_image.value
+            ):
+                if self._is_sana_model():
+                    result_images = self.pipeline(
+                        prompt=lcm_diffusion_setting.prompt,
+                        num_inference_steps=lcm_diffusion_setting.inference_steps,
+                        guidance_scale=guidance_scale,
+                        width=lcm_diffusion_setting.image_width,
+                        height=lcm_diffusion_setting.image_height,
+                        num_images_per_prompt=lcm_diffusion_setting.number_of_images,
+                    ).images
+                else:
+                    result_images = self.pipeline(
+                        prompt=lcm_diffusion_setting.prompt,
+                        negative_prompt=lcm_diffusion_setting.negative_prompt,
+                        num_inference_steps=lcm_diffusion_setting.inference_steps,
+                        guidance_scale=guidance_scale,
+                        width=lcm_diffusion_setting.image_width,
+                        height=lcm_diffusion_setting.image_height,
+                        num_images_per_prompt=lcm_diffusion_setting.number_of_images,
+                    ).images
+            elif (
+                lcm_diffusion_setting.diffusion_task
+                == DiffusionTask.image_to_image.value
+            ):
+                result_images = self.pipeline(
+                    image=lcm_diffusion_setting.init_image,
+                    strength=lcm_diffusion_setting.strength,
+                    prompt=lcm_diffusion_setting.prompt,
+                    negative_prompt=lcm_diffusion_setting.negative_prompt,
+                    num_inference_steps=img_to_img_inference_steps * 3,
+                    guidance_scale=guidance_scale,
+                    num_images_per_prompt=lcm_diffusion_setting.number_of_images,
+                ).images
+
+        else:
+            if (
+                lcm_diffusion_setting.diffusion_task
+                == DiffusionTask.text_to_image.value
+            ):
+                print(f"Using {self.pipeline.__class__.__name__}")
+                result_images = self.pipeline(
+                    prompt=lcm_diffusion_setting.prompt,
+                    negative_prompt=lcm_diffusion_setting.negative_prompt,
+                    num_inference_steps=lcm_diffusion_setting.inference_steps,
+                    guidance_scale=guidance_scale,
+                    width=lcm_diffusion_setting.image_width,
+                    height=lcm_diffusion_setting.image_height,
+                    num_images_per_prompt=lcm_diffusion_setting.number_of_images,
+                    timesteps=self._get_timesteps(),
+                    **pipeline_extra_args,
+                    **controlnet_args,
+                ).images
+
+            elif (
+                lcm_diffusion_setting.diffusion_task
+                == DiffusionTask.image_to_image.value
+            ):
+                print(f"Using {self.pipeline.__class__.__name__}")
+                result_images = self.img_to_img_pipeline(
+                    image=lcm_diffusion_setting.init_image,
+                    strength=lcm_diffusion_setting.strength,
+                    prompt=lcm_diffusion_setting.prompt,
+                    negative_prompt=lcm_diffusion_setting.negative_prompt,
+                    num_inference_steps=img_to_img_inference_steps,
+                    guidance_scale=guidance_scale,
+                    width=lcm_diffusion_setting.image_width,
+                    height=lcm_diffusion_setting.image_height,
+                    num_images_per_prompt=lcm_diffusion_setting.number_of_images,
+                    **pipeline_extra_args,
+                    **controlnet_args,
+                ).images
+
+        if self.txt2img_pipeline:  # In LCM or LCM-LoRA modes
+            self.pipeline = self.txt2img_pipeline
+            self.img_to_img_pipeline = self.img2img_pipeline
+
+        for i, seed in enumerate(seeds):
+            result_images[i].info["image_seed"] = seed
+
+        return result_images
+
+    def _init_gguf_diffusion(
+        self,
+        lcm_diffusion_setting: LCMDiffusionSetting,
+    ):
+        config = ModelConfig()
+        config.model_path = lcm_diffusion_setting.gguf_model.diffusion_path
+        config.diffusion_model_path = lcm_diffusion_setting.gguf_model.diffusion_path
+        config.clip_l_path = lcm_diffusion_setting.gguf_model.clip_path
+        config.t5xxl_path = lcm_diffusion_setting.gguf_model.t5xxl_path
+        config.vae_path = lcm_diffusion_setting.gguf_model.vae_path
+        config.n_threads = GGUF_THREADS
+        print(f"GGUF Threads : {GGUF_THREADS} ")
+        print("GGUF - Model config")
+        pprint(lcm_diffusion_setting.gguf_model.model_dump())
+        self.pipeline = GGUFDiffusion(
+            get_app_path(),  # Place DLL in fastsdcpu folder
+            config,
+            True,
+        )
+
+    def _generate_images_gguf(
+        self,
+        lcm_diffusion_setting: LCMDiffusionSetting,
+    ):
+        if lcm_diffusion_setting.diffusion_task == DiffusionTask.text_to_image.value:
+            t2iconfig = Txt2ImgConfig()
+            t2iconfig.prompt = lcm_diffusion_setting.prompt
+            t2iconfig.batch_count = lcm_diffusion_setting.number_of_images
+            t2iconfig.cfg_scale = lcm_diffusion_setting.guidance_scale
+            t2iconfig.height = lcm_diffusion_setting.image_height
+            t2iconfig.width = lcm_diffusion_setting.image_width
+            t2iconfig.sample_steps = lcm_diffusion_setting.inference_steps
+            t2iconfig.sample_method = SampleMethod.EULER
+            if lcm_diffusion_setting.use_seed:
+                t2iconfig.seed = lcm_diffusion_setting.seed
+            else:
+                t2iconfig.seed = -1
+
+            return self.pipeline.generate_text2mg(t2iconfig)

src/backend/lora.py

@@ -0,0 +1,170 @@
+import glob
+from os import path
+from paths import get_file_name, FastStableDiffusionPaths
+from pathlib import Path
+
+
+class _lora_info:
+    """
+    A basic class to keep track of the currently loaded LoRAs and their weights.
+
+    The diffusers function _get_active_adapters()_ returns a list of adapter
+    names but not their weights so we need a way to keep track of the current
+    LoRA weights to set whenever a new LoRA is loaded.
+    """
+
+    def __init__(
+        self,
+        path: str,
+        weight: float,
+    ):
+        self.path = path
+        self.adapter_name = get_file_name(path)
+        self.weight = weight
+
+    def __del__(self):
+        self.path = None
+        self.adapter_name = None
+
+
+_loaded_loras = []
+_current_pipeline = None
+
+
+def load_lora_weight(
+    pipeline,
+    lcm_diffusion_setting,
+):
+    """
+    Loads a LoRA from the LoRA path setting.
+
+    This function loads a LoRA from the LoRA path stored in the settings so
+    it's possible to load multiple LoRAs by calling this function more than
+    once with a different LoRA path setting; note that if you plan to load
+    multiple LoRAs and dynamically change their weights, you might want to
+    set the LoRA fuse option to _False_.
+    """
+    if not lcm_diffusion_setting.lora.path:
+        raise Exception("Empty lora model path")
+
+    if not path.exists(lcm_diffusion_setting.lora.path):
+        raise Exception("Lora model path is invalid")
+
+    # If the pipeline has been rebuilt since the last call, remove all
+    # references to previously loaded LoRAs and store the new pipeline
+    global _loaded_loras
+    global _current_pipeline
+    if pipeline != _current_pipeline:
+        reset_active_lora_weights()
+        _current_pipeline = pipeline
+
+    current_lora = _lora_info(
+        lcm_diffusion_setting.lora.path,
+        lcm_diffusion_setting.lora.weight,
+    )
+    _loaded_loras.append(current_lora)
+
+    if lcm_diffusion_setting.lora.enabled:
+        print(f"LoRA adapter name : {current_lora.adapter_name}")
+        pipeline.load_lora_weights(
+            FastStableDiffusionPaths.get_lora_models_path(),
+            weight_name=Path(lcm_diffusion_setting.lora.path).name,
+            local_files_only=True,
+            adapter_name=current_lora.adapter_name,
+        )
+        update_lora_weights(
+            pipeline,
+            lcm_diffusion_setting,
+        )
+
+        if lcm_diffusion_setting.lora.fuse:
+            pipeline.fuse_lora()
+
+
+def get_lora_models(root_dir: str):
+    lora_models = glob.glob(f"{root_dir}/**/*.safetensors", recursive=True)
+    lora_models_map = {}
+    for file_path in lora_models:
+        lora_name = get_file_name(file_path)
+        if lora_name is not None:
+            lora_models_map[lora_name] = file_path
+    return lora_models_map
+
+
+def get_active_lora_weights():
+    """
+    Returns a list of _(adapter_name, weight)_ tuples for the currently loaded LoRAs.
+    """
+    active_loras = []
+    for lora_info in _loaded_loras:
+        active_loras.append(
+            (
+                lora_info.adapter_name,
+                lora_info.weight,
+            )
+        )
+    return active_loras
+
+
+def reset_active_lora_weights():
+    """
+    Clears the global list of active LoRA weights.
+
+    This method clears the list of active LoRA weights but it doesn't actually
+    remove the active LoRA weights from the current generation pipeline.
+    This method is only meant to be called when rebuilding the generation pipeline
+    as it will also clear the _current_pipeline_ variable; setting the
+    _current_pipeline_ variable to _None_ is safe here since the active LoRA weights
+    list is being reset, but it also helps to remove the pipeline reference that
+    might prevent the garbage collector from releasing the current pipeline memory.
+    """
+    global _loaded_loras
+    for lora in _loaded_loras:
+        del lora
+    del _loaded_loras
+    _loaded_loras = []
+
+    global _current_pipeline
+    _current_pipeline = None
+
+
+def update_lora_weights(
+    pipeline,
+    lcm_diffusion_setting,
+    lora_weights=None,
+):
+    """
+    Updates the LoRA weights for the currently active LoRAs.
+
+    Args:
+        pipeline: The currently active pipeline.
+        lcm_diffusion_setting: The global settings, needed to verify if the
+            pipeline is running in LCM-LoRA mode.
+        lora_weights: An optional list of updated _(adapter_name, weight)_ tuples.
+    """
+    global _loaded_loras
+    global _current_pipeline
+    if pipeline != _current_pipeline:
+        print("Wrong pipeline when trying to update LoRA weights")
+        return
+    if lora_weights:
+        for idx, lora in enumerate(lora_weights):
+            if _loaded_loras[idx].adapter_name != lora[0]:
+                print("Wrong adapter name in LoRA enumeration!")
+                continue
+            _loaded_loras[idx].weight = lora[1]
+
+    adapter_names = []
+    adapter_weights = []
+    if lcm_diffusion_setting.use_lcm_lora:
+        adapter_names.append("lcm")
+        adapter_weights.append(1.0)
+    for lora in _loaded_loras:
+        adapter_names.append(lora.adapter_name)
+        adapter_weights.append(lora.weight)
+    pipeline.set_adapters(
+        adapter_names,
+        adapter_weights=adapter_weights,
+    )
+    adapter_weights = zip(adapter_names, adapter_weights)
+    print(f"Adapters: {list(adapter_weights)}")

src/backend/models/device.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel
+
+
+class DeviceInfo(BaseModel):
+    device_type: str
+    device_name: str
+    os: str
+    platform: str
+    processor: str

src/backend/models/gen_images.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel
+from enum import Enum
+from paths import FastStableDiffusionPaths
+
+
+class ImageFormat(str, Enum):
+    """Image format"""
+
+    JPEG = "jpeg"
+    PNG = "png"
+
+
+class GeneratedImages(BaseModel):
+    path: str = FastStableDiffusionPaths.get_results_path()
+    format: str = ImageFormat.PNG.value.upper()
+    save_image: bool = True
+    save_image_quality: int = 90

src/backend/models/lcmdiffusion_setting.py

@@ -0,0 +1,77 @@
+from enum import Enum
+from PIL import Image
+from typing import Any, Optional, Union
+
+from constants import LCM_DEFAULT_MODEL, LCM_DEFAULT_MODEL_OPENVINO
+from paths import FastStableDiffusionPaths
+from pydantic import BaseModel
+
+
+class LCMLora(BaseModel):
+    base_model_id: str = "Lykon/dreamshaper-8"
+    lcm_lora_id: str = "latent-consistency/lcm-lora-sdv1-5"
+
+
+class DiffusionTask(str, Enum):
+    """Diffusion task types"""
+
+    text_to_image = "text_to_image"
+    image_to_image = "image_to_image"
+
+
+class Lora(BaseModel):
+    models_dir: str = FastStableDiffusionPaths.get_lora_models_path()
+    path: Optional[Any] = None
+    weight: Optional[float] = 0.5
+    fuse: bool = True
+    enabled: bool = False
+
+
+class ControlNetSetting(BaseModel):
+    adapter_path: Optional[str] = None  # ControlNet adapter path
+    conditioning_scale: float = 0.5
+    enabled: bool = False
+    _control_image: Image = None  # Control image, PIL image
+
+
+class GGUFModel(BaseModel):
+    gguf_models: str = FastStableDiffusionPaths.get_gguf_models_path()
+    diffusion_path: Optional[str] = None
+    clip_path: Optional[str] = None
+    t5xxl_path: Optional[str] = None
+    vae_path: Optional[str] = None
+
+
+class LCMDiffusionSetting(BaseModel):
+    lcm_model_id: str = LCM_DEFAULT_MODEL
+    openvino_lcm_model_id: str = LCM_DEFAULT_MODEL_OPENVINO
+    use_offline_model: bool = False
+    use_lcm_lora: bool = False
+    lcm_lora: Optional[LCMLora] = LCMLora()
+    use_tiny_auto_encoder: bool = False
+    use_openvino: bool = False
+    prompt: str = ""
+    negative_prompt: str = ""
+    init_image: Any = None
+    strength: Optional[float] = 0.6
+    image_height: Optional[int] = 512
+    image_width: Optional[int] = 512
+    inference_steps: Optional[int] = 1
+    guidance_scale: Optional[float] = 1
+    clip_skip: Optional[int] = 1
+    token_merging: Optional[float] = 0
+    number_of_images: Optional[int] = 1
+    seed: Optional[int] = 123123
+    use_seed: bool = False
+    use_safety_checker: bool = False
+    diffusion_task: str = DiffusionTask.text_to_image.value
+    lora: Optional[Lora] = Lora()
+    controlnet: Optional[Union[ControlNetSetting, list[ControlNetSetting]]] = None
+    dirs: dict = {
+        "controlnet": FastStableDiffusionPaths.get_controlnet_models_path(),
+        "lora": FastStableDiffusionPaths.get_lora_models_path(),
+    }
+    rebuild_pipeline: bool = False
+    rebuild_controlnet_pipeline: bool = False
+    use_gguf_model: bool = False
+    gguf_model: Optional[GGUFModel] = GGUFModel()

src/backend/models/upscale.py

@@ -0,0 +1,9 @@
+from enum import Enum
+
+
+class UpscaleMode(str, Enum):
+    """Diffusion task types"""
+
+    normal = "normal"
+    sd_upscale = "sd_upscale"
+    aura_sr = "aura_sr"