Public/automatic
1
0
Fork 0
mirror of https://github.com/vladmandic/automatic synced 2026-08-26 15:16:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4706 from awsr/RUF013

Browse Source 
RUF013 / PEP 484 compatibility update
This commit is contained in:
Vladimir Mandic
2026-03-24 15:19:36 +01:00
committed by GitHub
parent d429e11d55 ff247d8fd2
commit 66d4d6ca3a
100 changed files with 550 additions and 524 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/api-checkpoint.py
+1 -1
View File
@@ -24,7 +24,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-control.py
+1 -1
View File
@@ -30,7 +30,7 @@ def auth():
return None
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-detect.py
+1 -1
View File
@@ -23,7 +23,7 @@ def auth():
return None
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-enhance.py
+1 -1
View File
@@ -24,7 +24,7 @@ def auth():
return None
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-faceid.py
+1 -1
View File
@@ -29,7 +29,7 @@ def auth():
return None
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-grid.py
+1 -1
View File
@@ -79,7 +79,7 @@ def generate(x: int, y: int): # pylint: disable=redefined-outer-name
return images
def merge(images: list[Image.Image], horizontal: bool, labels: list[str] = None):
def merge(images: list[Image.Image], horizontal: bool, labels: list[str] | None = None):
rows = 1 if horizontal else len(images)
cols = math.ceil(len(images) / rows)
w = max([i.size[0] for i in images])
cli/api-img2img.py
+1 -1
View File
@@ -29,7 +29,7 @@ def auth():
return None
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-info.py
+2 -2
View File
@@ -24,7 +24,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json=dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
@@ -32,7 +32,7 @@ def get(endpoint: str, dct: dict = None):
return req.json()
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-json.py
+1 -1
View File
@@ -28,7 +28,7 @@ def auth():
return None
def post(endpoint: str, payload: dict = None):
def post(endpoint: str, payload: dict | None = None):
if 'sdapi' not in endpoint:
endpoint = f'sdapi/v1/{endpoint}'
if 'http' not in endpoint:
cli/api-mask.py
+2 -2
View File
@@ -26,7 +26,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json=dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
@@ -34,7 +34,7 @@ def get(endpoint: str, dct: dict = None):
return req.json()
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-preprocess.py
+2 -2
View File
@@ -26,7 +26,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json=dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
@@ -34,7 +34,7 @@ def get(endpoint: str, dct: dict = None):
return req.json()
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-txt2img.py
+1 -1
View File
@@ -28,7 +28,7 @@ def auth():
return None
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-upscale.py
+2 -2
View File
@@ -24,7 +24,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json=dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
@@ -32,7 +32,7 @@ def get(endpoint: str, dct: dict = None):
return req.json()
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-vqa.py
+2 -2
View File
@@ -24,7 +24,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json=dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
@@ -32,7 +32,7 @@ def get(endpoint: str, dct: dict = None):
return req.json()
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-xyz.py
+1 -1
View File
@@ -30,7 +30,7 @@ def auth():
return None
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/api-xyzenum.py
+1 -1
View File
@@ -24,7 +24,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json = dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/civitai-search.py
+3 -3
View File
@@ -99,10 +99,10 @@ def search_civitai(
types:str = '', # (Checkpoint, TextualInversion, Hypernetwork, AestheticGradient, LORA, Controlnet, Poses)
sort:str = '', # (Highest Rated, Most Downloaded, Newest)
period:str = '', # (AllTime, Year, Month, Week, Day)
nsfw:bool = None, # optional:bool
nsfw:bool | None = None, # optional:bool
limit:int = 0,
base:list[str] = [], # list
token:str = None,
token:str | None = None,
exact:bool = True,
):
import requests
@@ -169,7 +169,7 @@ def search_civitai(
return exact_models if len(exact_models) > 0 else models
def models_to_dct(all_models:list, model_id:int=None):
def models_to_dct(all_models:list, model_id:int | None=None):
dct = []
for model in all_models:
if model_id is not None and model.id != model_id:
cli/gen-styles.py
+1 -1
View File
@@ -33,7 +33,7 @@ def pil_to_b64(img: Image, size: int, quality: int):
return f'data:image/jpeg;base64,{b64encoded}'
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(endpoint, json = dct, timeout=300, verify=False)
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
cli/image-search.py
+3 -3
View File
@@ -19,7 +19,7 @@ class ImageDB:
def __init__(self,
name:str='db',
fmt:str='json',
cache_dir:str=None,
cache_dir:str | None=None,
dtype:torch.dtype=torch.float16,
device:torch.device=torch.device('cpu'),
model:str='openai/clip-vit-large-patch14', # 'facebook/dinov2-small'
@@ -123,8 +123,8 @@ class ImageDB:
self.df = rec
self.index.add(embed)
def search(self, filename: str = None, metadata: str = None, embed: np.ndarray = None, k=10, d=1.0): # search by filename/metadata/prompt-embed/image-embed
def dct(record: pd.DataFrame, mode: str, distance: float = None):
def search(self, filename: str | None = None, metadata: str | None = None, embed: np.ndarray = None, k=10, d=1.0): # search by filename/metadata/prompt-embed/image-embed
def dct(record: pd.DataFrame, mode: str, distance: float | None = None):
if distance is not None:
return {'type': mode, 'filename': record[1]['filename'], 'metadata': record[1]['metadata'], 'distance': round(distance, 2)}
else:
cli/process.py
+2 -2
View File
@@ -21,7 +21,7 @@ all_images_by_type = {}
class Result():
def __init__(self, typ: str, fn: str, tag: str = None, requested: list = []):
def __init__(self, typ: str, fn: str, tag: str | None = None, requested: list = []):
self.type = typ
self.input = fn
self.output = ''
@@ -144,7 +144,7 @@ def upscale_restore_image(res: Result, upscale: bool = False):
return res
def caption_image(res: Result, tag: str = None):
def caption_image(res: Result, tag: str | None = None):
caption = ''
tags = []
for model in options.process.caption_model:
cli/sdapi.py
+4 -4
View File
@@ -93,7 +93,7 @@ def resultsync(req: requests.Response):
return res
async def get(endpoint: str, json: dict = None):
async def get(endpoint: str, json: dict | None = None):
global sess # pylint: disable=global-statement
sess = sess if sess is not None else await session()
try:
@@ -105,7 +105,7 @@ async def get(endpoint: str, json: dict = None):
return {}
def getsync(endpoint: str, json: dict = None):
def getsync(endpoint: str, json: dict | None = None):
try:
req = requests.get(f'{sd_url}{endpoint}', json=json, verify=False, auth=authsync()) # pylint: disable=missing-timeout
res = resultsync(req)
@@ -115,7 +115,7 @@ def getsync(endpoint: str, json: dict = None):
return {}
async def post(endpoint: str, json: dict = None):
async def post(endpoint: str, json: dict | None = None):
global sess # pylint: disable=global-statement
# sess = sess if sess is not None else await session()
if sess and not sess.closed:
@@ -130,7 +130,7 @@ async def post(endpoint: str, json: dict = None):
return {}
def postsync(endpoint: str, json: dict = None):
def postsync(endpoint: str, json: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json=json, verify=False, auth=authsync()) # pylint: disable=missing-timeout
res = resultsync(req)
return res
modules/apg/pipeline_stable_cascade_prior_apg.py
+2 -2
View File
@@ -372,7 +372,7 @@ class StableCascadePriorPipelineAPG(DiffusionPipeline):
height: int = 1024,
width: int = 1024,
num_inference_steps: int = 20,
timesteps: list[float] = None,
timesteps: list[float] | None = None,
guidance_scale: float = 4.0,
negative_prompt: str | list[str] | None = None,
prompt_embeds: torch.Tensor | None = None,
@@ -386,7 +386,7 @@ class StableCascadePriorPipelineAPG(DiffusionPipeline):
output_type: str | None = "pt",
return_dict: bool = True,
callback_on_step_end: Callable[[int, int, dict], None] | None = None,
callback_on_step_end_tensor_inputs: list[str] = None,
callback_on_step_end_tensor_inputs: list[str] | None = None,
):
"""
Function invoked when calling the pipeline for generation.
modules/apg/pipeline_stable_diffision_xl_apg.py
+4 -4
View File
@@ -793,13 +793,13 @@ class StableDiffusionXLPipelineAPG(
@replace_example_docstring(EXAMPLE_DOC_STRING)
def __call__(
self,
prompt: str | list[str] = None,
prompt: str | list[str] | None = None,
prompt_2: str | list[str] | None = None,
height: int | None = None,
width: int | None = None,
num_inference_steps: int = 50,
timesteps: list[int] = None,
sigmas: list[float] = None,
timesteps: list[int] | None = None,
sigmas: list[float] | None = None,
denoising_end: float | None = None,
guidance_scale: float = 5.0,
negative_prompt: str | list[str] | None = None,
@@ -826,7 +826,7 @@ class StableDiffusionXLPipelineAPG(
negative_target_size: tuple[int, int] | None = None,
clip_skip: int | None = None,
callback_on_step_end: Callable[[int, int, dict], None] | PipelineCallback | MultiPipelineCallbacks | None = None,
callback_on_step_end_tensor_inputs: list[str] = None,
callback_on_step_end_tensor_inputs: list[str] | None = None,
**kwargs,
):
r"""
modules/apg/pipeline_stable_diffusion_apg.py
+4 -4
View File
@@ -749,12 +749,12 @@ class StableDiffusionPipelineAPG(
@replace_example_docstring(EXAMPLE_DOC_STRING)
def __call__(
self,
prompt: str | list[str] = None,
prompt: str | list[str] | None = None,
height: int | None = None,
width: int | None = None,
num_inference_steps: int = 50,
timesteps: list[int] = None,
sigmas: list[float] = None,
timesteps: list[int] | None = None,
sigmas: list[float] | None = None,
guidance_scale: float = 7.5,
negative_prompt: str | list[str] | None = None,
num_images_per_prompt: int | None = 1,
@@ -771,7 +771,7 @@ class StableDiffusionPipelineAPG(
guidance_rescale: float = 0.0,
clip_skip: int | None = None,
callback_on_step_end: Callable[[int, int, dict], None] | PipelineCallback | MultiPipelineCallbacks | None = None,
callback_on_step_end_tensor_inputs: list[str] = None,
callback_on_step_end_tensor_inputs: list[str] | None = None,
**kwargs,
):
r"""
modules/api/docs.py
+4 -4
View File
@@ -10,11 +10,11 @@ def get_swagger_ui_html(*,
title: str,
swagger_js_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js",
swagger_css_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css",
swagger_extra_css_url: str = None,
swagger_extra_css_url: str | None = None,
swagger_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png",
oauth2_redirect_url: str = None,
init_oauth: dict = None,
swagger_ui_parameters: dict = None,
oauth2_redirect_url: str | None = None,
init_oauth: dict | None = None,
swagger_ui_parameters: dict | None = None,
) -> HTMLResponse:
current_swagger_ui_parameters = swagger_ui_default_parameters.copy()
if swagger_ui_parameters:
modules/api/endpoints.py
+1 -1
View File
@@ -251,7 +251,7 @@ def get_checkpoint():
checkpoint['hash'] = shared.sd_model.sd_checkpoint_info.shorthash
return checkpoint
def set_checkpoint(sd_model_checkpoint: str, dtype:str=None, force:bool=False):
def set_checkpoint(sd_model_checkpoint: str, dtype: str | None = None, force: bool = False):
"""Load a checkpoint by name. Optionally set dtype and force a clean reload."""
from modules import sd_models, devices
if force:
modules/api/models.py
+6 -6
View File
@@ -57,10 +57,10 @@ def underscore(name: str) -> str: # Convert CamelCase or PascalCase string to un
class PydanticModelGenerator:
def __init__(
self,
model_name: str = None,
class_instance = None,
additional_fields = None,
exclude_fields: list = None,
model_name: str,
class_instance,
additional_fields: list,
exclude_fields: list | None = None,
):
if exclude_fields is None:
exclude_fields = []
@@ -100,7 +100,7 @@ class PydanticModelGenerator:
self._model_def = [x for x in self._model_def if x.field != fld]
def generate_model(self):
model_fields = { d.field: (d.field_type, Field(default=d.field_value, alias=d.field_alias, exclude=d.field_exclude)) for d in self._model_def }
model_fields: dict[str, Any] = { d.field: (d.field_type, Field(default=d.field_value, alias=d.field_alias, exclude=d.field_exclude)) for d in self._model_def }
if PYDANTIC_V2:
config = ConfigDict(arbitrary_types_allowed=True, from_attributes=True, populate_by_name=True)
else:
@@ -511,7 +511,7 @@ class ItemLoadedModel(BaseModel):
# helper function
def create_model_from_signature(func: Callable, model_name: str, base_model: type[BaseModel] = BaseModel, additional_fields: list = None, exclude_fields: list[str] = None) -> type[BaseModel]:
def create_model_from_signature(func: Callable, model_name: str, base_model: type[BaseModel] = BaseModel, additional_fields: list | None = None, exclude_fields: list[str] | None = None) -> type[BaseModel]:
from PIL import Image
if exclude_fields is None:
modules/attention.py
+2 -2
View File
@@ -186,11 +186,11 @@ def set_sage_attention(backend: str, device: torch.device):
log.error(f'Torch attention: type="Sage attention" {err}')
def set_diffusers_attention(pipe, quiet:bool=False):
def set_diffusers_attention(pipe, quiet = False):
from modules import shared
import diffusers.models.attention_processor as p
def set_attn(pipe, attention, name:str=None):
def set_attn(pipe, attention, name: str):
if attention is None:
return
# other models uses their own attention processor
modules/caption/deepbooru.py
+8 -8
View File
@@ -53,13 +53,13 @@ class DeepDanbooru:
def tag_multi(
self,
pil_image,
general_threshold: float = None,
include_rating: bool = None,
exclude_tags: str = None,
max_tags: int = None,
sort_alpha: bool = None,
use_spaces: bool = None,
escape_brackets: bool = None,
general_threshold: float | None = None,
include_rating: bool | None = None,
exclude_tags: str | None = None,
max_tags: int | None = None,
sort_alpha: bool | None = None,
use_spaces: bool | None = None,
escape_brackets: bool | None = None,
):
"""Run inference and return formatted tag string.
@@ -134,7 +134,7 @@ def get_models() -> list:
return ["DeepBooru"]
def load_model(model_name: str = None) -> bool: # pylint: disable=unused-argument
def load_model(model_name: str = "") -> bool: # pylint: disable=unused-argument
"""Load the DeepBooru model."""
try:
model.load()
modules/caption/joycaption.py
+4 -4
View File
@@ -53,12 +53,12 @@ class JoyOptions:
return f'repo="{self.repo}" temp={self.temp} top_k={self.top_k} top_p={self.top_p} sample={self.sample} tokens={self.max_new_tokens}'
processor: AutoProcessor = None
llava_model: LlavaForConditionalGeneration = None
processor: AutoProcessor | None = None
llava_model: LlavaForConditionalGeneration | None = None
opts = JoyOptions()
def load(repo: str = None):
def load(repo: str | None = None):
"""Load JoyCaption model."""
global llava_model, processor # pylint: disable=global-statement
repo = repo or opts.repo
@@ -93,7 +93,7 @@ def unload():
log.debug('JoyCaption unload: no model loaded')
def predict(question: str, image, vqa_model: str = None) -> str:
def predict(question: str, image, vqa_model: str | None = None) -> str:
opts.max_new_tokens = shared.opts.caption_vlm_max_length
load(vqa_model)
modules/caption/moondream3.py
+5 -5
View File
@@ -87,7 +87,7 @@ def _image_hash(image: Image.Image) -> str:
return h.hexdigest()
def encode_image(image: Image.Image, cache_key: str = None):
def encode_image(image: Image.Image, cache_key: str | None = None):
"""
Encode image for reuse across multiple queries.
@@ -119,7 +119,7 @@ def encode_image(image: Image.Image, cache_key: str = None):
def query(image: Image.Image, question: str, repo: str, stream: bool = False,
temperature: float = None, top_p: float = None, max_tokens: int = None,
temperature: float | None = None, top_p: float | None = None, max_tokens: int | None = None,
use_cache: bool = False, reasoning: bool = True):
"""
Visual question answering with optional streaming.
@@ -180,7 +180,7 @@ def query(image: Image.Image, question: str, repo: str, stream: bool = False,
def caption(image: Image.Image, repo: str, length: str = 'normal', stream: bool = False,
temperature: float = None, top_p: float = None, max_tokens: int = None):
temperature: float | None = None, top_p: float | None = None, max_tokens: int | None = None):
"""
Generate image captions at different lengths.
@@ -290,8 +290,8 @@ def detect(image: Image.Image, object_name: str, repo: str, max_objects: int = 1
return detections
def predict(question: str, image: Image.Image, repo: str, model_name: str = None, thinking_mode: bool = False,
mode: str = None, stream: bool = False, use_cache: bool = False, **kwargs):
def predict(question: str, image: Image.Image, repo: str, model_name: str | None = None, thinking_mode: bool = False,
mode: str | None = None, stream: bool = False, use_cache: bool = False, **kwargs):
"""
Main entry point for Moondream 3 VQA - auto-detects mode from question.
modules/caption/tagger.py
+1 -1
View File
@@ -65,7 +65,7 @@ def unload_model():
waifudiffusion.unload_model()
def tag(image, model_name: str = None, **kwargs) -> str:
def tag(image, model_name: str | None = None, **kwargs) -> str:
"""Unified tagging - dispatch to correct backend.
Args:
modules/caption/vqa.py
+30 -19
View File
@@ -50,7 +50,7 @@ def get_prompts_for_model(model_name: str) -> list:
return vlm_prompts_common
def get_internal_prompt(friendly_name: str, user_prompt: str = None) -> str:
def get_internal_prompt(friendly_name: str, user_prompt: str | None = None) -> str:
"""Convert friendly prompt name to internal token/command."""
internal = vlm_prompt_mapping.get(friendly_name, friendly_name)
@@ -350,7 +350,7 @@ class VQA:
self.processor = None
devices.torch_gc(force=True, reason='vqa model switch')
def load(self, model_name: str = None):
def load(self, model_name: str | None = None):
"""Load VLM model into memory for the specified model name."""
model_name = model_name or shared.opts.caption_vlm_model
if not model_name:
@@ -444,7 +444,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _fastvlm(self, question: str, image: Image.Image, repo: str, model_name: str = None):
def _fastvlm(self, question: str, image: Image.Image, repo: str, model_name: str | None = None):
debug(f'VQA caption: handler=fastvlm model_name="{model_name}" repo="{repo}" question="{question}" image_size={image.size if image else None}')
self._load_fastvlm(repo)
move_aux_to_gpu('vqa')
@@ -521,7 +521,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _qwen(self, question: str, image: Image.Image, repo: str, system_prompt: str = None, model_name: str = None, prefill: str = None, thinking_mode: bool = False):
def _qwen(self, question: str, image: Image.Image, repo: str, system_prompt: str | None = None, model_name: str | None = None, prefill: str | None = None, thinking_mode: bool = False):
self._load_qwen(repo)
move_aux_to_gpu('vqa')
# Get model class name for logging
@@ -644,7 +644,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _gemma(self, question: str, image: Image.Image, repo: str, system_prompt: str = None, model_name: str = None, prefill: str = None, thinking_mode: bool = False):
def _gemma(self, question: str, image: Image.Image, repo: str, system_prompt: str | None = None, model_name: str | None = None, prefill: str | None = None, thinking_mode: bool = False):
self._load_gemma(repo)
move_aux_to_gpu('vqa')
# Get model class name for logging
@@ -757,7 +757,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _mistral(self, question: str, image: Image.Image, repo: str, system_prompt: str = None, model_name: str = None, prefill: str = None, thinking_mode: bool = False):
def _mistral(self, question: str, image: Image.Image, repo: str, system_prompt: str | None = None, model_name: str | None = None, prefill: str | None = None, thinking_mode: bool = False):
self._load_mistral(repo)
move_aux_to_gpu('vqa')
cls_name = self.model.__class__.__name__
@@ -828,7 +828,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _paligemma(self, question: str, image: Image.Image, repo: str, model_name: str = None): # pylint: disable=unused-argument
def _paligemma(self, question: str, image: Image.Image, repo: str, model_name: str | None = None): # pylint: disable=unused-argument
self._load_paligemma(repo)
move_aux_to_gpu('vqa')
question = question.replace('<', '').replace('>', '').replace('_', ' ')
@@ -870,7 +870,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _ovis(self, question: str, image: Image.Image, repo: str, model_name: str = None): # pylint: disable=unused-argument
def _ovis(self, question: str, image: Image.Image, repo: str, model_name: str | None = None): # pylint: disable=unused-argument
try:
pass # pylint: disable=unused-import
except Exception:
@@ -925,7 +925,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _smol(self, question: str, image: Image.Image, repo: str, system_prompt: str = None, model_name: str = None, prefill: str = None, thinking_mode: bool = False):
def _smol(self, question: str, image: Image.Image, repo: str, system_prompt: str | None = None, model_name: str | None = None, prefill: str | None = None, thinking_mode: bool = False):
self._load_smol(repo)
move_aux_to_gpu('vqa')
# Get model class name for logging
@@ -1019,7 +1019,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _git(self, question: str, image: Image.Image, repo: str, model_name: str = None): # pylint: disable=unused-argument
def _git(self, question: str, image: Image.Image, repo: str, model_name: str | None = None): # pylint: disable=unused-argument
self._load_git(repo)
move_aux_to_gpu('vqa')
pixel_values = self.processor(images=image, return_tensors="pt").pixel_values
@@ -1053,7 +1053,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _blip(self, question: str, image: Image.Image, repo: str, model_name: str = None): # pylint: disable=unused-argument
def _blip(self, question: str, image: Image.Image, repo: str, model_name: str | None = None): # pylint: disable=unused-argument
self._load_blip(repo)
move_aux_to_gpu('vqa')
inputs = self.processor(image, question, return_tensors="pt")
@@ -1081,7 +1081,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _vilt(self, question: str, image: Image.Image, repo: str, model_name: str = None): # pylint: disable=unused-argument
def _vilt(self, question: str, image: Image.Image, repo: str, model_name: str | None = None): # pylint: disable=unused-argument
self._load_vilt(repo)
move_aux_to_gpu('vqa')
inputs = self.processor(image, question, return_tensors="pt")
@@ -1111,7 +1111,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _pix(self, question: str, image: Image.Image, repo: str, model_name: str = None): # pylint: disable=unused-argument
def _pix(self, question: str, image: Image.Image, repo: str, model_name: str | None = None): # pylint: disable=unused-argument
self._load_pix(repo)
move_aux_to_gpu('vqa')
if len(question) > 0:
@@ -1144,7 +1144,7 @@ class VQA:
register_aux('vqa', self.model)
devices.torch_gc()
def _moondream(self, question: str, image: Image.Image, repo: str, model_name: str = None, thinking_mode: bool = False):
def _moondream(self, question: str, image: Image.Image, repo: str, model_name: str | None = None, thinking_mode: bool = False):
debug(f'VQA caption: handler=moondream model_name="{model_name}" repo="{repo}" question="{question}" thinking_mode={thinking_mode}')
self._load_moondream(repo)
move_aux_to_gpu('vqa')
@@ -1207,7 +1207,7 @@ class VQA:
# When keep_thinking is False, just use the answer (reasoning is discarded)
return response
def _load_florence(self, repo: str, revision: str = None):
def _load_florence(self, repo: str, revision: str | None = None):
"""Load Florence-2 model and processor."""
_get_imports = transformers.dynamic_module_utils.get_imports
@@ -1247,7 +1247,7 @@ class VQA:
self.loaded = cache_key
devices.torch_gc()
def _florence(self, question: str, image: Image.Image, repo: str, revision: str = None, model_name: str = None): # pylint: disable=unused-argument
def _florence(self, question: str, image: Image.Image, repo: str, revision: str | None = None, model_name: str | None = None): # pylint: disable=unused-argument
self._load_florence(repo, revision)
move_aux_to_gpu('vqa')
if question.startswith('<'):
@@ -1306,7 +1306,7 @@ class VQA:
self.loaded = repo
devices.torch_gc()
def _sa2(self, question: str, image: Image.Image, repo: str, model_name: str = None): # pylint: disable=unused-argument
def _sa2(self, question: str, image: Image.Image, repo: str, model_name: str | None = None): # pylint: disable=unused-argument
self._load_sa2(repo)
move_aux_to_gpu('vqa')
if question.startswith('<'):
@@ -1325,7 +1325,18 @@ class VQA:
response = return_dict["prediction"] # the text format answer
return response
def caption(self, question: str = '', system_prompt: str = None, prompt: str = None, image: Image.Image = None, model_name: str = None, prefill: str = None, thinking_mode: bool = None, quiet: bool = False, generation_kwargs: dict = None) -> str:
def caption(
self,
question: str = "",
system_prompt: str | None = None,
prompt: str | None = None,
image: list[Image.Image] | Image.Image | dict | None = None,
model_name: str | None = None,
prefill: str | None = None,
thinking_mode: bool | None = None,
quiet: bool = False,
generation_kwargs: dict | None = None,
) -> str:
"""
Main entry point for VQA captioning. Returns string answer.
Detection data stored in self.last_detection_data for annotated image creation.
@@ -1596,7 +1607,7 @@ def unload_model():
return get_instance().unload()
def load_model(model_name: str = None):
def load_model(model_name: str | None = None):
return get_instance().load(model_name)
modules/caption/vqa_detection.py
+3 -3
View File
@@ -38,7 +38,7 @@ def parse_points(result) -> list:
return points
def parse_detections(result, label: str, max_objects: int = None) -> list:
def parse_detections(result, label: str, max_objects: int | None = None) -> list:
"""Parse and validate detection bboxes from model result.
Args:
@@ -74,7 +74,7 @@ def parse_detections(result, label: str, max_objects: int = None) -> list:
return detections
def parse_florence_detections(response, image_size: tuple = None) -> list:
def parse_florence_detections(response, image_size: tuple | None = None) -> list:
"""Parse Florence-style detection response into standard detection format.
Florence returns detection data in two possible formats:
@@ -286,7 +286,7 @@ def calculate_eye_position(face_bbox: dict) -> tuple:
return (eye_x, eye_y)
def draw_bounding_boxes(image: Image.Image, detections: list, points: list = None) -> Image.Image:
def draw_bounding_boxes(image: Image.Image, detections: list, points: list | None = None) -> Image.Image:
"""
Draw bounding boxes and/or points on an image.
modules/caption/waifudiffusion.py
+12 -12
View File
@@ -50,7 +50,7 @@ class WaifuDiffusionTagger:
self.model_path = None
self.image_size = 448 # Standard for WD models
def load(self, model_name: str = None):
def load(self, model_name: str | None = None):
"""Load the ONNX model and tags from HuggingFace."""
import huggingface_hub
@@ -195,15 +195,15 @@ class WaifuDiffusionTagger:
def predict(
self,
image: Image.Image,
general_threshold: float = None,
character_threshold: float = None,
include_rating: bool = None,
exclude_tags: str = None,
max_tags: int = None,
sort_alpha: bool = None,
use_spaces: bool = None,
escape_brackets: bool = None,
image: Image.Image | list[Image.Image] | dict | None,
general_threshold: float | None = None,
character_threshold: float | None = None,
include_rating: bool | None = None,
exclude_tags: str | None = None,
max_tags: int | None = None,
sort_alpha: bool | None = None,
use_spaces: bool | None = None,
escape_brackets: bool | None = None,
) -> str:
"""Run inference and return formatted tag string.
@@ -352,7 +352,7 @@ def refresh_models() -> list:
return get_models()
def load_model(model_name: str = None) -> bool:
def load_model(model_name: str | None = None) -> bool:
"""Load the specified WaifuDiffusion model."""
return tagger.load(model_name)
@@ -362,7 +362,7 @@ def unload_model():
tagger.unload()
def tag(image: Image.Image, model_name: str = None, **kwargs) -> str:
def tag(image: Image.Image, model_name: str | None = None, **kwargs) -> str:
"""Tag an image using WaifuDiffusion tagger.
Args:
modules/civitai/api_civitai.py
+14 -14
View File
@@ -87,12 +87,12 @@ def get_search(
sort: str = '',
period: str = '',
base_models: str = '',
nsfw: bool = None,
nsfw: bool | None = None,
limit: int = 20,
cursor: str = None,
cursor: str | None = None,
username: str = '',
favorites: bool = False,
token: str = None,
token: str | None = None,
):
"""Search CivitAI models with pagination."""
from modules.civitai.client_civitai import client
@@ -110,7 +110,7 @@ def get_search(
return response.dict(by_alias=True)
def get_model(model_id: int, token: str = None):
def get_model(model_id: int, token: str | None = None):
"""Get a single model by ID (fresh fetch)."""
from modules.civitai.client_civitai import client
model = client.get_model(model_id, token=token)
@@ -119,7 +119,7 @@ def get_model(model_id: int, token: str = None):
return model_to_dict(model)
def get_version(version_id: int, token: str = None):
def get_version(version_id: int, token: str | None = None):
"""Get a single version by ID."""
from modules.civitai.client_civitai import client
version = client.get_version(version_id, token=token)
@@ -128,7 +128,7 @@ def get_version(version_id: int, token: str = None):
return version_to_dict(version)
def get_version_by_hash(hash_str: str, token: str = None):
def get_version_by_hash(hash_str: str, token: str | None = None):
"""Look up a version by file hash."""
from modules.civitai.client_civitai import client
version = client.get_version_by_hash(hash_str, token=token)
@@ -155,13 +155,13 @@ def get_creators(query: str = '', limit: int = 20, page: int = 1):
return client.get_creators(query=query, limit=limit, page=page).dict(by_alias=True)
def get_images(model_id: int = None, model_version_id: int = None, limit: int = 20):
def get_images(model_id: int | None = None, model_version_id: int | None = None, limit: int = 20):
"""Get images with generation metadata from CivitAI."""
from modules.civitai.client_civitai import client
return {"items": client.get_images_raw(model_id=model_id, model_version_id=model_version_id, limit=limit)}
def get_me(token: str = None):
def get_me(token: str | None = None):
"""Get authenticated CivitAI user profile."""
from modules.civitai.client_civitai import client
profile = client.get_me(token=token)
@@ -300,7 +300,7 @@ def get_resolve_path(
# Metadata
# ---------------------------------------------------------------------------
def post_metadata_scan(request: dict = None):
def post_metadata_scan(request: dict | None = None):
"""Scan local models for CivitAI metadata. Optional ``page`` filters by network type (e.g. 'lora', 'model')."""
from modules.civitai import metadata_civitai
page = (request or {}).get('page', None)
@@ -388,7 +388,7 @@ def delete_banned(name: str):
# User Data — Search History
# ---------------------------------------------------------------------------
def get_history(search_type: str = None):
def get_history(search_type: str | None = None):
from modules.civitai.userdata_civitai import search_history
return {"history": search_history.list(search_type)}
@@ -529,16 +529,16 @@ def post_check_local(request: dict):
# ---------------------------------------------------------------------------
def legacy_get_civitai(
model_id: int = None,
model_id: int | None = None,
query: str = '',
tag: str = '',
types: str = '',
sort: str = '',
period: str = '',
nsfw: bool = None,
nsfw: bool | None = None,
limit: int = 0,
base: str = '',
token: str = None,
token: str | None = None,
exact: bool = True,
):
"""Legacy GET /sdapi/v1/civitai — delegates to search or model lookup."""
@@ -559,7 +559,7 @@ def legacy_get_civitai(
return JSONResponse(content=[], status_code=200)
def legacy_post_civitai(page: str = None):
def legacy_post_civitai(page: str | None = None):
"""Legacy POST /sdapi/v1/civitai — scan metadata."""
from modules.civitai import metadata_civitai
result = []
modules/civitai/download_civitai.py
+1 -1
View File
@@ -406,7 +406,7 @@ def download_civit_preview(model_path: str, preview_url: str):
return 200, str(total_size), ''
def download_civit_model(model_url: str, model_name: str = '', model_path: str = '', model_type: str = '', token: str = None,
def download_civit_model(model_url: str, model_name: str = '', model_path: str = '', model_type: str = '', token: str | None = None,
base_model: str = '', model_id: int = 0, version_id: int = 0):
"""Legacy function — delegates to DownloadManager for non-blocking downloads."""
if not model_url:
modules/civitai/metadata_civitai.py
+1 -1
View File
@@ -169,7 +169,7 @@ def atomic_civit_search_metadata(item, results):
results.append(dict(result))
def civit_search_metadata(title: str = None, raw: bool = False):
def civit_search_metadata(title: str | None = None, raw: bool = False):
def create_search_metadata_table(rows):
html = """
<table class="simple-table">
modules/civitai/search_civitai.py
+2 -2
View File
@@ -15,10 +15,10 @@ def search_civitai(
types: str = '',
sort: str = '',
period: str = '',
nsfw: bool = None,
nsfw: bool | None = None,
limit: int = 0,
base: str = '',
token: str = None,
token: str | None = None,
exact: bool = True,
) -> list[CivitModel]:
if not query and not tag and not sort:
modules/civitai/userdata_civitai.py
+1 -1
View File
@@ -111,7 +111,7 @@ class SearchHistory:
self._entries = self._entries[:self._max_entries]
self._save()
def list(self, search_type: str = None) -> list[dict]:
def list(self, search_type: str | None = None) -> list[dict]:
with self._lock:
if search_type:
return [e for e in self._entries if e.get('type') == search_type]
modules/control/processor.py
+13 -13
View File
@@ -61,19 +61,19 @@ processors = [
def preprocess_image(
p:StableDiffusionProcessingControl,
pipe,
input_image:Image.Image = None,
init_image:Image.Image = None,
input_mask:Image.Image = None,
input_type:str = 0,
unit_type:str = 'controlnet',
active_process:list = None,
active_model:list = None,
selected_models:list = None,
has_models:bool = False,
active_units:list = None,
):
p: StableDiffusionProcessingControl,
pipe,
input_image: Image.Image | None= None,
init_image: Image.Image | None = None,
input_mask: Image.Image | None = None,
input_type = 0,
unit_type = "controlnet",
active_process: list | None = None,
active_model: list | None = None,
selected_models: list | None = None,
has_models = False,
active_units: list | None = None,
):
if selected_models is None:
selected_models = []
if active_model is None:
modules/control/processors.py
+7 -6
View File
@@ -195,20 +195,21 @@ def update_settings(*settings):
class Processor:
def __init__(self, processor_id: str = None, resize = True):
def __init__(self, processor_id: str | None = None, resize = True):
self.model = None
self.processor_id = None
self.override = None
self.processor_id: str | None = None
self.override: Image.Image | None = None
self.resize = resize
self.reset()
self.config(processor_id)
self.load_config: dict = {}
if processor_id is not None:
self.load()
def __str__(self):
return f' Processor(id={self.processor_id} model={self.model.__class__.__name__})' if self.processor_id and self.model else ''
def reset(self, processor_id: str = None):
def reset(self, processor_id: str | None = None):
if self.model is not None:
debug(f'Control Processor unloaded: id="{self.processor_id}"')
self.model = None
@@ -237,7 +238,7 @@ class Processor:
for k, v in from_config.items():
self.load_config[k] = v
def load(self, processor_id: str = None, force: bool = True) -> str:
def load(self, processor_id: str | None = None, force: bool = True) -> str:
from modules.shared import state
try:
t0 = time.time()
@@ -304,7 +305,7 @@ class Processor:
display(e, 'Control Processor load')
return f'Processor load filed: {processor_id}'
def __call__(self, image_input: Image, mode: str = 'RGB', width: int = 0, height: int = 0, resize_mode: int = 0, resize_name: str = 'None', scale_tab: int = 1, scale_by: float = 1.0, local_config: dict = None):
def __call__(self, image_input: Image, mode: str = 'RGB', width: int = 0, height: int = 0, resize_mode: int = 0, resize_name: str = 'None', scale_tab: int = 1, scale_by: float = 1.0, local_config: dict | None = None):
"""Run the preprocessor on an input image and return the processed control map.
Args:
modules/control/run.py
+46 -46
View File
@@ -140,7 +140,7 @@ def set_pipe(p, has_models, unit_type, selected_models, active_model, active_str
return pipe
def check_active(p, unit_type, units):
def check_active(p, unit_type: str, units: list[unit.Unit]):
active_process: list[processors.Processor] = [] # all active preprocessors
active_model: list[controlnet.ControlNet | xs.ControlNetXS | t2iadapter.Adapter] = [] # all active models
active_strength: list[float] = [] # strength factors for all active models
@@ -216,9 +216,9 @@ def check_active(p, unit_type, units):
return active_process, active_model, active_strength, active_start, active_end, active_units
def check_enabled(p, unit_type, units, active_model, active_strength, active_start, active_end):
def check_enabled(p, unit_type: str, units: list[unit.Unit], active_model: list[controlnet.ControlNet | xs.ControlNetXS | t2iadapter.Adapter], active_strength: list[float], active_start: list[float], active_end: list[float]):
has_models = False
selected_models: list[controlnet.ControlNetModel | xs.ControlNetXSModel | t2iadapter.AdapterModel] = None
selected_models: list[controlnet.ControlNetModel | xs.ControlNetXSModel | t2iadapter.AdapterModel] | None = None
control_conditioning = None
control_guidance_start = None
control_guidance_end = None
@@ -271,17 +271,17 @@ def init_units(units: list[unit.Unit]):
def control_run(state: str = '', # pylint: disable=keyword-arg-before-vararg
units: list[unit.Unit] = None, inputs: list[Image.Image] = None, inits: list[Image.Image] = None, mask: Image.Image = None, unit_type: str = None, is_generator: bool = True,
units: list[unit.Unit] | None = None, inputs: list[Image.Image] | None = None, inits: list[Image.Image] | None = None, mask: Image.Image = None, unit_type: str | None = None, is_generator: bool = True,
input_type: int = 0,
prompt: str = '', negative_prompt: str = '', styles: list[str] = None,
steps: int = 20, sampler_index: int = None,
prompt: str = '', negative_prompt: str = '', styles: list[str] | None = None,
steps: int = 20, sampler_index: int | None = None,
seed: int = -1, subseed: int = -1, subseed_strength: float = 0, seed_resize_from_h: int = -1, seed_resize_from_w: int = -1,
guidance_name: str = 'Default', guidance_scale: float = 6.0, guidance_rescale: float = 0.0, guidance_start: float = 0.0, guidance_stop: float = 1.0,
cfg_scale: float = 6.0, clip_skip: float = 1.0, image_cfg_scale: float = 6.0, diffusers_guidance_rescale: float = 0.7, pag_scale: float = 0.0, pag_adaptive: float = 0.5, cfg_end: float = 1.0,
vae_type: str = 'Full', tiling: bool = False, hidiffusion: bool = False,
detailer_enabled: bool = False, detailer_prompt: str = '', detailer_negative: str = '', detailer_steps: int = 10, detailer_strength: float = 0.3, detailer_resolution: int = 1024,
hdr_mode: int = 0, hdr_brightness: float = 0, hdr_color: float = 0, hdr_sharpen: float = 0, hdr_clamp: bool = False, hdr_boundary: float = 4.0, hdr_threshold: float = 0.95,
hdr_maximize: bool = False, hdr_max_center: float = 0.6, hdr_max_boundary: float = 1.0, hdr_color_picker: str = None, hdr_tint_ratio: float = 0, hdr_apply_hires: bool = True,
hdr_maximize: bool = False, hdr_max_center: float = 0.6, hdr_max_boundary: float = 1.0, hdr_color_picker: str | None = None, hdr_tint_ratio: float = 0, hdr_apply_hires: bool = True,
grading_brightness: float = 0.0, grading_contrast: float = 0.0, grading_saturation: float = 0.0, grading_hue: float = 0.0,
grading_gamma: float = 1.0, grading_sharpness: float = 0.0, grading_color_temp: float = 6500,
grading_shadows: float = 0.0, grading_midtones: float = 0.0, grading_highlights: float = 0.0,
@@ -293,54 +293,54 @@ def control_run(state: str = '', # pylint: disable=keyword-arg-before-vararg
resize_mode_after: int = 0, resize_name_after: str = 'None', resize_context_after: str = 'None', width_after: int = 0, height_after: int = 0, scale_by_after: float = 1.0, selected_scale_tab_after: int = 0,
resize_mode_mask: int = 0, resize_name_mask: str = 'None', resize_context_mask: str = 'None', width_mask: int = 0, height_mask: int = 0, scale_by_mask: float = 1.0, selected_scale_tab_mask: int = 0,
denoising_strength: float = 0.3, batch_count: int = 1, batch_size: int = 1,
enable_hr: bool = False, hr_sampler_index: int = None, hr_denoising_strength: float = 0.0, hr_resize_mode: int = 0, hr_resize_context: str = 'None', hr_upscaler: str = None, hr_force: bool = False, hr_second_pass_steps: int = 20,
enable_hr: bool = False, hr_sampler_index: int | None = None, hr_denoising_strength: float = 0.0, hr_resize_mode: int = 0, hr_resize_context: str = 'None', hr_upscaler: str | None = None, hr_force: bool = False, hr_second_pass_steps: int = 20,
hr_scale: float = 1.0, hr_resize_x: int = 0, hr_resize_y: int = 0, refiner_steps: int = 5, refiner_start: float = 0.0, refiner_prompt: str = '', refiner_negative: str = '',
video_skip_frames: int = 0, video_type: str = 'None', video_duration: float = 2.0, video_loop: bool = False, video_pad: int = 0, video_interpolate: int = 0,
override_script_name: str = None, override_script_args = None, extra: dict = None,
override_script_name: str | None = None, override_script_args = None, extra: dict | None = None,
*input_script_args,
# API-only params (keyword-only, not wired to Gradio)
detailer_segmentation: bool = None, detailer_include_detections: bool = None, detailer_merge: bool = None, detailer_sort: bool = None, detailer_classes: str = None,
detailer_conf: float = None, detailer_iou: float = None, detailer_max: int = None,
detailer_min_size: float = None, detailer_max_size: float = None,
detailer_blur: int = None, detailer_padding: int = None,
detailer_sigma_adjust: float = None, detailer_sigma_adjust_max: float = None,
detailer_models: list = None, detailer_augment: bool = None,
img2img_color_correction: bool = None, color_correction_method: str = None, img2img_background_color: str = None,
img2img_fix_steps: bool = None, mask_apply_overlay: bool = None,
include_mask: bool = None, inpainting_mask_weight: float = None,
detailer_segmentation: bool | None = None, detailer_include_detections: bool | None = None, detailer_merge: bool | None = None, detailer_sort: bool | None = None, detailer_classes: str | None = None,
detailer_conf: float | None = None, detailer_iou: float | None = None, detailer_max: int | None = None,
detailer_min_size: float | None = None, detailer_max_size: float | None = None,
detailer_blur: int | None = None, detailer_padding: int | None = None,
detailer_sigma_adjust: float | None = None, detailer_sigma_adjust_max: float | None = None,
detailer_models: list | None = None, detailer_augment: bool | None = None,
img2img_color_correction: bool | None = None, color_correction_method: str | None = None, img2img_background_color: str | None = None,
img2img_fix_steps: bool | None = None, mask_apply_overlay: bool | None = None,
include_mask: bool | None = None, inpainting_mask_weight: float | None = None,
# output and saving
samples_save: bool = None, samples_format: str = None,
save_images_before_highres_fix: bool = None, save_images_before_refiner: bool = None,
save_images_before_detailer: bool = None, save_images_before_color_correction: bool = None,
grid_save: bool = None, grid_format: str = None, return_grid: bool = None,
save_mask: bool = None, save_mask_composite: bool = None,
return_mask: bool = None, return_mask_composite: bool = None,
keep_incomplete: bool = None, image_metadata: bool = None, jpeg_quality: int = None,
samples_save: bool | None = None, samples_format: str | None = None,
save_images_before_highres_fix: bool | None = None, save_images_before_refiner: bool | None = None,
save_images_before_detailer: bool | None = None, save_images_before_color_correction: bool | None = None,
grid_save: bool | None = None, grid_format: str | None = None, return_grid: bool | None = None,
save_mask: bool | None = None, save_mask_composite: bool | None = None,
return_mask: bool | None = None, return_mask_composite: bool | None = None,
keep_incomplete: bool | None = None, image_metadata: bool | None = None, jpeg_quality: int | None = None,
# scheduler/noise overrides
schedulers_prediction_type: str = None, schedulers_beta_schedule: str = None, schedulers_timesteps: str = None,
schedulers_sigma: str = None, schedulers_use_thresholding: bool = None, schedulers_use_loworder: bool = None,
schedulers_solver_order: int = None, uni_pc_variant: str = None, schedulers_beta_start: float = None,
schedulers_beta_end: float = None, schedulers_shift: float = None, schedulers_dynamic_shift: bool = None,
schedulers_base_shift: float = None, schedulers_max_shift: float = None, schedulers_rescale_betas: bool = None,
schedulers_timestep_spacing: str = None, schedulers_timesteps_range: int = None,
schedulers_sigma_adjust: float = None, schedulers_sigma_adjust_min: float = None, schedulers_sigma_adjust_max: float = None,
scheduler_eta: float = None, eta_noise_seed_delta: int = None, enable_batch_seeds: bool = None,
diffusers_generator_device: str = None, nan_skip: bool = None,
sequential_seed: bool = None,
schedulers_prediction_type: str | None = None, schedulers_beta_schedule: str | None = None, schedulers_timesteps: str | None = None,
schedulers_sigma: str | None = None, schedulers_use_thresholding: bool | None = None, schedulers_use_loworder: bool | None = None,
schedulers_solver_order: int | None = None, uni_pc_variant: str | None = None, schedulers_beta_start: float | None = None,
schedulers_beta_end: float | None = None, schedulers_shift: float | None = None, schedulers_dynamic_shift: bool | None = None,
schedulers_base_shift: float | None = None, schedulers_max_shift: float | None = None, schedulers_rescale_betas: bool | None = None,
schedulers_timestep_spacing: str | None = None, schedulers_timesteps_range: int | None = None,
schedulers_sigma_adjust: float | None = None, schedulers_sigma_adjust_min: float | None = None, schedulers_sigma_adjust_max: float | None = None,
scheduler_eta: float | None = None, eta_noise_seed_delta: int | None = None, enable_batch_seeds: bool | None = None,
diffusers_generator_device: str | None = None, nan_skip: bool | None = None,
sequential_seed: bool | None = None,
# prompt/attention overrides
prompt_attention: str = None, prompt_mean_norm: bool = None, diffusers_zeros_prompt_pad: bool = None,
te_pooled_embeds: bool = None, lora_apply_te: bool = None, te_complex_human_instruction: str = None, te_use_mask: bool = None,
prompt_attention: str | None = None, prompt_mean_norm: bool | None = None, diffusers_zeros_prompt_pad: bool | None = None,
te_pooled_embeds: bool | None = None, lora_apply_te: bool | None = None, te_complex_human_instruction: str | None = None, te_use_mask: bool | None = None,
# generation modifier overrides (hijack)
freeu_enabled: bool = None, freeu_b1: float = None, freeu_b2: float = None, freeu_s1: float = None, freeu_s2: float = None,
hypertile_unet_enabled: bool = None, hypertile_hires_only: bool = None, hypertile_unet_tile: int = None, hypertile_unet_min_tile: int = None,
hypertile_unet_swap_size: int = None, hypertile_unet_depth: int = None,
hypertile_vae_enabled: bool = None, hypertile_vae_tile: int = None, hypertile_vae_swap_size: int = None,
teacache_enabled: bool = None, teacache_thresh: float = None,
token_merging_method: str = None, tome_ratio: float = None, todo_ratio: float = None,
freeu_enabled: bool | None = None, freeu_b1: float | None = None, freeu_b2: float | None = None, freeu_s1: float | None = None, freeu_s2: float | None = None,
hypertile_unet_enabled: bool | None = None, hypertile_hires_only: bool | None = None, hypertile_unet_tile: int | None = None, hypertile_unet_min_tile: int | None = None,
hypertile_unet_swap_size: int | None = None, hypertile_unet_depth: int | None = None,
hypertile_vae_enabled: bool | None = None, hypertile_vae_tile: int | None = None, hypertile_vae_swap_size: int | None = None,
teacache_enabled: bool | None = None, teacache_thresh: float | None = None,
token_merging_method: str | None = None, tome_ratio: float | None = None, todo_ratio: float | None = None,
# lora behavior
lora_fuse_native: bool = None, lora_fuse_diffusers: bool = None,
lora_force_reload: bool = None, extra_networks_default_multiplier: float = None,
lora_apply_tags: int = None,
lora_fuse_native: bool | None = None, lora_fuse_diffusers: bool | None = None,
lora_force_reload: bool | None = None, extra_networks_default_multiplier: float | None = None,
lora_apply_tags: int | None = None,
):
if override_script_args is None:
override_script_args = []
modules/control/unit.py
+20 -11
View File
@@ -1,3 +1,4 @@
from typing import TYPE_CHECKING
from PIL import Image
import gradio as gr
from modules.logger import log
@@ -33,10 +34,10 @@ class Unit: # mashup of gradio controls and mapping to actual implementation cla
def __init__(self,
# values
index: int = None,
enabled: bool = None,
strength: float = None,
unit_type: str = None,
index: int | None = None,
enabled: bool | None = None,
strength: float | None = None,
unit_type: str | None = None,
start: float = 0,
end: float = 1,
# ui bindings
@@ -55,7 +56,7 @@ class Unit: # mashup of gradio controls and mapping to actual implementation cla
control_mode = None,
control_tile = None,
result_txt = None,
extra_controls: list = None,
extra_controls: list | None = None,
):
if extra_controls is None:
extra_controls = []
@@ -71,15 +72,15 @@ class Unit: # mashup of gradio controls and mapping to actual implementation cla
self.end = end or 1
self.start = min(self.start, self.end)
self.end = max(self.start, self.end)
self.mode = None
self.mode: int | None = None
# processor always exists, adapter and controlnet are optional
self.model_name = None
self.process_name = None
self.model_name: str | None = None
self.process_name: str | None = None
self.process: processors.Processor = processors.Processor()
self.adapter: t2iadapter.Adapter = None
self.controlnet: controlnet.ControlNet | xs.ControlNetXS = None
self.adapter: t2iadapter.Adapter | None = None
self.controlnet: controlnet.ControlNet | xs.ControlNetXS | lite.ControlLLLite | None = None
# map to input image
self.override: Image = None
self.override: Image.Image | None = None
# global settings but passed per-unit
self.factor = 1.0
self.guess = False
@@ -177,6 +178,8 @@ class Unit: # mashup of gradio controls and mapping to actual implementation cla
# bind ui controls to properties if present
if self.type == 't2i adapter':
if TYPE_CHECKING:
assert isinstance(self.adapter, t2iadapter.Adapter)
if model_id is not None:
if isinstance(model_id, str):
self.adapter.load(model_id)
@@ -186,6 +189,8 @@ class Unit: # mashup of gradio controls and mapping to actual implementation cla
if extra_controls is not None and len(extra_controls) > 0:
extra_controls[0].change(fn=adapter_extra, inputs=extra_controls)
elif self.type == 'controlnet':
if TYPE_CHECKING:
assert isinstance(self.controlnet, controlnet.ControlNet)
if model_id is not None:
if isinstance(model_id, str):
self.controlnet.load(model_id)
@@ -196,6 +201,8 @@ class Unit: # mashup of gradio controls and mapping to actual implementation cla
if extra_controls is not None and len(extra_controls) > 0:
extra_controls[0].change(fn=controlnet_extra, inputs=extra_controls)
elif self.type == 'xs':
if TYPE_CHECKING:
assert isinstance(self.controlnet, xs.ControlNetXS)
if model_id is not None:
if isinstance(model_id, str):
self.controlnet.load(model_id)
@@ -205,6 +212,8 @@ class Unit: # mashup of gradio controls and mapping to actual implementation cla
if extra_controls is not None and len(extra_controls) > 0:
extra_controls[0].change(fn=controlnetxs_extra, inputs=extra_controls)
elif self.type == 'lite':
if TYPE_CHECKING:
assert isinstance(self.controlnet, lite.ControlLLLite)
if model_id is not None:
if isinstance(model_id, str):
self.controlnet.load(model_id)
modules/devices.py
+1 -1
View File
@@ -197,7 +197,7 @@ def get_optimal_device():
return torch.device(get_optimal_device_name())
def torch_gc(force:bool=False, fast:bool=False, reason:str=None):
def torch_gc(force: bool = False, fast: bool = False, reason: str | None = None):
def get_stats():
mem_dict = memstats.memory_stats()
gpu_dict = mem_dict.get('gpu', {})
modules/face/instantid_model.py
+5 -5
View File
@@ -545,9 +545,9 @@ class StableDiffusionXLInstantIDPipeline(StableDiffusionXLControlNetPipeline):
@replace_example_docstring(EXAMPLE_DOC_STRING)
def __call__(
self,
prompt: str | list[str] = None,
prompt: str | list[str] | None = None,
prompt_2: str | list[str] | None = None,
image: PipelineImageInput = None,
image: PipelineImageInput | None = None,
height: int | None = None,
width: int | None = None,
num_inference_steps: int = 50,
@@ -570,15 +570,15 @@ class StableDiffusionXLInstantIDPipeline(StableDiffusionXLControlNetPipeline):
guess_mode: bool = False,
control_guidance_start: float | list[float] = 0.0,
control_guidance_end: float | list[float] = 1.0,
original_size: tuple[int, int] = None,
original_size: tuple[int, int] | None = None,
crops_coords_top_left: tuple[int, int] = (0, 0),
target_size: tuple[int, int] = None,
target_size: tuple[int, int] | None = None,
negative_original_size: tuple[int, int] | None = None,
negative_crops_coords_top_left: tuple[int, int] = (0, 0),
negative_target_size: tuple[int, int] | None = None,
clip_skip: int | None = None,
callback_on_step_end: Callable[[int, int, dict], None] | None = None,
callback_on_step_end_tensor_inputs: list[str] = None,
callback_on_step_end_tensor_inputs: list[str] | None = None,
**kwargs,
):
r"""
modules/face/photomaker_pipeline.py
+4 -4
View File
@@ -445,13 +445,13 @@ class PhotoMakerStableDiffusionXLPipeline(StableDiffusionXLPipeline):
@torch.no_grad()
def __call__(
self,
prompt: str | list[str] = None,
prompt: str | list[str] | None = None,
prompt_2: str | list[str] | None = None,
height: int | None = None,
width: int | None = None,
num_inference_steps: int = 50,
timesteps: list[int] = None,
sigmas: list[float] = None,
timesteps: list[int] | None = None,
sigmas: list[float] | None = None,
denoising_end: float | None = None,
guidance_scale: float = 5.0,
negative_prompt: str | list[str] | None = None,
@@ -478,7 +478,7 @@ class PhotoMakerStableDiffusionXLPipeline(StableDiffusionXLPipeline):
negative_target_size: tuple[int, int] | None = None,
clip_skip: int | None = None,
callback_on_step_end: Callable[[int, int, dict], None] | PipelineCallback | MultiPipelineCallbacks | None = None,
callback_on_step_end_tensor_inputs: list[str] = None,
callback_on_step_end_tensor_inputs: list[str] | None = None,
# Added parameters (for PhotoMaker)
input_id_images: PipelineImageInput = None,
start_merge_step: int = 10,
modules/framepack/create-video.py
+2 -2
View File
@@ -26,7 +26,7 @@ def auth():
return None
def get(endpoint: str, dct: dict = None):
def get(endpoint: str, dct: dict | None = None):
req = requests.get(f'{sd_url}{endpoint}', json=dct, timeout=300, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
@@ -34,7 +34,7 @@ def get(endpoint: str, dct: dict = None):
return req.json()
def post(endpoint: str, dct: dict = None):
def post(endpoint: str, dct: dict | None = None):
req = requests.post(f'{sd_url}{endpoint}', json = dct, timeout=None, verify=False, auth=auth())
if req.status_code != 200:
return { 'error': req.status_code, 'reason': req.reason, 'url': req.url }
modules/framepack/framepack_hijack.py
+1 -1
View File
@@ -30,7 +30,7 @@ def set_progress_bar_config():
uni_pc_fm.sample_unipc = sample_unipc
def set_prompt_template(prompt, system_prompt:str=None, optimized_prompt:bool=True, unmodified_prompt:bool=False):
def set_prompt_template(prompt, system_prompt: str | None = None, optimized_prompt: bool = True, unmodified_prompt: bool = False):
from modules import shared
from modules.framepack.pipeline import hunyuan
mode = 'unknown'
modules/framepack/framepack_load.py
+2 -2
View File
@@ -35,7 +35,7 @@ def split_url(url):
return { 'repo': f'{url[0]}/{url[1]}', 'subfolder': url[2] }
def set_model(receipe: str=None):
def set_model(receipe: str | None = None):
if receipe is None or receipe == '':
return
lines = [line.strip() for line in receipe.split('\n') if line.strip() != '' and ':' in line]
@@ -62,7 +62,7 @@ def reset_model():
return ''
def load_model(variant:str=None, pipeline:str=None, text_encoder:str=None, text_encoder_2:str=None, feature_extractor:str=None, image_encoder:str=None, transformer:str=None):
def load_model(variant: str | None = None, pipeline: str | None = None, text_encoder: str | None = None, text_encoder_2: str | None = None, feature_extractor: str | None = None, image_encoder: str | None = None, transformer: str | None = None):
shared.state.begin('Load FramePack')
if variant is not None:
if variant not in models.keys():
modules/framepack/framepack_worker.py
+2 -2
View File
@@ -44,7 +44,7 @@ def worker(
mp4_fps, mp4_codec, mp4_sf, mp4_video, mp4_frames, mp4_opt, mp4_ext, mp4_interpolate,
vae_type,
variant,
metadata:dict=None,
metadata: dict | None = None,
):
if metadata is None:
metadata = {}
@@ -85,7 +85,7 @@ def worker(
if not is_f1:
prompts = list(reversed(prompts))
def text_encode(prompt, i:int=None):
def text_encode(prompt, i: int | None = None):
jobid = shared.state.begin('TE Encode')
pbar.update(task, description=f'text encode section={i}')
t0 = time.time()
modules/image/grid.py
+12 -4
View File
@@ -1,5 +1,6 @@
import math
from collections import namedtuple
from typing import TYPE_CHECKING
import numpy as np
from PIL import Image, ImageFont, ImageDraw
from modules import shared, script_callbacks
@@ -26,7 +27,7 @@ def check_grid_size(imgs):
return ok
def get_grid_size(imgs, batch_size=1, rows=None, cols=None):
def get_grid_size(imgs, batch_size=1, rows: int | None = None, cols: int | None = None):
if rows and rows > len(imgs):
rows = len(imgs)
if cols and cols > len(imgs):
@@ -34,12 +35,16 @@ def get_grid_size(imgs, batch_size=1, rows=None, cols=None):
if rows is None and cols is None:
if shared.opts.n_rows > 0:
rows = shared.opts.n_rows
if TYPE_CHECKING:
assert isinstance(rows, int)
cols = math.ceil(len(imgs) / rows)
elif shared.opts.n_rows == 0:
rows = batch_size
cols = math.ceil(len(imgs) / rows)
elif shared.opts.n_cols > 0:
cols = shared.opts.n_cols
if TYPE_CHECKING:
assert isinstance(cols, int)
rows = math.ceil(len(imgs) / cols)
elif shared.opts.n_cols == 0:
cols = batch_size
@@ -49,16 +54,19 @@ def get_grid_size(imgs, batch_size=1, rows=None, cols=None):
while len(imgs) % rows != 0:
rows -= 1
cols = math.ceil(len(imgs) / rows)
elif cols is None:
elif rows is not None and cols is None:
cols = math.ceil(len(imgs) / rows)
elif rows is None:
elif rows is None and cols is not None:
rows = math.ceil(len(imgs) / cols)
else:
if TYPE_CHECKING:
assert isinstance(rows, int)
assert isinstance(cols, int)
pass
return rows, cols
def image_grid(imgs, batch_size:int=1, rows:int=None, cols:int=None):
def image_grid(imgs, batch_size=1, rows=1, cols=1):
rows, cols = get_grid_size(imgs, batch_size, rows=rows, cols=cols)
params = script_callbacks.ImageGridLoopParams(imgs, cols, rows)
script_callbacks.image_grid_callback(params)
modules/image/resize.py
+2 -2
View File
@@ -8,7 +8,7 @@ from modules.logger import log
from modules.image import sharpfin
def resize_image(resize_mode: int, im: Image.Image | torch.Tensor, width: int, height: int, upscaler_name: str=None, output_type: str='image', context: str=None):
def resize_image(resize_mode: int, im: Image.Image | torch.Tensor, width: int, height: int, upscaler_name: str | None = None, output_type: str = 'image', context: str | None = None):
upscaler_name = upscaler_name or shared.opts.upscaler_for_img2img
def verify_image(image):
@@ -95,7 +95,7 @@ def resize_image(resize_mode: int, im: Image.Image | torch.Tensor, width: int, h
res.paste(im, box=((width - im.width)//2, (height - im.height)//2))
return res
def context_aware(im: Image.Image, width, height, context):
def context_aware(im: Image.Image, width: int, height: int, context: str):
from installer import install
install('seam-carving')
width, height = int(width), int(height)
modules/intel/openvino/__init__.py
+3 -3
View File
@@ -62,7 +62,7 @@ def warn_once(msg):
warned = True
class OpenVINOGraphModule(torch.nn.Module):
def __init__(self, gm, partition_id, use_python_fusion_cache, model_hash_str: str = None, file_name="", int_inputs=None):
def __init__(self, gm, partition_id, use_python_fusion_cache, model_hash_str: str | None = None, file_name="", int_inputs: list | None = None):
if int_inputs is None:
int_inputs = []
super().__init__()
@@ -190,7 +190,7 @@ def execute_cached(compiled_model, *args):
result = [torch.from_numpy(res[out]) for out in compiled_model.outputs]
return result
def openvino_compile(gm: GraphModule, *example_inputs, model_hash_str: str = None, file_name=""):
def openvino_compile(gm: GraphModule, *example_inputs, model_hash_str: str | None = None, file_name=""):
core = Core()
device = get_device()
@@ -352,7 +352,7 @@ def openvino_execute_partitioned(gm: GraphModule, *args, executor_parameters=Non
return shared.compiled_model_state.partitioned_modules[signature][0](*ov_inputs)
def partition_graph(gm: GraphModule, use_python_fusion_cache: bool, model_hash_str: str = None, file_name="", int_inputs=None):
def partition_graph(gm: GraphModule, use_python_fusion_cache: bool, model_hash_str: str | None = None, file_name="", int_inputs=None):
if int_inputs is None:
int_inputs = []
for node in gm.graph.nodes:
modules/logger.py
+2 -2
View File
@@ -40,7 +40,7 @@ def get_log():
return log
def install_traceback(suppress: list = None):
def install_traceback(suppress: list | None = None):
if suppress is None:
suppress = []
width = os.environ.get("SD_TRACEWIDTH", console.width if console else None)
@@ -143,7 +143,7 @@ def setup_logging(debug=None, trace=None, filename=None):
logging.Logger.trace = partialmethod(logging.Logger.log, logging.TRACE)
logging.trace = partial(logging.log, logging.TRACE)
def exception_hook(e: Exception, suppress=None):
def exception_hook(e: Exception, suppress: list | None = None):
from rich.traceback import Traceback
if suppress is None:
suppress = []
modules/lora/extra_networks_lora.py
+1 -1
View File
@@ -173,7 +173,7 @@ class ExtraNetworkLora(extra_networks.ExtraNetwork):
def signature(self, names: list[str], te_multipliers: list, unet_multipliers: list):
return [f'{name}:{te}:{unet}' for name, te, unet in zip(names, te_multipliers, unet_multipliers, strict=False)]
def changed(self, requested: list[str], include: list[str] = None, exclude: list[str] = None) -> bool:
def changed(self, requested: list[str], include: list[str] | None = None, exclude: list[str] | None = None) -> bool:
if shared.opts.lora_force_reload:
debug_log(f'Network check: type=LoRA requested={requested} status=forced')
return True
modules/masking.py
+1 -1
View File
@@ -378,7 +378,7 @@ def outpaint(input_image: Image.Image, outpaint_type: str = 'Edge'):
return image, mask
def run_mask(input_image: Image.Image, input_mask: Image.Image = None, return_type: str = None, mask_blur: int = None, mask_padding: int = None, invert=None):
def run_mask(input_image: Image.Image, input_mask: Image.Image | None = None, return_type: str | None = None, mask_blur: int | None = None, mask_padding: int | None = None, invert=None):
if isinstance(input_image, list) and len(input_image) > 0:
input_image = input_image[0]
elif isinstance(input_image, dict):
modules/merging/convert_sdxl.py
+1 -1
View File
@@ -260,7 +260,7 @@ def calculate_model_hash(state_dict):
return func.hexdigest()
def convert(model_path:str, checkpoint_path:str, metadata:dict=None):
def convert(model_path: str, checkpoint_path: str, metadata: dict | None = None):
if metadata is None:
metadata = {}
unet_path = os.path.join(model_path, "unet", "diffusion_pytorch_model.safetensors")
modules/merging/modules_sdxl.py
+7 -7
View File
@@ -65,7 +65,7 @@ def msg(text, err:bool=False):
return status
def load_base(override:str=None):
def load_base(override: str | None = None):
global pipeline # pylint: disable=global-statement
fn = override or recipe.base
yield msg(f'base={fn}')
@@ -79,7 +79,7 @@ def load_base(override:str=None):
pipeline.vae.register_to_config(force_upcast = False)
def load_unet(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
def load_unet(pipe: diffusers.StableDiffusionXLPipeline, override: str | None = None):
if (recipe.unet is None or len(recipe.unet) == 0) and override is None:
return
fn = override or recipe.unet
@@ -99,7 +99,7 @@ def load_unet(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
yield msg(f'unet: {e}')
def load_scheduler(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
def load_scheduler(pipe: diffusers.StableDiffusionXLPipeline, override: str | None = None):
if recipe.scheduler is None and override is None:
return
config = pipe.scheduler.config.__dict__
@@ -114,7 +114,7 @@ def load_scheduler(pipe: diffusers.StableDiffusionXLPipeline, override:str=None)
def load_vae(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
def load_vae(pipe: diffusers.StableDiffusionXLPipeline, override: str | None = None):
if (recipe.vae is None or len(recipe.vae) == 0)and override is None:
return
fn = override or recipe.vae
@@ -135,7 +135,7 @@ def load_vae(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
yield msg(f'vae: {e}')
def load_te1(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
def load_te1(pipe: diffusers.StableDiffusionXLPipeline, override: str | None = None):
if (recipe.te1 is None or len(recipe.te1) == 0) and override is None:
return
config = pipe.text_encoder.config.__dict__
@@ -156,7 +156,7 @@ def load_te1(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
yield msg(f'te1: {e}')
def load_te2(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
def load_te2(pipe: diffusers.StableDiffusionXLPipeline, override: str | None = None):
if (recipe.te2 is None or len(recipe.te2) == 0) and override is None:
return
config = pipe.text_encoder_2.config.__dict__
@@ -177,7 +177,7 @@ def load_te2(pipe: diffusers.StableDiffusionXLPipeline, override:str=None):
yield msg(f'te2: {e}')
def load_lora(pipe: diffusers.StableDiffusionXLPipeline, override: dict=None, fuse: float=None):
def load_lora(pipe: diffusers.StableDiffusionXLPipeline, override: dict | None = None, fuse: float | None = None):
if recipe.lora is None and override is None:
return
names = []
modules/model_quant.py
+5 -5
View File
@@ -49,7 +49,7 @@ def dont_quant():
return False
def create_trt_config(kwargs = None, allow: bool = True, module: str = 'Model', modules_to_not_convert: list = None):
def create_trt_config(kwargs = None, allow: bool = True, module: str = 'Model', modules_to_not_convert: list | None = None):
from modules import shared
if allow and (module == 'any' or module in shared.opts.trt_quantization):
load_trt()
@@ -97,7 +97,7 @@ def get_sdnq_devices(mode="pre"):
return quantization_device, return_device
def create_sdnq_config(kwargs = None, allow: bool = True, module: str = 'Model', weights_dtype: str = None, quantized_matmul_dtype: str = None, modules_to_not_convert: list = None, modules_dtype_dict: dict = None):
def create_sdnq_config(kwargs = None, allow: bool = True, module: str = 'Model', weights_dtype: str | None = None, quantized_matmul_dtype: str | None = None, modules_to_not_convert: list | None = None, modules_dtype_dict: dict | None = None):
from modules import shared
if allow and (shared.opts.sdnq_quantize_mode in {'pre', 'auto'}) and (module == 'any' or module in shared.opts.sdnq_quantize_weights):
from modules.sdnq import SDNQConfig
@@ -210,7 +210,7 @@ def check_nunchaku(module: str = ''):
return False
def create_config(kwargs = None, allow: bool = True, module: str = 'Model', modules_to_not_convert: list = None, modules_dtype_dict: dict = None):
def create_config(kwargs = None, allow: bool = True, module: str = 'Model', modules_to_not_convert: list | None = None, modules_dtype_dict: dict | None = None):
if kwargs is None:
kwargs = {}
if module == 'Model' and dont_quant():
@@ -345,7 +345,7 @@ def apply_layerwise(sd_model, quiet:bool=False):
log.error(f'Quantization: type=layerwise {e}')
def sdnq_quantize_model(model, op=None, sd_model=None, do_gc: bool = True, weights_dtype: str = None, quantized_matmul_dtype: str = None, modules_to_not_convert: list = None, modules_dtype_dict: dict = None):
def sdnq_quantize_model(model, op=None, sd_model=None, do_gc: bool = True, weights_dtype: str | None = None, quantized_matmul_dtype: str | None = None, modules_to_not_convert: list | None = None, modules_dtype_dict: dict | None = None):
global quant_last_model_name, quant_last_model_device # pylint: disable=global-statement
from modules import devices, shared, timer
from modules.sdnq import sdnq_post_load_quant
@@ -479,7 +479,7 @@ def sdnq_quantize_weights(sd_model):
return sd_model
def get_dit_args(load_config:dict=None, module:str=None, device_map:bool=False, allow_quant:bool=True, modules_to_not_convert: list = None, modules_dtype_dict: dict = None):
def get_dit_args(load_config: dict | None = None, module: str | None = None, device_map: bool = False, allow_quant: bool = True, modules_to_not_convert: list | None = None, modules_dtype_dict: dict | None = None):
from modules import shared, devices
config = {} if load_config is None else load_config.copy()
if 'torch_dtype' not in config:
modules/modelloader.py
+4 -4
View File
@@ -56,7 +56,7 @@ def hf_login(token=None):
return True
def download_diffusers_model(hub_id: str, cache_dir: str = None, download_config: dict[str, str] = None, token = None, variant = None, revision = None, mirror = None, custom_pipeline = None):
def download_diffusers_model(hub_id: str, cache_dir: str | None = None, download_config: dict[str, str | bool] | None = None, token = None, variant = None, revision = None, mirror = None, custom_pipeline = None):
if hub_id is None or len(hub_id) == 0:
return None
from diffusers import DiffusionPipeline
@@ -219,7 +219,7 @@ def get_reference_opts(name: str, quiet=False):
return model_opts
def load_reference(name: str, variant: str = None, revision: str = None, mirror: str = None, custom_pipeline: str = None):
def load_reference(name: str, variant: str | None = None, revision: str | None = None, mirror: str | None = None, custom_pipeline: str | None = None):
if '+' in name:
name = name.split('+')[0]
found = [r for r in diffuser_repos if name == r['name'] or name == r['friendly'] or name == r['path']]
@@ -337,7 +337,7 @@ def load_file_from_url(url: str, *, model_dir: str, progress: bool = True, file_
return None
def load_models(model_path: str, model_url: str = None, command_path: str = None, ext_filter=None, download_name=None, ext_blacklist=None) -> list:
def load_models(model_path: str, model_url: str | None = None, command_path: str | None = None, ext_filter=None, download_name=None, ext_blacklist=None) -> list:
"""
A one-and done loader to try finding the desired models in specified directories.
@param download_name: Specify to download from model_url immediately.
@@ -404,7 +404,7 @@ def cleanup_models():
move_files(src_path, dest_path)
def move_files(src_path: str, dest_path: str, ext_filter: str = None):
def move_files(src_path: str, dest_path: str, ext_filter: str | None = None):
try:
if not os.path.exists(dest_path):
os.makedirs(dest_path)
modules/onnx_impl/pipelines/onnx_stable_diffusion_pipeline.py
+1 -1
View File
@@ -30,7 +30,7 @@ class OnnxStableDiffusionPipeline(diffusers.OnnxStableDiffusionPipeline, Callabl
def __call__(
self,
prompt: str | list[str] = None,
prompt: str | list[str] | None = None,
height: int | None = 512,
width: int | None = 512,
num_inference_steps: int | None = 50,
modules/options_handler.py
+1 -1
View File
@@ -28,7 +28,7 @@ class Options:
debug = os.environ.get('SD_CONFIG_DEBUG', None) is not None
secrets_debug = os.environ.get("SD_SECRETS_DEBUG", None) is not None
def __init__(self, options_templates: dict[str, OptionInfo | LegacyOption] = None, restricted: set[str] | None = None, *, filename = '', secrets = ''):
def __init__(self, options_templates: dict[str, OptionInfo | LegacyOption] | None = None, restricted: set[str] | None = None, *, filename = '', secrets = ''):
if options_templates is None:
options_templates = {}
if restricted is None:
modules/postprocess/pixelart.py
+2 -2
View File
@@ -224,8 +224,8 @@ class JPEGEncoder(ImageProcessingMixin, ConfigMixin):
block_size: int = 16,
cbcr_downscale: int = 2,
norm: str = "ortho",
latents_std: list[float] = None,
latents_mean: list[float] = None,
latents_std: list[float] | None = None,
latents_mean: list[float] | None = None,
):
self.block_size = block_size
self.cbcr_downscale = cbcr_downscale
modules/postprocess/yolo.py
+3 -3
View File
@@ -96,11 +96,11 @@ class YoloRestorer(Detailer):
imgsz: int = 640,
half: bool = True,
device = devices.device,
augment: bool = None,
agnostic: bool = False,
retina: bool = False,
mask: bool = True,
offload: bool = None,
augment: bool | None = None,
offload: bool | None = None,
p = None,
) -> list[YoloResult]:
if augment is None:
@@ -201,7 +201,7 @@ class YoloRestorer(Detailer):
break
return result
def load(self, model_name: str = None):
def load(self, model_name: str | None = None):
with load_lock:
from modules import modelloader
model = None
modules/processing_callbacks.py
+1 -1
View File
@@ -51,7 +51,7 @@ def diffusers_callback_legacy(step: int, timestep: int, latents: torch.FloatTens
time.sleep(0.1)
def diffusers_callback(pipe, step: int = 0, timestep: int = 0, kwargs: dict = None):
def diffusers_callback(pipe, step: int = 0, timestep: int = 0, kwargs: dict | None = None):
if kwargs is None:
kwargs = {}
t0 = time.time()
modules/processing_class.py
+111 -111
View File
@@ -18,7 +18,7 @@ debug = log.trace if os.environ.get('SD_PROCESS_DEBUG', None) is not None else l
@dataclass(repr=False)
class StableDiffusionProcessing:
def __init__(self,
sd_model_checkpoint: str = None, # # used only to set sd_model
sd_model_checkpoint: str | None = None, # # used only to set sd_model
sd_model=None, # pylint: disable=unused-argument # local instance of sd_model
# base params
prompt: str = "",
@@ -35,10 +35,10 @@ class StableDiffusionProcessing:
width: int = 1024,
height: int = 1024,
# samplers
sampler_index: int = None, # pylint: disable=unused-argument # used only to set sampler_name
sampler_name: str = None,
hr_sampler_name: str = None,
eta: float = None,
sampler_index: int | None = None, # pylint: disable=unused-argument # used only to set sampler_name
sampler_name: str | None = None,
hr_sampler_name: str | None = None,
eta: float | None = None,
# modular guidance
guidance_name: str = 'Default',
guidance_scale: float = 6.0,
@@ -52,7 +52,7 @@ class StableDiffusionProcessing:
pag_scale: float = 0.0,
pag_adaptive: float = 0.5,
# styles
styles: list[str] = None,
styles: list[str] | None = None,
# vae
tiling: bool = False,
vae_type: str = 'Full',
@@ -66,53 +66,53 @@ class StableDiffusionProcessing:
detailer_steps: int = 10,
detailer_strength: float = 0.3,
detailer_resolution: int = 1024,
detailer_segmentation: bool = None,
detailer_include_detections: bool = None,
detailer_merge: bool = None,
detailer_sort: bool = None,
detailer_classes: str = None,
detailer_conf: float = None,
detailer_iou: float = None,
detailer_max: int = None,
detailer_min_size: float = None,
detailer_max_size: float = None,
detailer_blur: int = None,
detailer_padding: int = None,
detailer_sigma_adjust: float = None,
detailer_sigma_adjust_max: float = None,
detailer_models: list = None,
detailer_augment: bool = None,
detailer_segmentation: bool | None = None,
detailer_include_detections: bool | None = None,
detailer_merge: bool | None = None,
detailer_sort: bool | None = None,
detailer_classes: str | None = None,
detailer_conf: float | None = None,
detailer_iou: float | None = None,
detailer_max: int | None = None,
detailer_min_size: float | None = None,
detailer_max_size: float | None = None,
detailer_blur: int | None = None,
detailer_padding: int | None = None,
detailer_sigma_adjust: float | None = None,
detailer_sigma_adjust_max: float | None = None,
detailer_models: list | None = None,
detailer_augment: bool | None = None,
# img2img and mask
img2img_color_correction: bool = None,
color_correction_method: str = None,
img2img_background_color: str = None,
img2img_fix_steps: bool = None,
mask_apply_overlay: bool = None,
include_mask: bool = None,
inpainting_mask_weight: float = None,
img2img_color_correction: bool | None = None,
color_correction_method: str | None = None,
img2img_background_color: str | None = None,
img2img_fix_steps: bool | None = None,
mask_apply_overlay: bool | None = None,
include_mask: bool | None = None,
inpainting_mask_weight: float | None = None,
# output and saving
samples_save: bool = None,
samples_format: str = None,
save_images_before_highres_fix: bool = None,
save_images_before_refiner: bool = None,
save_images_before_detailer: bool = None,
save_images_before_color_correction: bool = None,
grid_save: bool = None,
grid_format: str = None,
return_grid: bool = None,
save_mask: bool = None,
save_mask_composite: bool = None,
return_mask: bool = None,
return_mask_composite: bool = None,
keep_incomplete: bool = None,
image_metadata: bool = None,
jpeg_quality: int = None,
samples_save: bool | None = None,
samples_format: str | None = None,
save_images_before_highres_fix: bool | None = None,
save_images_before_refiner: bool | None = None,
save_images_before_detailer: bool | None = None,
save_images_before_color_correction: bool | None = None,
grid_save: bool | None = None,
grid_format: str | None = None,
return_grid: bool | None = None,
save_mask: bool | None = None,
save_mask_composite: bool | None = None,
return_mask: bool | None = None,
return_mask_composite: bool | None = None,
keep_incomplete: bool | None = None,
image_metadata: bool | None = None,
jpeg_quality: int | None = None,
# lora behavior
lora_fuse_native: bool = None,
lora_fuse_diffusers: bool = None,
lora_force_reload: bool = None,
extra_networks_default_multiplier: float = None,
lora_apply_tags: int = None,
lora_fuse_native: bool | None = None,
lora_fuse_diffusers: bool | None = None,
lora_force_reload: bool | None = None,
extra_networks_default_multiplier: float | None = None,
lora_apply_tags: int | None = None,
# hdr corrections
hdr_mode: int = 0,
hdr_brightness: float = 0,
@@ -148,11 +148,11 @@ class StableDiffusionProcessing:
grading_lut_file: str = "",
grading_lut_strength: float = 1.0,
# img2img
init_images: list = None,
init_control: list = None,
denoising_strength: float = 0.3,
image_cfg_scale: float = None,
initial_noise_multiplier: float = None, # pylint: disable=unused-argument # a1111 compatibility
init_images: list | None = None,
init_control: list | None = None,
image_cfg_scale: float | None = None,
initial_noise_multiplier: float | None = None, # pylint: disable=unused-argument # a1111 compatibility
# resize
scale_by: float = 1,
selected_scale_tab: int = 0, # pylint: disable=unused-argument # a1111 compatibility
@@ -199,12 +199,12 @@ class StableDiffusionProcessing:
hr_force: bool = False,
hr_resize_mode: int = 0,
hr_resize_context: str = 'None',
hr_upscaler: str = None,
hr_second_pass_steps: int = 0,
hr_resize_x: int = 0,
hr_resize_y: int = 0,
hr_denoising_strength: float = 0.0,
refiner_steps: int = 5,
hr_upscaler: str | None = None,
refiner_start: float = 0,
refiner_prompt: str = '',
refiner_negative: str = '',
@@ -219,65 +219,65 @@ class StableDiffusionProcessing:
# xyz flag
xyz: bool = False,
# scripts
script_args: list = [],
script_args: list | None = None,
# scheduler/noise overrides
schedulers_prediction_type: str = None,
schedulers_beta_schedule: str = None,
schedulers_timesteps: str = None,
schedulers_sigma: str = None,
schedulers_use_thresholding: bool = None,
schedulers_use_loworder: bool = None,
schedulers_solver_order: int = None,
uni_pc_variant: str = None,
schedulers_beta_start: float = None,
schedulers_beta_end: float = None,
schedulers_shift: float = None,
schedulers_dynamic_shift: bool = None,
schedulers_base_shift: float = None,
schedulers_max_shift: float = None,
schedulers_rescale_betas: bool = None,
schedulers_timestep_spacing: str = None,
schedulers_timesteps_range: int = None,
schedulers_sigma_adjust: float = None,
schedulers_sigma_adjust_min: float = None,
schedulers_sigma_adjust_max: float = None,
scheduler_eta: float = None,
eta_noise_seed_delta: int = None,
enable_batch_seeds: bool = None,
diffusers_generator_device: str = None,
nan_skip: bool = None,
sequential_seed: bool = None,
schedulers_prediction_type: str | None = None,
schedulers_beta_schedule: str | None = None,
schedulers_timesteps: str | None = None,
schedulers_sigma: str | None = None,
schedulers_use_thresholding: bool | None = None,
schedulers_use_loworder: bool | None = None,
schedulers_solver_order: int | None = None,
uni_pc_variant: str | None = None,
schedulers_beta_start: float | None = None,
schedulers_beta_end: float | None = None,
schedulers_shift: float | None = None,
schedulers_dynamic_shift: bool | None = None,
schedulers_base_shift: float | None = None,
schedulers_max_shift: float | None = None,
schedulers_rescale_betas: bool | None = None,
schedulers_timestep_spacing: str | None = None,
schedulers_timesteps_range: int | None = None,
schedulers_sigma_adjust: float | None = None,
schedulers_sigma_adjust_min: float | None = None,
schedulers_sigma_adjust_max: float | None = None,
scheduler_eta: float | None = None,
eta_noise_seed_delta: int | None = None,
enable_batch_seeds: bool | None = None,
diffusers_generator_device: str | None = None,
nan_skip: bool | None = None,
sequential_seed: bool | None = None,
# prompt/attention overrides
prompt_attention: str = None,
prompt_mean_norm: bool = None,
diffusers_zeros_prompt_pad: bool = None,
te_pooled_embeds: bool = None,
lora_apply_te: bool = None,
te_complex_human_instruction: str = None,
te_use_mask: bool = None,
prompt_attention: str | None = None,
prompt_mean_norm: bool | None = None,
diffusers_zeros_prompt_pad: bool | None = None,
te_pooled_embeds: bool | None = None,
lora_apply_te: bool | None = None,
te_complex_human_instruction: str | None = None,
te_use_mask: bool | None = None,
# generation modifier overrides (hijack)
freeu_enabled: bool = None,
freeu_b1: float = None,
freeu_b2: float = None,
freeu_s1: float = None,
freeu_s2: float = None,
hypertile_unet_enabled: bool = None,
hypertile_hires_only: bool = None,
hypertile_unet_tile: int = None,
hypertile_unet_min_tile: int = None,
hypertile_unet_swap_size: int = None,
hypertile_unet_depth: int = None,
hypertile_vae_enabled: bool = None,
hypertile_vae_tile: int = None,
hypertile_vae_swap_size: int = None,
teacache_enabled: bool = None,
teacache_thresh: float = None,
token_merging_method: str = None,
tome_ratio: float = None,
todo_ratio: float = None,
freeu_enabled: bool | None = None,
freeu_b1: float | None = None,
freeu_b2: float | None = None,
freeu_s1: float | None = None,
freeu_s2: float | None = None,
hypertile_unet_enabled: bool | None = None,
hypertile_hires_only: bool | None = None,
hypertile_unet_tile: int | None = None,
hypertile_unet_min_tile: int | None = None,
hypertile_unet_swap_size: int | None = None,
hypertile_unet_depth: int | None = None,
hypertile_vae_enabled: bool | None = None,
hypertile_vae_tile: int | None = None,
hypertile_vae_swap_size: int | None = None,
teacache_enabled: bool | None = None,
teacache_thresh: float | None = None,
token_merging_method: str | None = None,
tome_ratio: float | None = None,
todo_ratio: float | None = None,
# overrides
override_settings: dict[str, Any] = None,
override_settings_restore_afterwards: bool = True,
override_settings: dict[str, Any] | None = None,
# metadata
# extra_generation_params: Dict[Any, Any] = {},
# task_args: Dict[str, Any] = {},
@@ -834,7 +834,7 @@ class StableDiffusionProcessingControl(StableDiffusionProcessingImg2Img):
debug(f'Process init: mode={self.__class__.__name__} kwargs={kwargs}') # pylint: disable=protected-access
super().__init__(**kwargs)
def init_hr(self, scale:float=None, upscaler:str=None, force:bool=False):
def init_hr(self, scale: float | None = None, upscaler: str | None = None, force = False):
scale = scale or self.scale_by or self.scale_by_before
upscaler = upscaler or self.hr_upscaler or self.resize_name or self.resize_name_before
if upscaler is None:
@@ -852,7 +852,7 @@ class StableDiffusionProcessingControl(StableDiffusionProcessingImg2Img):
self.hr_upscale_to_x, self.hr_upscale_to_y = int(self.hr_resize_x), int(self.hr_resize_y)
def switch_class(p: StableDiffusionProcessing, new_class: type, dct: dict = None):
def switch_class(p: StableDiffusionProcessing, new_class: type, dct: dict | None = None):
kwargs = {}
signature = inspect.signature(StableDiffusionProcessing.__init__, follow_wrapped=True) # base class
possible = list(signature.parameters)
modules/processing_helpers.py
+1 -1
View File
@@ -155,7 +155,7 @@ def images_tensor_to_samples(image, approximation=None, model=None): # pylint: d
return x_latent
def get_sampler_name(sampler_index: int, img: bool = False) -> str:
def get_sampler_name(sampler_index: int | None = None, img: bool = False) -> str:
sampler_index = sampler_index or 0
if len(sd_samplers.samplers) > sampler_index:
sampler_name = sd_samplers.samplers[sampler_index].name
modules/prompt_parser_diffusers.py
+2 -2
View File
@@ -600,7 +600,7 @@ def split_prompts(pipe, prompt, SD3 = False):
return prompt, prompt2, prompt3, prompt4
def get_weighted_text_embeddings(pipe, prompt: str = "", neg_prompt: str = "", clip_skip: int = None, prompt_mean_norm=None, diffusers_zeros_prompt_pad=None, te_pooled_embeds=None):
def get_weighted_text_embeddings(pipe, prompt: str = "", neg_prompt: str = "", clip_skip: int | None = None, prompt_mean_norm=None, diffusers_zeros_prompt_pad=None, te_pooled_embeds=None):
device = devices.device
if prompt is None:
prompt = ''
@@ -753,7 +753,7 @@ def get_weighted_text_embeddings(pipe, prompt: str = "", neg_prompt: str = "", c
return prompt_embeds, pooled_prompt_embeds, None, negative_prompt_embeds, negative_pooled_prompt_embeds, None
def get_xhinker_text_embeddings(pipe, prompt: str = "", neg_prompt: str = "", clip_skip: int = None):
def get_xhinker_text_embeddings(pipe, prompt: str = "", neg_prompt: str = "", clip_skip: int | None = None):
is_sd3 = hasattr(pipe, 'text_encoder_3')
prompt, prompt_2, _prompt_3, _ = split_prompts(pipe, prompt, is_sd3)
neg_prompt, neg_prompt_2, _neg_prompt_3, _ = split_prompts(pipe, neg_prompt, is_sd3)
modules/prompt_parser_xhinker.py
+3 -17
View File
@@ -27,10 +27,7 @@ from diffusers import ChromaPipeline
from modules.prompt_parser import parse_prompt_attention # use built-in A1111 parser
def get_prompts_tokens_with_weights(
clip_tokenizer: CLIPTokenizer
, prompt: str = None
):
def get_prompts_tokens_with_weights(clip_tokenizer: CLIPTokenizer, prompt: str | None = None):
"""
Get prompt token ids and weights, this function works for both prompt and negative prompt
@@ -754,13 +751,7 @@ def get_weighted_text_embeddings_sdxl_refiner(
return prompt_embeds, negative_prompt_embeds, pooled_prompt_embeds, negative_pooled_prompt_embeds
def get_weighted_text_embeddings_sdxl_2p(
pipe: StableDiffusionXLPipeline
, prompt: str = ""
, prompt_2: str = None
, neg_prompt: str = ""
, neg_prompt_2: str = None
):
def get_weighted_text_embeddings_sdxl_2p(pipe: StableDiffusionXLPipeline, prompt: str = "", prompt_2: str | None = None, neg_prompt: str = "", neg_prompt_2: str | None = None):
"""
This function can process long prompt with weights, no length limitation
for Stable Diffusion XL, support two prompt sets.
@@ -1345,12 +1336,7 @@ def get_weighted_text_embeddings_sd3(
return sd3_prompt_embeds, sd3_neg_prompt_embeds, pooled_prompt_embeds, negative_pooled_prompt_embeds
def get_weighted_text_embeddings_flux1(
pipe: FluxPipeline
, prompt: str = ""
, prompt2: str = None
, device=None
):
def get_weighted_text_embeddings_flux1(pipe: FluxPipeline, prompt: str = "", prompt2: str | None = None, device=None):
"""
This function can process long prompt with weights for flux1 model
modules/ras/ras_forward.py
+4 -4
View File
@@ -22,10 +22,10 @@ from . import ras_manager
def ras_forward(
self,
hidden_states: torch.FloatTensor,
encoder_hidden_states: torch.FloatTensor = None,
pooled_projections: torch.FloatTensor = None,
timestep: torch.LongTensor = None,
block_controlnet_hidden_states: list = None,
encoder_hidden_states: torch.FloatTensor | None = None,
pooled_projections: torch.FloatTensor | None = None,
timestep: torch.LongTensor | None = None,
block_controlnet_hidden_states: list | None = None,
joint_attention_kwargs: dict[str, Any] | None = None,
return_dict: bool = True,
skip_layers: list[int] | None = None,
modules/sd_models.py
+3 -1
View File
@@ -1418,7 +1418,7 @@ def hf_auth_check(checkpoint_info, force:bool=False):
return False
def save_model(name: str, path: str = None, shard: str = None, overwrite: bool = False):
def save_model(name: str, path: str | None = None, shard: str = "5GB", overwrite = False):
if (name is None) or len(name.strip()) == 0:
log.error('Save model: invalid model name')
return 'Invalid model name'
@@ -1432,6 +1432,8 @@ def save_model(name: str, path: str = None, shard: str = None, overwrite: bool =
if os.path.exists(model_name) and not overwrite:
log.error(f'Save model: path="{model_name}" exists')
return f'Path exists: {model_name}'
if not shard.strip():
shard = "5GB" # Guard against empty input
try:
t0 = time.time()
save_sdnq_model(
modules/sd_offload.py
+1 -1
View File
@@ -467,7 +467,7 @@ def report_model_stats(module_name, module):
log.error(f'Module stats: name={module_name} {e}')
def apply_balanced_offload(sd_model=None, exclude:list[str]=None, force:bool=False, silent:bool=False):
def apply_balanced_offload(sd_model=None, exclude: list[str] | None = None, force: bool = False, silent: bool = False):
global offload_hook_instance # pylint: disable=global-statement
if shared.opts.diffusers_offload_mode != "balanced":
return sd_model
modules/sd_offload_aux.py
+1 -1
View File
@@ -33,7 +33,7 @@ def deregister_aux(name: str) -> None:
debug_move(f'Offload: type=aux op=deregister name={name}')
def evict_aux(exclude: str = None, reason: str = 'evict') -> None:
def evict_aux(exclude: str | None = None, reason: str = 'evict') -> None:
for name, entry in aux_models.items():
if name == exclude:
continue
modules/sd_te_remote.py
+1 -1
View File
@@ -8,7 +8,7 @@ from modules.logger import log
def get_t5_prompt_embeds(
prompt: str | list[str] = None,
prompt: str | list[str] | None = None,
num_images_per_prompt: int = 1, # pylint: disable=unused-argument
max_sequence_length: int = 512, # pylint: disable=unused-argument
device: torch.device | None = None,
modules/sd_unet.py
+4 -4
View File
@@ -35,11 +35,11 @@ def load_unet_sdxl_nunchaku(repo_id):
return unet
def load_unet(model, repo_id:str=None):
global loaded_unet # pylint: disable=global-statement
def load_unet(model, repo_id: str | None = None):
global loaded_unet # pylint: disable=global-statement
if ("StableDiffusionXLPipeline" in model.__class__.__name__) and (('stable-diffusion-xl-base' in repo_id) or ('sdxl-turbo' in repo_id)):
if model_quant.check_nunchaku('Model'):
if ("StableDiffusionXLPipeline" in model.__class__.__name__) and repo_id is not None and (("stable-diffusion-xl-base" in repo_id) or ("sdxl-turbo" in repo_id)):
if model_quant.check_nunchaku("Model"):
unet = load_unet_sdxl_nunchaku(repo_id)
if unet is not None:
model.unet = unet
modules/sdnq/file_loader.py
+4 -4
View File
@@ -13,7 +13,7 @@ def map_keys(key: str, key_mapping: dict) -> str:
return new_key
def load_safetensors(files: list[str], state_dict: dict = None, key_mapping: dict = None, device: torch.device = "cpu") -> dict:
def load_safetensors(files: list[str], state_dict: dict | None = None, key_mapping: dict | None = None, device: torch.device = "cpu") -> dict:
from safetensors.torch import safe_open
if state_dict is None:
state_dict = {}
@@ -23,7 +23,7 @@ def load_safetensors(files: list[str], state_dict: dict = None, key_mapping: dic
state_dict[map_keys(key, key_mapping)] = f.get_tensor(key)
def load_threaded(files: list[str], state_dict: dict = None, key_mapping: dict = None, device: torch.device = "cpu") -> dict:
def load_threaded(files: list[str], state_dict: dict | None = None, key_mapping: dict | None = None, device: torch.device = "cpu") -> dict:
future_items = {}
if state_dict is None:
state_dict = {}
@@ -34,7 +34,7 @@ def load_threaded(files: list[str], state_dict: dict = None, key_mapping: dict =
future.result()
def load_streamer(files: list[str], state_dict: dict = None, key_mapping: dict = None, device: torch.device = "cpu") -> dict:
def load_streamer(files: list[str], state_dict: dict | None = None, key_mapping: dict | None = None, device: torch.device = "cpu") -> dict:
# requires pip install runai_model_streamer
from runai_model_streamer import SafetensorsStreamer
if state_dict is None:
@@ -45,7 +45,7 @@ def load_streamer(files: list[str], state_dict: dict = None, key_mapping: dict =
state_dict[map_keys(key, key_mapping)] = tensor.to(device)
def load_files(files: list[str], state_dict: dict = None, key_mapping: dict = None, device: torch.device = "cpu", method: str = None) -> dict:
def load_files(files: list[str], state_dict: dict | None = None, key_mapping: dict | None = None, device: torch.device = "cpu", method: str | None = None) -> dict:
# note: files is list-of-files within a module for chunked loading, not accross model
if isinstance(files, str):
files = [files]
modules/sdnq/layers/conv/conv_fp16.py
+5 -5
View File
@@ -20,11 +20,11 @@ def conv_fp16_matmul(
padding_mode: str, conv_type: int,
groups: int, stride: list[int],
padding: list[int], dilation: list[int],
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
return_dtype = input.dtype
input, mm_output_shape = process_conv_input(conv_type, input, reversed_padding_repeated_twice, padding_mode, result_shape, stride, padding, dilation)
modules/sdnq/layers/conv/conv_fp8.py
+5 -5
View File
@@ -19,11 +19,11 @@ def conv_fp8_matmul(
padding_mode: str, conv_type: int,
groups: int, stride: list[int],
padding: list[int], dilation: list[int],
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
return_dtype = input.dtype
input, mm_output_shape = process_conv_input(conv_type, input, reversed_padding_repeated_twice, padding_mode, result_shape, stride, padding, dilation)
modules/sdnq/layers/conv/conv_fp8_tensorwise.py
+5 -5
View File
@@ -20,11 +20,11 @@ def conv_fp8_matmul_tensorwise(
padding_mode: str, conv_type: int,
groups: int, stride: list[int],
padding: list[int], dilation: list[int],
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
return_dtype = input.dtype
input, mm_output_shape = process_conv_input(conv_type, input, reversed_padding_repeated_twice, padding_mode, result_shape, stride, padding, dilation)
modules/sdnq/layers/conv/conv_int8.py
+5 -5
View File
@@ -20,11 +20,11 @@ def conv_int8_matmul(
padding_mode: str, conv_type: int,
groups: int, stride: list[int],
padding: list[int], dilation: list[int],
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
return_dtype = input.dtype
input, mm_output_shape = process_conv_input(conv_type, input, reversed_padding_repeated_twice, padding_mode, result_shape, stride, padding, dilation)
modules/sdnq/layers/conv/forward.py
+3 -3
View File
@@ -76,16 +76,16 @@ def quantized_conv_forward(self, input) -> torch.FloatTensor:
return self._conv_forward(input, self.sdnq_dequantizer(self.weight, self.scale, self.zero_point, self.svd_up, self.svd_down), self.bias)
def quantized_conv_transpose_1d_forward(self, input: torch.FloatTensor, output_size: list[int] = None) -> torch.FloatTensor:
def quantized_conv_transpose_1d_forward(self, input: torch.FloatTensor, output_size: list[int] | None = None) -> torch.FloatTensor:
output_padding = self._output_padding(input, output_size, self.stride, self.padding, self.kernel_size, 1, self.dilation)
return torch.nn.functional.conv_transpose1d(input, self.sdnq_dequantizer(self.weight, self.scale, self.zero_point, self.svd_up, self.svd_down), self.bias, self.stride, self.padding, output_padding, self.groups, self.dilation)
def quantized_conv_transpose_2d_forward(self, input: torch.FloatTensor, output_size: list[int] = None) -> torch.FloatTensor:
def quantized_conv_transpose_2d_forward(self, input: torch.FloatTensor, output_size: list[int] | None = None) -> torch.FloatTensor:
output_padding = self._output_padding(input, output_size, self.stride, self.padding, self.kernel_size, 2, self.dilation)
return torch.nn.functional.conv_transpose2d(input, self.sdnq_dequantizer(self.weight, self.scale, self.zero_point, self.svd_up, self.svd_down), self.bias, self.stride, self.padding, output_padding, self.groups, self.dilation)
def quantized_conv_transpose_3d_forward(self, input: torch.FloatTensor, output_size: list[int] = None) -> torch.FloatTensor:
def quantized_conv_transpose_3d_forward(self, input: torch.FloatTensor, output_size: list[int] | None = None) -> torch.FloatTensor:
output_padding = self._output_padding(input, output_size, self.stride, self.padding, self.kernel_size, 3, self.dilation)
return torch.nn.functional.conv_transpose3d(input, self.sdnq_dequantizer(self.weight, self.scale, self.zero_point, self.svd_up, self.svd_down), self.bias, self.stride, self.padding, output_padding, self.groups, self.dilation)
modules/sdnq/layers/linear/linear_fp16.py
+5 -5
View File
@@ -14,11 +14,11 @@ def fp16_matmul(
input: torch.FloatTensor,
weight: torch.Tensor,
scale: torch.FloatTensor,
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
if quantized_weight_shape is not None:
weight = unpack_float(weight, weights_dtype, quantized_weight_shape).to(dtype=torch.float16).t_()
modules/sdnq/layers/linear/linear_fp8.py
+5 -5
View File
@@ -19,11 +19,11 @@ def fp8_matmul(
input: torch.FloatTensor,
weight: torch.Tensor,
scale: torch.FloatTensor,
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
if quantized_weight_shape is not None:
weight = unpack_float(weight, weights_dtype, quantized_weight_shape).to(dtype=torch.float8_e4m3fn).t_()
modules/sdnq/layers/linear/linear_fp8_tensorwise.py
+5 -5
View File
@@ -22,11 +22,11 @@ def fp8_matmul_tensorwise(
input: torch.FloatTensor,
weight: torch.Tensor,
scale: torch.FloatTensor,
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
if quantized_weight_shape is not None:
weight = unpack_float(weight, weights_dtype, quantized_weight_shape).to(dtype=torch.float8_e4m3fn).t_()
modules/sdnq/layers/linear/linear_int8.py
+5 -5
View File
@@ -22,11 +22,11 @@ def int8_matmul(
input: torch.FloatTensor,
weight: torch.Tensor,
scale: torch.FloatTensor,
bias: torch.FloatTensor = None,
svd_up: torch.FloatTensor = None,
svd_down: torch.FloatTensor = None,
quantized_weight_shape: torch.Size = None,
weights_dtype: str = None,
bias: torch.FloatTensor | None = None,
svd_up: torch.FloatTensor | None = None,
svd_down: torch.FloatTensor | None = None,
quantized_weight_shape: torch.Size | None = None,
weights_dtype: str | None = None,
) -> torch.FloatTensor:
if quantized_weight_shape is not None:
weight = unpack_int(weight, weights_dtype, quantized_weight_shape, dtype=torch.int8).t_()
modules/sdnq/loader.py
+4 -4
View File
@@ -25,7 +25,7 @@ def unset_config_on_save(quantization_config: SDNQConfig) -> SDNQConfig:
return quantization_config
def save_sdnq_model(model: ModelMixin, model_path: str, max_shard_size: str = "5GB", is_pipeline: bool = False, sdnq_config: SDNQConfig = None) -> None:
def save_sdnq_model(model: ModelMixin, model_path: str, max_shard_size: str = "5GB", is_pipeline: bool = False, sdnq_config: SDNQConfig | None = None) -> None:
if is_pipeline:
for module_name in get_module_names(model):
module = getattr(model, module_name, None)
@@ -63,7 +63,7 @@ def save_sdnq_model(model: ModelMixin, model_path: str, max_shard_size: str = "5
model.config.quantization_config.to_json_file(quantization_config_path)
def load_sdnq_model(model_path: str, model_cls: ModelMixin = None, file_name: str = None, dtype: torch.dtype = None, device: torch.device = "cpu", dequantize_fp32: bool = None, use_quantized_matmul: bool = None, model_config: dict = None, quantization_config: dict = None, load_method: str = "safetensors") -> ModelMixin:
def load_sdnq_model(model_path: str, model_cls: ModelMixin | None = None, file_name: str | None = None, dtype: torch.dtype | None = None, device: torch.device = "cpu", dequantize_fp32: bool | None = None, use_quantized_matmul: bool | None = None, model_config: dict | None = None, quantization_config: dict | None = None, load_method: str = "safetensors") -> ModelMixin:
from accelerate import init_empty_weights
with init_empty_weights():
@@ -162,7 +162,7 @@ def post_process_model(model):
return model
def apply_sdnq_options_to_module(model, dtype: torch.dtype = None, dequantize_fp32: bool = None, use_quantized_matmul: bool = None):
def apply_sdnq_options_to_module(model, dtype: torch.dtype | None = None, dequantize_fp32: bool | None = None, use_quantized_matmul: bool | None = None):
has_children = list(model.children())
if not has_children:
if dtype is not None and getattr(model, "dtype", torch.float32) not in {torch.float32, torch.float64}:
@@ -231,7 +231,7 @@ def apply_sdnq_options_to_module(model, dtype: torch.dtype = None, dequantize_fp
return model
def apply_sdnq_options_to_model(model, dtype: torch.dtype = None, dequantize_fp32: bool = None, use_quantized_matmul: bool = None):
def apply_sdnq_options_to_model(model, dtype: torch.dtype | None = None, dequantize_fp32: bool | None = None, use_quantized_matmul: bool | None = None):
if use_quantized_matmul and not check_torch_compile():
raise RuntimeError("SDNQ Quantized MatMul requires a working Triton install.")
model = apply_sdnq_options_to_module(model, dtype=dtype, dequantize_fp32=dequantize_fp32, use_quantized_matmul=use_quantized_matmul)
modules/sdnq/quantizer.py
+21 -21
View File
@@ -189,7 +189,7 @@ def get_quant_kwargs(quant_kwargs: dict, modules_quant_config: dict[str, dict])
return quant_kwargs
def add_module_skip_keys(model, modules_to_not_convert: list[str] = None, modules_dtype_dict: dict[str, list[str]] = None):
def add_module_skip_keys(model, modules_to_not_convert: list[str] | None = None, modules_dtype_dict: dict[str, list[str]] | None = None):
if modules_to_not_convert is None:
modules_to_not_convert = []
if modules_dtype_dict is None:
@@ -547,7 +547,7 @@ def sdnq_quantize_layer(layer, weights_dtype="int8", quantized_matmul_dtype=None
@devices.inference_context()
def apply_sdnq_to_module(model, weights_dtype="int8", quantized_matmul_dtype=None, torch_dtype=None, group_size=0, svd_rank=32, svd_steps=8, dynamic_loss_threshold=1e-2, use_svd=False, quant_conv=False, use_quantized_matmul=False, use_quantized_matmul_conv=False, use_dynamic_quantization=False, use_stochastic_rounding=False, dequantize_fp32=True, non_blocking=False, modules_to_not_convert: list[str] = None, modules_dtype_dict: dict[str, list[str]] = None, modules_quant_config: dict[str, dict] = None, quantization_device=None, return_device=None, full_param_name=""): # pylint: disable=unused-argument
def apply_sdnq_to_module(model, weights_dtype="int8", quantized_matmul_dtype=None, torch_dtype=None, group_size=0, svd_rank=32, svd_steps=8, dynamic_loss_threshold=1e-2, use_svd=False, quant_conv=False, use_quantized_matmul=False, use_quantized_matmul_conv=False, use_dynamic_quantization=False, use_stochastic_rounding=False, dequantize_fp32=True, non_blocking=False, modules_to_not_convert: list[str] | None = None, modules_dtype_dict: dict[str, list[str]] | None = None, modules_quant_config: dict[str, dict] | None = None, quantization_device=None, return_device=None, full_param_name=""): # pylint: disable=unused-argument
has_children = list(model.children())
if not has_children:
return model, modules_to_not_convert, modules_dtype_dict
@@ -628,8 +628,8 @@ def apply_sdnq_to_module(model, weights_dtype="int8", quantized_matmul_dtype=Non
def sdnq_post_load_quant(
model: torch.nn.Module,
weights_dtype: str = "int8",
quantized_matmul_dtype: str = None,
torch_dtype: torch.dtype = None,
quantized_matmul_dtype: str | None = None,
torch_dtype: torch.dtype | None = None,
group_size: int = 0,
svd_rank: int = 32,
svd_steps: int = 8,
@@ -643,11 +643,11 @@ def sdnq_post_load_quant(
dequantize_fp32: bool = True,
non_blocking: bool = False,
add_skip_keys:bool = True,
quantization_device: torch.device = None,
return_device: torch.device = None,
modules_to_not_convert: list[str] = None,
modules_dtype_dict: dict[str, list[str]] = None,
modules_quant_config: dict[str, dict] = None,
quantization_device: torch.device | None = None,
return_device: torch.device | None = None,
modules_to_not_convert: list[str] | None = None,
modules_dtype_dict: dict[str, list[str]] | None = None,
modules_quant_config: dict[str, dict] | None = None,
):
if modules_to_not_convert is None:
modules_to_not_convert = []
@@ -735,9 +735,9 @@ class SDNQQuantize:
def convert(
self,
input_dict: dict[str, list[torch.Tensor]],
model: torch.nn.Module = None,
full_layer_name: str = None,
missing_keys: list[str] = None, # pylint: disable=unused-argument
model: torch.nn.Module | None = None,
full_layer_name: str | None = None,
missing_keys: list[str] | None = None, # pylint: disable=unused-argument
**kwargs, # pylint: disable=unused-argument
) -> dict[str, torch.FloatTensor]:
_module_name, value = tuple(input_dict.items())[0]
@@ -898,11 +898,11 @@ class SDNQQuantizer(DiffusersQuantizer, HfQuantizer):
def adjust_target_dtype(self, target_dtype: torch.dtype) -> torch.dtype: # pylint: disable=unused-argument,arguments-renamed
return dtype_dict[self.quantization_config.weights_dtype]["target_dtype"]
def update_torch_dtype(self, torch_dtype: torch.dtype = None) -> torch.dtype:
def update_torch_dtype(self, torch_dtype: torch.dtype | None = None) -> torch.dtype:
self.torch_dtype = torch_dtype
return torch_dtype
def update_dtype(self, dtype: torch.dtype = None) -> torch.dtype:
def update_dtype(self, dtype: torch.dtype | None = None) -> torch.dtype:
"""
needed for transformers compatibilty, returns self.update_torch_dtype
"""
@@ -912,7 +912,7 @@ class SDNQQuantizer(DiffusersQuantizer, HfQuantizer):
self,
model,
device_map, # pylint: disable=unused-argument
keep_in_fp32_modules: list[str] = None,
keep_in_fp32_modules: list[str] | None = None,
**kwargs, # pylint: disable=unused-argument
):
if self.pre_quantized:
@@ -1055,7 +1055,7 @@ class SDNQConfig(QuantizationConfigMixin):
def __init__( # pylint: disable=super-init-not-called
self,
weights_dtype: str = "int8",
quantized_matmul_dtype: str = None,
quantized_matmul_dtype: str | None = None,
group_size: int = 0,
svd_rank: int = 32,
svd_steps: int = 8,
@@ -1071,11 +1071,11 @@ class SDNQConfig(QuantizationConfigMixin):
dequantize_fp32: bool = True,
non_blocking: bool = False,
add_skip_keys: bool = True,
quantization_device: torch.device = None,
return_device: torch.device = None,
modules_to_not_convert: list[str] = None,
modules_dtype_dict: dict[str, list[str]] = None,
modules_quant_config: dict[str, dict] = None,
quantization_device: torch.device | None = None,
return_device: torch.device | None = None,
modules_to_not_convert: list[str] | None = None,
modules_dtype_dict: dict[str, list[str]] | None = None,
modules_quant_config: dict[str, dict] | None = None,
is_training: bool = False,
**kwargs, # pylint: disable=unused-argument
):
modules/shared_state.py
+2 -2
View File
@@ -148,7 +148,7 @@ class State:
return job
return None
def history(self, op:str, task_id:str=None, results:list=None):
def history(self, op: str, task_id: str | None = None, results: list | None = None):
if results is None:
results = []
job = {
@@ -174,7 +174,7 @@ class State:
if len(self.results) > 0:
self.history('output', self.id, results=self.results)
def get_id(self, task_id:str=None):
def get_id(self, task_id: str | None = None):
if task_id is None or task_id == 0:
task_id = uuid.uuid4().hex[:15]
if not isinstance(task_id, str):
modules/taesd/hybrid_small.py
+2 -2
View File
@@ -76,8 +76,8 @@ class AutoencoderSmall(ModelMixin, ConfigMixin, FromOriginalModelMixin):
down_block_types: tuple[str] = ("DownEncoderBlock2D",),
up_block_types: tuple[str] = ("UpDecoderBlock2D",),
block_out_channels: tuple[int] = (64,),
encoder_block_out_channels: tuple[int] = None,
decoder_block_out_channels: tuple[int] = None,
encoder_block_out_channels: tuple[int] | None = None,
decoder_block_out_channels: tuple[int] | None = None,
layers_per_block: int = 1,
act_fn: str = "silu",
latent_channels: int = 4,
modules/textual_inversion.py
+1 -1
View File
@@ -254,7 +254,7 @@ class EmbeddingDatabase:
self.ids_lookup[first_id] = sorted(self.ids_lookup[first_id] + [(ids, embedding)], key=lambda x: len(x[0]), reverse=True)
return embedding
def load_diffusers_embedding(self, filename: str | list[str] = None, data: dict = None):
def load_diffusers_embedding(self, filename: str | list[str] | None = None, data: dict | None = None):
"""
File names take precidence over bundled embeddings passed as a dict.
Bundled embeddings are automatically set to overwrite previous embeddings.
modules/todo/todo_merge.py
+1 -1
View File
@@ -38,7 +38,7 @@ def init_generator(device: torch.device, fallback: torch.Generator = None):
return fallback
def do_nothing(x: torch.Tensor, mode: str = None): # pylint: disable=unused-argument
def do_nothing(x: torch.Tensor, mode: str | None = None): # pylint: disable=unused-argument
return x
modules/ui_control.py
+2 -2
View File
@@ -19,7 +19,7 @@ debug('Trace: CONTROL')
use_generator = os.environ.get('SD_USE_GENERATOR', None) is not None
def return_stats(t: float = None):
def return_stats(t: float | None = None):
if t is None:
elapsed_text = ''
else:
@@ -48,7 +48,7 @@ def return_stats(t: float = None):
return f"<div class='performance'><p>{elapsed_text} {summary} {gpu} {cpu}</p></div>"
def return_controls(res, t: float = None):
def return_controls(res, t: float | None = None):
# return preview, image, video, gallery, text
debug(f'Control received: type={type(res)} {res}')
if t is None:
modules/ui_models.py
+1 -1
View File
@@ -84,7 +84,7 @@ def create_ui():
with gr.Row():
save_path = gr.Textbox(label="Model base path", placeholder="Path to save model to", value=opts.diffusers_dir)
with gr.Row():
save_shard = gr.Textbox(label="Max shard size", placeholder="Maximum shard size", value="10GB")
save_shard = gr.Textbox(label="Max shard size", placeholder="Maximum shard size", value="5GB")
save_overwrite = gr.Checkbox(label="Overwrite existing", value=False)
with gr.Row():
save_result = gr.HTML(value="", elem_id="model_save_outcome")
modules/ui_sections.py
+2 -2
View File
@@ -5,7 +5,7 @@ from modules.ui_components import ToolButton
from modules.caption import caption
def create_toprow(is_img2img: bool = False, id_part: str = None, generate_visible: bool = True, negative_visible: bool = True, reprocess_visible: bool = True):
def create_toprow(is_img2img: bool = False, id_part: str | None = None, generate_visible: bool = True, negative_visible: bool = True, reprocess_visible: bool = True):
def apply_styles(prompt, prompt_neg, styles):
prompt = shared.prompt_styles.apply_styles_to_prompt(prompt, styles, wildcards=not shared.opts.extra_networks_apply_unparsed)
prompt_neg = shared.prompt_styles.apply_negative_styles_to_prompt(prompt_neg, styles, wildcards=not shared.opts.extra_networks_apply_unparsed)
@@ -92,7 +92,7 @@ def create_resolution_inputs(tab, default_width=1024, default_height=1024):
return width, height
def create_caption_button(tab: str, inputs: list = None, outputs: str = None, what: str = ''):
def create_caption_button(tab: str, inputs: list | None = None, outputs: str | None = None, what: str = ''):
button_caption = gr.Button(ui_symbols.caption, elem_id=f"{tab}_caption_{what}", elem_classes=['caption'])
if inputs is not None and outputs is not None:
button_caption.click(fn=caption.caption, inputs=inputs, outputs=[outputs])
modules/ui_video_vlm.py
+1 -1
View File
@@ -21,7 +21,7 @@ system_prompts = {
}
def enhance_prompt(enable:bool, model:str=None, image=None, prompt:str='', system_prompt:str='', nsfw:bool=True):
def enhance_prompt(enable: bool, model: str | None = None, image=None, prompt: str = "", system_prompt: str = "", nsfw: bool = True):
from modules.caption import vqa
if not enable:
return prompt
modules/upscaler.py
+10 -4
View File
@@ -1,10 +1,16 @@
from __future__ import annotations
import os
from abc import abstractmethod
from typing import TYPE_CHECKING
from PIL import Image
from modules import modelloader, shared, paths
from modules.logger import log
if TYPE_CHECKING:
from torch import Tensor
models = None
@@ -92,10 +98,10 @@ class Upscaler:
return scalers
@abstractmethod
def do_upscale(self, img: Image, selected_model: str):
def do_upscale(self, img: Image.Image | Tensor, selected_model: str):
return img
def upscale(self, img: Image, scale, selected_model: str = None):
def upscale(self, img: Image.Image | Tensor, scale, selected_model: str | None = None):
jobid = shared.state.begin('Upscale')
self.scale = scale
if isinstance(img, Image.Image):
@@ -153,10 +159,10 @@ class UpscalerData:
name = None
data_path = None
scale: int = 4
scaler: Upscaler = None
scaler: Upscaler | None = None
model: None
def __init__(self, name: str, path: str = None, upscaler: Upscaler = None, scale: int = 4, model=None):
def __init__(self, name: str, path: str | None = None, upscaler: Upscaler | None = None, scale: int = 4, model=None):
self.name = name
self.data_path = path
self.local_data_path = path
modules/vae/sd_vae_fal.py
+4 -4
View File
@@ -50,12 +50,12 @@ class Flux2TinyAutoEncoder(ModelMixin, ConfigMixin):
in_channels: int = 3,
out_channels: int = 3,
latent_channels: int = 128,
encoder_block_out_channels: list[int] = None,
decoder_block_out_channels: list[int] = None,
encoder_block_out_channels: list[int] | None = None,
decoder_block_out_channels: list[int] | None = None,
act_fn: str = "silu",
upsampling_scaling_factor: int = 2,
num_encoder_blocks: list[int] = None,
num_decoder_blocks: list[int] = None,
num_encoder_blocks: list[int] | None = None,
num_decoder_blocks: list[int] | None = None,
latent_magnitude: float = 3.0,
latent_shift: float = 0.5,
force_upcast: bool = False,
modules/video_models/video_save.py
+30 -29
View File
@@ -29,7 +29,7 @@ def get_video_filename(p:processing.StableDiffusionProcessingVideo):
return filename
def save_params(p, filename: str = None):
def save_params(p, filename: str | None = None):
from modules.paths import params_path
if p is None:
dct = {}
@@ -129,17 +129,18 @@ def write_audio(
container.mux(packet)
def atomic_save_video(filename: str,
tensor:torch.Tensor,
audio:torch.Tensor=None,
fps:float=24,
codec:str='libx264',
pix_fmt:str='yuv420p',
options:str='',
aac:int=24000,
metadata:dict=None,
pbar=None,
):
def atomic_save_video(
filename: str,
tensor: torch.Tensor,
audio: torch.Tensor | None = None,
fps: float = 24,
codec: str = "libx264",
pix_fmt: str = "yuv420p",
options: str = "",
aac: int = 24000,
metadata: dict | None = None,
pbar=None,
):
if metadata is None:
metadata = {}
av = check_av()
@@ -212,23 +213,23 @@ def save_thumbnail(video_path, tensor=None):
def save_video(
p:processing.StableDiffusionProcessingVideo,
pixels:torch.Tensor=None,
audio:torch.Tensor=None,
binary:bytes=None,
mp4_fps:int=24,
mp4_codec:str='libx264',
mp4_opt:str='',
mp4_ext:str='mp4',
mp4_sf:bool=False, # save safetensors
mp4_video:bool=True, # save video
mp4_frames:bool=False, # save frames
mp4_interpolate:int=0, # rife interpolation
aac_sample_rate:int=24000, # audio sample rate
stream=None, # async progress reporting stream
metadata:dict=None, # metadata for video
pbar=None, # progress bar for video
):
p: processing.StableDiffusionProcessingVideo,
pixels: torch.Tensor | None = None,
audio: torch.Tensor | None = None,
binary: bytes | None = None,
mp4_fps: int = 24,
mp4_codec: str = "libx264",
mp4_opt: str = "",
mp4_ext: str = "mp4",
mp4_sf: bool = False, # save safetensors
mp4_video: bool = True, # save video
mp4_frames: bool = False, # save frames
mp4_interpolate: int = 0, # rife interpolation
aac_sample_rate: int = 24000, # audio sample rate
stream=None, # async progress reporting stream
metadata: dict | None = None, # metadata for video
pbar=None, # progress bar for video
):
if metadata is None:
metadata = {}
output_video = None
pyproject.toml
+3 -1
View File
@@ -97,7 +97,6 @@ ignore = [
"RUF008", # Do not use mutable default values for dataclass
"RUF010", # Use explicit conversion flag
"RUF012", # Mutable class attributes
"RUF013", # PEP 484 prohibits implicit `Optional`
"RUF015", # Prefer `next(...)` over single element slice
"RUF022", # All is not sorted
"RUF046", # Value being cast to `int` is already an integer
@@ -108,6 +107,9 @@ fixable = ["ALL"]
unfixable = []
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
[tool.ruff.lint.per-file-ignores]
"modules/caption/joytag.py" = ["RUF013"] # Per header comment: "Do not modify directly — sync from upstream"
[tool.ruff.format]
quote-style = "double"
indent-style = "space"