mirror of
https://github.com/vladmandic/automatic
synced 2026-09-19 17:24:32 +02:00
lint fixes
This commit is contained in:
@@ -9,10 +9,8 @@ from modules import devices
|
||||
|
||||
class DeepDanbooruModel(nn.Module):
|
||||
def __init__(self):
|
||||
super(DeepDanbooruModel, self).__init__()
|
||||
|
||||
super().__init__()
|
||||
self.tags = []
|
||||
|
||||
self.n_Conv_0 = nn.Conv2d(kernel_size=(7, 7), in_channels=3, out_channels=64, stride=(2, 2))
|
||||
self.n_MaxPool_0 = nn.MaxPool2d(kernel_size=(3, 3), stride=(2, 2))
|
||||
self.n_Conv_1 = nn.Conv2d(kernel_size=(1, 1), in_channels=64, out_channels=256)
|
||||
@@ -673,5 +671,4 @@ class DeepDanbooruModel(nn.Module):
|
||||
|
||||
def load_state_dict(self, state_dict, **kwargs): # pylint: disable=arguments-differ,unused-argument
|
||||
self.tags = state_dict.get('tags', [])
|
||||
|
||||
super(DeepDanbooruModel, self).load_state_dict({k: v for k, v in state_dict.items() if k != 'tags'})
|
||||
super(DeepDanbooruModel, self).load_state_dict({k: v for k, v in state_dict.items() if k != 'tags'}) # pylint: disable=R1725
|
||||
|
||||
+1
-1
@@ -422,7 +422,7 @@ class FilenameGenerator:
|
||||
shared.log.error(f'Filename apply pattern: {e}')
|
||||
if replacement == NOTHING:
|
||||
continue
|
||||
elif replacement is not None:
|
||||
if replacement is not None:
|
||||
res += text + str(replacement).replace('/', '-').replace('\\', '-')
|
||||
continue
|
||||
else:
|
||||
|
||||
+6
-6
@@ -1,4 +1,5 @@
|
||||
import os
|
||||
import itertools # SBM Batch frames
|
||||
import numpy as np
|
||||
from PIL import Image, ImageOps, ImageFilter, ImageEnhance, ImageChops, UnidentifiedImageError
|
||||
import modules.scripts
|
||||
@@ -6,7 +7,6 @@ from modules import sd_samplers, shared, processing, images
|
||||
from modules.generation_parameters_copypaste import create_override_settings_dict
|
||||
from modules.ui import plaintext_to_html
|
||||
from modules.memstats import memory_stats
|
||||
import itertools # SBM Batch frames
|
||||
|
||||
|
||||
def process_batch(p, input_files, input_dir, output_dir, inpaint_mask_dir, args):
|
||||
@@ -121,17 +121,17 @@ def img2img(id_task: str, mode: int, prompt: str, negative_prompt: str, prompt_s
|
||||
|
||||
if mode == 0: # img2img
|
||||
if init_img is None:
|
||||
return
|
||||
return [], '', '', 'Error: init image not provided'
|
||||
image = init_img.convert("RGB")
|
||||
mask = None
|
||||
elif mode == 1: # img2img sketch
|
||||
if sketch is None:
|
||||
return
|
||||
return [], '', '', 'Error: sketch image not provided'
|
||||
image = sketch.convert("RGB")
|
||||
mask = None
|
||||
elif mode == 2: # inpaint
|
||||
if init_img_with_mask is None:
|
||||
return
|
||||
return [], '', '', 'Error: init image with mask not provided'
|
||||
image = init_img_with_mask["image"]
|
||||
mask = init_img_with_mask["mask"]
|
||||
alpha_mask = ImageOps.invert(image.split()[-1]).convert('L').point(lambda x: 255 if x > 0 else 0, mode='1')
|
||||
@@ -139,7 +139,7 @@ def img2img(id_task: str, mode: int, prompt: str, negative_prompt: str, prompt_s
|
||||
image = image.convert("RGB")
|
||||
elif mode == 3: # inpaint sketch
|
||||
if inpaint_color_sketch is None:
|
||||
return
|
||||
return [], '', '', 'Error: color sketch image not provided'
|
||||
image = inpaint_color_sketch
|
||||
orig = inpaint_color_sketch_orig or inpaint_color_sketch
|
||||
pred = np.any(np.array(image) != np.array(orig), axis=-1)
|
||||
@@ -150,7 +150,7 @@ def img2img(id_task: str, mode: int, prompt: str, negative_prompt: str, prompt_s
|
||||
image = image.convert("RGB")
|
||||
elif mode == 4: # inpaint upload mask
|
||||
if init_img_inpaint is None:
|
||||
return
|
||||
return [], '', '', 'Error: inpaint image not provided'
|
||||
image = init_img_inpaint
|
||||
mask = init_mask_inpaint
|
||||
else:
|
||||
|
||||
@@ -173,6 +173,7 @@ def download_civit_model_thread(model_name, model_url, model_path, model_type, p
|
||||
else:
|
||||
os.rename(temp_file, model_file)
|
||||
shared.state.end()
|
||||
return res
|
||||
|
||||
|
||||
def download_civit_model(model_url: str, model_name: str, model_path: str, model_type: str, preview):
|
||||
@@ -184,7 +185,7 @@ def download_civit_model(model_url: str, model_name: str, model_path: str, model
|
||||
|
||||
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):
|
||||
if hub_id is None or len(hub_id) == 0:
|
||||
return
|
||||
return None
|
||||
from diffusers import DiffusionPipeline
|
||||
import huggingface_hub as hf
|
||||
shared.state.begin('huggingface-download-model')
|
||||
@@ -225,7 +226,7 @@ def download_diffusers_model(hub_id: str, cache_dir: str = None, download_config
|
||||
|
||||
if pipeline_dir is None:
|
||||
shared.log.error(f"Diffusers no pipeline folder: {hub_id}")
|
||||
return
|
||||
return None
|
||||
try:
|
||||
model_info_dict = hf.model_info(hub_id).cardData if pipeline_dir is not None else None # pylint: disable=no-member # TODO Diffusers is this real error?
|
||||
except Exception:
|
||||
@@ -399,19 +400,19 @@ def download_url_to_file(url: str, dst: str):
|
||||
|
||||
file_size = None
|
||||
req = Request(url, headers={"User-Agent": "sdnext"})
|
||||
u = urlopen(req)
|
||||
u = urlopen(req) # pylint: disable=R1732
|
||||
meta = u.info()
|
||||
if hasattr(meta, 'getheaders'):
|
||||
content_length = meta.getheaders("Content-Length")
|
||||
else:
|
||||
content_length = meta.get_all("Content-Length")
|
||||
content_length = meta.get_all("Content-Length") # pylint: disable=R1732
|
||||
if content_length is not None and len(content_length) > 0:
|
||||
file_size = int(content_length[0])
|
||||
dst = os.path.expanduser(dst)
|
||||
for _seq in range(tempfile.TMP_MAX):
|
||||
tmp_dst = dst + '.' + uuid.uuid4().hex + '.partial'
|
||||
try:
|
||||
f = open(tmp_dst, 'w+b')
|
||||
f = open(tmp_dst, 'w+b') # pylint: disable=R1732
|
||||
except FileExistsError:
|
||||
continue
|
||||
break
|
||||
|
||||
@@ -51,8 +51,6 @@ def undo(key, obj, field):
|
||||
original_func = originals[key].pop(patch_key)
|
||||
setattr(obj, field, original_func)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def original(key, obj, field):
|
||||
"""Returns the original function for the patch created by the patch() function"""
|
||||
|
||||
@@ -51,7 +51,7 @@ def run_postprocessing(extras_mode, image, image_folder: List[tempfile.NamedTemp
|
||||
outpath = output_dir
|
||||
else:
|
||||
outpath = opts.outdir_samples or opts.outdir_extras_samples
|
||||
for image, name, ext in zip(image_data, image_names, image_ext):
|
||||
for image, name, ext in zip(image_data, image_names, image_ext): # pylint: disable=redefined-argument-from-local
|
||||
shared.log.debug(f'process: image={image} {args}')
|
||||
infotext = ''
|
||||
if shared.state.interrupted:
|
||||
|
||||
@@ -456,7 +456,7 @@ def process_diffusers(p: StableDiffusionProcessing, seeds, prompts, negative_pro
|
||||
# results.append(image)
|
||||
# return results
|
||||
noise_level = round(350 * p.denoising_strength)
|
||||
output_type='latent' if hasattr(shared.sd_refiner, 'vae') else 'np',
|
||||
output_type='latent' if hasattr(shared.sd_refiner, 'vae') else 'np'
|
||||
if shared.sd_refiner.__class__.__name__ == 'StableDiffusionUpscalePipeline':
|
||||
image = vae_decode(latents=image, model=shared.sd_model, full_quality=p.full_quality, output_type='pil')
|
||||
p.extra_generation_params['Noise level'] = noise_level
|
||||
|
||||
@@ -9,10 +9,10 @@ class ScriptPostprocessingForMainUI(scripts.Script):
|
||||
def title(self):
|
||||
return self.script.name
|
||||
|
||||
def show(self, is_img2img):
|
||||
def show(self, is_img2img): # pylint: disable=unused-argument
|
||||
return scripts.AlwaysVisible
|
||||
|
||||
def ui(self, is_img2img):
|
||||
def ui(self, is_img2img): # pylint: disable=unused-argument
|
||||
self.postprocessing_controls = self.script.ui()
|
||||
return self.postprocessing_controls.values()
|
||||
|
||||
|
||||
@@ -22,7 +22,7 @@ class FrozenOpenCLIPEmbedderWithCustomWords(sd_hijack_clip.FrozenCLIPEmbedderWit
|
||||
z = self.wrapped.encode_with_transformer(tokens)
|
||||
return z
|
||||
|
||||
def encode_embedding_init_text(self, init_text, nvpt):
|
||||
def encode_embedding_init_text(self, init_text, nvpt): # pylint: disable=unused-argument
|
||||
ids = tokenizer.encode(init_text)
|
||||
ids = torch.asarray([ids], device=devices.device, dtype=torch.int)
|
||||
embedded = self.wrapped.model.token_embedding.wrapped(ids).squeeze(0)
|
||||
|
||||
@@ -96,7 +96,7 @@ class CheckpointInfo:
|
||||
def calculate_shorthash(self):
|
||||
self.sha256 = hashes.sha256(self.filename, f"checkpoint/{self.name}")
|
||||
if self.sha256 is None:
|
||||
return
|
||||
return None
|
||||
self.shorthash = self.sha256[0:10]
|
||||
checkpoints_list.pop(self.title)
|
||||
self.title = f'{self.name} [{self.shorthash}]'
|
||||
@@ -309,14 +309,13 @@ def scrub_dict(dict_obj, keys):
|
||||
for key in list(dict_obj.keys()):
|
||||
if not isinstance(dict_obj, dict):
|
||||
continue
|
||||
elif key in keys:
|
||||
if key in keys:
|
||||
dict_obj.pop(key, None)
|
||||
elif isinstance(dict_obj[key], dict):
|
||||
scrub_dict(dict_obj[key], keys)
|
||||
elif isinstance(dict_obj[key], list):
|
||||
for item in dict_obj[key]:
|
||||
scrub_dict(item, keys)
|
||||
return
|
||||
|
||||
|
||||
def read_metadata_from_safetensors(filename):
|
||||
|
||||
@@ -248,7 +248,7 @@ class KDiffusionSampler:
|
||||
if self.config.options.get('scheduler', None) is not None:
|
||||
self.config.options['scheduler'] = shared.opts.data.get('schedulers_sigma', None)
|
||||
if p is None:
|
||||
return
|
||||
return {}
|
||||
self.model_wrap_cfg.mask = p.mask if hasattr(p, 'mask') else None
|
||||
self.model_wrap_cfg.nmask = p.nmask if hasattr(p, 'nmask') else None
|
||||
self.model_wrap_cfg.image_cfg_scale = getattr(p, 'image_cfg_scale', None)
|
||||
|
||||
+2
-2
@@ -233,7 +233,7 @@ def reload_vae_weights(sd_model=None, vae_file=unspecified):
|
||||
if not sd_model:
|
||||
sd_model = shared.sd_model
|
||||
if sd_model is None:
|
||||
return
|
||||
return None
|
||||
global checkpoint_info # pylint: disable=global-statement
|
||||
checkpoint_info = sd_model.sd_checkpoint_info
|
||||
checkpoint_file = checkpoint_info.filename
|
||||
@@ -242,7 +242,7 @@ def reload_vae_weights(sd_model=None, vae_file=unspecified):
|
||||
else:
|
||||
vae_source = "function-argument"
|
||||
if loaded_vae_file == vae_file:
|
||||
return
|
||||
return None
|
||||
if not getattr(sd_model, 'has_accelerate', False):
|
||||
if shared.cmd_opts.lowvram or shared.cmd_opts.medvram:
|
||||
lowvram.send_everything_to_cpu()
|
||||
|
||||
+1
-1
@@ -850,7 +850,7 @@ class Options:
|
||||
d = {k: self.data.get(k, self.data_labels.get(k).default) for k in self.data_labels.keys()}
|
||||
metadata = {
|
||||
k: {
|
||||
"is_stored": k in self.data and self.data[k] != self.data_labels[k].default,
|
||||
"is_stored": k in self.data and self.data[k] != self.data_labels[k].default, # pylint: disable=unnecessary-dict-index-lookup
|
||||
"tab_name": v.section[0]
|
||||
} for k, v in self.data_labels.items()
|
||||
}
|
||||
|
||||
+2
-2
@@ -117,7 +117,7 @@ def process_interrogate(interrogation_function, mode, ii_input_files, ii_input_d
|
||||
else:
|
||||
if not os.path.isdir(ii_input_dir):
|
||||
log.error(f"Input directory not found: {ii_input_dir}")
|
||||
return
|
||||
return [gr.update(), None]
|
||||
images = modules.shared.listfiles(ii_input_dir)
|
||||
if ii_output_dir != "":
|
||||
os.makedirs(ii_output_dir, exist_ok=True)
|
||||
@@ -959,7 +959,7 @@ def create_ui(startup_timer = None):
|
||||
for key, value, comp in zip(opts.data_labels.keys(), args, components):
|
||||
if comp == dummy_component:
|
||||
continue
|
||||
elif not opts.same_type(value, opts.data_labels[key].default):
|
||||
if not opts.same_type(value, opts.data_labels[key].default):
|
||||
log.error(f'Setting bad value: {key}={value} expecting={type(opts.data_labels[key].default).__name__}')
|
||||
continue
|
||||
if opts.set(key, value):
|
||||
|
||||
@@ -181,6 +181,7 @@ def install_extension_from_url(dirname, url, branch_name, search_text, sort_colu
|
||||
shared.log.error(f'Error installing extension: {url} {e}')
|
||||
finally:
|
||||
shutil.rmtree(tmpdir, True)
|
||||
return []
|
||||
|
||||
|
||||
def install_extension(extension_to_install, search_text, sort_column):
|
||||
|
||||
@@ -39,8 +39,8 @@ def register_page(page):
|
||||
# registers extra networks page for the UI; recommend doing it in on_before_ui() callback for extensions
|
||||
shared.extra_networks.append(page)
|
||||
allowed_dirs.clear()
|
||||
for page in shared.extra_networks:
|
||||
for folder in page.allowed_directories_for_previews():
|
||||
for pg in shared.extra_networks:
|
||||
for folder in pg.allowed_directories_for_previews():
|
||||
if folder not in allowed_dirs:
|
||||
allowed_dirs.append(os.path.abspath(folder))
|
||||
|
||||
@@ -249,6 +249,7 @@ class ExtraNetworksPage:
|
||||
return ''
|
||||
shared.log.debug(f"Extra networks: page='{self.name}' items={len(self.items)} subdirs={len(subdirs)} tab={tabname} dirs={self.allowed_directories_for_previews()} time={self.list_time}s")
|
||||
threading.Thread(target=self.create_thumb).start()
|
||||
return self.html
|
||||
|
||||
def list_items(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@@ -58,10 +58,10 @@ def pil_to_temp_file(self, img, dir: str, format="png") -> str: # pylint: disabl
|
||||
if isinstance(key, str) and isinstance(value, str):
|
||||
metadata.add_text(key, value)
|
||||
use_metadata = True
|
||||
file_obj = tempfile.NamedTemporaryFile(delete=False, suffix=".png", dir=dir)
|
||||
img.save(file_obj, pnginfo=(metadata if use_metadata else None))
|
||||
name = file_obj.name
|
||||
shared.log.debug(f'Saving temp: image="{name}"')
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".png", dir=dir) as file_obj:
|
||||
img.save(file_obj, pnginfo=(metadata if use_metadata else None))
|
||||
name = file_obj.name
|
||||
shared.log.debug(f'Saving temp: image="{name}"')
|
||||
return name
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user