FastAPI powered API for Fooocus
from typing import List, Optional
from fastapi import Depends, FastAPI, Header, Query, UploadFile
from fastapi.params import File
import uvicorn
from fooocusapi.api_utils import generation_output, req_to_params
from fooocusapi.models import GeneratedImageBase64, ImgInpaintOrOutpaintRequest, ImgPromptRequest, ImgUpscaleOrVaryRequest, Text2ImgRequest
from fooocusapi.task_queue import TaskQueue
from fooocusapi.worker import process_generate

app = FastAPI()

task_queue = TaskQueue()

img_generate_responses = {
    "200": {
        "description": "PNG bytes if request's 'Accept' header is 'image/png', otherwise JSON",
        "content": {
            "application/json": {
                "example": [{
                    "base64": "...very long string...",
                    "seed": 1050625087,
                    "finish_reason": "SUCCESS"
                }]
            },
            "image/png": {
                "example": "PNG bytes, what did you expect?"
            }
        }
    }
}

@app.get("/")
def home():
    return "Swagger-UI to: /docs"


@app.post("/v1/generation/text-to-image", response_model=List[GeneratedImageBase64], responses=img_generate_responses)
def text2img_generation(req: Text2ImgRequest, accept: str = Header(None),
                        accept_query: str | None = Query(None, alias='accept', description="Parameter to overvide 'Accept' header, 'image/png' for output bytes")):
    if accept_query is not None and len(accept_query) > 0:
        accept = accept_query

    if accept == 'image/png':
        streaming_output = True
        # image_number auto set to 1 in streaming mode
        req.image_number = 1
    else:
        streaming_output = False

    results = process_generate(req_to_params(req))
    return generation_output(results, streaming_output)


@app.post("/v1/generation/image-upscale-vary", response_model=List[GeneratedImageBase64], responses=img_generate_responses)
def img_upscale_or_vary(input_image: UploadFile, req: ImgUpscaleOrVaryRequest = Depends(ImgUpscaleOrVaryRequest.as_form),
                        accept: str = Header(None),
                        accept_query: str | None = Query(None, alias='accept', description="Parameter to overvide 'Accept' header, 'image/png' for output bytes")):
    if accept_query is not None and len(accept_query) > 0:
        accept = accept_query

    if accept == 'image/png':
        streaming_output = True
        # image_number auto set to 1 in streaming mode
        req.image_number = 1
    else:
        streaming_output = False

    results = process_generate(req_to_params(req))
    return generation_output(results, streaming_output)


@app.post("/v1/generation/image-inpait-outpaint", response_model=List[GeneratedImageBase64], responses=img_generate_responses)
def img_inpaint_or_outpaint(input_image: UploadFile, req: ImgInpaintOrOutpaintRequest = Depends(ImgInpaintOrOutpaintRequest.as_form),
                            accept: str = Header(None),
                            accept_query: str | None = Query(None, alias='accept', description="Parameter to overvide 'Accept' header, 'image/png' for output bytes")):
    if accept_query is not None and len(accept_query) > 0:
        accept = accept_query

    if accept == 'image/png':
        streaming_output = True
        # image_number auto set to 1 in streaming mode
        req.image_number = 1
    else:
        streaming_output = False

    results = process_generate(req_to_params(req))
    return generation_output(results, streaming_output)


@app.post("/v1/generation/image-prompt", response_model=List[GeneratedImageBase64], responses=img_generate_responses)
def img_prompt(cn_img1: Optional[UploadFile] = File(None),
               req: ImgPromptRequest = Depends(ImgPromptRequest.as_form),
               accept: str = Header(None),
               accept_query: str | None = Query(None, alias='accept', description="Parameter to overvide 'Accept' header, 'image/png' for output bytes")):
    if accept_query is not None and len(accept_query) > 0:
        accept = accept_query

    if accept == 'image/png':
        streaming_output = True
        # image_number auto set to 1 in streaming mode
        req.image_number = 1
    else:
        streaming_output = False

    results = process_generate(req_to_params(req))
    return generation_output(results, streaming_output)


def start_app(args):
    uvicorn.run("fooocusapi.api:app", host=args.host,
                port=args.port, log_level=args.log_level)

fooocusapi/api.py
konieshadow/Fooocus-API
import base64
import io
from io import BytesIO
from typing import List

import numpy as np
from fastapi import Response, UploadFile
from PIL import Image
from fooocusapi.models import GeneratedImageBase64, GenerationFinishReason, ImgInpaintOrOutpaintRequest, ImgPromptRequest, ImgUpscaleOrVaryRequest, Text2ImgRequest
from fooocusapi.parameters import ImageGenerationParams, ImageGenerationResult
import modules.flags as flags


def narray_to_base64img(narray: np.ndarray) -> str:
    if narray is None:
        return None

    img = Image.fromarray(narray)
    output_buffer = BytesIO()
    img.save(output_buffer, format='PNG')
    byte_data = output_buffer.getvalue()
    base64_str = base64.b64encode(byte_data)
    return base64_str


def narray_to_bytesimg(narray) -> bytes:
    if narray is None:
        return None

    img = Image.fromarray(narray)
    output_buffer = BytesIO()
    img.save(output_buffer, format='PNG')
    byte_data = output_buffer.getvalue()
    return byte_data


def read_input_image(input_image: UploadFile) -> np.ndarray:
    input_image_bytes = input_image.file.read()
    pil_image = Image.open(io.BytesIO(input_image_bytes))
    image = np.array(pil_image)
    return image


def req_to_params(req: Text2ImgRequest) -> ImageGenerationParams:
    prompt = req.prompt
    negative_prompt = req.negative_prompt
    style_selections = [s.value for s in req.style_selections]
    performance_selection = req.performance_selection.value
    aspect_ratios_selection = req.aspect_ratios_selection.value
    image_number = req.image_number
    image_seed = None if req.image_seed == -1 else req.image_seed
    sharpness = req.sharpness
    guidance_scale = req.guidance_scale
    base_model_name = req.base_model_name
    refiner_model_name = req.refiner_model_name
    loras = [(lora.model_name, lora.weight) for lora in req.loras]
    uov_input_image = None if not isinstance(
        req, ImgUpscaleOrVaryRequest) else read_input_image(req.input_image)
    uov_method = flags.disabled if not isinstance(
        req, ImgUpscaleOrVaryRequest) else req.uov_method.value
    outpaint_selections = [] if not isinstance(req, ImgInpaintOrOutpaintRequest) else [
        s.value for s in req.outpaint_selections]

    inpaint_input_image = None
    if isinstance(req, ImgInpaintOrOutpaintRequest):
        input_image = read_input_image(req.input_image)
        if req.input_mask is not None:
            input_mask = read_input_image(req.input_mask)
        else:
            input_mask = np.zeros(input_image.shape)
        inpaint_input_image = {
            'image': input_image,
            'mask': input_mask
        }

    image_prompts = []
    if isinstance(req, ImgPromptRequest):
        for img_prompt in req.image_prompts:
            if img_prompt.cn_img is not None:
                cn_img = read_input_image(img_prompt.cn_img)
                image_prompts.append(
                    (cn_img, img_prompt.cn_stop, img_prompt.cn_weight, img_prompt.cn_type.value))

    return ImageGenerationParams(prompt=prompt,
                                 negative_prompt=negative_prompt,
                                 style_selections=style_selections,
                                 performance_selection=performance_selection,
                                 aspect_ratios_selection=aspect_ratios_selection,
                                 image_number=image_number,
                                 image_seed=image_seed,
                                 sharpness=sharpness,
                                 guidance_scale=guidance_scale,
                                 base_model_name=base_model_name,
                                 refiner_model_name=refiner_model_name,
                                 loras=loras,
                                 uov_input_image=uov_input_image,
                                 uov_method=uov_method,
                                 outpaint_selections=outpaint_selections,
                                 inpaint_input_image=inpaint_input_image,
                                 image_prompts=image_prompts
                                 )


def generation_output(results: List[ImageGenerationResult], streaming_output: bool) -> Response | List[GeneratedImageBase64]:
    if streaming_output:
        if len(results) == 0 or results[0].finish_reason != GenerationFinishReason.success:
            return Response(status_code=500)
        bytes = narray_to_bytesimg(results[0].im)
        return Response(bytes, media_type='image/png')
    else:
        results = [GeneratedImageBase64(base64=narray_to_base64img(
            item.im), seed=item.seed, finish_reason=item.finish_reason) for item in results]
        return results


class QueueReachLimitException(Exception):
    pass

fooocusapi/api_utils.py
konieshadow/Fooocus-API
from fastapi import Form, UploadFile
from fastapi.params import File
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, ConfigDict, Field
from typing import List
from enum import Enum

from pydantic_core import InitErrorDetails
from fooocusapi.parameters import GenerationFinishReason
import modules.flags as flags


class Lora(BaseModel):
    model_name: str
    weight: float = Field(default=0.5, min=-2, max=2)

    model_config = ConfigDict(
        protected_namespaces=('protect_me_', 'also_protect_')
    )


class PerfomanceSelection(str, Enum):
    speed = 'Speed'
    quality = 'Quality'


class FooocusStyle(str, Enum):
    fooocus_expansion = 'Fooocus V2'
    default = 'Default (Slightly Cinematic)'
    sai_3d_model = 'SAI 3D Model'
    sai_analog_film = 'SAI Analog Film'
    sai_anime = 'SAI Anime'
    sai_cinematic = 'SAI Cinematic'
    sai_comic_book = 'SAI Comic Book'
    sai_ccraft_clay = 'SAI Craft Clay'
    sai_digital_art = 'SAI Digital Art'
    sai_enhance = 'SAI Enhance'
    sai_fantasy_art = 'SAI Fantasy Art'
    sai_isometric = 'SAI Isometric'
    sai_line_art = 'SAI Line Art'
    sai_lowpoly = 'SAI Lowpoly'
    sai_neonpunk = 'SAI Neonpunk'
    sai_origami = 'SAI Prigami'
    sai_photographic = 'SAI Photographic'
    sai_pixel_art = 'SAI Pixel Art'
    sai_texture = 'SAI Texture'
    ads_advertising = 'Ads Advertising'
    ads_automotive = 'Ads Automotive'
    ads_corporate = 'Ads Corporate'
    ads_fashion_editorial = 'Ads Fashion Editorial'
    adsfood_photography = 'Ads Food Photography'
    ads_luxury = 'Ads Luxury'
    ads_real_estate = 'Ads Real Estate'
    ads_retail = 'Ads Retail'
    artstyle_abstract = 'Artstyle Abstract'
    artstyle_abstract_expressionism = 'Artstyle Abstract Expressionism'
    artstyle_art_deco = 'Artstyle Art Deco'
    artstyle_art_nouveau = 'Artstyle Art Nouveau'
    artstyle_constructivist = 'Artstyle Constructivist'
    artstyle_cubist = 'Artstyle Cubist'
    artstyle_expressionist = 'Artstyle Expressionist'
    artstyle_graffiti = 'Artstyle Graffiti'
    artstyle_hyperrealism = 'Artstyle Hyperrealism'
    artstyle_impressionist = 'Artstyle Impressionist'
    artstyle_pointillism = 'Artstyle Pointillism'
    artstyle_pop_art = 'Artstyle Pop Art'
    artstyle_psychedelic = 'Artstyle Psychedelic'
    artstyle_renaissance = 'Artstyle Renaissance'
    artstyle_steampunk = 'Artstyle Steampunk'
    artstyle_surrealist = 'Artstyle Surrealist'
    artstyle_typography = 'Artstyle Typography'
    artstyle_watercolor = 'Artstyle Watercolor'
    futuristic_biomechanical = 'Futuristic Biomechanical'
    futuristic_biomechanical_cyberpunk = 'Futuristic Biomechanical Cyberpunk'
    futuristic_cybernetic = 'Futuristic Cybernetic'
    futuristic_cybernetic_robot = 'Futuristic Cybernetic Robot'
    futuristic_cyberpunk_cityscape = 'Futuristic Cyberpunk Cityscape'
    futuristic_futuristic = 'Futuristic Futuristic'
    futuristic_retro_cyberpunk = 'Futuristic Retro Cyberpunk'
    futuristic_retro_futurism = 'Futuristic Retro Futurism'
    futuristic_sci_fi = 'Futuristic Sci Fi'
    futuristic_vaporwave = 'Futuristic Vaporwave'
    game_bubble_bobble = 'Game Bubble Bobble'
    game_cyberpunk_game = 'Game Cyberpunk Game'
    game_fighting_game = 'Game Fighting Game'
    game_gta = 'Game Gta'
    game_mario = 'Game Mario'
    game_minecraft = 'Game Minecraft'
    game_pokemon = 'Game Pokemon'
    game_retro_arcade = 'Game Retro Arcade'
    game_retro_game = 'Game Retro Game'
    game_rpg_fantasy_game = 'Game Rpg Fantasy Game'
    game_strategy_game = 'Game Strategy Game'
    game_streetfighter = 'Game Streetfighter'
    game_zelda = 'Game Zelda'
    misc_architectural = 'Misc Architectural'
    misc_disco = 'Misc Disco'
    misc_dreamscape = 'Misc Dreamscape'
    misc_dystopian = 'Misc Dystopian'
    misc_fairy_tale = 'Misc Fairy Tale'
    misc_gothic = 'Misc Gothic'
    misc_grunge = 'Misc Grunge'
    misc_horror = 'Misc Horror'
    misc_kawaii = 'Misc Kawaii'
    misc_lovecraftian = 'Misc Lovecraftian'
    misc_macabre = 'Misc Macabre'
    misc_manga = 'Misc Manga'
    misc_metropolis = 'Misc Metropolis'
    misc_minimalist = 'Misc Minimalist'
    misc_monochrome = 'Misc Monochrome'
    misc_nautical = 'Misc Nautical'
    misc_space = 'Misc Space'
    misc_stained_glass = 'Misc Stained Glass'
    misc_techwear_fashion = 'Misc Techwear Fashion'
    misc_tribal = 'Misc Tribal'
    misc_zentangle = 'Misc Zentangle'
    papercraft_collage = 'Papercraft Collage'
    papercraft_flat_papercut = 'Papercraft Flat Papercut'
    papercraft_kirigami = 'Papercraft Kirigami'
    papercraft_paper_mache = 'Papercraft Paper Mache'
    papercraft_paper_quilling = 'Papercraft Paper Quilling'
    papercraft_papercut_collage = 'Papercraft Papercut Collage'
    papercraft_papercut_shadow_box = 'Papercraft Papercut Shadow Box'
    papercraft_stacked_papercut = 'Papercraft Stacked Papercut'
    papercraft_thick_layered_papercut = 'Papercraft Thick Layered Papercut'
    photo_alien = 'Photo Alien'
    photo_film_noir = 'Photo Film Noir'
    photo_hdr = 'Photo Hdr'
    photo_long_exposure = 'Photo Long Exposure'
    photo_neon_noir = 'Photo Neon Noir'
    photo_silhouette = 'Photo Silhouette'
    photo_tilt_shift = 'Photo Tilt Shift'
    cinematic_diva = 'Cinematic Diva'
    abstract_expressionism = 'Abstract Expressionism'
    academia = 'Academia'
    action_figure = 'Action Figure'
    adorable_3d_character = 'Adorable 3D Character'
    adorable_kawaii = 'Adorable Kawaii'
    art_deco = 'Art Deco'
    art_nouveau = 'Art Nouveau'
    astral_aura = 'Astral Aura'
    avant_garde = 'Avant Garde'
    baroque = 'Baroque'
    bauhaus_style_poster = 'Bauhaus Style Poster'
    blueprint_schematic_drawing = 'Blueprint Schematic Drawing'
    caricature = 'Caricature'
    cel_shaded_art = 'Cel Shaded Art'
    character_design_sheet = 'Character Design Sheet'
    classicism_art = 'Classicism Art'
    color_field_painting = 'Color Field Painting'
    colored_pencil_art = 'Colored Pencil Art'
    conceptual_art = 'Conceptual Art'
    constructivism = 'Constructivism'
    cubism = 'Cubism'
    dadaism = 'Dadaism'
    dark_fantasy = 'Dark Fantasy'
    dark_moody_atmosphere = 'Dark Moody Atmosphere'
    dmt_art_style = 'DMT Art Style'
    doodle_art = 'Doodle Art'
    double_exposure = 'Double Exposure'
    dripping_paint_splatter_art = 'Dripping Paint Splatter Art'
    expressionism = 'Expressionism'
    faded_polaroid_Photo = 'Faded Polaroid Photo'
    fauvism = 'Fauvism'
    flat_2d_art = 'Flat 2D Art'
    fortnite_art_style = 'Fortnite Art Style'
    futurism = 'Futurism'
    glitchcore = 'Glitchcore'
    glo_fi = 'Glo Fi'
    googie_art_style = 'Googie Art Style'
    graffiti_art = 'Graffiti Art'
    harlem_renaissance_art = 'Harlem Renaissance Art'
    high_fashion = 'High Fashion'
    idyllic = 'Idyllic'
    impressionism = 'Impressionism'
    infographic_drawing = 'Infographic Drawing'
    ink_dripping_drawing = 'Ink Dripping Drawing'
    japanese_ink_drawing = 'Japanese Ink Drawing'
    knolling_photography = 'Knolling Photography'
    light_cheery_atmosphere = 'Light Cheery Atmosphere'
    logo_design = 'Logo Design'
    luxurious_elegance = 'Luxurious Elegance'
    macro_photography = 'Macro Photography'
    mandola_art = 'Mandola Art'
    marker_drawing = 'Marker Drawing'
    medievalism = 'Medievalism'
    minimalism = 'Minimalism'
    neo_baroque = 'Neo Baroque'
    neo_byzantine = 'Neo Byzantine'
    neo_futurism = 'Neo Futurism'
    neo_impressionism = 'Neo Impressionism'
    neo_rococo = 'Neo Rococo'
    neoclassicism = 'Neoclassicism'
    op_art = 'Op Art'
    ornate_and_intricate = 'Ornate And Intricate'
    pencil_sketch_drawing = 'Pencil Sketch Drawing'
    pop_art_2 = 'Pop Art 2'
    rococo = 'Rococo'
    silhouette_art = 'Silhouette Art'
    simple_vector_art = 'Simple Vector Art'
    sketchup = 'Sketchup'
    steampunk_2 = 'Steampunk 2'
    surrealism = 'Surrealism'
    suprematism = 'Suprematism'
    terragen = 'Terragen'
    tranquil_relaxing_atmosphere = 'Tranquil Relaxing Atmosphere'
    sticker_designs = 'Sticker Designs'
    vibrant_rim_light = 'Vibrant Rim Light'
    volumetric_lighting = 'Volumetric Lighting'
    watercolor_2 = 'Watercolor 2'
    whimsical_and_playful = 'Whimsical And Playful'


class AspectRatio(str, Enum):
    a_0_5 = '704×1408'
    a_0_52 = '704×1344'
    a_0_57 = '768×1344'
    a_0_6 = '768×1280'
    a_0_68 = '832×1216'
    a_0_72 = '832×1152'
    a_0_78 = '896×1152'
    a_0_82 = '896×1088'
    a_0_88 = '960×1088'
    a_0_94 = '960×1024'
    a_1_0 = '1024×1024'
    a_1_07 = '1024×960'
    a_1_13 = '1088×960'
    a_1_21 = '1088×896'
    a_1_29 = '1152×896'
    a_1_38 = '1152×832'
    a_1_46 = '1216×832'
    a_1_67 = '1280×768'
    a_1_75 = '1344×768'
    a_1_91 = '1344×704'
    a_2_0 = '1408×704'
    a_2_09 = '1472×704'
    a_2_4 = '1536×640'
    a_2_5 = '1600×640'
    a_2_89 = '1664×576'
    a_3_0 = '1728×576'


class UpscaleOrVaryMethod(str, Enum):
    subtle_variation = 'Vary (Subtle)'
    strong_variation = 'Vary (Strong)'
    upscale_15 = 'Upscale (1.5x)'
    upscale_2 = 'Upscale (2x)'
    upscale_fast = 'Upscale (Fast 2x)'


class OutpaintExpansion(str, Enum):
    left = 'Left'
    right = 'Right'
    top = 'Top'
    bottom = 'Bottom'


class ControlNetType(str, Enum):
    cn_ip = 'Image Prompt'
    cn_canny = 'PyraCanny'
    cn_cpds = 'CPDS'


class ImagePrompt(BaseModel):
    cn_img: UploadFile | None = Field(default=None)
    cn_stop: float = Field(default=0.4, min=0, max=1)
    cn_weight: float | None = Field(
        default=None, min=0, max=2, description="None for default value")
    cn_type: ControlNetType = Field(default=ControlNetType.cn_ip)


class Text2ImgRequest(BaseModel):
    prompt: str = ''
    negative_prompt: str = ''
    style_selections: List[FooocusStyle] = [
        FooocusStyle.fooocus_expansion, FooocusStyle.default]
    performance_selection: PerfomanceSelection = PerfomanceSelection.speed
    aspect_ratios_selection: AspectRatio = AspectRatio.a_1_29
    image_number: int = Field(
        default=1, description="Image number", min=1, max=32)
    image_seed: int = Field(default=-1, description="Seed to generate image, -1 for random")
    sharpness: float = Field(default=2.0, min=0.0, max=30.0)
    guidance_scale: float = Field(default=7.0, min=1.0, max=30.0)
    base_model_name: str = 'sd_xl_base_1.0_0.9vae.safetensors'
    refiner_model_name: str = 'sd_xl_refiner_1.0_0.9vae.safetensors'
    loras: List[Lora] = Field(default=[
        Lora(model_name='sd_xl_offset_example-lora_1.0.safetensors', weight=0.5)])


class ImgUpscaleOrVaryRequest(Text2ImgRequest):
    input_image: UploadFile
    uov_method: UpscaleOrVaryMethod

    @classmethod
    def as_form(cls, input_image: UploadFile = Form(description="Init image for upsacale or outpaint"),
                uov_method: UpscaleOrVaryMethod = Form(),
                prompt: str = Form(''),
                negative_prompt: str = Form(''),
                style_selections: List[str] = Form([
                    FooocusStyle.fooocus_expansion, FooocusStyle.default], description="Fooocus style selections, seperated by comma"),
                performance_selection: PerfomanceSelection = Form(
                    PerfomanceSelection.speed),
                aspect_ratios_selection: AspectRatio = Form(
                    AspectRatio.a_1_29),
                image_number: int = Form(
                    default=1, description="Image number", ge=1, le=32),
                image_seed: int = Form(default=-1, description="Seed to generate image, -1 for random"),
                sharpness: float = Form(default=2.0, ge=0.0, le=30.0),
                guidance_scale: float = Form(default=7.0, ge=1.0, le=30.0),
                base_model_name: str = Form(
                    'sd_xl_base_1.0_0.9vae.safetensors'),
                refiner_model_name: str = Form(
                    'sd_xl_refiner_1.0_0.9vae.safetensors'),
                l1: str | None = Form(
                    'sd_xl_offset_example-lora_1.0.safetensors'),
                w1: float = Form(default=0.5, ge=-2, le=2),
                l2: str | None = Form(None),
                w2: float = Form(default=0.5, ge=-2, le=2),
                l3: str | None = Form(None),
                w3: float = Form(default=0.5, ge=-2, le=2),
                l4: str | None = Form(None),
                w4: float = Form(default=0.5, ge=-2, le=2),
                l5: str | None = Form(None),
                w5: float = Form(default=0.5, ge=-2, le=2),
                ):
        style_selection_arr: List[FooocusStyle] = []
        for part in style_selections:
            if len(part) > 0:
                for s in part.split(','):
                    try:
                        style = FooocusStyle(s)
                        style_selection_arr.append(style)
                    except ValueError as ve:
                        err = InitErrorDetails(type='enum', loc=['style_selections'], input=style_selections, ctx={
                            'expected': 'Valid fooocus styles seperated by comma'})
                        raise RequestValidationError(errors=[err])

        loras: List[Lora] = []
        lora_config = [(l1, w1), (l2, w2), (l3, w3), (l4, w4), (l5, w5)]
        for config in lora_config:
            lora_model, lora_weight = config
            if lora_model is not None and len(lora_model) > 0:
                loras.append(Lora(model_name=lora_model, weight=lora_weight))

        return cls(input_image=input_image, uov_method=uov_method, prompt=prompt, negative_prompt=negative_prompt, style_selections=style_selection_arr,
                   performance_selection=performance_selection, aspect_ratios_selection=aspect_ratios_selection,
                   image_number=image_number, image_seed=image_seed, sharpness=sharpness, guidance_scale=guidance_scale,
                   base_model_name=base_model_name, refiner_model_name=refiner_model_name,
                   loras=loras)


class ImgInpaintOrOutpaintRequest(Text2ImgRequest):
    input_image: UploadFile
    input_mask: UploadFile | None
    outpaint_selections: List[OutpaintExpansion]

    @classmethod
    def as_form(cls, input_image: UploadFile = Form(description="Init image for inpaint or outpaint"),
                input_mask: UploadFile = Form(
                    File(None), description="Inpaint or outpaint mask"),
                outpaint_selections: List[str] = Form(
                    [], description="Outpaint expansion selections, literal 'Left', 'Right', 'Top', 'Bottom' seperated by comma"),
                prompt: str = Form(''),
                negative_prompt: str = Form(''),
                style_selections: List[str] = Form([
                    FooocusStyle.fooocus_expansion, FooocusStyle.default], description="Fooocus style selections, seperated by comma"),
                performance_selection: PerfomanceSelection = Form(
                    PerfomanceSelection.speed),
                aspect_ratios_selection: AspectRatio = Form(
                    AspectRatio.a_1_29),
                image_number: int = Form(
                    default=1, description="Image number", ge=1, le=32),
                image_seed: int = Form(default=-1, description="Seed to generate image, -1 for random"),
                sharpness: float = Form(default=2.0, ge=0.0, le=30.0),
                guidance_scale: float = Form(default=7.0, ge=1.0, le=30.0),
                base_model_name: str = Form(
                    'sd_xl_base_1.0_0.9vae.safetensors'),
                refiner_model_name: str = Form(
                    'sd_xl_refiner_1.0_0.9vae.safetensors'),
                l1: str | None = Form(
                    'sd_xl_offset_example-lora_1.0.safetensors'),
                w1: float = Form(default=0.5, ge=-2, le=2),
                l2: str | None = Form(None),
                w2: float = Form(default=0.5, ge=-2, le=2),
                l3: str | None = Form(None),
                w3: float = Form(default=0.5, ge=-2, le=2),
                l4: str | None = Form(None),
                w4: float = Form(default=0.5, ge=-2, le=2),
                l5: str | None = Form(None),
                w5: float = Form(default=0.5, ge=-2, le=2),
                ):

        if isinstance(input_mask, File):
            input_mask = None
        
        outpaint_selections_arr: List[OutpaintExpansion] = []
        for part in outpaint_selections:
            if len(part) > 0:
                for s in part.split(','):
                    try:
                        expansion = OutpaintExpansion(s)
                        outpaint_selections_arr.append(expansion)
                    except ValueError as ve:
                        err = InitErrorDetails(type='enum', loc=['outpaint_selections'], input=outpaint_selections, ctx={
                            'expected': "Literal 'Left', 'Right', 'Top', 'Bottom' seperated by comma"})
                        raise RequestValidationError(errors=[err])

        style_selection_arr: List[FooocusStyle] = []
        for part in style_selections:
            if len(part) > 0:
                for s in part.split(','):
                    try:
                        expansion = FooocusStyle(s)
                        style_selection_arr.append(expansion)
                    except ValueError as ve:
                        err = InitErrorDetails(type='enum', loc=['style_selections'], input=style_selections, ctx={
                            'expected': 'Valid fooocus styles seperated by comma'})
                        raise RequestValidationError(errors=[err])

        loras: List[Lora] = []
        lora_config = [(l1, w1), (l2, w2), (l3, w3), (l4, w4), (l5, w5)]
        for config in lora_config:
            lora_model, lora_weight = config
            if lora_model is not None and len(lora_model) > 0:
                loras.append(Lora(model_name=lora_model, weight=lora_weight))

        return cls(input_image=input_image, input_mask=input_mask, outpaint_selections=outpaint_selections_arr, prompt=prompt, negative_prompt=negative_prompt, style_selections=style_selection_arr,
                   performance_selection=performance_selection, aspect_ratios_selection=aspect_ratios_selection,
                   image_number=image_number, image_seed=image_seed, sharpness=sharpness, guidance_scale=guidance_scale,
                   base_model_name=base_model_name, refiner_model_name=refiner_model_name,
                   loras=loras)


class ImgPromptRequest(Text2ImgRequest):
    image_prompts: List[ImagePrompt]

    @classmethod
    def as_form(cls, cn_img1: UploadFile = Form(File(None), description="Input image for image prompt"),
                cn_stop1: float | None = Form(
                    default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
                cn_weight1: float | None = Form(
                    default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
                cn_type1: ControlNetType = Form(
                    default=ControlNetType.cn_ip, description="ControlNet type for image prompt"),
                cn_img2: UploadFile = Form(
                    File(None), description="Input image for image prompt"),
                cn_stop2: float | None = Form(
                    default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
                cn_weight2: float | None = Form(
                    default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
                cn_type2: ControlNetType = Form(
                    default=ControlNetType.cn_ip, description="ControlNet type for image prompt"),
                cn_img3: UploadFile = Form(
                    File(None), description="Input image for image prompt"),
                cn_stop3: float | None = Form(
                    default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
                cn_weight3: float | None = Form(
                    default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
                cn_type3: ControlNetType = Form(
                    default=ControlNetType.cn_ip, description="ControlNet type for image prompt"),
                cn_img4: UploadFile = Form(
                    File(None), description="Input image for image prompt"),
                cn_stop4: float | None = Form(
                    default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
                cn_weight4: float | None = Form(
                    default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
                cn_type4: ControlNetType = Form(
                    default=ControlNetType.cn_ip, description="ControlNet type for image prompt"),
                prompt: str = Form(''),
                negative_prompt: str = Form(''),
                style_selections: List[str] = Form([
                    FooocusStyle.fooocus_expansion, FooocusStyle.default], description="Fooocus style selections, seperated by comma"),
                performance_selection: PerfomanceSelection = Form(
                    PerfomanceSelection.speed),
                aspect_ratios_selection: AspectRatio = Form(
                    AspectRatio.a_1_29),
                image_number: int = Form(
                    default=1, description="Image number", ge=1, le=32),
                image_seed: int = Form(default=-1, description="Seed to generate image, -1 for random"),
                sharpness: float = Form(default=2.0, ge=0.0, le=30.0),
                guidance_scale: float = Form(default=7.0, ge=1.0, le=30.0),
                base_model_name: str = Form(
                    'sd_xl_base_1.0_0.9vae.safetensors'),
                refiner_model_name: str = Form(
                    'sd_xl_refiner_1.0_0.9vae.safetensors'),
                l1: str | None = Form(
                    'sd_xl_offset_example-lora_1.0.safetensors'),
                w1: float = Form(default=0.5, ge=-2, le=2),
                l2: str | None = Form(None),
                w2: float = Form(default=0.5, ge=-2, le=2),
                l3: str | None = Form(None),
                w3: float = Form(default=0.5, ge=-2, le=2),
                l4: str | None = Form(None),
                w4: float = Form(default=0.5, ge=-2, le=2),
                l5: str | None = Form(None),
                w5: float = Form(default=0.5, ge=-2, le=2),
                ):
        if isinstance(cn_img1, File):
            cn_img1 = None
        if isinstance(cn_img2, File):
            cn_img2 = None
        if isinstance(cn_img3, File):
            cn_img3 = None
        if isinstance(cn_img4, File):
            cn_img4 = None

        image_prompts: List[ImagePrompt] = []
        image_prompt_config = [(cn_img1, cn_stop1, cn_weight1, cn_type1), (cn_img2, cn_stop2, cn_weight2, cn_type2),
                               (cn_img3, cn_stop3, cn_weight3, cn_type3), (cn_img4, cn_stop4, cn_weight4, cn_type4)]
        for config in image_prompt_config:
            cn_img, cn_stop, cn_weight, cn_type = config
            if cn_stop is None:
                cn_stop = flags.default_parameters[cn_type.value][0]
            if cn_weight is None:
                cn_weight = flags.default_parameters[cn_type.value][1]
            image_prompts.append(ImagePrompt(
                cn_img=cn_img, cn_stop=cn_stop, cn_weight=cn_weight, cn_type=cn_type))

        style_selection_arr: List[FooocusStyle] = []
        for part in style_selections:
            if len(part) > 0:
                for s in part.split(','):
                    try:
                        expansion = FooocusStyle(s)
                        style_selection_arr.append(expansion)
                    except ValueError as ve:
                        err = InitErrorDetails(type='enum', loc=['style_selections'], input=style_selections, ctx={
                            'expected': 'Valid fooocus styles seperated by comma'})
                        raise RequestValidationError(errors=[err])

        loras: List[Lora] = []
        lora_config = [(l1, w1), (l2, w2), (l3, w3), (l4, w4), (l5, w5)]
        for config in lora_config:
            lora_model, lora_weight = config
            if lora_model is not None and len(lora_model) > 0:
                loras.append(Lora(model_name=lora_model, weight=lora_weight))

        return cls(image_prompts=image_prompts, prompt=prompt, negative_prompt=negative_prompt, style_selections=style_selection_arr,
                   performance_selection=performance_selection, aspect_ratios_selection=aspect_ratios_selection,
                   image_number=image_number, image_seed=image_seed, sharpness=sharpness, guidance_scale=guidance_scale,
                   base_model_name=base_model_name, refiner_model_name=refiner_model_name,
                   loras=loras)


class GeneratedImageBase64(BaseModel):
    base64: str | None = Field(
        description="Image encoded in base64, or null if finishReasen is not 'SUCCESS'")
    seed: int = Field(description="The seed associated with this image")
    finish_reason: GenerationFinishReason

fooocusapi/models.py
konieshadow/Fooocus-API
from enum import Enum
from typing import BinaryIO, Dict, List, Tuple
import numpy as np


inpaint_model_version = 'v1'


fooocus_styles = [
    'Fooocus V2',
    'Default (Slightly Cinematic)',
    'SAI 3D Model',
    'SAI Analog Film',
    'SAI Anime',
    'SAI Cinematic',
    'SAI Comic Book',
    'SAI Craft Clay',
    'SAI Digital Art',
    'SAI Enhance',
    'SAI Fantasy Art',
    'SAI Isometric',
    'SAI Line Art',
    'SAI Lowpoly',
    'SAI Neonpunk',
    'SAI Prigami',
    'SAI Photographic',
    'SAI Pixel Art',
    'SAI Texture',
    'Ads Advertising',
    'Ads Automotive',
    'Ads Corporate',
    'Ads Fashion Editorial',
    'Ads Food Photography',
    'Ads Luxury',
    'Ads Real Estate',
    'Ads Retail',
    'Artstyle Abstract',
    'Artstyle Abstract Expressionism',
    'Artstyle Art Deco',
    'Artstyle Art Nouveau',
    'Artstyle Constructivist',
    'Artstyle Cubist',
    'Artstyle Expressionist',
    'Artstyle Graffiti',
    'Artstyle Hyperrealism',
    'Artstyle Impressionist',
    'Artstyle Pointillism',
    'Artstyle Pop Art',
    'Artstyle Psychedelic',
    'Artstyle Renaissance',
    'Artstyle Steampunk',
    'Artstyle Surrealist',
    'Artstyle Typography',
    'Artstyle Watercolor',
    'Futuristic Biomechanical',
    'Futuristic Biomechanical Cyberpunk',
    'Futuristic Cybernetic',
    'Futuristic Cybernetic Robot',
    'Futuristic Cyberpunk Cityscape',
    'Futuristic Futuristic',
    'Futuristic Retro Cyberpunk',
    'Futuristic Retro Futurism',
    'Futuristic Sci Fi',
    'Futuristic Vaporwave',
    'Game Bubble Bobble',
    'Game Cyberpunk Game',
    'Game Fighting Game',
    'Game Gta',
    'Game Mario',
    'Game Minecraft',
    'Game Pokemon',
    'Game Retro Arcade',
    'Game Retro Game',
    'Game Rpg Fantasy Game',
    'Game Strategy Game',
    'Game Streetfighter',
    'Game Zelda',
    'Misc Architectural',
    'Misc Disco',
    'Misc Dreamscape',
    'Misc Dystopian',
    'Misc Fairy Tale',
    'Misc Gothic',
    'Misc Grunge',
    'Misc Horror',
    'Misc Kawaii',
    'Misc Lovecraftian',
    'Misc Macabre',
    'Misc Manga',
    'Misc Metropolis',
    'Misc Minimalist',
    'Misc Monochrome',
    'Misc Nautical',
    'Misc Space',
    'Misc Stained Glass',
    'Misc Techwear Fashion',
    'Misc Tribal',
    'Misc Zentangle',
    'Papercraft Collage',
    'Papercraft Flat Papercut',
    'Papercraft Kirigami',
    'Papercraft Paper Mache',
    'Papercraft Paper Quilling',
    'Papercraft Papercut Collage',
    'Papercraft Papercut Shadow Box',
    'Papercraft Stacked Papercut',
    'Papercraft Thick Layered Papercut',
    'Photo Alien',
    'Photo Film Noir',
    'Photo Hdr',
    'Photo Long Exposure',
    'Photo Neon Noir',
    'Photo Silhouette',
    'Photo Tilt Shift',
    'Cinematic Diva',
    'Abstract Expressionism',
    'Academia',
    'Action Figure',
    'Adorable 3D Character',
    'Adorable Kawaii',
    'Art Deco',
    'Art Nouveau',
    'Astral Aura',
    'Avant Garde',
    'Baroque',
    'Bauhaus Style Poster',
    'Blueprint Schematic Drawing',
    'Caricature',
    'Cel Shaded Art',
    'Character Design Sheet',
    'Classicism Art',
    'Color Field Painting',
    'Colored Pencil Art',
    'Conceptual Art',
    'Constructivism',
    'Cubism',
    'Dadaism',
    'Dark Fantasy',
    'Dark Moody Atmosphere',
    'DMT Art Style',
    'Doodle Art',
    'Double Exposure',
    'Dripping Paint Splatter Art',
    'Expressionism',
    'Faded Polaroid Photo',
    'Fauvism',
    'Flat 2D Art',
    'Fortnite Art Style',
    'Futurism',
    'Glitchcore',
    'Glo Fi',
    'Googie Art Style',
    'Graffiti Art',
    'Harlem Renaissance Art',
    'High Fashion',
    'Idyllic',
    'Impressionism',
    'Infographic Drawing',
    'Ink Dripping Drawing',
    'Japanese Ink Drawing',
    'Knolling Photography',
    'Light Cheery Atmosphere',
    'Logo Design',
    'Luxurious Elegance',
    'Macro Photography',
    'Mandola Art',
    'Marker Drawing',
    'Medievalism',
    'Minimalism',
    'Neo Baroque',
    'Neo Byzantine',
    'Neo Futurism',
    'Neo Impressionism',
    'Neo Rococo',
    'Neoclassicism',
    'Op Art',
    'Ornate And Intricate',
    'Pencil Sketch Drawing',
    'Pop Art 2',
    'Rococo',
    'Silhouette Art',
    'Simple Vector Art',
    'Sketchup',
    'Steampunk 2',
    'Surrealism',
    'Suprematism',
    'Terragen',
    'Tranquil Relaxing Atmosphere',
    'Sticker Designs',
    'Vibrant Rim Light',
    'Volumetric Lighting',
    'Watercolor 2',
    'Whimsical And Playful',
]


aspect_ratios = [
    '704×1408',
    '704×1344',
    '768×1344',
    '768×1280',
    '832×1216',
    '832×1152',
    '896×1152',
    '896×1088',
    '960×1088',
    '960×1024',
    '1024×1024',
    '1024×960',
    '1088×960',
    '1088×896',
    '1152×896',
    '1152×832',
    '1216×832',
    '1280×768',
    '1344×768',
    '1344×704',
    '1408×704',
    '1472×704',
    '1536×640',
    '1600×640',
    '1664×576',
    '1728×576',
]

uov_methods = [
    'Disabled', 'Vary (Subtle)', 'Vary (Strong)', 'Upscale (1.5x)', 'Upscale (2x)', 'Upscale (Fast 2x)'
]


outpaint_expansions = [
    'Left', 'Right', 'Top', 'Bottom'
]


class GenerationFinishReason(str, Enum):
    success = 'SUCCESS'
    queue_is_full = 'QUEUE_IS_FULL'
    user_cancel = 'USER_CANCEL'
    error = 'ERROR'


class ImageGenerationResult(object):
    def __init__(self, im: np.ndarray | None, seed: int, finish_reason: GenerationFinishReason):
        self.im = im
        self.seed = seed
        self.finish_reason = finish_reason


class ImageGenerationParams(object):
    def __init__(self, prompt: str,
                 negative_prompt: str,
                 style_selections: List[str],
                 performance_selection: List[str],
                 aspect_ratios_selection: str,
                 image_number: int,
                 image_seed: int | None,
                 sharpness: float,
                 guidance_scale: float,
                 base_model_name: str,
                 refiner_model_name: str,
                 loras: List[Tuple[str, float]],
                 uov_input_image: np.ndarray | None,
                 uov_method: str,
                 outpaint_selections: List[str],
                 inpaint_input_image: Dict[str, np.ndarray] | None,
                 image_prompts: List[Tuple[np.ndarray, float, float, str]]):
        self.prompt = prompt
        self.negative_prompt = negative_prompt
        self.style_selections = style_selections
        self.performance_selection = performance_selection
        self.aspect_ratios_selection = aspect_ratios_selection
        self.image_number = image_number
        self.image_seed = image_seed
        self.sharpness = sharpness
        self.guidance_scale = guidance_scale
        self.base_model_name = base_model_name
        self.refiner_model_name = refiner_model_name
        self.loras = loras
        self.uov_input_image = uov_input_image
        self.uov_method = uov_method
        self.outpaint_selections = outpaint_selections
        self.inpaint_input_image = inpaint_input_image
        self.image_prompts = image_prompts

fooocusapi/parameters.py
konieshadow/Fooocus-API
import os

fooocus_version = '2.1.44'
comfy_commit_hash = os.environ.get(
    'COMFY_COMMIT_HASH', "d1a0abd40b86f3f079b0cc71e49f9f4604831457")
fooocus_commit_hash = os.environ.get(
    'FOOOCUS_COMMIT_HASH', "5e8f2f96e91817cc6691642e71edec8b8264b114")

fooocusapi/repositories_versions.py
konieshadow/Fooocus-API
from enum import Enum
import time
from typing import List


class TaskType(str, Enum):
    text2img = 'text2img'


class QueueTask(object):
    is_finished: bool = False
    start_millis: int = 0
    finish_millis: int = 0
    finish_with_error: bool = False
    task_result: any = None

    def __init__(self, seq: int, type: TaskType, req_param: dict, in_queue_millis: int):
        self.seq = seq
        self.type = type
        self.req_param = req_param
        self.in_queue_millis = in_queue_millis


class TaskQueue(object):
    queue: List[QueueTask] = []
    history: List[QueueTask] = []
    last_seq = 0

    def __init__(self, queue_size: int = 3):
        self.queue_size = queue_size

    def add_task(self, type: TaskType, req_param: dict) -> QueueTask | None:
        """
        Create and add task to queue
        :returns: The created task's seq, or None if reach the queue size limit
        """
        if len(self.queue) >= self.queue_size:
            return None

        task = QueueTask(seq=self.last_seq+1, type=type, req_param=req_param,
                         in_queue_millis=int(round(time.time() * 1000)))
        self.queue.append(task)
        self.last_seq += task.seq
        return task.seq

    def get_task(self, seq: int, include_history: bool = False) -> QueueTask | None:
        for task in self.queue:
            if task.seq == seq:
                return task

        if include_history:
            for task in self.history:
                if task.seq == seq:
                    return task

        return None

    def is_task_ready_to_start(self, seq: int) -> bool:
        task = self.get_task(seq)
        if task is None:
            return False

        return self.queue[0].seq == seq

    def start_task(self, seq: int):
        task = self.get_task(seq)
        if task is not None:
            task.start_millis = int(round(time.time() * 1000))

    def finish_task(self, seq: int, task_result: any, finish_with_error: bool):
        task = self.get_task(seq)
        if task is not None:
            task.is_finished = True
            task.finish_millis = int(round(time.time() * 1000))
            task.finish_with_error = finish_with_error
            task.task_result = task_result

            # Move task to history
            self.queue.remove(task)
            self.history.append(task)

fooocusapi/task_queue.py
konieshadow/Fooocus-API
import copy
import random
import time
import numpy as np
import torch
from typing import List
from fooocusapi.parameters import inpaint_model_version, GenerationFinishReason, ImageGenerationParams, ImageGenerationResult
from fooocusapi.task_queue import TaskQueue, TaskType

save_log = True
task_queue = TaskQueue()


@torch.no_grad()
@torch.inference_mode()
def process_generate(params: ImageGenerationParams) -> List[ImageGenerationResult]:
    import modules.default_pipeline as pipeline
    import modules.patch as patch
    import modules.flags as flags
    import modules.core as core
    import modules.inpaint_worker as inpaint_worker
    import modules.path as path
    import comfy.model_management as model_management
    import modules.advanced_parameters as advanced_parameters
    import fooocus_extras.preprocessors as preprocessors
    import fooocus_extras.ip_adapter as ip_adapter
    from modules.util import join_prompts, remove_empty_str, image_is_generated_in_current_ui, resize_image, HWC3, make_sure_that_image_is_not_too_large
    from modules.private_logger import log
    from modules.upscaler import perform_upscale
    from modules.expansion import safe_str
    from modules.sdxl_styles import apply_style, fooocus_expansion, aspect_ratios

    outputs = []

    def progressbar(number, text):
        print(f'[Fooocus] {text}')
        outputs.append(['preview', (number, text, None)])

    def make_results_from_outputs():
        results: List[ImageGenerationResult] = []
        for item in outputs:
            if item[0] == 'results':
                for im in item[1]:
                    if isinstance(im, np.ndarray):
                        results.append(ImageGenerationResult(im=im, seed=item[2], finish_reason=GenerationFinishReason.success))
        return results

    task_seq = task_queue.add_task(TaskType.text2img, {
        'body': params.__dict__})
    if task_seq is None:
        print("[Task Queue] The task queue has reached limit")
        results = [ImageGenerationResult(im=None, seed=0,
                           finish_reason=GenerationFinishReason.queue_is_full)]
        return results

    try:
        waiting_sleep_steps: int = 0
        waiting_start_time = time.perf_counter()
        while not task_queue.is_task_ready_to_start(task_seq):
            if waiting_sleep_steps == 0:
                print(
                    f"[Task Queue] Waiting for task queue become free, seq={task_seq}")
            delay = 0.1
            time.sleep(delay)
            waiting_sleep_steps += 1
            if waiting_sleep_steps % int(10 / delay) == 0:
                waiting_time = time.perf_counter() - waiting_start_time
                print(
                    f"[Task Queue] Already waiting for {waiting_time}S, seq={task_seq}")

        print(f"[Task Queue] Task queue is free, start task, seq={task_seq}")

        task_queue.start_task(task_seq)

        execution_start_time = time.perf_counter()

        # Transform pamameters
        prompt = params.prompt
        negative_prompt = params.negative_prompt
        style_selections = params.style_selections
        performance_selection = params.performance_selection
        aspect_ratios_selection = params.aspect_ratios_selection
        image_number = params.image_number
        image_seed = None if params.image_seed == -1 else params.image_seed
        sharpness = params.sharpness
        guidance_scale = params.guidance_scale
        base_model_name = params.base_model_name
        refiner_model_name = params.refiner_model_name
        loras = params.loras
        input_image_checkbox = params.uov_input_image is not None or params.inpaint_input_image is not None or len(params.image_prompts) > 0
        current_tab = 'uov' if params.uov_method != flags.disabled else 'inpaint' if params.inpaint_input_image is not None else 'ip' if len(params.image_prompts) > 0 else None
        uov_method = params.uov_method
        uov_input_image = params.uov_input_image
        outpaint_selections = params.outpaint_selections
        inpaint_input_image = params.inpaint_input_image

        cn_tasks = {flags.cn_ip: [], flags.cn_canny: [], flags.cn_cpds: []}
        for img_prompt in params.image_prompts:
            cn_img, cn_stop, cn_weight, cn_type = img_prompt
            cn_tasks[cn_type].append([cn_img, cn_stop, cn_weight])

        def build_advanced_parameters():
            adm_scaler_positive = 1.5
            adm_scaler_negative = 0.8
            adm_scaler_end = 0.3
            adaptive_cfg = 7.0
            sampler_name = path.default_sampler
            scheduler_name = path.default_scheduler
            overwrite_step = -1
            overwrite_switch = -1
            overwrite_width = -1
            overwrite_height = -1
            overwrite_vary_strength = -1
            overwrite_upscale_strength = -1
            mixing_image_prompt_and_vary_upscale = False
            mixing_image_prompt_and_inpaint = False
            debugging_cn_preprocessor = False
            controlnet_softness = 0.25
            canny_low_threshold = 64
            canny_high_threshold = 128
            inpaint_engine = inpaint_model_version
            freeu_enabled = False
            freeu_b1, freeu_b2, freeu_s1, freeu_s2 = [None] * 4
            return [adm_scaler_positive, adm_scaler_negative, adm_scaler_end, adaptive_cfg, sampler_name,
                               scheduler_name, overwrite_step, overwrite_switch, overwrite_width, overwrite_height,
                               overwrite_vary_strength, overwrite_upscale_strength,
                               mixing_image_prompt_and_vary_upscale, mixing_image_prompt_and_inpaint,
                                debugging_cn_preprocessor, controlnet_softness, canny_low_threshold, canny_high_threshold, inpaint_engine,
                                freeu_enabled, freeu_b1, freeu_b2, freeu_s1, freeu_s2]
        
        advanced_parameters.set_all_advanced_parameters(*build_advanced_parameters())

        # Fooocus async_worker.py code start

        outpaint_selections = [o.lower() for o in outpaint_selections]
        loras_raw = copy.deepcopy(loras)
        raw_style_selections = copy.deepcopy(style_selections)
        uov_method = uov_method.lower()

        if fooocus_expansion in style_selections:
            use_expansion = True
            style_selections.remove(fooocus_expansion)
        else:
            use_expansion = False

        use_style = len(style_selections) > 0

        patch.adaptive_cfg = advanced_parameters.adaptive_cfg
        print(f'[Parameters] Adaptive CFG = {patch.adaptive_cfg}')

        patch.sharpness = sharpness
        print(f'[Parameters] Sharpness = {patch.sharpness}')

        patch.positive_adm_scale = advanced_parameters.adm_scaler_positive
        patch.negative_adm_scale = advanced_parameters.adm_scaler_negative
        patch.adm_scaler_end = advanced_parameters.adm_scaler_end
        print(f'[Parameters] ADM Scale = {patch.positive_adm_scale} : {patch.negative_adm_scale} : {patch.adm_scaler_end}')

        cfg_scale = float(guidance_scale)
        print(f'[Parameters] CFG = {cfg_scale}')

        initial_latent = None
        denoising_strength = 1.0
        tiled = False
        inpaint_worker.current_task = None
        width, height = aspect_ratios[aspect_ratios_selection]
        skip_prompt_processing = False

        raw_prompt = prompt
        raw_negative_prompt = negative_prompt

        inpaint_image = None
        input_mask = None
        inpaint_head_model_path = None
        controlnet_canny_path = None
        controlnet_cpds_path = None
        clip_vision_path, ip_negative_path, ip_adapter_path = None, None, None

        seed = image_seed
        max_seed = int(1024 * 1024 * 1024)
        if not isinstance(seed, int):
            seed = random.randint(1, max_seed)
        if seed < 0:
            seed = - seed
        seed = seed % max_seed

        if performance_selection == 'Speed':
            steps = 30
            switch = 20
        else:
            steps = 60
            switch = 40

        sampler_name = advanced_parameters.sampler_name
        scheduler_name = advanced_parameters.scheduler_name

        goals = []
        tasks = []

        if input_image_checkbox:
            if (current_tab == 'uov' or (current_tab == 'ip' and advanced_parameters.mixing_image_prompt_and_vary_upscale)) \
                    and uov_method != flags.disabled and uov_input_image is not None:
                uov_input_image = HWC3(uov_input_image)
                if 'vary' in uov_method:
                    goals.append('vary')
                elif 'upscale' in uov_method:
                    goals.append('upscale')
                    if 'fast' in uov_method:
                        skip_prompt_processing = True
                    else:
                        if performance_selection == 'Speed':
                            steps = 18
                            switch = 12
                        else:
                            steps = 36
                            switch = 24
                    progressbar(1, 'Downloading upscale models ...')
                    path.downloading_upscale_model()
            if (current_tab == 'inpaint' or (current_tab == 'ip' and advanced_parameters.mixing_image_prompt_and_inpaint))\
                    and isinstance(inpaint_input_image, dict):
                inpaint_image = inpaint_input_image['image']
                input_mask = inpaint_input_image['mask'][:, :, 0]
                inpaint_image = HWC3(inpaint_image)
                if isinstance(inpaint_image, np.ndarray) and isinstance(input_mask, np.ndarray) \
                        and (np.any(input_mask > 127) or len(outpaint_selections) > 0):
                    progressbar(1, 'Downloading inpainter ...')
                    inpaint_head_model_path, inpaint_patch_model_path = path.downloading_inpaint_models(advanced_parameters.inpaint_engine)
                    loras += [(inpaint_patch_model_path, 1.0)]
                    print(f'[Inpaint] Current inpaint model is {inpaint_patch_model_path}')
                    goals.append('inpaint')
                    sampler_name = 'dpmpp_fooocus_2m_sde_inpaint_seamless'
            if current_tab == 'ip' or \
                    advanced_parameters.mixing_image_prompt_and_inpaint or \
                    advanced_parameters.mixing_image_prompt_and_vary_upscale:
                goals.append('cn')
                progressbar(1, 'Downloading control models ...')
                if len(cn_tasks[flags.cn_canny]) > 0:
                    controlnet_canny_path = path.downloading_controlnet_canny()
                if len(cn_tasks[flags.cn_cpds]) > 0:
                    controlnet_cpds_path = path.downloading_controlnet_cpds()
                if len(cn_tasks[flags.cn_ip]) > 0:
                    clip_vision_path, ip_negative_path, ip_adapter_path = path.downloading_ip_adapters()
                progressbar(1, 'Loading control models ...')

        # Load or unload CNs
        pipeline.refresh_controlnets([controlnet_canny_path, controlnet_cpds_path])
        ip_adapter.load_ip_adapter(clip_vision_path, ip_negative_path, ip_adapter_path)

        if advanced_parameters.overwrite_step > 0:
            steps = advanced_parameters.overwrite_step

        if advanced_parameters.overwrite_switch > 0:
            switch = advanced_parameters.overwrite_switch

        if advanced_parameters.overwrite_width > 0:
            width = advanced_parameters.overwrite_width

        if advanced_parameters.overwrite_height > 0:
            height = advanced_parameters.overwrite_height

        print(f'[Parameters] Sampler = {sampler_name} - {scheduler_name}')
        print(f'[Parameters] Steps = {steps} - {switch}')

        progressbar(1, 'Initializing ...')

        if not skip_prompt_processing:

            prompts = remove_empty_str([safe_str(p) for p in prompt.split('\n')], default='')
            negative_prompts = remove_empty_str([safe_str(p) for p in negative_prompt.split('\n')], default='')

            prompt = prompts[0]
            negative_prompt = negative_prompts[0]

            extra_positive_prompts = prompts[1:] if len(prompts) > 1 else []
            extra_negative_prompts = negative_prompts[1:] if len(negative_prompts) > 1 else []

            progressbar(3, 'Loading models ...')
            pipeline.refresh_everything(refiner_model_name=refiner_model_name, base_model_name=base_model_name, loras=loras)

            progressbar(3, 'Processing prompts ...')
            positive_basic_workloads = []
            negative_basic_workloads = []

            if use_style:
                for s in style_selections:
                    p, n = apply_style(s, positive=prompt)
                    positive_basic_workloads.append(p)
                    negative_basic_workloads.append(n)
            else:
                positive_basic_workloads.append(prompt)

            negative_basic_workloads.append(negative_prompt)  # Always use independent workload for negative.

            positive_basic_workloads = positive_basic_workloads + extra_positive_prompts
            negative_basic_workloads = negative_basic_workloads + extra_negative_prompts

            positive_basic_workloads = remove_empty_str(positive_basic_workloads, default=prompt)
            negative_basic_workloads = remove_empty_str(negative_basic_workloads, default=negative_prompt)

            positive_top_k = len(positive_basic_workloads)
            negative_top_k = len(negative_basic_workloads)

            tasks = [dict(
                task_seed=seed + i,
                positive=positive_basic_workloads,
                negative=negative_basic_workloads,
                expansion='',
                c=None,
                uc=None,
            ) for i in range(image_number)]

            if use_expansion:
                for i, t in enumerate(tasks):
                    progressbar(5, f'Preparing Fooocus text #{i + 1} ...')
                    expansion = pipeline.final_expansion(prompt, t['task_seed'])
                    print(f'[Prompt Expansion] New suffix: {expansion}')
                    t['expansion'] = expansion
                    t['positive'] = copy.deepcopy(t['positive']) + [join_prompts(prompt, expansion)]  # Deep copy.

            for i, t in enumerate(tasks):
                progressbar(7, f'Encoding positive #{i + 1} ...')
                t['c'] = pipeline.clip_encode(texts=t['positive'], pool_top_k=positive_top_k)

            for i, t in enumerate(tasks):
                progressbar(10, f'Encoding negative #{i + 1} ...')
                t['uc'] = pipeline.clip_encode(texts=t['negative'], pool_top_k=negative_top_k)

        if len(goals) > 0:
            progressbar(13, 'Image processing ...')

        if 'vary' in goals:
            if not image_is_generated_in_current_ui(uov_input_image, ui_width=width, ui_height=height):
                uov_input_image = resize_image(uov_input_image, width=width, height=height)
                print(f'Resolution corrected - users are uploading their own images.')
            else:
                print(f'Processing images generated by Fooocus.')
            if 'subtle' in uov_method:
                denoising_strength = 0.5
            if 'strong' in uov_method:
                denoising_strength = 0.85
            if advanced_parameters.overwrite_vary_strength > 0:
                denoising_strength = advanced_parameters.overwrite_vary_strength

            uov_input_image = make_sure_that_image_is_not_too_large(uov_input_image)
            initial_pixels = core.numpy_to_pytorch(uov_input_image)
            progressbar(13, 'VAE encoding ...')
            initial_latent = core.encode_vae(vae=pipeline.final_vae, pixels=initial_pixels)
            B, C, H, W = initial_latent['samples'].shape
            width = W * 8
            height = H * 8
            print(f'Final resolution is {str((height, width))}.')

        if 'upscale' in goals:
            H, W, C = uov_input_image.shape
            progressbar(13, f'Upscaling image from {str((H, W))} ...')

            uov_input_image = core.numpy_to_pytorch(uov_input_image)
            uov_input_image = perform_upscale(uov_input_image)
            uov_input_image = core.pytorch_to_numpy(uov_input_image)[0]
            print(f'Image upscaled.')

            if '1.5x' in uov_method:
                f = 1.5
            elif '2x' in uov_method:
                f = 2.0
            else:
                f = 1.0

            width_f = int(width * f)
            height_f = int(height * f)

            if image_is_generated_in_current_ui(uov_input_image, ui_width=width_f, ui_height=height_f):
                uov_input_image = resize_image(uov_input_image, width=int(W * f), height=int(H * f))
                print(f'Processing images generated by Fooocus.')
            else:
                uov_input_image = resize_image(uov_input_image, width=width_f, height=height_f)
                print(f'Resolution corrected - users are uploading their own images.')

            H, W, C = uov_input_image.shape
            image_is_super_large = H * W > 2800 * 2800

            if 'fast' in uov_method:
                direct_return = True
            elif image_is_super_large:
                print('Image is too large. Directly returned the SR image. '
                      'Usually directly return SR image at 4K resolution '
                      'yields better results than SDXL diffusion.')
                direct_return = True
            else:
                direct_return = False

            if direct_return:
                d = [('Upscale (Fast)', '2x')]
                if save_log:
                    log(uov_input_image, d, single_line_number=1)
                outputs.append(['results', [uov_input_image], -1 if len(tasks) == 0 else tasks[0]['task_seed']])
                results = make_results_from_outputs()
                task_queue.finish_task(task_seq, results, False)
                return results * image_number

            tiled = True
            denoising_strength = 0.382

            if advanced_parameters.overwrite_upscale_strength > 0:
                denoising_strength = advanced_parameters.overwrite_upscale_strength

            initial_pixels = core.numpy_to_pytorch(uov_input_image)
            progressbar(13, 'VAE encoding ...')

            initial_latent = core.encode_vae(vae=pipeline.final_vae, pixels=initial_pixels, tiled=True)
            B, C, H, W = initial_latent['samples'].shape
            width = W * 8
            height = H * 8
            print(f'Final resolution is {str((height, width))}.')

        if 'inpaint' in goals:
            if len(outpaint_selections) > 0:
                H, W, C = inpaint_image.shape
                if 'top' in outpaint_selections:
                    inpaint_image = np.pad(inpaint_image, [[int(H * 0.3), 0], [0, 0], [0, 0]], mode='edge')
                    input_mask = np.pad(input_mask, [[int(H * 0.3), 0], [0, 0]], mode='constant',
                                          constant_values=255)
                if 'bottom' in outpaint_selections:
                    inpaint_image = np.pad(inpaint_image, [[0, int(H * 0.3)], [0, 0], [0, 0]], mode='edge')
                    input_mask = np.pad(input_mask, [[0, int(H * 0.3)], [0, 0]], mode='constant',
                                          constant_values=255)

                H, W, C = inpaint_image.shape
                if 'left' in outpaint_selections:
                    inpaint_image = np.pad(inpaint_image, [[0, 0], [int(H * 0.3), 0], [0, 0]], mode='edge')
                    input_mask = np.pad(input_mask, [[0, 0], [int(H * 0.3), 0]], mode='constant',
                                          constant_values=255)
                if 'right' in outpaint_selections:
                    inpaint_image = np.pad(inpaint_image, [[0, 0], [0, int(H * 0.3)], [0, 0]], mode='edge')
                    input_mask = np.pad(input_mask, [[0, 0], [0, int(H * 0.3)]], mode='constant',
                                          constant_values=255)

                inpaint_image = np.ascontiguousarray(inpaint_image.copy())
                input_mask = np.ascontiguousarray(input_mask.copy())

            inpaint_worker.current_task = inpaint_worker.InpaintWorker(image=inpaint_image, mask=input_mask,
                                                                       is_outpaint=len(outpaint_selections) > 0)

            # print(f'Inpaint task: {str((height, width))}')
            # outputs.append(['results', inpaint_worker.current_task.visualize_mask_processing()])
            # return

            progressbar(13, 'VAE encoding ...')
            inpaint_pixels = core.numpy_to_pytorch(inpaint_worker.current_task.image_ready)
            initial_latent = core.encode_vae(vae=pipeline.final_vae, pixels=inpaint_pixels)
            inpaint_latent = initial_latent['samples']
            B, C, H, W = inpaint_latent.shape
            input_mask = core.numpy_to_pytorch(inpaint_worker.current_task.mask_ready[None])
            input_mask = torch.nn.functional.avg_pool2d(input_mask, (8, 8))
            input_mask = torch.nn.functional.interpolate(input_mask, (H, W), mode='bilinear')
            inpaint_worker.current_task.load_latent(latent=inpaint_latent, mask=input_mask)

            progressbar(13, 'VAE inpaint encoding ...')

            input_mask = (inpaint_worker.current_task.mask_ready > 0).astype(np.float32)
            input_mask = torch.tensor(input_mask).float()

            vae_dict = core.encode_vae_inpaint(
                mask=input_mask, vae=pipeline.final_vae, pixels=inpaint_pixels)

            inpaint_latent = vae_dict['samples']
            input_mask = vae_dict['noise_mask']
            inpaint_worker.current_task.load_inpaint_guidance(latent=inpaint_latent, mask=input_mask,
                                                              model_path=inpaint_head_model_path)

            B, C, H, W = inpaint_latent.shape
            final_height, final_width = inpaint_worker.current_task.image_raw.shape[:2]
            height, width = H * 8, W * 8
            print(f'Final resolution is {str((final_height, final_width))}, latent is {str((height, width))}.')

        if 'cn' in goals:
            for task in cn_tasks[flags.cn_canny]:
                cn_img, cn_stop, cn_weight = task
                cn_img = resize_image(HWC3(cn_img), width=width, height=height)
                cn_img = preprocessors.canny_pyramid(cn_img)
                cn_img = HWC3(cn_img)
                task[0] = core.numpy_to_pytorch(cn_img)
                if advanced_parameters.debugging_cn_preprocessor:
                    outputs.append(['results', [cn_img], task['task_seed']])
                    results = make_results_from_outputs()
                    task_queue.finish_task(task_seq, results, False)
                    return results
            for task in cn_tasks[flags.cn_cpds]:
                cn_img, cn_stop, cn_weight = task
                cn_img = resize_image(HWC3(cn_img), width=width, height=height)
                cn_img = preprocessors.cpds(cn_img)
                cn_img = HWC3(cn_img)
                task[0] = core.numpy_to_pytorch(cn_img)
                if advanced_parameters.debugging_cn_preprocessor:
                    outputs.append(['results', [cn_img], task['task_seed']])
                    results = make_results_from_outputs()
                    task_queue.finish_task(task_seq, results, False)
                    return results
            for task in cn_tasks[flags.cn_ip]:
                cn_img, cn_stop, cn_weight = task
                cn_img = HWC3(cn_img)

                # https://github.com/tencent-ailab/IP-Adapter/blob/d580c50a291566bbf9fc7ac0f760506607297e6d/README.md?plain=1#L75
                cn_img = resize_image(cn_img, width=224, height=224, resize_mode=0)

                task[0] = ip_adapter.preprocess(cn_img)
                if advanced_parameters.debugging_cn_preprocessor:
                    outputs.append(['results', [cn_img], task['task_seed']])
                    results = make_results_from_outputs()
                    task_queue.finish_task(task_seq, results, False)
                    return results

            if len(cn_tasks[flags.cn_ip]) > 0:
                pipeline.final_unet = ip_adapter.patch_model(pipeline.final_unet, cn_tasks[flags.cn_ip])

        if advanced_parameters.freeu_enabled:
            print(f'FreeU is enabled!')
            pipeline.final_unet = core.apply_freeu(
                pipeline.final_unet,
                advanced_parameters.freeu_b1,
                advanced_parameters.freeu_b2,
                advanced_parameters.freeu_s1,
                advanced_parameters.freeu_s2
            )

        results = []
        all_steps = steps * image_number

        preparation_time = time.perf_counter() - execution_start_time
        print(f'Preparation time: {preparation_time:.2f} seconds')

        outputs.append(['preview', (13, 'Moving model to GPU ...', None)])
        execution_start_time = time.perf_counter()
        model_management.load_models_gpu([pipeline.final_unet])
        moving_time = time.perf_counter() - execution_start_time
        print(f'Moving model to GPU: {moving_time:.2f} seconds')

        outputs.append(['preview', (13, 'Starting tasks ...', None)])

        def callback(step, x0, x, total_steps, y):
            done_steps = current_task_id * steps + step
            outputs.append(['preview', (
                int(15.0 + 85.0 * float(done_steps) / float(all_steps)),
                f'Step {step}/{total_steps} in the {current_task_id + 1}-th Sampling',
                y)])

        for current_task_id, task in enumerate(tasks):
            execution_start_time = time.perf_counter()

            try:
                positive_cond, negative_cond = task['c'], task['uc']

                if 'cn' in goals:
                    for cn_flag, cn_path in [
                        (flags.cn_canny, controlnet_canny_path),
                        (flags.cn_cpds, controlnet_cpds_path)
                    ]:
                        for cn_img, cn_stop, cn_weight in cn_tasks[cn_flag]:
                            positive_cond, negative_cond = core.apply_controlnet(
                                positive_cond, negative_cond,
                                pipeline.loaded_ControlNets[cn_path], cn_img, cn_weight, 0, cn_stop)

                imgs = pipeline.process_diffusion(
                    positive_cond=positive_cond,
                    negative_cond=negative_cond,
                    steps=steps,
                    switch=switch,
                    width=width,
                    height=height,
                    image_seed=task['task_seed'],
                    callback=callback,
                    sampler_name=sampler_name,
                    scheduler_name=scheduler_name,
                    latent=initial_latent,
                    denoise=denoising_strength,
                    tiled=tiled,
                    cfg_scale=cfg_scale
                )

                del task['c'], task['uc'], positive_cond, negative_cond  # Save memory

                if inpaint_worker.current_task is not None:
                    imgs = [inpaint_worker.current_task.post_process(x) for x in imgs]

                for x in imgs:
                    d = [
                        ('Prompt', raw_prompt),
                        ('Negative Prompt', raw_negative_prompt),
                        ('Fooocus V2 Expansion', task['expansion']),
                        ('Styles', str(raw_style_selections)),
                        ('Performance', performance_selection),
                        ('Resolution', str((width, height))),
                        ('Sharpness', sharpness),
                        ('Guidance Scale', guidance_scale),
                        ('ADM Guidance', str((patch.positive_adm_scale, patch.negative_adm_scale))),
                        ('Base Model', base_model_name),
                        ('Refiner Model', refiner_model_name),
                        ('Sampler', sampler_name),
                        ('Scheduler', scheduler_name),
                        ('Seed', task['task_seed'])
                    ]
                    for n, w in loras_raw:
                        if n != 'None':
                            d.append((f'LoRA [{n}] weight', w))
                    if save_log:
                        log(x, d, single_line_number=3)
                
                # Fooocus async_worker.py code end

                results.append(ImageGenerationResult(
                    im=imgs[0], seed=task['task_seed'], finish_reason=GenerationFinishReason.success))
            except model_management.InterruptProcessingException as e:
                print('User stopped')
                results.append(ImageGenerationResult(
                        im=None, seed=task['task_seed'], finish_reason=GenerationFinishReason.user_cancel))
                break
            except Exception as e:
                print('Process failed:', e)
                results.append(ImageGenerationResult(
                    im=None, seed=task['task_seed'], finish_reason=GenerationFinishReason.error))

            execution_time = time.perf_counter() - execution_start_time
            print(f'Generating and saving time: {execution_time:.2f} seconds')

        pipeline.prepare_text_encoder(async_call=True)

        print(f"[Task Queue] Finish task, seq={task_seq}")
        task_queue.finish_task(task_seq, results, False)

        return results
    except Exception as e:
        print('Worker error:', e)
        print(f"[Task Queue] Finish task, seq={task_seq}")
        task_queue.finish_task(task_seq, [], True)
        raise e

fooocusapi/worker.py
konieshadow/Fooocus-API
import argparse
import os
import shutil
import subprocess
import sys
from importlib.util import find_spec

from fooocus_api_version import version
from fooocusapi.repositories_versions import (comfy_commit_hash,
                                              fooocus_commit_hash,
                                              fooocus_version)

os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1"

python = sys.executable
default_command_live = True
index_url = os.environ.get('INDEX_URL', "")

comfyui_name = 'ComfyUI-from-StabilityAI-Official'
fooocus_name = 'Fooocus'

modules_path = os.path.dirname(os.path.realpath(__file__))
script_path = modules_path
dir_repos = "repositories"


# This function was copied from [Fooocus](https://github.com/lllyasviel/Fooocus) repository.
def onerror(func, path, exc_info):
    import stat
    if not os.access(path, os.W_OK):
        os.chmod(path, stat.S_IWUSR)
        func(path)
    else:
        raise 'Failed to invoke "shutil.rmtree", git management failed.'


# This function was copied from [Fooocus](https://github.com/lllyasviel/Fooocus) repository.
def git_clone(url, dir, name, hash=None):
    import pygit2

    try:
        try:
            repo = pygit2.Repository(dir)
            remote_url = repo.remotes['origin'].url
            if remote_url != url:
                print(f'{name} exists but remote URL will be updated.')
                del repo
                raise url
            else:
                print(f'{name} exists and URL is correct.')
        except:
            if os.path.isdir(dir) or os.path.exists(dir):
                shutil.rmtree(dir, onerror=onerror)
            os.makedirs(dir, exist_ok=True)
            repo = pygit2.clone_repository(url, dir)
            print(f'{name} cloned from {url}.')

        remote = repo.remotes['origin']
        remote.fetch()

        commit = repo.get(hash)

        repo.checkout_tree(commit, strategy=pygit2.GIT_CHECKOUT_FORCE)
        repo.set_head(commit.id)

        print(f'{name} checkout finished for {hash}.')
    except Exception as e:
        print(f'Git clone failed for {name}: {str(e)}')


# This function was copied from [Fooocus](https://github.com/lllyasviel/Fooocus) repository.
def repo_dir(name):
    return os.path.join(script_path, dir_repos, name)


# This function was copied from [Fooocus](https://github.com/lllyasviel/Fooocus) repository.
def run(command, desc=None, errdesc=None, custom_env=None, live: bool = default_command_live) -> str:
    if desc is not None:
        print(desc)

    run_kwargs = {
        "args": command,
        "shell": True,
        "env": os.environ if custom_env is None else custom_env,
        "encoding": 'utf8',
        "errors": 'ignore',
    }

    if not live:
        run_kwargs["stdout"] = run_kwargs["stderr"] = subprocess.PIPE

    result = subprocess.run(**run_kwargs)

    if result.returncode != 0:
        error_bits = [
            f"{errdesc or 'Error running command'}.",
            f"Command: {command}",
            f"Error code: {result.returncode}",
        ]
        if result.stdout:
            error_bits.append(f"stdout: {result.stdout}")
        if result.stderr:
            error_bits.append(f"stderr: {result.stderr}")
        raise RuntimeError("\n".join(error_bits))

    return (result.stdout or "")


# This function was copied from [Fooocus](https://github.com/lllyasviel/Fooocus) repository.
def run_pip(command, desc=None, live=default_command_live):
    try:
        index_url_line = f' --index-url {index_url}' if index_url != '' else ''
        return run(f'"{python}" -m pip {command} --prefer-binary{index_url_line}', desc=f"Installing {desc}",
                   errdesc=f"Couldn't install {desc}", live=live)
    except Exception as e:
        print(e)
        print(f'CMD Failed {desc}: {command}')
        return None


def download_repositories():
    import pygit2

    pygit2.option(pygit2.GIT_OPT_SET_OWNER_VALIDATION, 0)

    http_proxy = os.environ.get('HTTP_PROXY')
    https_proxy = os.environ.get('HTTPS_PROXY')
    
    if http_proxy != None:
        print(f"Using http proxy for git clone: {http_proxy}")
        os.environ['http_proxy'] = http_proxy

    if https_proxy != None:
        print(f"Using https proxy for git clone: {https_proxy}")
        os.environ['https_proxy'] = https_proxy

    # Check and download ComfyUI
    comfy_repo = os.environ.get(
        'COMFY_REPO', "https://github.com/comfyanonymous/ComfyUI")
    git_clone(comfy_repo, repo_dir(comfyui_name),
              "Inference Engine", comfy_commit_hash)
    
    # Check and download Fooocus
    fooocus_repo = os.environ.get(
        'FOOOCUS_REPO', 'https://github.com/lllyasviel/Fooocus')
    git_clone(fooocus_repo, repo_dir(fooocus_name),
              "Fooocus", fooocus_commit_hash)
    

def is_installed(package):
    try:
        spec = find_spec(package)
    except ModuleNotFoundError:
        return False

    return spec is not None


def download_models():
    model_filenames = [
        ('sd_xl_base_1.0_0.9vae.safetensors',
         'https://huggingface.co/stabilityai/stable-diffusion-xl-base-1.0/resolve/main/sd_xl_base_1.0_0.9vae.safetensors'),
        ('sd_xl_refiner_1.0_0.9vae.safetensors',
         'https://huggingface.co/stabilityai/stable-diffusion-xl-refiner-1.0/resolve/main/sd_xl_refiner_1.0_0.9vae.safetensors')
    ]

    lora_filenames = [
        ('sd_xl_offset_example-lora_1.0.safetensors',
         'https://huggingface.co/stabilityai/stable-diffusion-xl-base-1.0/resolve/main/sd_xl_offset_example-lora_1.0.safetensors')
    ]

    vae_approx_filenames = [
        ('xlvaeapp.pth',
         'https://huggingface.co/lllyasviel/misc/resolve/main/xlvaeapp.pth')
    ]

    from modules.model_loader import load_file_from_url
    from modules.path import (fooocus_expansion_path, lorafile_path,
                              modelfile_path, upscale_models_path,
                              vae_approx_path)

    for file_name, url in model_filenames:
        load_file_from_url(url=url, model_dir=modelfile_path,
                           file_name=file_name)
    for file_name, url in lora_filenames:
        load_file_from_url(url=url, model_dir=lorafile_path,
                           file_name=file_name)
    for file_name, url in vae_approx_filenames:
        load_file_from_url(
            url=url, model_dir=vae_approx_path, file_name=file_name)

    load_file_from_url(
        url='https://huggingface.co/lllyasviel/misc/resolve/main/fooocus_expansion.bin',
        model_dir=fooocus_expansion_path,
        file_name='pytorch_model.bin'
    )


def run_pip_install():
    print("Run pip install")
    run_pip("install -r requirements.txt", "requirements")
    run_pip("install torch torchvision --extra-index-url https://download.pytorch.org/whl/cu118", "torch")
    run_pip("install xformers", "xformers")


def prepare_environments(args) -> bool:
    # Check if need pip install
    if not is_installed('xformers'):
        run_pip_install()

    skip_sync_repo = False
    if args.sync_repo is not None:
        if args.sync_repo == 'only':
            print("Only download and sync depent repositories")
            download_repositories()
            models_path = os.path.join(
                script_path, dir_repos, fooocus_name, "models")
            print(
                f"Sync repositories successful. Now you can put model files in subdirectories of '{models_path}'")
            return False
        elif args.sync_repo == 'skip':
            skip_sync_repo = True
        else:
            print(
                f"Invalid value for argument '--sync-repo', acceptable value are 'skip' and 'only'")
            exit(1)

    if not skip_sync_repo:
        download_repositories()

    # Add indent repositories to import path
    sys.path.append(os.path.join(script_path, dir_repos, comfyui_name))
    sys.path.append(os.path.join(script_path, dir_repos, fooocus_name))

    download_models()

    sys.argv = [sys.argv[0]]
    ini_comfy_args()

    if args.preload_pipeline:
        print("Preload pipeline")
        import modules.default_pipeline as _

    return True

def pre_setup(skip_sync_repo: bool=False, disable_private_log: bool=False, load_all_models: bool=False, preload_pipeline: bool=False):
    class Args(object):
        sync_repo = None
        preload_pipeline = False

    print("[Pre Setup] Prepare environments")

    args = Args()
    args.preload_pipeline = preload_pipeline
    if skip_sync_repo:
        args.sync_repo = 'skip'
    prepare_environments(args)

    sys.argv = [sys.argv[0]]
    ini_comfy_args()

    if disable_private_log:
        import fooocusapi.worker as worker
        worker.save_log = False

    if load_all_models:
        import modules.path as path
        from fooocusapi.parameters import inpaint_model_version
        path.downloading_upscale_model()
        path.downloading_inpaint_models(inpaint_model_version)
        path.downloading_controlnet_canny()
        path.downloading_controlnet_cpds()
        path.downloading_ip_adapters()
    print("[Pre Setup] Finished")


# This function was copied from [Fooocus](https://github.com/lllyasviel/Fooocus) repository.
def ini_comfy_args():
    from args_manager import args
    return args


if __name__ == "__main__":
    print(f"Python {sys.version}")
    print(f"Fooocus-API version: {version}")

    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=8888,
                        help="Set the listen port")
    parser.add_argument("--host", type=str,
                        default='127.0.0.1', help="Set the listen host")
    parser.add_argument("--log-level", type=str,
                        default='info', help="Log info for Uvicorn")
    parser.add_argument("--sync-repo", default=None,
                        help="Sync dependent git repositories to local, 'skip' for skip sync action, 'only' for only do the sync action and not launch app")
    parser.add_argument("--preload-pipeline", default=False, action="store_true", help="True for preload pipeline before start http server")

    args = parser.parse_args()

    if prepare_environments(args):
        argv = sys.argv
        sys.argv = [sys.argv[0]]

        # Start api server
        from fooocusapi.api import start_app
        start_app(args)

main.py
konieshadow/Fooocus-API
# Prediction interface for Cog ⚙️
# https://github.com/replicate/cog/blob/main/docs/python.md

import os
import sys
from typing import List
from cog import BasePredictor, Input, Path

from fooocusapi.parameters import inpaint_model_version, GenerationFinishReason, ImageGenerationParams, fooocus_styles, aspect_ratios, uov_methods, outpaint_expansions
from fooocusapi.worker import process_generate
import numpy as np
from PIL import Image


class Args(object):
    sync_repo = None


class Predictor(BasePredictor):
    def setup(self) -> None:
        """Load the model into memory to make running multiple predictions efficient"""
        from main import pre_setup
        pre_setup(disable_private_log=True, preload_pipeline=True)

    def predict(
        self,
        prompt: str = Input(default='', description="Prompt for image generation"),
        negative_prompt: str = Input(default='', description="Negtive prompt for image generation"),
        style_selections: str = Input(default='Fooocus V2,Default (Slightly Cinematic)', description="Fooocus styles applied for image generation, seperated by comma"),
        performance_selection: str = Input(default='Speed', description="Performance selection", choices=['Speed', 'Quality']),
        aspect_ratios_selection: str = Input(default='1152×896', description="The generated image's size", choices=aspect_ratios),
        image_number: int = Input(default=1, description="How many image to generate", ge=1, le=8),
        image_seed: int = Input(default=-1, description="Seed to generate image, -1 for random"),
        sharpness: float = Input(default=2.0, ge=0.0, le=30.0),
        guidance_scale: float = Input(default=7.0, ge=1.0, le=30.0),
        uov_input_image: Path = Input(default=None, description="Input image for upscale or variation, keep None for not upscale or variation"),
        uov_method: str = Input(default='Disabled', choices=uov_methods),
        inpaint_input_image: Path = Input(default=None, description="Input image for inpaint or outpaint, keep None for not inpaint or outpaint. Please noticed, `uov_input_image` has bigger priority is not None."),
        inpaint_input_mask: Path = Input(default=None, description="Input mask for inpaint"),
        outpaint_selections: str = Input(default='', description="Outpaint expansion selections, literal 'Left', 'Right', 'Top', 'Bottom' seperated by comma"),
        cn_img1: Path = Input(default=None, description="Input image for image prompt. If all cn_img[n] are None, image prompt will not applied."),
        cn_stop1: float = Input(default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
        cn_weight1: float = Input(default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
        cn_type1: str = Input(default='Image Prompt', description="ControlNet type for image prompt", choices=['Image Prompt', 'PyraCanny', 'CPDS']),
        cn_img2: Path = Input(default=None, description="Input image for image prompt. If all cn_img[n] are None, image prompt will not applied."),
        cn_stop2: float = Input(default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
        cn_weight2: float = Input(default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
        cn_type2: str = Input(default='Image Prompt', description="ControlNet type for image prompt", choices=['Image Prompt', 'PyraCanny', 'CPDS']),
        cn_img3: Path = Input(default=None, description="Input image for image prompt. If all cn_img[n] are None, image prompt will not applied."),
        cn_stop3: float = Input(default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
        cn_weight3: float = Input(default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
        cn_type3: str = Input(default='Image Prompt', description="ControlNet type for image prompt", choices=['Image Prompt', 'PyraCanny', 'CPDS']),
        cn_img4: Path = Input(default=None, description="Input image for image prompt. If all cn_img[n] are None, image prompt will not applied."),
        cn_stop4: float = Input(default=None, ge=0, le=1, description="Stop at for image prompt, None for default value"),
        cn_weight4: float = Input(default=None, ge=0, le=2, description="Weight for image prompt, None for default value"),
        cn_type4: str = Input(default='Image Prompt', description="ControlNet type for image prompt", choices=['Image Prompt', 'PyraCanny', 'CPDS']),
    ) -> List[Path]:
        """Run a single prediction on the model"""
        from modules.util import generate_temp_filename
        import modules.flags as flags

        base_model_name = 'sd_xl_base_1.0_0.9vae.safetensors'
        refiner_model_name = 'sd_xl_refiner_1.0_0.9vae.safetensors'
        loras = [('sd_xl_offset_example-lora_1.0.safetensors', 0.5)]

        style_selections_arr = []
        for s in style_selections.strip().split(','):
            style = s.strip()
            if style in fooocus_styles:
                style_selections_arr.append(style)

        if uov_input_image is not None:
            im = Image.open(str(uov_input_image))
            uov_input_image = np.array(im)

        inpaint_input_image_dict = None
        if inpaint_input_image is not None:
            im = Image.open(str(inpaint_input_image))
            inpaint_input_image = np.array(im)

            if inpaint_input_mask is not None:
                im = Image.open(str(inpaint_input_mask))
                inpaint_input_mask = np.array(im)
            else:
                inpaint_input_mask = np.zeros(inpaint_input_image.shape)
            
            inpaint_input_image_dict = {
                'image': inpaint_input_image,
                'mask': inpaint_input_mask
            }

        outpaint_selections_arr = []
        for e in outpaint_selections.strip().split(','):
            expansion = e.strip()
            if expansion in outpaint_expansions:
                outpaint_selections_arr.append(expansion)

        image_prompts = []
        image_prompt_config = [(cn_img1, cn_stop1, cn_weight1, cn_type1), (cn_img2, cn_stop2, cn_weight2, cn_type2),
                               (cn_img3, cn_stop3, cn_weight3, cn_type3), (cn_img4, cn_stop4, cn_weight4, cn_type4)]
        for config in image_prompt_config:
            cn_img, cn_stop, cn_weight, cn_type = config
            if cn_img is not None:
                im = Image.open(str(cn_img))
                cn_img = np.array(im)
                if cn_stop is None:
                    cn_stop = flags.default_parameters[cn_type][0]
                if cn_weight is None:
                    cn_weight = flags.default_parameters[cn_type][1]
                image_prompts.append((cn_img, cn_stop, cn_weight, cn_type))


        params = ImageGenerationParams(prompt=prompt,
                                 negative_prompt=negative_prompt,
                                 style_selections=style_selections_arr,
                                 performance_selection=performance_selection,
                                 aspect_ratios_selection=aspect_ratios_selection,
                                 image_number=image_number,
                                 image_seed=image_seed,
                                 sharpness=sharpness,
                                 guidance_scale=guidance_scale,
                                 base_model_name=base_model_name,
                                 refiner_model_name=refiner_model_name,
                                 loras=loras,
                                 uov_input_image=uov_input_image,
                                 uov_method=uov_method,
                                 outpaint_selections=outpaint_selections_arr,
                                 inpaint_input_image=inpaint_input_image_dict,
                                 image_prompts=image_prompts
                                 )
        
        print(f"[Predictor Predict] Params: {params.__dict__}")

        results = process_generate(params)

        output_paths: List[Path] = []
        for r in results:
            if r.finish_reason == GenerationFinishReason.success and r.im is not None:
                _, local_temp_filename, _ = generate_temp_filename('/tmp')
                os.makedirs(os.path.dirname(local_temp_filename), exist_ok=True)
                Image.fromarray(r.im).save(local_temp_filename)
                output_paths.append(Path(local_temp_filename))

        print(f"[Predictor Predict] Finished with {len(output_paths)} images")

        if len(output_paths) == 0:
            raise Exception(
                f"Process failed."
            )

        return output_paths

predict.py
konieshadow/Fooocus-API