Upload 8 files

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()

constants.py

@@ -0,0 +1,26 @@
+from os import environ, cpu_count
+
+cpu_cores = cpu_count()
+cpus = cpu_cores // 2 if cpu_cores else 0
+APP_VERSION = "v1.0.0 beta 252"
+LCM_DEFAULT_MODEL = "stabilityai/sd-turbo"
+LCM_DEFAULT_MODEL_OPENVINO = "rupeshs/sd-turbo-openvino"
+APP_NAME = "FastSD CPU"
+APP_SETTINGS_FILE = "settings.yaml"
+RESULTS_DIRECTORY = "results"
+CONFIG_DIRECTORY = "configs"
+DEVICE = environ.get("DEVICE", "cpu")
+SD_MODELS_FILE = "stable-diffusion-models.txt"
+LCM_LORA_MODELS_FILE = "lcm-lora-models.txt"
+OPENVINO_LCM_MODELS_FILE = "openvino-lcm-models.txt"
+TAESD_MODEL = "madebyollin/taesd"
+TAESDXL_MODEL = "madebyollin/taesdxl"
+TAESD_MODEL_OPENVINO = "rupeshs/taesd-ov"
+LCM_MODELS_FILE = "lcm-models.txt"
+TAESDXL_MODEL_OPENVINO = "rupeshs/taesdxl-openvino"
+LORA_DIRECTORY = "lora_models"
+CONTROLNET_DIRECTORY = "controlnet_models"
+MODELS_DIRECTORY = "models"
+GGUF_THREADS = environ.get("GGUF_THREADS", cpus)
+TAEF1_MODEL_OPENVINO = "rupeshs/taef1-openvino"
+SAFETY_CHECKER_MODEL = "Falconsai/nsfw_image_detection"

context.py

@@ -0,0 +1,109 @@
+from pprint import pprint
+from time import perf_counter
+from traceback import print_exc
+from typing import Any
+
+from app_settings import Settings
+from backend.image_saver import ImageSaver
+from backend.lcm_text_to_image import LCMTextToImage
+from backend.models.lcmdiffusion_setting import DiffusionTask
+from backend.utils import get_blank_image
+from models.interface_types import InterfaceType
+
+
+class Context:
+    def __init__(
+        self,
+        interface_type: InterfaceType,
+        device="cpu",
+    ):
+        self.interface_type = interface_type.value
+        self.lcm_text_to_image = LCMTextToImage(device)
+        self._latency = 0
+        self._error = ""
+
+    @property
+    def latency(self):
+        return self._latency
+
+    @property
+    def error(self):
+        return self._error
+
+    def generate_text_to_image(
+        self,
+        settings: Settings,
+        reshape: bool = False,
+        device: str = "cpu",
+        save_config=True,
+    ) -> Any:
+        try:
+            self._error = ""
+            tick = perf_counter()
+            from state import get_settings
+
+            if (
+                settings.lcm_diffusion_setting.diffusion_task
+                == DiffusionTask.text_to_image.value
+            ):
+                settings.lcm_diffusion_setting.init_image = None
+
+            if save_config:
+                get_settings().save()
+
+            pprint(settings.lcm_diffusion_setting.model_dump())
+            if not settings.lcm_diffusion_setting.lcm_lora:
+                return None
+            self.lcm_text_to_image.init(
+                device,
+                settings.lcm_diffusion_setting,
+            )
+
+            images = self.lcm_text_to_image.generate(
+                settings.lcm_diffusion_setting,
+                reshape,
+            )
+
+            elapsed = perf_counter() - tick
+            self._latency = elapsed
+            print(f"Latency : {elapsed:.2f} seconds")
+            if settings.lcm_diffusion_setting.controlnet:
+                if settings.lcm_diffusion_setting.controlnet.enabled:
+                    images.append(
+                        settings.lcm_diffusion_setting.controlnet._control_image
+                    )
+
+            if settings.lcm_diffusion_setting.use_safety_checker:
+                print("Safety Checker is enabled")
+                from state import get_safety_checker
+
+                safety_checker = get_safety_checker()
+                blank_image = get_blank_image(
+                    settings.lcm_diffusion_setting.image_width,
+                    settings.lcm_diffusion_setting.image_height,
+                )
+                for idx, image in enumerate(images):
+                    if not safety_checker.is_safe(image):
+                        images[idx] = blank_image
+        except Exception as exception:
+            print(f"Error in generating images: {exception}")
+            self._error = str(exception)
+            print_exc()
+            return None
+        return images
+
+    def save_images(
+        self,
+        images: Any,
+        settings: Settings,
+    ) -> list[str]:
+        saved_images = []
+        if images and settings.generated_images.save_image:
+            saved_images = ImageSaver.save_images(
+                settings.generated_images.path,
+                images=images,
+                lcm_diffusion_setting=settings.lcm_diffusion_setting,
+                format=settings.generated_images.format,
+                jpeg_quality=settings.generated_images.save_image_quality,
+            )
+        return saved_images

image_ops.py

@@ -0,0 +1,15 @@
+from PIL import Image
+
+
+def resize_pil_image(
+    pil_image: Image,
+    image_width,
+    image_height,
+):
+    return pil_image.convert("RGB").resize(
+        (
+            image_width,
+            image_height,
+        ),
+        Image.Resampling.LANCZOS,
+    )

paths.py

@@ -0,0 +1,110 @@
+import os
+import constants
+from pathlib import Path
+from time import time
+from utils import get_image_file_extension
+
+
+def join_paths(
+    first_path: str,
+    second_path: str,
+) -> str:
+    return os.path.join(first_path, second_path)
+
+
+def get_file_name(file_path: str) -> str:
+    return Path(file_path).stem
+
+
+def get_app_path() -> str:
+    app_dir = os.path.dirname(__file__)
+    work_dir = os.path.dirname(app_dir)
+    return work_dir
+
+
+def get_configs_path() -> str:
+    config_path = join_paths(get_app_path(), constants.CONFIG_DIRECTORY)
+    return config_path
+
+
+class FastStableDiffusionPaths:
+    @staticmethod
+    def get_app_settings_path() -> str:
+        configs_path = get_configs_path()
+        settings_path = join_paths(
+            configs_path,
+            constants.APP_SETTINGS_FILE,
+        )
+        return settings_path
+
+    @staticmethod
+    def get_results_path() -> str:
+        results_path = join_paths(get_app_path(), constants.RESULTS_DIRECTORY)
+        return results_path
+
+    @staticmethod
+    def get_css_path() -> str:
+        app_dir = os.path.dirname(__file__)
+        css_path = os.path.join(
+            app_dir,
+            "frontend",
+            "webui",
+            "css",
+            "style.css",
+        )
+        return css_path
+
+    @staticmethod
+    def get_models_config_path(model_config_file: str) -> str:
+        configs_path = get_configs_path()
+        models_path = join_paths(
+            configs_path,
+            model_config_file,
+        )
+        return models_path
+
+    @staticmethod
+    def get_upscale_filepath(
+        file_path_src: str,
+        scale_factor: int,
+        format: str,
+    ) -> str:
+        if file_path_src:
+            file_name_src = get_file_name(file_path_src)
+        else:
+            file_name_src = "fastsdcpu"
+
+        extension = get_image_file_extension(format)
+        upscaled_filepath = join_paths(
+            FastStableDiffusionPaths.get_results_path(),
+            f"{file_name_src}_{int(scale_factor)}x_upscale_{int(time())}{extension}",
+        )
+        return upscaled_filepath
+
+    @staticmethod
+    def get_lora_models_path() -> str:
+        lora_models_path = join_paths(get_app_path(), constants.LORA_DIRECTORY)
+        return lora_models_path
+
+    @staticmethod
+    def get_controlnet_models_path() -> str:
+        controlnet_models_path = join_paths(
+            get_app_path(), constants.CONTROLNET_DIRECTORY
+        )
+        return controlnet_models_path
+
+    @staticmethod
+    def get_gguf_models_path() -> str:
+        models_path = join_paths(get_app_path(), constants.MODELS_DIRECTORY)
+        guuf_models_path = join_paths(models_path, "gguf")
+        return guuf_models_path
+
+
+def get_base_folder_name(path: str) -> str:
+    return os.path.basename(path)
+
+
+def ensure_path(path: str) -> None:
+    """Ensure that the directory exists."""
+    if not os.path.exists(path):
+        os.makedirs(path, exist_ok=True)

state.py

@@ -0,0 +1,42 @@
+from app_settings import AppSettings
+from typing import Optional
+
+from context import Context
+from models.interface_types import InterfaceType
+from backend.safety_checker import SafetyChecker
+
+
+class _AppState:
+    _instance: Optional["_AppState"] = None
+    settings: Optional[AppSettings] = None
+    context: Optional[Context] = None
+    safety_checker: Optional[SafetyChecker] = None
+
+
+def get_state() -> _AppState:
+    if _AppState._instance is None:
+        _AppState._instance = _AppState()
+    return _AppState._instance
+
+
+def get_settings(skip_file: bool = False) -> AppSettings:
+    state = get_state()
+    if state.settings is None:
+        state.settings = AppSettings()
+        state.settings.load(skip_file)
+    return state.settings
+
+
+def get_context(interface_type: InterfaceType) -> Context:
+    state = get_state()
+    if state.context is None:
+        state.context = Context(interface_type)
+    return state.context
+
+
+def get_safety_checker() -> SafetyChecker:
+    state = get_state()
+    if state.safety_checker is None:
+        print("Initializing safety checker")
+        state.safety_checker = SafetyChecker()
+    return state.safety_checker

utils.py

@@ -0,0 +1,38 @@
+from os import path, listdir
+import platform
+from typing import List
+
+
+def show_system_info():
+    try:
+        print(f"Running on {platform.system()} platform")
+        print(f"OS: {platform.platform()}")
+        print(f"Processor: {platform.processor()}")
+    except Exception as ex:
+        print(f"Error occurred while getting system information {ex}")
+
+
+def get_models_from_text_file(file_path: str) -> List:
+    models = []
+    with open(file_path, "r") as file:
+        lines = file.readlines()
+    for repo_id in lines:
+        if repo_id.strip() != "":
+            models.append(repo_id.strip())
+    return models
+
+
+def get_image_file_extension(image_format: str) -> str:
+    if image_format == "JPEG":
+        return ".jpg"
+    elif image_format == "PNG":
+        return ".png"
+
+
+def get_files_in_dir(root_dir: str) -> List:
+    models = []
+    models.append("None")
+    for file in listdir(root_dir):
+        if file.endswith((".gguf", ".safetensors")):
+            models.append(path.join(root_dir, file))
+    return models