Update readfile type safety

This commit is contained in:
awsr
2025-12-17 16:43:54 -08:00
parent 18bfa4b031
commit dcb7164d30
16 changed files with 33 additions and 31 deletions
+1 -1
View File
@@ -77,7 +77,7 @@ def civit_update_metadata(raw:bool=False):
model.id = d['modelId']
download_civit_meta(model.fn, model.id)
fn = os.path.splitext(item['filename'])[0] + '.json'
model.meta = readfile(fn, silent=True, dict_only=True)
model.meta = readfile(fn, silent=True, as_type="dict")
model.name = model.meta.get('name', model.name)
model.versions = len(model.meta.get('modelVersions', []))
versions = model.meta.get('modelVersions', [])
+2 -2
View File
@@ -12,7 +12,7 @@ progress_ok = True
def init_cache():
global cache_data # pylint: disable=global-statement
if cache_data is None:
cache_data = {} if not os.path.isfile(cache_filename) else shared.readfile(cache_filename, lock=True, dict_only=True)
cache_data = {} if not os.path.isfile(cache_filename) else shared.readfile(cache_filename, lock=True, as_type="dict")
def dump_cache():
@@ -22,7 +22,7 @@ def dump_cache():
def cache(subsection):
global cache_data # pylint: disable=global-statement
if cache_data is None:
cache_data = {} if not os.path.isfile(cache_filename) else shared.readfile(cache_filename, lock=True, dict_only=True)
cache_data = {} if not os.path.isfile(cache_filename) else shared.readfile(cache_filename, lock=True, as_type="dict")
s = cache_data.get(subsection, {})
cache_data[subsection] = s
return s
+8 -6
View File
@@ -12,12 +12,12 @@ locking_available = True # used by file read/write locking
@overload
def readfile(filename: str, silent: bool = False, lock: bool = False) -> dict | list: ...
def readfile(filename: str, silent: bool = False, lock: bool = False, *, as_type: Literal["dict"]) -> dict: ...
@overload
def readfile(filename: str, silent: bool = False, lock: bool = False, *, dict_only: Literal[True]) -> dict: ...
def readfile(filename: str, silent=False, lock=False, *, dict_only=False) -> dict | list:
def readfile(filename: str, silent: bool = False, lock: bool = False, *, as_type: Literal["list"]) -> list: ...
@overload
def readfile(filename: str, silent: bool = False, lock: bool = False) -> dict | list: ...
def readfile(filename: str, silent: bool = False, lock: bool = False, *, as_type="") -> dict | list:
global locking_available # pylint: disable=global-statement
data = {}
lock_file = None
@@ -58,11 +58,13 @@ def readfile(filename: str, silent=False, lock=False, *, dict_only=False) -> dic
os.remove(f"{filename}.lock")
except Exception:
locking_available = False
if isinstance(data, list) and dict_only:
if isinstance(data, list) and as_type == "dict":
data0 = data[0]
if isinstance(data0, dict):
return data0
return {}
if isinstance(data, dict) and as_type == "list":
return [data]
return data
+1 -1
View File
@@ -107,7 +107,7 @@ class NetworkOnDisk:
if self.filename is not None:
fn = os.path.splitext(self.filename)[0] + '.json'
if os.path.exists(fn):
data = shared.readfile(fn, silent=True, dict_only=True)
data = shared.readfile(fn, silent=True, as_type="dict")
return data
def get_desc(self):
+1 -1
View File
@@ -163,7 +163,7 @@ class Options():
log.debug(f'Settings: fn="{filename}" created')
self.save(filename)
return
self.data = readfile(filename, lock=True, dict_only=True)
self.data = readfile(filename, lock=True, as_type="dict")
if self.data.get('quicksettings') is not None and self.data.get('quicksettings_list') is None:
self.data['quicksettings_list'] = [i.strip() for i in self.data.get('quicksettings').split(',')]
unknown_settings = []
+1 -1
View File
@@ -410,7 +410,7 @@ def get_tokens(pipe, msg, prompt):
fn = os.path.join(fn, 'vocab.json')
else:
fn = os.path.join(fn, 'tokenizer', 'vocab.json')
token_dict = shared.readfile(fn, silent=True, dict_only=True)
token_dict = shared.readfile(fn, silent=True, as_type="dict")
added_tokens = getattr(tokenizer, 'added_tokens_decoder', {})
for k, v in added_tokens.items():
token_dict[str(v)] = k
+2 -2
View File
@@ -329,7 +329,7 @@ def select_checkpoint(op='model', sd_model_checkpoint=None):
def init_metadata():
global sd_metadata # pylint: disable=global-statement
if sd_metadata is None:
sd_metadata = shared.readfile(sd_metadata_file, lock=True, dict_only=True) if os.path.isfile(sd_metadata_file) else {}
sd_metadata = shared.readfile(sd_metadata_file, lock=True, as_type="dict") if os.path.isfile(sd_metadata_file) else {}
def extract_thumbnail(filename, data):
@@ -349,7 +349,7 @@ def extract_thumbnail(filename, data):
def read_metadata_from_safetensors(filename):
global sd_metadata # pylint: disable=global-statement
if sd_metadata is None:
sd_metadata = shared.readfile(sd_metadata_file, lock=True, dict_only=True) if os.path.isfile(sd_metadata_file) else {}
sd_metadata = shared.readfile(sd_metadata_file, lock=True, as_type="dict") if os.path.isfile(sd_metadata_file) else {}
res = sd_metadata.get(filename, None)
if res is not None:
return res
+3 -3
View File
@@ -152,7 +152,7 @@ def guess_by_diffusers(fn, current_guess):
return current_guess, None
index = os.path.join(fn, 'model_index.json')
if os.path.exists(index) and os.path.isfile(index):
index = shared.readfile(index, silent=True, dict_only=True)
index = shared.readfile(index, silent=True, as_type="dict")
name = index.get('_name_or_path', None)
if name is not None and name in exclude_by_name:
return current_guess, None
@@ -171,7 +171,7 @@ def guess_by_diffusers(fn, current_guess):
is_quant = True
break
if folder.endswith('config.json'):
quantization_config = shared.readfile(folder, silent=True, dict_only=True).get("quantization_config", None)
quantization_config = shared.readfile(folder, silent=True, as_type="dict").get("quantization_config", None)
if quantization_config is not None:
is_quant = True
break
@@ -182,7 +182,7 @@ def guess_by_diffusers(fn, current_guess):
is_quant = True
break
if f.endswith('config.json'):
quantization_config = shared.readfile(f, silent=True, dict_only=True).get("quantization_config", None)
quantization_config = shared.readfile(f, silent=True, as_type="dict").get("quantization_config", None)
if quantization_config is not None:
is_quant = True
break
+2 -2
View File
@@ -608,9 +608,9 @@ def load_sdnq_module(fn: str, module_name: str, load_method: str):
quantization_config_path = os.path.join(fn, module_name, 'quantization_config.json')
model_config_path = os.path.join(fn, module_name, 'config.json')
if os.path.exists(quantization_config_path):
quantization_config = shared.readfile(quantization_config_path, silent=True, dict_only=True)
quantization_config = shared.readfile(quantization_config_path, silent=True, as_type="dict")
elif os.path.exists(model_config_path):
quantization_config = shared.readfile(model_config_path, silent=True, dict_only=True).get("quantization_config", None)
quantization_config = shared.readfile(model_config_path, silent=True, as_type="dict").get("quantization_config", None)
if quantization_config is None:
return None, module_name, 0
model_name = os.path.join(fn, module_name)
+1 -1
View File
@@ -52,7 +52,7 @@ def load_unet(model, repo_id:str=None):
config_file = os.path.splitext(unet_dict[shared.opts.sd_unet])[0] + '.json'
if os.path.exists(config_file):
config = shared.readfile(config_file, dict_only=True)
config = shared.readfile(config_file, as_type="dict")
else:
config = None
config_file = 'default'
+3 -3
View File
@@ -128,13 +128,13 @@ def apply_vae_config(model_file, vae_file, sd_model):
def get_vae_config():
config_file = os.path.join(paths.sd_configs_path, os.path.splitext(os.path.basename(model_file))[0] + '_vae.json')
if config_file is not None and os.path.exists(config_file):
return shared.readfile(config_file, dict_only=True)
return shared.readfile(config_file, as_type="dict")
config_file = os.path.join(paths.sd_configs_path, os.path.splitext(os.path.basename(vae_file))[0] + '.json') if vae_file else None
if config_file is not None and os.path.exists(config_file):
return shared.readfile(config_file, dict_only=True)
return shared.readfile(config_file, as_type="dict")
config_file = os.path.join(paths.sd_configs_path, shared.sd_model_type, 'vae', 'config.json')
if config_file is not None and os.path.exists(config_file):
return shared.readfile(config_file, dict_only=True)
return shared.readfile(config_file, as_type="dict")
return {}
if hasattr(sd_model, 'vae') and hasattr(sd_model.vae, 'config'):
+1 -1
View File
@@ -834,7 +834,7 @@ log.info(f'Engine: backend={backend} compute={devices.backend} device={devices.g
profiler = None
prompt_styles = modules.styles.StyleDatabase(opts)
reference_models = readfile(os.path.join('html', 'reference.json'), dict_only=True) if opts.extra_network_reference_enable else {}
reference_models = readfile(os.path.join('html', 'reference.json'), as_type="dict") if opts.extra_network_reference_enable else {}
cmd_opts.disable_extension_access = (cmd_opts.share or cmd_opts.listen or (cmd_opts.server_name or False)) and not cmd_opts.insecure
devices.args = cmd_opts
devices.opts = opts
+4 -4
View File
@@ -440,7 +440,7 @@ class ExtraNetworksPage:
def update_all_previews(self, items):
global preview_map # pylint: disable=global-statement
if preview_map is None:
preview_map = shared.readfile('html/previews.json', silent=True, dict_only=True)
preview_map = shared.readfile('html/previews.json', silent=True, as_type="dict")
t0 = time.time()
reference_path = os.path.abspath(os.path.join('models', 'Reference'))
possible_paths = list(set([os.path.dirname(item['filename']) for item in items] + [reference_path]))
@@ -520,10 +520,10 @@ class ExtraNetworksPage:
t0 = time.time()
fn = os.path.splitext(path)[0] + '.json'
if not data and os.path.exists(fn):
data = shared.readfile(fn, silent=True, dict_only=True)
data = shared.readfile(fn, silent=True, as_type="dict")
fn = os.path.join(path, 'model_index.json')
if not data and os.path.exists(fn):
data = shared.readfile(fn, silent=True, dict_only=True)
data = shared.readfile(fn, silent=True, as_type="dict")
t1 = time.time()
self.info_time += t1-t0
return data
@@ -866,7 +866,7 @@ def create_ui(container, button_parent, tabname, skip_indexing = False):
if hasattr(item, 'mtime') and item.mtime is not None:
stat_mtime = item.mtime
desc = item.description
fullinfo = shared.readfile(os.path.splitext(item.filename)[0] + '.json', silent=True, dict_only=True)
fullinfo = shared.readfile(os.path.splitext(item.filename)[0] + '.json', silent=True, as_type="dict")
if 'modelVersions' in fullinfo: # sanitize massive objects
fullinfo['modelVersions'] = []
info = fullinfo
+1 -1
View File
@@ -40,7 +40,7 @@ class ExtraNetworksPageCheckpoints(ui_extra_networks.ExtraNetworksPage):
shared.log.debug(f'Networks: type="reference" autodownload={shared.opts.sd_checkpoint_autodownload} enable={shared.opts.extra_network_reference_enable}')
return []
count = { 'total': 0, 'ready': 0, 'hidden': 0, 'experimental': 0, 'base': 0 }
shared.reference_models = readfile(os.path.join('html', 'reference.json'), dict_only=True)
shared.reference_models = readfile(os.path.join('html', 'reference.json'), as_type="dict")
for k, v in shared.reference_models.items():
count['total'] += 1
url = v['path']
+1 -1
View File
@@ -118,7 +118,7 @@ class UiLoadsave:
def read_from_file(self):
from modules.shared import readfile
return readfile(self.filename, dict_only=True)
return readfile(self.filename, as_type="dict")
def write_to_file(self, current_ui_settings):
from modules.shared import writefile
+1 -1
View File
@@ -23,7 +23,7 @@ class Upscaler:
def __init__(self, create_dirs=True):
global models # pylint: disable=global-statement
if models is None:
models = shared.readfile('html/upscalers.json', dict_only=True)
models = shared.readfile('html/upscalers.json', as_type="dict")
self.mod_pad_h = None
self.tile_size = shared.opts.upscaler_tile_size
self.tile_pad = shared.opts.upscaler_tile_overlap