strict typechecking

Signed-off-by: Vladimir Mandic <mandic00@live.com>
This commit is contained in:
Vladimir Mandic
2026-08-16 11:32:47 +02:00
parent f30e6baa67
commit 76941efc7e
44 changed files with 165 additions and 155 deletions
+3 -3
View File
@@ -263,7 +263,7 @@ def validate_image(image_b64: str):
return image.convert('RGB')
def build_clip_overrides(req) -> dict:
def build_clip_overrides(req) -> dict | None:
"""Build clip_interrogator overrides dict from request fields."""
overrides = {}
for key in ('max_length', 'chunk_size', 'min_flavors', 'max_flavors', 'flavor_count', 'num_beams'):
@@ -304,7 +304,7 @@ def do_openclip(image, req):
return caption, get_top_item(results[0]), get_top_item(results[1]), get_top_item(results[2]), get_top_item(results[3]), get_top_item(results[4])
def build_vqa_kwargs(req) -> dict:
def build_vqa_kwargs(req) -> dict | None:
"""Build generation kwargs dict from VQA request fields."""
kwargs = {}
for key in ('max_tokens', 'temperature', 'top_k', 'top_p', 'num_beams', 'do_sample', 'keep_thinking', 'keep_prefill'):
@@ -377,7 +377,7 @@ def do_analyze(image, req):
return answer, annotated_b64
def parse_tagger_scores(tags: str) -> dict:
def parse_tagger_scores(tags: str) -> dict | None:
"""Parse confidence scores from tagger output string."""
scores = {}
for item in tags.split(', '):
+2 -2
View File
@@ -62,8 +62,8 @@ class PydanticModelGenerator:
additional_fields: list[dict[str, Any]] | None = None,
exclude_fields: list | None = None,
):
if exclude_fields is None:
exclude_fields = []
additional_fields = additional_fields or []
exclude_fields = exclude_fields or []
def field_type_generator(_k, v):
field_type = v.annotation
return Optional[field_type]
+5 -5
View File
@@ -23,7 +23,7 @@ def set_sdnq_attention():
from sdnq.kernels.triton_atten import sdnq_triton_atten
sdpa_pre_sdnq_atten = torch.nn.functional.scaled_dot_product_attention
@wraps(sdpa_pre_sdnq_atten)
def sdpa_sdnq_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.FloatTensor:
def sdpa_sdnq_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.Tensor:
if (
query.device.type != "cpu"
and (query.shape[-2] >= 32 and key.shape[-2] >= 32)
@@ -57,7 +57,7 @@ def set_triton_flash_attention(backend: str):
sdpa_pre_triton_flash_atten = torch.nn.functional.scaled_dot_product_attention
@wraps(sdpa_pre_triton_flash_atten)
def sdpa_triton_flash_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.FloatTensor:
def sdpa_triton_flash_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.Tensor:
use_triton = (
query.shape[-1] <= 128
and attn_mask is None
@@ -98,7 +98,7 @@ def set_flex_attention():
sdpa_pre_flex_atten = torch.nn.functional.scaled_dot_product_attention
@wraps(sdpa_pre_flex_atten)
def sdpa_flex_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.FloatTensor: # pylint: disable=unused-argument
def sdpa_flex_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.Tensor: # pylint: disable=unused-argument
score_mod = None
block_mask = None
if attn_mask is not None:
@@ -140,7 +140,7 @@ def set_ck_flash_attention(backend: str, device: torch.device):
sdpa_pre_flash_atten = torch.nn.functional.scaled_dot_product_attention
@wraps(sdpa_pre_flash_atten)
def sdpa_flash_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.FloatTensor:
def sdpa_flash_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.Tensor:
use_flash = (
query.shape[-1] <= 128
and attn_mask is None
@@ -215,7 +215,7 @@ def set_sage_attention(backend: str, device: torch.device):
sdpa_pre_sage_atten = torch.nn.functional.scaled_dot_product_attention
@wraps(sdpa_pre_sage_atten)
def sdpa_sage_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.FloatTensor:
def sdpa_sage_atten(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.Tensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.Tensor:
use_sage = (
query.shape[-1] in {128, 96, 64}
and attn_mask is None
+1 -1
View File
@@ -288,7 +288,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 = None) -> Image.Image:
def draw_bounding_boxes(image: Image.Image, detections: list, points: list | None = None) -> Image.Image | None:
"""
Draw bounding boxes and/or points on an image.
+2 -1
View File
@@ -270,7 +270,8 @@ class WaifuDiffusionTagger:
character_count = 0
rating_count = 0
for i, (tag_name, prob) in enumerate(zip(self.tags, probs, strict=False)):
tags = self.tags or []
for i, (tag_name, prob) in enumerate(zip(tags, probs, strict=False)):
category = self.tag_categories[i]
tag_lower = tag_name.lower()
+2 -1
View File
@@ -242,7 +242,8 @@ def preprocess_image(
p.init_images = [init_image] * len(active_model)
if hasattr(shared.sd_model, 'controlnet') and 'control_image' in p.task_args and len(p.task_args['control_image']) > 1 and (shared.sd_model.__class__.__name__ == 'StableDiffusionXLControlNetUnionPipeline'): # special case for controlnet-union
p.task_args['control_image'] = [[x] for x in p.task_args['control_image']]
p.task_args['control_mode'] = [[x] for x in p.task_args['control_mode']]
control_mode = p.task_args.get('control_mode') or []
p.task_args['control_mode'] = [[x] for x in control_mode]
# determine txt2img, img2img, inpaint pipeline
if unit_type == 'reference' and has_models: # special case
+1 -1
View File
@@ -134,7 +134,7 @@ class Detailer():
log.debug(f'Detailer: items={len(items)} filtered={len(filtered)}')
return filtered
def draw_masks(self, image: Image.Image, items: list[DetailerResult], p=None) -> Image.Image:
def draw_masks(self, image: Image.Image, items: list[DetailerResult], p=None) -> Image.Image | np.ndarray:
if not isinstance(image, Image.Image):
image = Image.fromarray(image)
image = image.convert('RGBA')
+1 -1
View File
@@ -19,7 +19,7 @@ def format_grounding_dino_prompt(prompt: str) -> str:
return ". ".join(items) + "."
def load(self, model_name: str | None = None) -> tuple[str, transformers.AutoModelForZeroShotObjectDetection]: # pylint: disable=unused-argument
def load(self, model_name: str | None = None) -> tuple[str | None, transformers.AutoModelForZeroShotObjectDetection | object]: # pylint: disable=unused-argument
cached = sd_offload_aux.get_aux_model(model_name)
if cached is not None:
return model_name, cached
+1 -1
View File
@@ -30,7 +30,7 @@ def select_florence_task(prompt: str) -> tuple[str, str]:
return task, formatted
def load(self, model_name: str | None = None) -> tuple[str, transformers.AutoModelForCausalLM]: # pylint: disable=unused-argument
def load(self, model_name: str | None = None) -> tuple[str | None, transformers.AutoModelForCausalLM | object]: # pylint: disable=unused-argument
cached = sd_offload_aux.get_aux_model(model_name)
if cached is not None:
return model_name, cached
+2 -2
View File
@@ -56,7 +56,7 @@ def template(prompt: str, schema: str, min_confidence: float) -> list[dict]:
]
def load(self, model_name: str | None = None) -> tuple[str, transformers.Qwen3VLForConditionalGeneration]: # pylint: disable=unused-argument
def load(self, model_name: str | None = None) -> tuple[str | None, transformers.Qwen3VLForConditionalGeneration | object]: # pylint: disable=unused-argument
cached = sd_offload_aux.get_aux_model(model_name)
if cached is not None:
return model_name, cached
@@ -70,7 +70,7 @@ def load(self, model_name: str | None = None) -> tuple[str, transformers.Qwen3VL
quant_args = model_quant.create_config(module='LLM', modules_to_not_convert=['conv1d', 'linear_attn.conv1d'])
model = transformers.Qwen3VLForConditionalGeneration.from_pretrained(**load_kwargs, **quant_args, attn_implementation="sdpa")
model = model.eval()
model.processor: transformers.Qwen3VLProcessor = transformers.Qwen3VLProcessor.from_pretrained(**load_kwargs)
model.processor = transformers.Qwen3VLProcessor.from_pretrained(**load_kwargs)
sd_offload_aux.register_aux(model_name, model)
if shared.opts.detailer_unload:
sd_offload_aux.offload_aux(model_name)
+1 -1
View File
@@ -17,7 +17,7 @@ def format_rex_prompt(prompt: str) -> str:
return f"<|grounding|>{clean_prompt}"
def load(self, model_name: str | None = None) -> tuple[str, transformers.AutoModelForCausalLM]: # pylint: disable=unused-argument
def load(self, model_name: str | None = None) -> tuple[str | None, transformers.AutoModelForCausalLM | object]: # pylint: disable=unused-argument
cached = sd_offload_aux.get_aux_model(model_name)
if cached is not None:
return model_name, cached
+1 -1
View File
@@ -6,7 +6,7 @@ from modules.detailer import DetailerResult, detailer_opt, get_mask
from modules.logger import log
def load(self, model_name: str | None = None) -> tuple[str, transformers.Sam3Model]: # pylint: disable=unused-argument
def load(self, model_name: str | None = None) -> tuple[str | None, transformers.Sam3Model | object]: # pylint: disable=unused-argument
cached = sd_offload_aux.get_aux_model(model_name)
if cached is not None:
return model_name, cached
+1 -1
View File
@@ -762,7 +762,7 @@ def llm_context():
yield
def torch_reset() -> bool:
def torch_reset() -> None:
"""
Resets PyTorch execution graph, flushes VRAM caches, and syncs streams.
"""
+3 -2
View File
@@ -1,7 +1,7 @@
### original <https://github.com/TencentARC/PhotoMaker/blob/main/photomaker/pipeline.py>
import inspect
from typing import Any, Union
from typing import Any, Union, cast
from collections.abc import Callable
import PIL
import torch
@@ -813,7 +813,8 @@ class PhotoMakerStableDiffusionXLPipeline(StableDiffusionXLPipeline):
callback_kwargs = {}
for k in callback_on_step_end_tensor_inputs:
callback_kwargs[k] = locals()[k]
callback_outputs = callback_on_step_end(self, i, t, callback_kwargs)
callback_on_step_end_fn = cast(Any, callback_on_step_end)
callback_outputs = callback_on_step_end_fn(self, i, t, callback_kwargs)
latents = callback_outputs.pop("latents", latents)
prompt_embeds = callback_outputs.pop("prompt_embeds", prompt_embeds)
+7 -8
View File
@@ -1,3 +1,5 @@
from __future__ import annotations
from typing import Union
import os
from collections import UserDict
@@ -8,16 +10,13 @@ from modules.logger import log
do_cache_folders = os.environ.get('SD_NO_CACHE', None) is None
class Directory: # forward declaration
...
FilePathList = list[str]
FilePathIterator = Iterator[str]
DirectoryPathList = list[str]
DirectoryPathIterator = Iterator[str]
DirectoryList = list[Directory]
DirectoryIterator = Iterator[Directory]
DirectoryCollection = dict[str, Directory]
DirectoryList = list['Directory']
DirectoryIterator = Iterator['Directory']
DirectoryCollection = dict[str, 'Directory']
ExtensionFilter = Callable
ExtensionList = list[str]
RecursiveType = Union[bool, Callable]
@@ -35,7 +34,7 @@ def real_path(directory_path: str) -> str | None:
@dataclass(frozen=True)
class Directory(Directory): # pylint: disable=E0102
class Directory:
path: str = field(default_factory=str)
files: FilePathList = field(default_factory=list)
directories: DirectoryPathList = field(default_factory=list)
@@ -265,7 +264,7 @@ def unique_paths(directory_paths: DirectoryPathList) -> DirectoryPathIterator:
yield r
def get_directories(*directory_paths: DirectoryPathList, fetch: bool = True, recursive: RecursiveType = True) -> DirectoryCollection:
def get_directories(*directory_paths: DirectoryPathList, fetch: bool = True, recursive: RecursiveType = True) -> list[Directory]:
dirs = unique_directories(directory_paths, recursive=recursive)
return [d for d in (get_directory(p, fetch=fetch) for p in dirs) if d]
+1 -1
View File
@@ -175,7 +175,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 = None, exclude: list[str] | None = None) -> bool:
def changed(self, requested: list[str], include: list[str] | None = None, exclude: list[str] | None = None) -> tuple[bool, str]:
if shared.opts.lora_force_reload:
debug_log(f'Network check: type=LoRA requested={requested} status="forced"')
return True, "forced"
+2 -2
View File
@@ -45,8 +45,8 @@ def install_nunchaku(force=False):
if arch not in ['linux', 'windows']:
log.error(f'Nunchaku: platform={arch} unsupported')
return False
if not force and devices.backend not in ['cuda']:
log.error(f'Nunchaku: backend={devices.backend} unsupported')
if not force and devices.backend not in ['cuda']: # ty: ignore
log.error(f'Nunchaku: backend={devices.backend} unsupported') # ty: ignore
return False
url = os.environ.get('NUNCHAKU_COMMAND', None)
+2 -1
View File
@@ -1,4 +1,5 @@
import inspect
from collections.abc import Callable
import diffusers
import transformers
import safetensors.torch
@@ -48,7 +49,7 @@ def get_safetensor_keys(filename):
return keys
def get_modules(model: callable):
def get_modules(model: Callable):
signature = inspect.signature(model.__init__, follow_wrapped=True)
params = {param.name: param.annotation for param in signature.parameters.values() if param.annotation != inspect._empty and hasattr(param.annotation, 'from_pretrained')} # pylint: disable=protected-access
for name, cls in params.items():
+2 -1
View File
@@ -1,4 +1,5 @@
from __future__ import annotations
import builtins
import os
import sys
import json
@@ -28,7 +29,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 = None, restricted: set[str] | None = None, *, filename = '', secrets = ''):
def __init__(self, options_templates: dict[str, OptionInfo | LegacyOption] | None = None, restricted: builtins.set[str] | None = None, *, filename = '', secrets = ''):
if options_templates is None:
options_templates = {}
if restricted is None:
+5 -5
View File
@@ -5,9 +5,9 @@ from modules.logger import log
class LinuxUtils():
@staticmethod
def get_status() -> dict[str, float] | None:
def get_status() -> dict[str, int | float | str] | None:
lines = []
status = {}
status: dict[str, int | float | str] = {}
try:
with open("/proc/self/status", encoding="utf-8") as handle:
lines = handle.readlines()
@@ -26,7 +26,7 @@ class LinuxUtils():
return status
@staticmethod
def get_smaps(limit: int = 8) -> list[dict[str, float | str]] | None:
def get_smaps(limit: int = 8) -> list[dict[str, int | float | str]] | None:
try:
with open("/proc/self/smaps", encoding="utf-8") as handle:
lines = handle.readlines()
@@ -70,7 +70,7 @@ class LinuxUtils():
current["shared"] += amount
if current is not None:
entries.append(current)
merged = {}
merged: dict[str, dict[str, int | float | str]] = {}
for entry in entries:
path = entry["path"]
if path not in merged:
@@ -80,7 +80,7 @@ class LinuxUtils():
merged[path]["pss"] += entry["pss"]
merged[path]["private"] += entry["private"]
merged[path]["shared"] += entry["shared"]
top = sorted(merged.values(), key=lambda item: item["rss"], reverse=True)[:limit]
top: list[dict[str, int | float | str]] = sorted(merged.values(), key=lambda item: item["rss"], reverse=True)[:limit]
for entry in top:
entry["rss"] = round(entry["rss"] / 1024 / 1024, 3)
entry["pss"] = round(entry["pss"] / 1024 / 1024, 3)
+12 -12
View File
@@ -74,21 +74,21 @@ def edge_detect_for_pixelart(image: PipelineImageInput, image_weight: float = 1.
return new_image
def get_dct_harmonics(N: int, device: torch.device) -> torch.FloatTensor:
def get_dct_harmonics(N: int, device: torch.device) -> torch.Tensor:
k = torch.arange(N, dtype=torch.float32, device=device)
spatial = torch.add(1, k.unsqueeze(1), alpha=2)
spectral = k.unsqueeze(0) * (torch.pi / (2 * N))
return torch.cos(torch.mm(spatial, spectral))
def get_dct_norm(N: int, device: torch.device) -> torch.FloatTensor:
def get_dct_norm(N: int, device: torch.device) -> torch.Tensor:
n = torch.ones((N, 1), dtype=torch.float32, device=device)
n[0, 0] = 1 / math.sqrt(2)
n = torch.mm(n, n.t())
return n
def dct_2d(x: torch.FloatTensor, norm: str="ortho") -> torch.FloatTensor:
def dct_2d(x: torch.FloatTensor, norm: str="ortho") -> torch.Tensor:
x_shape = x.shape
N = x_shape[-1]
x = x.contiguous().view(-1, N, N)
@@ -102,7 +102,7 @@ def dct_2d(x: torch.FloatTensor, norm: str="ortho") -> torch.FloatTensor:
return coeff
def idct_2d(coeff: torch.FloatTensor, norm: str="ortho") -> torch.FloatTensor:
def idct_2d(coeff: torch.FloatTensor, norm: str="ortho") -> torch.Tensor:
x_shape = coeff.shape
N = x_shape[-1]
coeff = coeff.contiguous().view(-1, N, N)
@@ -116,7 +116,7 @@ def idct_2d(coeff: torch.FloatTensor, norm: str="ortho") -> torch.FloatTensor:
return x
def encode_single_channel_dct_2d(img: torch.FloatTensor, block_size: int=16, norm: str="ortho") -> torch.FloatTensor:
def encode_single_channel_dct_2d(img: torch.FloatTensor, block_size: int=16, norm: str="ortho") -> torch.Tensor:
batch_size, height, width = img.shape
h_blocks = int(height//block_size)
w_blocks = int(width//block_size)
@@ -130,7 +130,7 @@ def encode_single_channel_dct_2d(img: torch.FloatTensor, block_size: int=16, nor
return dct_tensor
def decode_single_channel_dct_2d(img: torch.FloatTensor, norm: str="ortho") -> torch.FloatTensor:
def decode_single_channel_dct_2d(img: torch.FloatTensor, norm: str="ortho") -> torch.Tensor:
batch_size, combined_block_size, h_blocks, w_blocks = img.shape
block_size = int(math.sqrt(combined_block_size))
height = int(h_blocks*block_size)
@@ -142,19 +142,19 @@ def decode_single_channel_dct_2d(img: torch.FloatTensor, norm: str="ortho") -> t
return img_tensor
def rgb_to_ycbcr_tensor(image: torch.ByteTensor) -> torch.FloatTensor:
def rgb_to_ycbcr_tensor(image: torch.ByteTensor) -> torch.Tensor:
rgb_weights = torch.tensor([[0.002345098, -0.001323419, 0.003921569], [0.004603922, -0.00259815, -0.003283824], [0.000894118, 0.003921569, -0.000637744]], device=image.device)
ycbcr = torch.einsum("cv,...chw->...vhw", [rgb_weights, image.permute(0,3,1,2).to(dtype=torch.float32)])
ycbcr[:,0,:,:] = ycbcr[:,0,:,:].add(-1)
return ycbcr
def ycbcr_tensor_to_rgb(ycbcr: torch.FloatTensor) -> torch.ByteTensor:
def ycbcr_tensor_to_rgb(ycbcr: torch.FloatTensor) -> torch.Tensor:
ycbcr_weights = torch.tensor([[127.5, 127.5, 127.5], [0, -43.877376465, 225.93], [178.755, -91.052376465, 0]], device=ycbcr.device)
return torch.einsum("cv,...chw->...vhw", [ycbcr_weights, ycbcr]).add(127.5).round().clamp(0,255).permute(0,2,3,1).to(dtype=torch.uint8)
def encode_jpeg_tensor(img: torch.FloatTensor, block_size: int=16, cbcr_downscale: int=2, norm: str="ortho") -> torch.FloatTensor:
def encode_jpeg_tensor(img: torch.FloatTensor, block_size: int=16, cbcr_downscale: int=2, norm: str="ortho") -> torch.Tensor:
img = img[:, :, :(img.shape[-2]//block_size)*block_size, :(img.shape[-1]//block_size)*block_size] # crop to a multiply of block_size
cbcr_block_size = block_size//cbcr_downscale
_, _, height, width = img.shape
@@ -165,7 +165,7 @@ def encode_jpeg_tensor(img: torch.FloatTensor, block_size: int=16, cbcr_downscal
return torch.cat([y,cb,cr], dim=1)
def decode_jpeg_tensor(jpeg_img: torch.FloatTensor, block_size: int=16, cbcr_downscale: int=2, norm: str="ortho") -> torch.FloatTensor:
def decode_jpeg_tensor(jpeg_img: torch.FloatTensor, block_size: int=16, cbcr_downscale: int=2, norm: str="ortho") -> torch.Tensor:
_, _, h_blocks, w_blocks = jpeg_img.shape
y_block_size = block_size*block_size
cbcr_block_size = int((block_size//cbcr_downscale) ** 2)
@@ -181,7 +181,7 @@ def decode_jpeg_tensor(jpeg_img: torch.FloatTensor, block_size: int=16, cbcr_dow
return torch.stack([y,cb,cr], dim=1)
def process_image_input(images: PipelineImageInput) -> torch.ByteTensor:
def process_image_input(images: PipelineImageInput) -> torch.Tensor:
if isinstance(images, list):
combined_images = []
for img in images:
@@ -235,7 +235,7 @@ class JPEGEncoder(ImageProcessingMixin, ConfigMixin):
self.latents_mean = latents_mean
super().__init__()
def encode(self, images: PipelineImageInput, device: str="cpu") -> torch.FloatTensor:
def encode(self, images: PipelineImageInput, device: str="cpu") -> torch.Tensor:
"""
Encode RGB 0-255 image to JPEG Latents.
+1 -1
View File
@@ -11,7 +11,7 @@ from modules.paths import resolve_output_path
def run_postprocessing(extras_mode,
image,
image_folder: list[tempfile.NamedTemporaryFile],
image_folder: list[object],
input_dir,
output_dir,
video,
+2 -2
View File
@@ -42,8 +42,8 @@ class Processed:
self.prompt = p.prompt or ''
self.negative_prompt = p.negative_prompt or ''
self.prompt = self.prompt if type(self.prompt) != list else self.prompt[0]
self.negative_prompt = self.negative_prompt if type(self.negative_prompt) != list else self.negative_prompt[0]
self.prompt = self.prompt[0] if isinstance(self.prompt, list) and self.prompt else self.prompt
self.negative_prompt = self.negative_prompt[0] if isinstance(self.negative_prompt, list) and self.negative_prompt else self.negative_prompt
self.styles = p.styles
self.bytes = binary
+3 -1
View File
@@ -148,7 +148,9 @@ def task_specific_kwargs(p, model):
task_args['image'] = p.init_images
if ('QwenImageLayeredPipeline' in model_cls) and (task_args.get('image', None) is not None):
task_args['image'] = [i.convert('RGBA') for i in task_args['image']]
image_items = task_args['image']
if isinstance(image_items, list):
task_args['image'] = [i.convert('RGBA') for i in image_items]
if ('LatentConsistencyModelPipeline' in model_cls) and (len(p.init_images) > 0):
p.ops.append('lcm')
init_latents = [processing_vae.vae_encode(image, model=shared.sd_model, vae_type=p.vae_type).squeeze(dim=0) for image in p.init_images]
+1
View File
@@ -168,6 +168,7 @@ class PromptEmbedder:
self.negative_prompt_attention_masks = [self.negative_prompt_attention_masks[0]] * self.batchsize
debug(f"Prompt cache: get={key}")
return True
return False
def compare_prompts(self):
same = (self.prompts == [self.prompts[0]] * len(self.prompts) and self.negative_prompts == [self.negative_prompts[0]] * len(self.negative_prompts))
+1 -1
View File
@@ -167,7 +167,7 @@ class RASJointAttnProcessor2_0:
attention_mask: torch.FloatTensor | None = None,
*args,
**kwargs,
) -> torch.FloatTensor:
) -> torch.FloatTensor | tuple[torch.FloatTensor, torch.FloatTensor]:
residual = hidden_states
batch_size = hidden_states.shape[0]
+1 -1
View File
@@ -29,7 +29,7 @@ def ras_forward(
joint_attention_kwargs: dict[str, Any] | None = None,
return_dict: bool = True,
skip_layers: list[int] | None = None,
) -> torch.FloatTensor | Transformer2DModelOutput:
) -> tuple[torch.Tensor] | torch.Tensor | Transformer2DModelOutput:
"""
The [`SD3Transformer2DModel`] forward method.
+11 -7
View File
@@ -96,14 +96,18 @@ class Agent:
@overload
def __init__(self, name: str): ...
@overload
def __init__(self, device: 'torch.types.Device'): ...
def __init__(self, name: 'torch.types.Device'): ...
def __init__(self, arg):
if isinstance(arg, str):
name = arg
else: # assume arg is device-like object
def __init__(self, name: str | 'torch.types.Device' | None = None):
if isinstance(name, str):
arch_name = name
elif name is not None:
import torch
name = getattr(torch.cuda.get_device_properties(arg), "gcnArchName", "gfx0000")
arch_name = getattr(torch.cuda.get_device_properties(name), "gcnArchName", "gfx0000")
else:
raise ValueError("ROCm Agent requires a GPU name or device")
arch_name = str(arch_name)
self.name = arch_name.split(':')[0]
self.name = name.split(':')[0]
self.gfx_version = Agent.parse_gfx_version(self.name)
if self.gfx_version > 0x1000:
@@ -221,7 +225,7 @@ def find() -> ROCmEnvironment | None:
return ROCmEnvironment(resolve_link("/opt/rocm"))
def get_version() -> str:
def get_version() -> str | None:
try:
if isinstance(environment, ROCmEnvironment):
# We don't load the hip library that will not be used by PyTorch.
+1 -1
View File
@@ -105,7 +105,7 @@ def load(filename, *args, **kwargs):
return load_with_extra(filename, *args, extra_handler=global_extra_handler, **kwargs)
def load_with_extra(filename, extra_handler=None, *args, **kwargs): # pylint: disable=keyword-arg-before-vararg
def load_with_extra(filename, *args, extra_handler=None, **kwargs):
"""
this function is intended to be used by extensions that want to load models with
some extra classes in them that the usual unpickler would find suspicious.
@@ -345,7 +345,7 @@ class PeRFlowScheduler(SchedulerMixin, ConfigMixin):
sample: torch.FloatTensor,
timestep: Union[float, torch.FloatTensor],
noise: Optional[torch.FloatTensor] = None,
) -> torch.FloatTensor:
) -> torch.Tensor:
if noise is None:
noise = torch.randn_like(sample)
if not isinstance(timestep, torch.Tensor):
@@ -367,7 +367,7 @@ class PeRFlowScheduler(SchedulerMixin, ConfigMixin):
original_samples: torch.FloatTensor,
noise: torch.FloatTensor,
timesteps: torch.IntTensor,
) -> torch.FloatTensor:
) -> torch.Tensor:
# Make sure alphas_cumprod and timestep have same device and dtype as original_samples
alphas_cumprod = self.alphas_cumprod.to(device=original_samples.device, dtype=original_samples.dtype)
timesteps = timesteps.to(original_samples.device) - 1 # indexing from 0
+2 -2
View File
@@ -71,14 +71,14 @@ class Script:
"""this function should return the title of the script. This is what will be displayed in the dropdown menu."""
raise NotImplementedError
def ui(self, is_img2img) -> list[IOComponent]:
def ui(self, is_img2img) -> list[IOComponent]: # ty: ignore
"""this function should create gradio UI elements. See https://gradio.app/docs/#components
The return value should be an array of all components that are used in processing.
Values of those returned components will be passed to run() and process() functions.
"""
pass # pylint: disable=unnecessary-pass
def show(self, is_img2img) -> bool | AlwaysVisible: # pylint: disable=unused-argument
def show(self, is_img2img) -> bool | AlwaysVisible: # pylint: disable=unused-argument # ty: ignore
"""
is_img2img is True if this function is called for the img2img interface, and False otherwise
This function should return:
+3 -2
View File
@@ -36,8 +36,9 @@ def hijack_set_module_tensor(
if tensor_name in module._buffers: # pylint: disable=protected-access
module._buffers[tensor_name] = value.to(device, old_value.dtype) # pylint: disable=protected-access
elif value is not None or not devices.same_device(device, module._parameters[tensor_name].device): # pylint: disable=protected-access
param_cls = type(module._parameters[tensor_name]) # pylint: disable=protected-access
module._parameters[tensor_name] = param_cls(value, requires_grad=old_value.requires_grad).to(device, old_value.dtype) # pylint: disable=protected-access
param = module._parameters[tensor_name] # pylint: disable=protected-access
param_cls = type(param) if param is not None else torch.nn.Parameter
module._parameters[tensor_name] = param_cls(value, requires_grad=old_value.requires_grad).to(device, old_value.dtype) # pylint: disable=protected-access
t1 = time.time()
tensor_to_timer += (t1 - t0)
+2 -2
View File
@@ -19,7 +19,7 @@ def find_split_size(original_size: int, slice_block_size: int, slice_rate: int =
# Find slice sizes for SDPA
@cache
def find_sdpa_slice_sizes(query_shape: tuple[int], key_shape: tuple[int], query_element_size: int, slice_rate: int = 2, trigger_rate: int = 3) -> tuple[bool, int]:
def find_sdpa_slice_sizes(query_shape: tuple[int], key_shape: tuple[int], query_element_size: int, slice_rate: int = 2, trigger_rate: int = 3) -> tuple[bool, bool, bool, int, int, int]:
batch_size, attn_heads, query_len, _ = query_shape
_, _, key_len, _ = key_shape
@@ -53,7 +53,7 @@ def find_sdpa_slice_sizes(query_shape: tuple[int], key_shape: tuple[int], query_
if devices.sdpa_pre_dyanmic_atten is None:
devices.sdpa_pre_dyanmic_atten = torch.nn.functional.scaled_dot_product_attention
@wraps(devices.sdpa_pre_dyanmic_atten)
def dynamic_scaled_dot_product_attention(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.FloatTensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.FloatTensor:
def dynamic_scaled_dot_product_attention(query: torch.FloatTensor, key: torch.FloatTensor, value: torch.FloatTensor, attn_mask: torch.FloatTensor | None = None, dropout_p: float = 0.0, is_causal: bool = False, scale: float | None = None, enable_gqa: bool = False, **kwargs) -> torch.Tensor:
is_unsqueezed = False
if query.dim() == 3:
query = query.unsqueeze(0)
+6 -6
View File
@@ -253,7 +253,7 @@ class AutoencoderSmall(ModelMixin, ConfigMixin, FromOriginalModelMixin):
@apply_forward_hook
def encode(
self, x: torch.FloatTensor, return_dict: bool = True
) -> AutoencoderKLOutput | tuple[DiagonalGaussianDistribution]:
) -> AutoencoderKLOutput | tuple[DiagonalGaussianDistribution] | tuple[torch.Tensor]:
"""
Encode a batch of images into latents.
@@ -283,7 +283,7 @@ class AutoencoderSmall(ModelMixin, ConfigMixin, FromOriginalModelMixin):
return AutoencoderKLOutput(latent_dist=posterior)
def _decode(self, z: torch.FloatTensor, return_dict: bool = True) -> DecoderOutput | torch.FloatTensor:
def _decode(self, z: torch.FloatTensor, return_dict: bool = True) -> DecoderOutput | tuple[torch.Tensor] | torch.Tensor:
if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
return self.tiled_decode(z, return_dict=return_dict)
@@ -298,7 +298,7 @@ class AutoencoderSmall(ModelMixin, ConfigMixin, FromOriginalModelMixin):
@apply_forward_hook
def decode(
self, z: torch.FloatTensor, return_dict: bool = True, generator=None
) -> DecoderOutput | torch.FloatTensor:
) -> DecoderOutput | tuple[torch.Tensor] | torch.Tensor:
"""
Decode a batch of images.
@@ -336,7 +336,7 @@ class AutoencoderSmall(ModelMixin, ConfigMixin, FromOriginalModelMixin):
b[:, :, :, x] = a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
return b
def tiled_encode(self, x: torch.FloatTensor, return_dict: bool = True) -> AutoencoderKLOutput:
def tiled_encode(self, x: torch.FloatTensor, return_dict: bool = True) -> AutoencoderKLOutput | tuple[DiagonalGaussianDistribution] | tuple[torch.Tensor]:
r"""Encode a batch of images using a tiled encoder.
When this option is enabled, the VAE will split the input tensor into tiles to compute encoding in several
@@ -390,7 +390,7 @@ class AutoencoderSmall(ModelMixin, ConfigMixin, FromOriginalModelMixin):
return AutoencoderKLOutput(latent_dist=posterior)
def tiled_decode(self, z: torch.FloatTensor, return_dict: bool = True) -> DecoderOutput | torch.FloatTensor:
def tiled_decode(self, z: torch.FloatTensor, return_dict: bool = True) -> DecoderOutput | tuple[torch.Tensor] | torch.Tensor:
r"""
Decode a batch of images using a tiled decoder.
@@ -444,7 +444,7 @@ class AutoencoderSmall(ModelMixin, ConfigMixin, FromOriginalModelMixin):
sample_posterior: bool = False,
return_dict: bool = True,
generator: torch.Generator | None = None,
) -> DecoderOutput | torch.FloatTensor:
) -> DecoderOutput | tuple[torch.Tensor] | torch.Tensor:
r"""
Args:
sample (`torch.FloatTensor`): Input sample.
+9 -11
View File
@@ -1,8 +1,6 @@
from typing import Any, Dict, Optional, Union
import torch
import torch.nn as nn
import numpy as np
from typing import Any, Dict, Optional, Union, List
from diffusers.models.modeling_outputs import Transformer2DModelOutput
from diffusers.utils import USE_PEFT_BACKEND, logging, scale_lora_layers, unscale_lora_layers
@@ -102,13 +100,13 @@ def teacache_lumina2_forward(
if self.enable_teacache and not should_calc:
if max_seq_len in self.cache and "previous_residual" in self.cache[max_seq_len] and self.cache[max_seq_len]["previous_residual"] is not None:
processed_hidden_states = input_to_main_loop + self.cache[max_seq_len]["previous_residual"]
processed_hidden_states = input_to_main_loop + self.cache[max_seq_len]["previous_residual"]
else:
should_calc = True
current_processing_states = input_to_main_loop
for layer in self.layers:
current_processing_states = layer(current_processing_states, attention_mask_for_main_loop_arg, joint_rotary_emb, temb)
processed_hidden_states = current_processing_states
should_calc = True
current_processing_states = input_to_main_loop
for layer in self.layers:
current_processing_states = layer(current_processing_states, attention_mask_for_main_loop_arg, joint_rotary_emb, temb)
processed_hidden_states = current_processing_states
if not (self.enable_teacache and not should_calc) :
@@ -118,9 +116,9 @@ def teacache_lumina2_forward(
if self.enable_teacache:
if max_seq_len in self.cache:
self.cache[max_seq_len]["previous_residual"] = current_processing_states - input_to_main_loop
self.cache[max_seq_len]["previous_residual"] = current_processing_states - input_to_main_loop
else:
logger.warning(f"TeaCache: Cache key {max_seq_len} not found when trying to save residual.")
logger.warning(f"TeaCache: Cache key {max_seq_len} not found when trying to save residual.")
processed_hidden_states = current_processing_states
+6 -6
View File
@@ -88,10 +88,10 @@ def teacache_mochi_forward(
if torch.is_grad_enabled() and self.gradient_checkpointing:
def create_custom_forward(module):
def custom_forward(*inputs):
return module(*inputs)
def custom_forward(*inputs):
return module(*inputs)
return custom_forward
return custom_forward
ckpt_kwargs: Dict[str, Any] = {"use_reentrant": False} if is_torch_version(">=", "1.11.0") else {}
hidden_states, encoder_hidden_states = torch.utils.checkpoint.checkpoint(
@@ -117,10 +117,10 @@ def teacache_mochi_forward(
for _i, block in enumerate(self.transformer_blocks):
if torch.is_grad_enabled() and self.gradient_checkpointing:
def create_custom_forward(module):
def custom_forward(*inputs):
return module(*inputs)
def custom_forward(*inputs):
return module(*inputs)
return custom_forward
return custom_forward
ckpt_kwargs: Dict[str, Any] = {"use_reentrant": False} if is_torch_version(">=", "1.11.0") else {}
hidden_states, encoder_hidden_states = torch.utils.checkpoint.checkpoint(
+2 -2
View File
@@ -109,7 +109,7 @@ class Pages:
self.pages.append(page)
self.size = sum(page.size for page in self.pages)
def search(self, text: str, topk: int = 10, full: bool = True) -> list[Page]:
def search(self, text: str, topk: int = 10, full: bool = True) -> list[tuple[float, Page]]:
if not text or len(text) < 2:
return []
if len(self.pages) == 0:
@@ -146,7 +146,7 @@ def get_docs_page(page_title: str) -> str:
return content
def search_html(pages: list[Page]) -> str:
def search_html(pages: list[tuple[float, Page]]) -> str:
html = ''
for score, page in pages:
if score > 0.0:
+2 -1
View File
@@ -122,7 +122,8 @@ class Component:
self.type = 'variable'
elif 'enum' in self.str:
self.type = 'enum'
self.enum = [v.name for v in self.cls]
enum_values = self.cls if self.cls is not None else ()
self.enum = [v.name for v in enum_values]
elif inspect.isclass(signature.annotation):
self.type = 'class'
elif inspect.ismodule(signature.annotation):
+10 -5
View File
@@ -248,21 +248,26 @@ def create_ui(disabled_tabs=None):
sections = []
options_count = len(shared.opts.data_labels)
for item in shared.opts.data_labels.values(): # get unique sections from all items
if len(item.section) == 2:
section_id, section_text = item.section
elif len(item.section) == 3: # compatibility item with a1111 extensions
_category, section_id, section_text = item.section
section = item.section or (None, 'Hidden')
if len(section) == 2:
section_id, section_text = section
elif len(section) == 3: # compatibility item with a1111 extensions
_category, section_id, section_text = section
item.section = section_id, section_text
else:
section_id = None
item.section = None, 'Hidden'
section_text = 'Hidden'
if (section_id, section_text) not in sections:
sections.append((section_id, section_text))
with gr.Tabs(elem_id="settings"):
quicksettings_list.clear()
for (section_id, section_text) in sections:
items = [item for item in shared.opts.data_labels.items() if item[1].section[0] == section_id] # find all items in this section
items = [
item for item in shared.opts.data_labels.items()
if item[1].section is not None and item[1].section[0] == section_id
] # find all items in this section
hidden = section_id is None or 'hidden' in section_id.lower() or 'hidden' in section_text.lower()
# log.trace(f'Settings: section="{section_id}" title="{section_text}" items={len(items)} hidden={hidden}')
if hidden:
+2
View File
@@ -70,6 +70,8 @@ class NattenAttnProcessor:
a = torch.softmax(qk, dim=-1)
hidden_states = natten.functional.na2d_av(a, v, self.kernel_size, 1) # natten2dav
hidden_states = rearrange(hidden_states, "n nh h w e -> n h w (nh e)")
if attn.to_out is None:
return hidden_states
linear_proj, dropout = attn.to_out
hidden_states = linear_proj(hidden_states)
hidden_states = dropout(hidden_states)
+1 -1
View File
@@ -35,7 +35,7 @@ def google_requirements():
# reload('pydantic', '2.11.7')
def get_size_buckets(width: int, height: int) -> str:
def get_size_buckets(width: int, height: int) -> tuple[str, str]:
aspect_ratio = width / height
closest_aspect_ratio = min(aspect_ratios_buckets.items(), key=lambda x: abs(x[1] - aspect_ratio))[0]
pixel_count = width * height
+38 -46
View File
@@ -390,58 +390,50 @@ include = [
exclude = [
"venv/",
"*.git/",
"scripts/pulid/*",
"scripts/pixelsmith/*",
"scripts/mod/*",
"scripts/layerdiffuse/*",
"scripts/lbm/*",
"scripts/daam/*",
"scripts/infiniteyou/*",
"scripts/ctrlx/*",
"scripts/consistory/*",
"scripts/freescale/*",
"scripts/instantir/*",
"scripts/softfill.py",
"scripts/custom_code.py",
"pipelines/anima",
"pipelines/boogu",
"pipelines/bria",
"pipelines/ernie",
"pipelines/f_lite",
"pipelines/flex2",
"pipelines/hidream",
"pipelines/lumina_dimmo",
"pipelines/mageflow",
"pipelines/meissonic",
"pipelines/model_stablecascade.py",
"pipelines/omnigen2",
"pipelines/sefi",
"pipelines/step1x",
"pipelines/ultraflux",
"pipelines/vibe",
"pipelines/xomni",
"pipelines/zetachroma",
"extensions-builtin/sd-extension-chainner/nodes",
"extensions-builtin/sdnq",
"modules/apg",
"modules/cfgzero",
"modules/control/proc",
"modules/framepack",
"modules/ggml",
"modules/hidiffusion",
"modules/intel",
"modules/onnx_impl",
"modules/pag",
"modules/postprocess",
"modules/control/units/*.py",
"modules/res4lyf",
"modules/schedulers/scheduler_*.py",
"modules/seedvr",
"modules/sharpfin",
"modules/teacache",
"modules/face/*.py",
"modules/sub_quadratic_attention.py",
"pipelines/**/*.py",
"scripts/**/*.py",
]
[tool.ty.rules]
invalid-method-override = "ignore"
invalid-argument-type = "ignore"
unresolved-import = "ignore"
unresolved-attribute = "ignore"
invalid-assignment = "ignore"
unsupported-operator = "ignore"
no-matching-overload = "ignore"
unsupported-base = "ignore"
possibly-missing-attribute = "ignore"
invalid-parameter-default = "ignore"
call-non-callable = "ignore"
not-subscriptable = "ignore"
possibly-missing-submodule = "ignore"
missing-argument = "ignore"
unknown-argument = "ignore"
invalid-attribute-access = "ignore"
call-top-callable = "ignore"
deprecated = "ignore"
invalid-argument-type = "ignore"
invalid-assignment = "ignore"
invalid-attribute-access = "ignore"
invalid-method-override = "ignore"
invalid-parameter-default = "ignore"
missing-argument = "ignore"
no-matching-overload = "ignore"
not-subscriptable = "ignore"
possibly-missing-attribute = "ignore"
possibly-missing-submodule = "ignore"
unknown-argument = "ignore"
unresolved-attribute = "ignore"
unresolved-import = "ignore"
unsupported-base = "ignore"
unsupported-operator = "ignore"
unused-type-ignore-comment = "ignore"
[tool.codespell]
skip = "./venv, ./.git, __pycache__, ./tmp, ./extensions-builtin, ./configs, ./models, ./outputs, *.txt, *.json, *.yaml, *.map, *.mjs, *.log"
+1 -1
View File
@@ -75,7 +75,7 @@ class ConsiStoryScript(scripts_manager.Script):
shared.sd_model = sd_models.switch_pipe(cs.ConsistoryExtendAttnSDXLPipeline, shared.sd_model)
shared.sd_model.unet = cs.ConsistorySDXLUNet2DConditionModel.from_config(shared.sd_model.unet.config)
shared.sd_model.unet.load_state_dict(state_dict) # now load it into new class
shared.sd_model.unet.to(dtype=devices.dtype) # ty: ignore
shared.sd_model.unet.to(dtype=devices.dtype)
state_dict = None
# sd_models.set_diffuser_options(shared.sd_model)
sd_models.move_model(shared.sd_model, devices.device)
+1 -1
View File
@@ -90,7 +90,7 @@ def set_template(
pass
elif options.processor is None:
log.error('Prompt enhance: image not supported by model')
return prompt # Return original text part if image cannot be processed
return prompt if prompt is not None else '' # Return original text part if image cannot be processed
if has_image:
chat_template = get_image_template(system, prompt, options, nsfw, has_system, has_prompt, has_processor, is_video, image)