diff --git a/cli/civitai-search.py b/cli/civitai-search.py new file mode 100644 index 000000000..e7600fd06 --- /dev/null +++ b/cli/civitai-search.py @@ -0,0 +1,191 @@ +import os +import sys +import json +import time +import logging +import bs4 + + +debug = False +logging.basicConfig(level = logging.INFO, format = '%(asctime)s %(levelname)s: %(message)s') +log = logging.getLogger(__name__) + + +class ModelImage(object): + def __init__(self, dct: dict): + if isinstance(dct, str): + dct = json.loads(dct) + self.dct: dict = dct + self.id: int = dct.get('id', 0) + self.url: str = dct.get('url', '') + self.width: int = dct.get('width', 0) + self.height: int = dct.get('height', 0) + self.type: str = dct.get('type', 'Unknown') + + def __str__(self): + return f'ModelImage(id={self.id} url="{self.url}" width={self.width} height={self.height} type="{self.type}")' + +class ModelFile(object): + def __init__(self, dct: dict): + if isinstance(dct, str): + dct = json.loads(dct) + self.dct: dict = dct + self.id: int = dct.get('id', 0) + self.size: int = int(1024 * dct.get('sizeKB', 0)) + self.name: str = dct.get('name', 'Unknown') + self.type: str = dct.get('type', 'Unknown') + self.hashes: list[str] = dct.get('hashes', {}).values() + self.url: str = dct.get('downloadUrl', '') + + def __str__(self): + return f'ModelFile(id={self.id} name="{self.name}" size={self.size} type="{self.type}" url="{self.url}")' + + +class ModelVersion(object): + def __init__(self, dct: dict): + if isinstance(dct, str): + dct = json.loads(dct) + self.dct = dct + self.id = dct.get('id', 0) + self.name = dct.get('name', 'Unknown') + self.base = dct.get('baseModel', 'Unknown') + self.mtime = dct.get('publishedAt', '') + self.downloads = dct.get('stats', {}).get('downloadCount', 0) + self.availability = dct.get('availability', 'Unknown') + self.html = dct.get('description', '') or '' + self.desc = bs4.BeautifulSoup(self.html, features="html.parser").get_text() + self.files = [ModelFile(f) for f in dct.get('files', [])] + self.images = [ModelImage(i) for i in dct.get('images', [])] + + def __str__(self): + return f'ModelVersion(id={self.id} name="{self.name}" base="{self.base}" mtime="{self.mtime}" downloads={self.downloads} availability={self.availability} desc="{self.desc[:30]}...")' + + +class Model(object): + def __init__(self, dct: dict): + if isinstance(dct, str): + dct = json.loads(dct) + self.id = dct.get('id', 0) + self.dct = dct + self.url = f'https://civitai.com/models/{self.id}' + self.type = dct.get('type', 'Unknown') + self.name = dct.get('name', 'Unknown') + self.html = dct.get('description', '') + self.desc = bs4.BeautifulSoup(self.html, features="html.parser").get_text() + self.tags = dct.get('tags', []) + self.nsfw = dct.get('nsfw', False) + self.level = dct.get('nsfwLevel', 0) + self.availability = dct.get('availability', 'Unknown') + self.downloads = dct.get('stats', {}).get('downloadCount', 0) + self.creator = dct.get('creator', {}).get('username', 'Unknown') + self.versions = [ModelVersion(v) for v in dct.get('modelVersions', [])] + + def __str__(self): + return f'Model(id={self.id} type={self.type} name="{self.name}" versions={len(self.versions)} nsfw={self.nsfw}/{self.level} downloads={self.downloads} author="{self.creator}" tags={self.tags} desc="{self.desc[:30]}...")' + + +def search_civitai( + query:str, + tag:str = '', # optional:tag name + 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 + limit:int = 0, + base:list[str] = [], # list + token:str = None, + exact:bool = True, +): + import requests + from urllib.parse import urlencode + + if len(query) == 0: + log.error('CivitAI: empty query') + return [] + + t0 = time.time() + dct = { 'query': query } + if len(tag) > 0: + dct['tag'] = tag + if nsfw is not None: + dct['nsfw'] = 'true' if nsfw else 'false' + if limit > 0: + dct['limit'] = limit + if len(types) > 0: + dct['types'] = types + if len(sort) > 0: + dct['sort'] = sort + if len(period) > 0: + dct['period'] = period + if len(base) > 0: + dct['baseModels'] = ','.join(base) + encoded = urlencode(dct) + + headers = {} + if token is None: + token = os.environ.get('CIVITAI_TOKEN', None) + if token is not None and len(token) > 0: + headers['Authorization'] = f'Bearer {token}' + + url = 'https://civitai.com/api/v1/models' + uri = f'{url}?{encoded}' + log.info(f'CivitAI request: uri="{uri}" dct={dct} token={token is not None}') + result = requests.get(uri, headers=headers, timeout=60) + + if result.status_code != 200: + log.error(f'CivitAI: code={result.status_code} reason={result.reason} uri={result.url}') + return [] + + models: list[Model] = [] + exact_models: list[Model] = [] + items = result.json().get('items', []) + for item in items: + models.append(Model(item)) + + if exact: + for model in models: + model_names = [model.name.lower()] + version_names = [v.name.lower() for v in model.versions] + file_names = [f.name.lower() for v in model.versions for f in v.files] + if any([query.lower() in name for name in model_names + version_names + file_names]): + exact_models.append(model) + + t1 = time.time() + log.info(f'CivitAI result: code={result.status_code} exact={len(exact_models)} total={len(models)} time={t1-t0:.2f}') + return exact_models if len(exact_models) > 0 else models + + +def print_models(models: list[Model]): + if debug: + from rich import print as dbg + else: + dbg = lambda *args, **kwargs: None # pylint: disable=unnecessary-lambda-assignment + for model in models: + log.info(f' {model}') + dbg('Model', model.dct) + for version in model.versions: + log.info(f' {version}') + dbg('ModelVersion', version.dct) + for file in version.files: + log.info(f' {file}') + dbg('ModelFile', file.dct) + for image in version.images: + log.info(f' {image}') + dbg('ModelImage', image.dct) + + +if __name__ == "__main__": + sys.argv.pop(0) + txt = ' '.join(sys.argv) + res = search_civitai( + query=txt, + # tag = '', + # types = '', + # sort = 'Most Downloaded', + # period = 'Year', + # nsfw = True, + # base = [], + # exact= True, + # limit=100, + ) + print_models(res) diff --git a/modules/models_civitai.py b/modules/models_civitai.py new file mode 100644 index 000000000..9b2f5420e --- /dev/null +++ b/modules/models_civitai.py @@ -0,0 +1,293 @@ +import os +import re +import time +import json +import gradio as gr +from modules.shared import log, opts, req, readfile, max_workers + + +data = [] +selected_model = None +update_data = [] + + +class CivitModel: + def __init__(self, name, fn, sha = None, meta = {}): + self.name = name + self.id = meta.get('id', 0) + self.fn = fn + self.sha = sha + self.meta = meta + self.versions = 0 + self.vername = '' + self.latest = '' + self.latest_hashes = [] + self.latest_name = '' + self.url = None + self.status = 'Not found' + def array(self): + return [self.id, self.fn, self.name, self.versions, self.vername, self.latest, self.status] + + +def civit_update_metadata(): + log.debug('CivitAI update metadata: models') + from modules import ui_extra_networks, modelloader + res = [] + pages = ui_extra_networks.get_pages('Model') + if len(pages) == 0: + return 'CivitAI update metadata: no models found' + page: ui_extra_networks.ExtraNetworksPage = pages[0] + table_data = [] + update_data.clear() + all_hashes = [(item.get('hash', None) or 'XXXXXXXX').upper()[:8] for item in page.list_items()] + for item in page.list_items(): + model = CivitModel(name=item['name'], fn=item['filename'], sha=item.get('hash', None), meta=item.get('metadata', {})) + if model.sha is None or len(model.sha) == 0: + res.append(f'CivitAI skip search: name="{model.name}" hash=None') + else: + r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{model.sha}') + res.append(f'CivitAI search: name="{model.name}" hash={model.sha} status={r.status_code}') + if r.status_code == 200: + d = r.json() + model.id = d['modelId'] + modelloader.download_civit_meta(model.fn, model.id) + fn = os.path.splitext(item['filename'])[0] + '.json' + model.meta = readfile(fn, silent=True) + model.name = model.meta.get('name', model.name) + model.versions = len(model.meta.get('modelVersions', [])) + versions = model.meta.get('modelVersions', []) + if len(versions) > 0: + model.latest = versions[0].get('name', '') + model.latest_hashes.clear() + for v in versions[0].get('files', []): + for h in v.get('hashes', {}).values(): + model.latest_hashes.append(h[:8].upper()) + for ver in versions: + for f in ver.get('files', []): + for h in f.get('hashes', {}).values(): + if h[:8].upper() == model.sha[:8].upper(): + model.vername = ver.get('name', '') + model.url = f.get('downloadUrl', None) + model.latest_name = f.get('name', '') + if model.vername == model.latest: + model.status = 'Latest' + elif any(map(lambda v: v in model.latest_hashes, all_hashes)): # pylint: disable=cell-var-from-loop # noqa: C417 + model.status = 'Downloaded' + else: + model.status = 'Available' + break + log.debug(res[-1]) + update_data.append(model) + table_data.append(model.array()) + yield gr.update(value=table_data), '
'.join([r for r in res if len(r) > 0]) + return '
'.join([r for r in res if len(r) > 0]) + +def civit_update_select(evt: gr.SelectData, in_data): + global selected_model # pylint: disable=global-statement + try: + selected_model = next([m for m in update_data if m.fn == in_data[evt.index[0]][1]]) + except Exception: + selected_model = None + if selected_model is None or selected_model.url is None or selected_model.status != 'Available': + return [gr.update(value='Model update not available'), gr.update(visible=False)] + else: + return [gr.update(), gr.update(visible=True)] + +def civit_update_download(): + if selected_model is None or selected_model.url is None or selected_model.status != 'Available': + return 'Model update not available' + if selected_model.latest_name is None or len(selected_model.latest_name) == 0: + model_name = f'{selected_model.name} {selected_model.latest}.safetensors' + else: + model_name = selected_model.latest_name + return civit_download_model(selected_model.url, model_name, model_path='', model_type='Model') + + +def civit_search_model(name, tag, model_type): + # types = 'LORA' if model_type == 'LoRA' else 'Checkpoint' + url = 'https://civitai.com/api/v1/models?limit=25&Sort=Newest' + if model_type == 'Model': + url += '&types=Checkpoint' + elif model_type == 'LoRA': + url += '&types=LORA&types=DoRA&types=LoCon' + elif model_type == 'Embedding': + url += '&types=TextualInversion' + elif model_type == 'VAE': + url += '&types=VAE' + if name is not None and len(name) > 0: + url += f'&query={name}' + if tag is not None and len(tag) > 0: + url += f'&tag={tag}' + r = req(url) + log.debug(f'CivitAI search: type={model_type} name="{name}" tag={tag or "none"} url="{url}" status={r.status_code}') + if r.status_code != 200: + log.warning(f'CivitAI search: name="{name}" tag={tag} status={r.status_code}') + return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None) + try: + body = r.json() + except Exception as e: + log.error(f'CivitAI search: name="{name}" tag={tag} {e}') + return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None) + global data # pylint: disable=global-statement + data = body.get('items', []) + data1 = [] + for model in data: + found = 0 + if model_type == 'LoRA' and model['type'].lower() in ['lora', 'locon', 'dora', 'lycoris']: + found += 1 + elif model_type == 'Embedding' and model['type'].lower() in ['textualinversion', 'embedding']: + found += 1 + elif model_type == 'Model' and model['type'].lower() in ['checkpoint']: + found += 1 + elif model_type == 'VAE' and model['type'].lower() in ['vae']: + found += 1 + elif model_type == 'Other': + found += 1 + if found > 0: + data1.append([ + model['id'], + model['name'], + ', '.join(model['tags']), + model['stats']['downloadCount'], + model['stats']['rating'] + ]) + res = f'Search result: name={name} tag={tag or "none"} type={model_type} models={len(data1)}' + return res, gr.update(visible=len(data1) > 0, value=data1 if len(data1) > 0 else []), gr.update(visible=False, value=None), gr.update(visible=False, value=None) + + +def civit_select1(evt: gr.SelectData, in_data): + model_id = in_data[evt.index[0]][0] + data2 = [] + preview_img = None + for model in data: + if model['id'] == model_id: + for d in model['modelVersions']: + try: + if d.get('images') is not None and len(d['images']) > 0 and len(d['images'][0]['url']) > 0: + preview_img = d['images'][0]['url'] + data2.append([d.get('id', None), d.get('modelId', None) or model_id, d.get('name', None), d.get('baseModel', None), d.get('createdAt', None) or d.get('publishedAt', None)]) + except Exception as e: + log.error(f'CivitAI select: model="{in_data[evt.index[0]]}" {e}') + log.error(f'CivitAI version data={type(d)}: {d}') + log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" versions={len(data2)}') + return data2, None, preview_img + + +def civit_select2(evt: gr.SelectData, in_data): + variant_id = in_data[evt.index[0]][0] + model_id = in_data[evt.index[0]][1] + data3 = [] + for model in data: + if model['id'] == model_id: + for variant in model['modelVersions']: + if variant['id'] == variant_id: + for f in variant['files']: + try: + if os.path.splitext(f['name'])[1].lower() in ['.safetensors', '.ckpt', '.pt', '.pth', '.bin']: + data3.append([f['name'], round(f['sizeKB']), json.dumps(f['metadata']), f['downloadUrl']]) + except Exception: + pass + log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" files={len(data3)}') + return data3 + + +def civit_select3(evt: gr.SelectData, in_data): + log.debug(f'CivitAI select: variant={in_data[evt.index[0]]}') + return in_data[evt.index[0]][3], in_data[evt.index[0]][0], gr.update(interactive=True) + + +def civit_download_model(model_url: str, model_name: str, model_path: str, model_type: str, token: str = None): + if model_url is None or len(model_url) == 0: + return 'No model selected' + try: + from modules.modelloader import download_civit_model + res = download_civit_model(model_url, model_name, model_path, model_type, token=token) + except Exception as e: + res = f"CivitAI model downloaded error: model={model_url} {e}" + log.error(res) + return res + from modules.sd_models import list_models # pylint: disable=W0621 + list_models() + return res + + +def atomic_civit_search_metadata(item, res, rehash): + from modules.modelloader import download_civit_preview, download_civit_meta + if item is None: + return + meta = os.path.splitext(item['filename'])[0] + '.json' + has_meta = os.path.isfile(meta) and os.stat(meta).st_size > 0 + if ('card-no-preview.png' in item['preview'] or not has_meta) and os.path.isfile(item['filename']): + sha = item.get('hash', None) + found = False + if sha is not None and len(sha) > 0: + r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}') + log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}') + if r.status_code == 200: + d = r.json() + res.append(download_civit_meta(item['filename'], d['modelId'])) + if d.get('images') is not None: + for i in d['images']: + preview_url = i['url'] + img_res = download_civit_preview(item['filename'], preview_url) + res.append(img_res) + if 'error' not in img_res: + found = True + break + if not found and rehash and os.stat(item['filename']).st_size < (1024 * 1024 * 1024): + from modules import hashes + sha = hashes.calculate_sha256(item['filename'], quiet=True)[:10] + r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}') + log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}') + if r.status_code == 200: + d = r.json() + res.append(download_civit_meta(item['filename'], d['modelId'])) + if d.get('images') is not None: + for i in d['images']: + preview_url = i['url'] + img_res = download_civit_preview(item['filename'], preview_url) + res.append(img_res) + if 'error' not in img_res: + found = True + break + + +def civit_search_metadata(rehash, title): + log.debug(f'CivitAI search metadata: type={title if type(title) == str else "all"}') + from modules.ui_extra_networks import get_pages + res = [] + scanned, skipped = 0, 0 + t0 = time.time() + candidates = [] + re_skip = [r.strip() for r in opts.extra_networks_scan_skip.split(',') if len(r.strip()) > 0] + log.debug(f'CivitAI search metadata: skip={re_skip}') + for page in get_pages(): + if type(title) == str: + if page.title != title: + continue + if page.name == 'style': + continue + for item in page.list_items(): + if item is None: + continue + if any(re.search(re_str, item.get('name', '') + item.get('filename', '')) for re_str in re_skip): + skipped += 1 + continue + scanned += 1 + candidates.append(item) + # atomic_civit_search_metadata(item, res, rehash) + import concurrent + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + for fn in candidates: + executor.submit(atomic_civit_search_metadata, fn, res, rehash) + atomic_civit_search_metadata(None, res, rehash) + t1 = time.time() + log.debug(f'CivitAI search metadata: scanned={scanned} skipped={skipped} time={t1-t0:.2f}') + txt = '
'.join([r for r in res if len(r) > 0]) + return txt + + +def civitai_update_token(token): + log.debug('CivitAI update token') + opts.civitai_token = token + opts.save() diff --git a/modules/models_hf.py b/modules/models_hf.py new file mode 100644 index 000000000..801fafc04 --- /dev/null +++ b/modules/models_hf.py @@ -0,0 +1,42 @@ +import os +import gradio as gr +from modules.shared import log, opts + + +def hf_init(): + os.environ.setdefault('HF_HUB_DISABLE_EXPERIMENTAL_WARNING', '1') + os.environ.setdefault('HF_HUB_DISABLE_SYMLINKS_WARNING', '1') + os.environ.setdefault('HF_HUB_DISABLE_IMPLICIT_TOKEN', '1') + os.environ.setdefault('HUGGINGFACE_HUB_VERBOSITY', 'warning') + + +def hf_search(keyword): + hf_init() + import huggingface_hub as hf + hf_api = hf.HfApi() + models = hf_api.list_models(model_name=keyword, full=True, library="diffusers", limit=50, sort="downloads", direction=-1) + data = [] + for model in models: + tags = [t for t in model.tags if not t.startswith('diffusers') and not t.startswith('license') and not t.startswith('arxiv') and len(t) > 2] + data.append([model.id, model.pipeline_tag, tags, model.downloads, model.lastModified, f'https://huggingface.co/{model.id}']) + return data + + +def hf_select(evt: gr.SelectData, data): + return data[evt.index[0]][0] + + +def hf_download_model(hub_id: str, token, variant, revision, mirror, custom_pipeline): + hf_init() + from modules.modelloader import download_diffusers_model + download_diffusers_model(hub_id, cache_dir=opts.diffusers_dir, token=token, variant=variant, revision=revision, mirror=mirror, custom_pipeline=custom_pipeline) + from modules.sd_models import list_models # pylint: disable=W0621 + list_models() + log.info(f'Diffuser model downloaded: model="{hub_id}"') + return f'Diffuser model downloaded: model="{hub_id}"' + + +def hf_update_token(token): + log.debug('Huggingface update token') + opts.huggingface_token = token + opts.save() diff --git a/modules/prompt_parser.py b/modules/prompt_parser.py index 92389c1fd..cb8789aea 100644 --- a/modules/prompt_parser.py +++ b/modules/prompt_parser.py @@ -334,7 +334,7 @@ def parse_prompt_attention(text): whitespace = '' else: re_attention = re_attention_v2 - if native and opts.sd_textencder_linebreak: + if opts.sd_textencder_linebreak: text = text.replace('\n', ' BREAK ') else: text = text.replace('\n', ' ') diff --git a/modules/sd_checkpoint.py b/modules/sd_checkpoint.py index 1c9bfe117..b54eb701f 100644 --- a/modules/sd_checkpoint.py +++ b/modules/sd_checkpoint.py @@ -127,7 +127,7 @@ def list_models(): global checkpoints_list # pylint: disable=global-statement checkpoints_list.clear() checkpoint_aliases.clear() - ext_filter = [".safetensors"] if shared.opts.sd_disable_ckpt or shared.native else [".ckpt", ".safetensors"] + ext_filter = [".safetensors"] model_list = list(modelloader.load_models(model_path=model_path, model_url=None, command_path=shared.opts.ckpt_dir, ext_filter=ext_filter, download_name=None, ext_blacklist=[".vae.ckpt", ".vae.safetensors"])) safetensors_list = [] for filename in sorted(model_list, key=str.lower): @@ -136,21 +136,16 @@ def list_models(): if checkpoint_info.name is not None: checkpoint_info.register() diffusers_list = [] - if shared.native: - for repo in modelloader.load_diffusers_models(clear=True): - checkpoint_info = CheckpointInfo(repo['name'], sha=repo['hash']) - diffusers_list.append(checkpoint_info) - if checkpoint_info.name is not None: - checkpoint_info.register() + for repo in modelloader.load_diffusers_models(clear=True): + checkpoint_info = CheckpointInfo(repo['name'], sha=repo['hash']) + diffusers_list.append(checkpoint_info) + if checkpoint_info.name is not None: + checkpoint_info.register() if shared.cmd_opts.ckpt is not None: - if not os.path.exists(shared.cmd_opts.ckpt) and not shared.native: - if shared.cmd_opts.ckpt.lower() != "none": - shared.log.warning(f'Load model: path="{shared.cmd_opts.ckpt}" not found') - else: - checkpoint_info = CheckpointInfo(shared.cmd_opts.ckpt) - if checkpoint_info.name is not None: - checkpoint_info.register() - shared.opts.data['sd_model_checkpoint'] = checkpoint_info.title + checkpoint_info = CheckpointInfo(shared.cmd_opts.ckpt) + if checkpoint_info.name is not None: + checkpoint_info.register() + shared.opts.data['sd_model_checkpoint'] = checkpoint_info.title elif shared.cmd_opts.ckpt != shared.default_sd_model_file and shared.cmd_opts.ckpt is not None: shared.log.warning(f'Load model: path="{shared.cmd_opts.ckpt}" not found') shared.log.info(f'Available Models: safetensors="{shared.opts.ckpt_dir}":{len(safetensors_list)} diffusers="{shared.opts.diffusers_dir}":{len(diffusers_list)} items={len(checkpoints_list)} time={time.time()-t0:.2f}') diff --git a/modules/shared.py b/modules/shared.py index 6fed4d343..f9edba6a4 100644 --- a/modules/shared.py +++ b/modules/shared.py @@ -264,7 +264,7 @@ options_templates.update(options_section(('cuda', "Compute Settings"), { "diffusers_generator_device": OptionInfo("GPU", "Generator device", gr.Radio, {"choices": ["GPU", "CPU", "Unset"]}), "cross_attention_sep": OptionInfo("

Cross Attention

", "", gr.HTML), - "cross_attention_optimization": OptionInfo(startup_cross_attention, "Attention optimization method", gr.Radio, lambda: {"choices": shared_items.list_crossattention(native)}), + "cross_attention_optimization": OptionInfo(startup_cross_attention, "Attention optimization method", gr.Radio, lambda: {"choices": shared_items.list_crossattention()}), "sdp_options": OptionInfo(startup_sdp_options, "SDP options", gr.CheckboxGroup, {"choices": ['Flash attention', 'Memory attention', 'Math attention', 'Dynamic attention', 'CK Flash attention', 'Sage attention']}), "xformers_options": OptionInfo(['Flash attention'], "xFormers options", gr.CheckboxGroup, {"choices": ['Flash attention'] }), "dynamic_attention_slice_rate": OptionInfo(0.5, "Dynamic Attention slicing rate in GB", gr.Slider, {"minimum": 0.01, "maximum": max(gpu_memory,4), "step": 0.01}), diff --git a/modules/shared_items.py b/modules/shared_items.py index 483186905..74b087739 100644 --- a/modules/shared_items.py +++ b/modules/shared_items.py @@ -105,26 +105,15 @@ def refresh_te_list(): modules.model_te.refresh_te_list() -def list_crossattention(native:bool=True): - if native: - return [ - "Disabled", - "Scaled-Dot-Product", - "xFormers", - "Batch matrix-matrix", - "Split attention", - "Dynamic Attention BMM" - ] - else: - return [ - "Disabled", - "Scaled-Dot-Product", - "xFormers", - "Doggettx's", - "InvokeAI's", - "Sub-quadratic", - "Split attention" - ] +def list_crossattention(): + return [ + "Disabled", + "Scaled-Dot-Product", + "xFormers", + "Batch matrix-matrix", + "Split attention", + "Dynamic Attention BMM" + ] def get_pipelines(): from installer import log diff --git a/modules/textual_inversion.py b/modules/textual_inversion.py index f6b1c558d..4d7b76a77 100644 --- a/modules/textual_inversion.py +++ b/modules/textual_inversion.py @@ -13,7 +13,7 @@ supported_models = ['ldm', 'sd', 'sdxl'] def list_embeddings(*dirs): - is_ext = extension_filter(['.SAFETENSORS', '.PT' ] + ( ['.PNG', '.WEBP', '.JXL', '.AVIF', '.BIN' ] if not shared.native else [] )) + is_ext = extension_filter(['.SAFETENSORS', '.PT' ]) is_not_preview = lambda fp: not next(iter(os.path.splitext(fp))).upper().endswith('.PREVIEW') # pylint: disable=unnecessary-lambda-assignment return list(filter(lambda fp: is_ext(fp) and is_not_preview(fp) and os.stat(fp).st_size > 0, directory_files(*dirs))) diff --git a/modules/ui_extra_networks.py b/modules/ui_extra_networks.py index 7787ea1fc..b8c777f92 100644 --- a/modules/ui_extra_networks.py +++ b/modules/ui_extra_networks.py @@ -932,9 +932,8 @@ def create_ui(container, button_parent, tabname, skip_indexing = False): return pages def ui_scan_click(title): - from modules import ui_models - if ui_models.search_metadata_civit is not None: - ui_models.search_metadata_civit(True, title) + from modules.models_civitai import civit_search_metadata + civit_search_metadata(True, title) return ui_refresh_click(title) def ui_save_click(): diff --git a/modules/ui_models.py b/modules/ui_models.py index 1a16d1b44..bb83199c7 100644 --- a/modules/ui_models.py +++ b/modules/ui_models.py @@ -1,21 +1,15 @@ import os -import re -import time import json import inspect from datetime import datetime import gradio as gr -from modules import errors, sd_models, sd_vae, extras, sd_samplers, ui_symbols, hashes +from modules import errors, sd_models, sd_vae, extras, sd_samplers, ui_symbols from modules.ui_components import ToolButton from modules.ui_common import create_refresh_button from modules.call_queue import wrap_gradio_gpu_call -from modules.shared import opts, log, req, readfile, max_workers, native -from modules.merging import merge_methods -from modules.merging.merge_utils import BETA_METHODS, TRIPLE_METHODS, interpolate -from modules.merging.merge_presets import BLOCK_WEIGHTS_PRESETS, SDXL_BLOCK_WEIGHTS_PRESETS +from modules.shared import opts, log -search_metadata_civit = None extra_ui = [] @@ -67,6 +61,10 @@ def create_ui(): ui_models_load.create_ui(models_outcome, models_file) with gr.Tab(label="Merge"): + from modules.merging import merge_methods + from modules.merging.merge_utils import BETA_METHODS, TRIPLE_METHODS, interpolate + from modules.merging.merge_presets import BLOCK_WEIGHTS_PRESETS, SDXL_BLOCK_WEIGHTS_PRESETS + def sd_model_choices(): return ['None'] + sd_models.checkpoint_titles() @@ -420,38 +418,7 @@ def create_ui(): model_list_btn.click(fn=list_models, inputs=[], outputs=[model_table, models_outcome]) with gr.Tab(label="Huggingface"): - data = [] - os.environ.setdefault('HF_HUB_DISABLE_EXPERIMENTAL_WARNING', '1') - os.environ.setdefault('HF_HUB_DISABLE_SYMLINKS_WARNING', '1') - os.environ.setdefault('HF_HUB_DISABLE_IMPLICIT_TOKEN', '1') - os.environ.setdefault('HUGGINGFACE_HUB_VERBOSITY', 'warning') - - def hf_search(keyword): - import huggingface_hub as hf - hf_api = hf.HfApi() - models = hf_api.list_models(model_name=keyword, full=True, library="diffusers", limit=50, sort="downloads", direction=-1) - data.clear() - for model in models: - tags = [t for t in model.tags if not t.startswith('diffusers') and not t.startswith('license') and not t.startswith('arxiv') and len(t) > 2] - data.append([model.id, model.pipeline_tag, tags, model.downloads, model.lastModified, f'https://huggingface.co/{model.id}']) - return data - - def hf_select(evt: gr.SelectData, data): - return data[evt.index[0]][0] - - def hf_download_model(hub_id: str, token, variant, revision, mirror, custom_pipeline): - from modules.modelloader import download_diffusers_model - download_diffusers_model(hub_id, cache_dir=opts.diffusers_dir, token=token, variant=variant, revision=revision, mirror=mirror, custom_pipeline=custom_pipeline) - from modules.sd_models import list_models # pylint: disable=W0621 - list_models() - log.info(f'Diffuser model downloaded: model="{hub_id}"') - return f'Diffuser model downloaded: model="{hub_id}"' - - def hf_update_token(token): - log.debug('Huggingface update token') - opts.huggingface_token = token - opts.save() - + from modules.models_hf import hf_search, hf_select, hf_download_model, hf_update_token with gr.Column(scale=6): with gr.Row(): gr.HTML('

 Download model from huggingface

') @@ -486,192 +453,7 @@ def create_ui(): hf_token.change(fn=hf_update_token, inputs=[hf_token], outputs=[]) with gr.Tab(label="CivitAI"): - data = [] - - def civit_search_model(name, tag, model_type): - # types = 'LORA' if model_type == 'LoRA' else 'Checkpoint' - url = 'https://civitai.com/api/v1/models?limit=25&Sort=Newest' - if model_type == 'Model': - url += '&types=Checkpoint' - elif model_type == 'LoRA': - url += '&types=LORA&types=DoRA&types=LoCon' - elif model_type == 'Embedding': - url += '&types=TextualInversion' - elif model_type == 'VAE': - url += '&types=VAE' - if name is not None and len(name) > 0: - url += f'&query={name}' - if tag is not None and len(tag) > 0: - url += f'&tag={tag}' - r = req(url) - log.debug(f'CivitAI search: type={model_type} name="{name}" tag={tag or "none"} url="{url}" status={r.status_code}') - if r.status_code != 200: - log.warning(f'CivitAI search: name="{name}" tag={tag} status={r.status_code}') - return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None) - try: - body = r.json() - except Exception as e: - log.error(f'CivitAI search: name="{name}" tag={tag} {e}') - return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None) - nonlocal data - data = body.get('items', []) - data1 = [] - for model in data: - found = 0 - if model_type == 'LoRA' and model['type'].lower() in ['lora', 'locon', 'dora', 'lycoris']: - found += 1 - elif model_type == 'Embedding' and model['type'].lower() in ['textualinversion', 'embedding']: - found += 1 - elif model_type == 'Model' and model['type'].lower() in ['checkpoint']: - found += 1 - elif model_type == 'VAE' and model['type'].lower() in ['vae']: - found += 1 - elif model_type == 'Other': - found += 1 - if found > 0: - data1.append([ - model['id'], - model['name'], - ', '.join(model['tags']), - model['stats']['downloadCount'], - model['stats']['rating'] - ]) - res = f'Search result: name={name} tag={tag or "none"} type={model_type} models={len(data1)}' - return res, gr.update(visible=len(data1) > 0, value=data1 if len(data1) > 0 else []), gr.update(visible=False, value=None), gr.update(visible=False, value=None) - - def civit_select1(evt: gr.SelectData, in_data): - model_id = in_data[evt.index[0]][0] - data2 = [] - preview_img = None - for model in data: - if model['id'] == model_id: - for d in model['modelVersions']: - try: - if d.get('images') is not None and len(d['images']) > 0 and len(d['images'][0]['url']) > 0: - preview_img = d['images'][0]['url'] - data2.append([d.get('id', None), d.get('modelId', None) or model_id, d.get('name', None), d.get('baseModel', None), d.get('createdAt', None) or d.get('publishedAt', None)]) - except Exception as e: - log.error(f'CivitAI select: model="{in_data[evt.index[0]]}" {e}') - log.error(f'CivitAI version data={type(d)}: {d}') - log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" versions={len(data2)}') - return data2, None, preview_img - - def civit_select2(evt: gr.SelectData, in_data): - variant_id = in_data[evt.index[0]][0] - model_id = in_data[evt.index[0]][1] - data3 = [] - for model in data: - if model['id'] == model_id: - for variant in model['modelVersions']: - if variant['id'] == variant_id: - for f in variant['files']: - try: - if os.path.splitext(f['name'])[1].lower() in ['.safetensors', '.ckpt', '.pt', '.pth', '.bin']: - data3.append([f['name'], round(f['sizeKB']), json.dumps(f['metadata']), f['downloadUrl']]) - except Exception: - pass - log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" files={len(data3)}') - return data3 - - def civit_select3(evt: gr.SelectData, in_data): - log.debug(f'CivitAI select: variant={in_data[evt.index[0]]}') - return in_data[evt.index[0]][3], in_data[evt.index[0]][0], gr.update(interactive=True) - - def civit_download_model(model_url: str, model_name: str, model_path: str, model_type: str, token: str = None): - if model_url is None or len(model_url) == 0: - return 'No model selected' - try: - from modules.modelloader import download_civit_model - res = download_civit_model(model_url, model_name, model_path, model_type, token=token) - except Exception as e: - res = f"CivitAI model downloaded error: model={model_url} {e}" - log.error(res) - return res - from modules.sd_models import list_models # pylint: disable=W0621 - list_models() - return res - - def atomic_civit_search_metadata(item, res, rehash): - from modules.modelloader import download_civit_preview, download_civit_meta - if item is None: - return - meta = os.path.splitext(item['filename'])[0] + '.json' - has_meta = os.path.isfile(meta) and os.stat(meta).st_size > 0 - if ('card-no-preview.png' in item['preview'] or not has_meta) and os.path.isfile(item['filename']): - sha = item.get('hash', None) - found = False - if sha is not None and len(sha) > 0: - r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}') - log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}') - if r.status_code == 200: - d = r.json() - res.append(download_civit_meta(item['filename'], d['modelId'])) - if d.get('images') is not None: - for i in d['images']: - preview_url = i['url'] - img_res = download_civit_preview(item['filename'], preview_url) - res.append(img_res) - if 'error' not in img_res: - found = True - break - if not found and rehash and os.stat(item['filename']).st_size < (1024 * 1024 * 1024): - sha = hashes.calculate_sha256(item['filename'], quiet=True)[:10] - r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}') - log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}') - if r.status_code == 200: - d = r.json() - res.append(download_civit_meta(item['filename'], d['modelId'])) - if d.get('images') is not None: - for i in d['images']: - preview_url = i['url'] - img_res = download_civit_preview(item['filename'], preview_url) - res.append(img_res) - if 'error' not in img_res: - found = True - break - - def civit_search_metadata(rehash, title): - log.debug(f'CivitAI search metadata: type={title if type(title) == str else "all"}') - from modules.ui_extra_networks import get_pages - res = [] - scanned, skipped = 0, 0 - t0 = time.time() - candidates = [] - re_skip = [r.strip() for r in opts.extra_networks_scan_skip.split(',') if len(r.strip()) > 0] - log.debug(f'CivitAI search metadata: skip={re_skip}') - for page in get_pages(): - if type(title) == str: - if page.title != title: - continue - if page.name == 'style': - continue - for item in page.list_items(): - if item is None: - continue - if any(re.search(re_str, item.get('name', '') + item.get('filename', '')) for re_str in re_skip): - skipped += 1 - continue - scanned += 1 - candidates.append(item) - # atomic_civit_search_metadata(item, res, rehash) - import concurrent - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - for fn in candidates: - executor.submit(atomic_civit_search_metadata, fn, res, rehash) - atomic_civit_search_metadata(None, res, rehash) - t1 = time.time() - log.debug(f'CivitAI search metadata: scanned={scanned} skipped={skipped} time={t1-t0:.2f}') - txt = '
'.join([r for r in res if len(r) > 0]) - return txt - - global search_metadata_civit # pylint: disable=global-statement - search_metadata_civit = civit_search_metadata - - def civitai_update_token(token): - log.debug('CivitAI update token') - opts.civitai_token = token - opts.save() - + from modules.models_civitai import civitai_update_token, civit_search_model, civit_search_metadata, civit_select1, civit_select2, civit_select3, civit_download_model with gr.Row(): gr.HTML('

 CivitAI fetch metadata

') gr.HTML('Fetches preview and metadata information for all models with missing information
Models with existing previews and information are not updated
') @@ -737,6 +519,7 @@ def create_ui(): civit_previews_btn.click(fn=civit_search_metadata, inputs=[civit_previews_rehash, civit_previews_rehash], outputs=[models_outcome]) with gr.Tab(label="Update"): + from modules.models_civitai import civit_update_metadata, civit_update_select, civit_update_download with gr.Row(): gr.HTML('

 Scan CivitAI for information on latest available model versions

') with gr.Row(): @@ -753,107 +536,12 @@ def create_ui(): with gr.Row(): civit_update_download_btn = gr.Button(value="Download", variant='primary', visible=False) - class CivitModel: - def __init__(self, name, fn, sha = None, meta = {}): - self.name = name - self.id = meta.get('id', 0) - self.fn = fn - self.sha = sha - self.meta = meta - self.versions = 0 - self.vername = '' - self.latest = '' - self.latest_hashes = [] - self.latest_name = '' - self.url = None - self.status = 'Not found' - def array(self): - return [self.id, self.fn, self.name, self.versions, self.vername, self.latest, self.status] - - selected_model: CivitModel = None - update_data = [] - - def civit_update_metadata(): - nonlocal update_data - log.debug('CivitAI update metadata: models') - from modules import ui_extra_networks, modelloader - res = [] - pages = ui_extra_networks.get_pages('Model') - if len(pages) == 0: - return 'CivitAI update metadata: no models found' - page: ui_extra_networks.ExtraNetworksPage = pages[0] - table_data = [] - update_data.clear() - all_hashes = [(item.get('hash', None) or 'XXXXXXXX').upper()[:8] for item in page.list_items()] - for item in page.list_items(): - model = CivitModel(name=item['name'], fn=item['filename'], sha=item.get('hash', None), meta=item.get('metadata', {})) - if model.sha is None or len(model.sha) == 0: - res.append(f'CivitAI skip search: name="{model.name}" hash=None') - else: - r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{model.sha}') - res.append(f'CivitAI search: name="{model.name}" hash={model.sha} status={r.status_code}') - if r.status_code == 200: - d = r.json() - model.id = d['modelId'] - modelloader.download_civit_meta(model.fn, model.id) - fn = os.path.splitext(item['filename'])[0] + '.json' - model.meta = readfile(fn, silent=True) - model.name = model.meta.get('name', model.name) - model.versions = len(model.meta.get('modelVersions', [])) - versions = model.meta.get('modelVersions', []) - if len(versions) > 0: - model.latest = versions[0].get('name', '') - model.latest_hashes.clear() - for v in versions[0].get('files', []): - for h in v.get('hashes', {}).values(): - model.latest_hashes.append(h[:8].upper()) - for ver in versions: - for f in ver.get('files', []): - for h in f.get('hashes', {}).values(): - if h[:8].upper() == model.sha[:8].upper(): - model.vername = ver.get('name', '') - model.url = f.get('downloadUrl', None) - model.latest_name = f.get('name', '') - if model.vername == model.latest: - model.status = 'Latest' - elif any(map(lambda v: v in model.latest_hashes, all_hashes)): # pylint: disable=cell-var-from-loop # noqa: C417 - model.status = 'Downloaded' - else: - model.status = 'Available' - break - log.debug(res[-1]) - update_data.append(model) - table_data.append(model.array()) - yield gr.update(value=table_data), '
'.join([r for r in res if len(r) > 0]) - return '
'.join([r for r in res if len(r) > 0]) - - def civit_update_select(evt: gr.SelectData, in_data): - nonlocal selected_model, update_data - try: - selected_model = next([m for m in update_data if m.fn == in_data[evt.index[0]][1]]) - except Exception: - selected_model = None - if selected_model is None or selected_model.url is None or selected_model.status != 'Available': - return [gr.update(value='Model update not available'), gr.update(visible=False)] - else: - return [gr.update(), gr.update(visible=True)] - - def civit_update_download(): - if selected_model is None or selected_model.url is None or selected_model.status != 'Available': - return 'Model update not available' - if selected_model.latest_name is None or len(selected_model.latest_name) == 0: - model_name = f'{selected_model.name} {selected_model.latest}.safetensors' - else: - model_name = selected_model.latest_name - return civit_download_model(selected_model.url, model_name, model_path='', model_type='Model') - civit_update_btn.click(fn=civit_update_metadata, inputs=[], outputs=[civit_results4, models_outcome]) civit_results4.select(fn=civit_update_select, inputs=[civit_results4], outputs=[models_outcome, civit_update_download_btn]) civit_update_download_btn.click(fn=civit_update_download, inputs=[], outputs=[models_outcome]) - if native: - from modules.lora.lora_extract import create_ui as lora_extract_ui - lora_extract_ui() + from modules.lora.lora_extract import create_ui as lora_extract_ui + lora_extract_ui() for ui in extra_ui: if callable(ui):